#!/usr/bin/env python3 """Rotate only Django's signing key, preserving existing JWT/integration keys. Existing sessions use Django SECRET_KEY_FALLBACKS. Remove the old fallback after the session/password-reset validity window. No key material is printed. Run after deploying settings support, under the deployment lock, as root. """ import base64 import datetime import fcntl import hashlib import json import os from pathlib import Path import re import secrets import stat import subprocess def parent_directory_fd(path): """Pin the parent directory without following replaceable path symlinks.""" if not path.is_absolute(): raise ValueError("An absolute environment path is required") descriptor = os.open(path.anchor, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: for component in path.parent.parts[1:]: child = os.open(component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor) os.close(descriptor) descriptor = child return descriptor except BaseException: os.close(descriptor) raise def read_environment(path): parent_fd = parent_directory_fd(path) try: descriptor = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent_fd) with os.fdopen(descriptor, "r") as stream: if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode): raise RuntimeError("Environment must be a regular file") return stream.read() finally: os.close(parent_fd) def atomic_environment_update(path, expected_source, content): """Never write/chown by a staging pathname in the deploy-writable directory.""" parent_fd = parent_directory_fd(path) try: original_fd = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent_fd) except BaseException: os.close(parent_fd) raise try: original_stat = os.fstat(original_fd) if not stat.S_ISREG(original_stat.st_mode): raise RuntimeError("Environment must remain a regular file") with os.fdopen(os.dup(original_fd), "r") as stream: if stream.read() != expected_source: raise RuntimeError("Environment changed concurrently; no rotation applied") attributes = { name: os.getxattr(original_fd, name) for name in os.listxattr(original_fd) } name = f".env.signing-key-{secrets.token_hex(16)}" descriptor = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent_fd) try: with os.fdopen(descriptor, "w") as stream: stream.write(content) stream.flush() os.fchown(stream.fileno(), original_stat.st_uid, original_stat.st_gid) for attribute, value in attributes.items(): os.setxattr(stream.fileno(), attribute, value) os.fchmod(stream.fileno(), stat.S_IMODE(original_stat.st_mode)) os.fsync(stream.fileno()) expected = os.fstat(stream.fileno()) observed = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) if (observed.st_dev, observed.st_ino) != (expected.st_dev, expected.st_ino): raise RuntimeError("Environment staging file was replaced; aborting") current = os.stat(path.name, dir_fd=parent_fd, follow_symlinks=False) if (current.st_dev, current.st_ino, current.st_size, current.st_mtime_ns) != ( original_stat.st_dev, original_stat.st_ino, original_stat.st_size, original_stat.st_mtime_ns ): raise RuntimeError("Environment was replaced concurrently; aborting") os.replace(name, path.name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd) os.fsync(parent_fd) finally: try: os.unlink(name, dir_fd=parent_fd) except FileNotFoundError: pass finally: os.close(original_fd) os.close(parent_fd) def parse_fallbacks(value): fallbacks = json.loads(value or "[]") if not isinstance(fallbacks, list) or not all( isinstance(key, str) and key for key in fallbacks ): raise ValueError("Existing key fallbacks must be a JSON list of nonempty strings") return fallbacks def render_values(source, values): for key, value in values.items(): if "\n" in value or "\r" in value: raise ValueError("Multiline key encoding is unsupported; no changes applied") # python-dotenv single quotes decode both backslash and quote escapes. escaped = value.replace("\\", "\\\\").replace("'", "\\'") line = f"{key}='{escaped}'" pattern = rf"(?m)^{re.escape(key)}=.*$" if re.search(pattern, source): source = re.sub(pattern, lambda _: line, source) else: source = source.rstrip() + "\n" + line + "\n" return source def main(): if os.geteuid() != 0: raise SystemExit("Run with sudo on CORP.") with open("/run/lock/eusrr-deploy.lock", "w") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) env_path = Path("/srv/eusrr/backend/.env") if env_path.is_symlink() or not env_path.is_file(): raise SystemExit("Unexpected environment file.") env_source = read_environment(env_path) # Parse dotenv without running application code as root or printing values. result = subprocess.run( [ "setpriv", "--no-new-privs", "runuser", "-u", "deploy", "--", "/srv/eusrr-current/backend/.venv/bin/python", "-c", "import ast, hashlib, io, json; from pathlib import Path; from dotenv import dotenv_values; " "data=Path('/srv/eusrr/backend/.env').read_text(); c=dotenv_values(stream=io.StringIO(data)); " "tree=ast.parse(Path('/srv/eusrr-current/backend/eusrr_backend/settings.py').read_text()); " "assignments={target.id:node.value for node in tree.body if isinstance(node,ast.Assign) " "for target in node.targets if isinstance(target,ast.Name)}; " "jwt=assignments.get('SIMPLE_JWT'); " "supported='SECRET_KEY_FALLBACKS' in assignments and isinstance(jwt,ast.Dict) " "and any(isinstance(k,ast.Constant) and k.value=='SIGNING_KEY' for k in jwt.keys); " "result={k:c.get(k) for k in ['DJANGO_SECRET_KEY','JWT_SIGNING_KEY','COMMERCE_CREDENTIAL_KEY','DJANGO_SECRET_KEY_FALLBACKS']}; " "result['settings_support_rotation']=supported; result['source_hash']=hashlib.sha256(data.encode()).hexdigest(); " "print(json.dumps(result))", ], check=True, capture_output=True, text=True, timeout=15, ) config = json.loads(result.stdout) if config.get("source_hash") != hashlib.sha256(env_source.encode()).hexdigest(): raise SystemExit("Environment changed during inspection; no changes applied.") if not config.get("settings_support_rotation"): raise SystemExit("Deploy key-fallback and explicit JWT signing support first; no changes applied.") old_key = config["DJANGO_SECRET_KEY"] if not old_key or len(old_key) < 32: raise SystemExit( "Existing key needs explicit compromise/weak-key recovery; aborting." ) if not old_key.startswith("django-insecure-"): print("Django key already has a production format; no rotation performed.") return commerce_key = ( config["COMMERCE_CREDENTIAL_KEY"] or base64.urlsafe_b64encode( hashlib.sha256(f"commerce-integrations:{old_key}".encode()).digest() ).decode() ) fallbacks = parse_fallbacks(config["DJANGO_SECRET_KEY_FALLBACKS"]) values = { "DJANGO_SECRET_KEY": secrets.token_urlsafe(64), "DJANGO_SECRET_KEY_FALLBACKS": json.dumps( list(dict.fromkeys([old_key, *fallbacks])) ), "JWT_SIGNING_KEY": config["JWT_SIGNING_KEY"] or old_key, "COMMERCE_CREDENTIAL_KEY": commerce_key, } source = render_values(env_source, values) stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") backup = Path("/var/lib/eusrr-deploy") / f"signing-key-backup-{stamp}" backup.mkdir(mode=0o700) (backup / "backend.env").write_text(env_source) (backup / "backend.env").chmod(0o600) atomic_environment_update(env_path, env_source, source) print( f"New Django key staged; existing JWT and integration keys preserved. Backup: {backup}" ) print( "Effective on next standard deployment. Keep fallback for existing sessions, then remove it after expiry." ) if __name__ == "__main__": main()