fix(delivery): add bounded explicit TLS rotation

Keep immutable overwrite refusal as the default. Explicit previous fingerprints authorize a stopped-node directory switch with retained private history and resumable publication. Preserve all installed versions with a 65536-entry backup inventory under the existing byte and resource limits.

backend/tests/test_delivery_io.py

@@ -1350,3 +1350,90 @@ def test_startup_rejects_unknown_or_duplicate_groups(ordinary_lifecycle_model, n
     with pytest.raises(policy.DeliveryError, match="unknown startup dependency group"):
         maintenance.start_group(names)
     assert trace == []
+
+
+@pytest.mark.parametrize("rotation", [False, True])
+def test_proxy_install_cli_keeps_legacy_inputs_and_requires_explicit_rotation(
+    monkeypatch, capsys, rotation
+):
+    calls = []
+
+    def route(*args, **kwargs):
+        calls.append((args, kwargs))
+        return {"routed": True}
+
+    monkeypatch.setattr(proxy_install, "install_proxy", route)
+    arguments = [
+        "--source",
+        "/ordinary/proxy",
+        "--receipt-sha256",
+        "a" * 64,
+        "--certificate",
+        "/ordinary/certificate",
+        "--private-key",
+        "/ordinary/key",
+    ]
+    expected = {}
+    if rotation:
+        arguments += ["--previous-certificate-sha256", "b" * 64, "--previous-key-sha256", "c" * 64]
+        expected = {"previous_certificate_sha256": "b" * 64, "previous_key_sha256": "c" * 64}
+    proxy_install.main(arguments)
+    assert calls == [
+        (
+            (
+                Path("/ordinary/proxy"),
+                "a" * 64,
+                Path("/ordinary/certificate"),
+                Path("/ordinary/key"),
+            ),
+            expected,
+        )
+    ]
+    assert json.loads(capsys.readouterr().out) == {"routed": True}
+
+
+def test_proxy_install_cli_rejects_an_incomplete_rotation_before_dispatch(monkeypatch, capsys):
+    calls = []
+    monkeypatch.setattr(proxy_install, "install_proxy", lambda *args, **kwargs: calls.append(args))
+    with pytest.raises(SystemExit) as failure:
+        proxy_install.main(
+            [
+                "--source",
+                "/ordinary/proxy",
+                "--receipt-sha256",
+                "a" * 64,
+                "--certificate",
+                "/ordinary/certificate",
+                "--private-key",
+                "/ordinary/key",
+                "--previous-certificate-sha256",
+                "b" * 64,
+            ]
+        )
+    assert failure.value.code == 1
+    assert calls == []
+    assert "paired previous TLS fingerprints required" in capsys.readouterr().err
+
+
+def test_proxy_previous_snapshot_retains_the_legacy_unit_and_binding_format(tmp_path, monkeypatch):
+    # This is a trusted unit-format fixture; Root ownership is verified only in VM.
+    service = proxy_install.proxy_service
+    selected = {
+        "schema_version": 1,
+        "release_id": "sd-" + "1" * 12,
+        "configuration_id": "cfg-" + "2" * 24,
+        "manifest_sha256": "3" * 64,
+        "receipt_sha256": "4" * 64,
+    }
+    monkeypatch.setattr(service, "UNITS", tmp_path / "units")
+    monkeypatch.setattr(proxy_install.preflight, "protected", lambda *args, **kwargs: None)
+    directory = service.UNITS / service.batch(selected).name
+    directory.mkdir(parents=True)
+    binding = directory / "binding.json"
+    binding.write_bytes(policy.canonical(selected))
+    binding.chmod(0o600)
+    unit = directory / service.UNIT
+    unit.write_bytes(service.render(selected))
+    unit.chmod(0o644)
+    assert proxy_install.known_snapshot(directory) == selected
+    assert "--previous-certificate" not in unit.read_text()

scripts/release/backup_manifest.py

@@ -19,7 +19,7 @@ from .policy import (
 FORMAT = "kf.private.cold-backup.v3"
 MAX_BYTES = 7 * 1024**3
 MAX_FILE_BYTES = 256 * 1024**2
-MAX_ENTRIES = 50000
+MAX_ENTRIES = 65536
 MAX_METADATA_BYTES = 16 * 1024**2
 DOMAINS = {
     "postgres",

scripts/release/proxy_install.py

@@ -47,7 +47,10 @@ def account(*, create: bool = False, check_lock: bool = False) -> None:
             ],
         ):
             p = subprocess.run(
-                args, env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"}, capture_output=True, timeout=10
+                args,
+                env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"},
+                capture_output=True,
+                timeout=10,
             )
             if p.returncode:
                 raise DeliveryError("proxy account provisioning refused")
@@ -117,6 +120,39 @@ def storage(*, allow_read_only: bool = False) -> dict:
     }
 
 
+def known_snapshot(directory: Path) -> dict:
+    if directory.parent != proxy_service.UNITS:
+        raise DeliveryError("unknown proxy unit snapshot")
+    target = directory / proxy_service.UNIT
+    preflight.protected(directory / "binding.json")
+    configuration.private(directory / "binding.json")
+    selected = read_json(directory / "binding.json", 4096)
+    fields(
+        selected,
+        {
+            "schema_version",
+            "release_id",
+            "configuration_id",
+            "manifest_sha256",
+            "receipt_sha256",
+        },
+        "proxy unit binding",
+    )
+    if type(selected["schema_version"]) is not int or selected["schema_version"] != 1:
+        raise DeliveryError("proxy unit binding version differs")
+    sha256(selected["manifest_sha256"])
+    sha256(selected["receipt_sha256"])
+    if directory.name != proxy_service.batch(selected).name:
+        raise DeliveryError("proxy unit snapshot binding differs")
+    preflight.protected(target)
+    if (
+        target.read_bytes() != proxy_service.render(selected)
+        or stat.S_IMODE(target.stat().st_mode) != 0o644
+    ):
+        raise DeliveryError("proxy unit snapshot changed")
+    return selected
+
+
 def known_link(path: Path) -> Path | None:
     try:
         info = path.lstat()
@@ -131,27 +167,7 @@ def known_link(path: Path) -> Path | None:
         or target.as_posix() != os.readlink(path)
     ):
         raise DeliveryError("unknown proxy unit pointer")
-    preflight.protected(target.parent / "binding.json")
-    configuration.private(target.parent / "binding.json")
-    selected = read_json(target.parent / "binding.json", 4096)
-    fields(
-        selected,
-        {"schema_version", "release_id", "configuration_id", "manifest_sha256", "receipt_sha256"},
-        "proxy unit binding",
-    )
-    if type(selected["schema_version"]) is not int or selected["schema_version"] != 1:
-        raise DeliveryError("proxy unit binding version differs")
-    sha256(selected["manifest_sha256"])
-    sha256(selected["receipt_sha256"])
-    if target.parent.name != proxy_service.batch(selected).name:
-        raise DeliveryError("proxy unit snapshot binding differs")
-    preflight.protected(target)
-    preflight.protected(target.parent / "binding.json")
-    if (
-        target.read_bytes() != proxy_service.render(selected)
-        or stat.S_IMODE(target.stat().st_mode) != 0o644
-    ):
-        raise DeliveryError("proxy unit snapshot changed")
+    known_snapshot(target.parent)
     return target.parent
 
 
@@ -310,9 +326,22 @@ def temporary_directories() -> None:
 
 
 @manifest.verified_operation
-def install_proxy(source: Path, expected: str, certificate: Path, key: Path) -> dict:
+def install_proxy(
+    source: Path,
+    expected: str,
+    certificate: Path,
+    key: Path,
+    *,
+    previous_certificate_sha256: str | None = None,
+    previous_key_sha256: str | None = None,
+) -> dict:
     if os.getuid() != 0 or os.geteuid() != 0 or os.getgid() != 0:
         raise DeliveryError("root proxy installer required")
+    if (previous_certificate_sha256 is None) != (previous_key_sha256 is None):
+        raise DeliveryError("paired previous TLS fingerprints required for rotation")
+    if previous_certificate_sha256 is not None:
+        sha256(previous_certificate_sha256)
+        sha256(previous_key_sha256)
     preflight.protected(source, directory=True)
     receipt_path = source / "receipt.json"
     preflight.protected(receipt_path)
@@ -330,6 +359,7 @@ def install_proxy(source: Path, expected: str, certificate: Path, key: Path) ->
         "manifest.py",
         "proxy_guard.py",
         "proxy_runtime.py",
+        "tls_rotation.py",
     ):
         p = payload / "delivery/scripts/release" / name
         preflight.protected(p)
@@ -360,7 +390,10 @@ def install_proxy(source: Path, expected: str, certificate: Path, key: Path) ->
         "configuration_sha256"
     ]:
         raise DeliveryError("proxy is not fixed trusted composition")
-    for p, fingerprint in ((certificate, spec["certificate_sha256"]), (key, spec["key_sha256"])):
+    for p, fingerprint in (
+        (certificate, spec["certificate_sha256"]),
+        (key, spec["key_sha256"]),
+    ):
         preflight.protected(p)
         configuration.private(p)
         if not 1 <= p.stat().st_size <= 65536 or digest(p) != fingerprint:
@@ -378,8 +411,23 @@ def install_proxy(source: Path, expected: str, certificate: Path, key: Path) ->
 
     ctx.load_cert_chain(certificate, key, password=no_password_prompt)
     system = service_install.SYSTEM_UNITS / proxy_service.UNIT
-    known_link(system)
+    old_directory = known_link(system)
     install.directory(proxy_service.ROOT, 0o711)
+    rotation = None
+    if previous_certificate_sha256 is not None:
+        from . import tls_rotation
+
+        rotation = tls_rotation.rotate(
+            selected,
+            spec,
+            certificate,
+            key,
+            previous_certificate_sha256,
+            previous_key_sha256,
+            old_directory,
+            known_snapshot,
+        )
+    else:
         install.directory(proxy_service.ROOT / "tls", 0o711)
     directory = proxy_service.batch(selected)
     install.directory(directory, 0o755)
@@ -427,7 +475,9 @@ def install_proxy(source: Path, expected: str, certificate: Path, key: Path) ->
     props = dict(
         line.split("=", 1)
         for line in service_install.systemctl(
-            "show", proxy_service.UNIT, "--property=LoadState,ActiveState,MainPID,FragmentPath"
+            "show",
+            proxy_service.UNIT,
+            "--property=LoadState,ActiveState,MainPID,FragmentPath",
         ).splitlines()
         if "=" in line
     )
@@ -444,6 +494,7 @@ def install_proxy(source: Path, expected: str, certificate: Path, key: Path) ->
         "source_sha": receipt["source_sha"],
         "configuration_id": selected["configuration_id"],
         "tls_material_parsed": True,
+        "tls_rotation": rotation,
         "tls_verified": False,
         "services_started": False,
         "execution_verified": False,
@@ -459,13 +510,24 @@ def main(argv: list[str] | None = None) -> None:
     parser.add_argument("--receipt-sha256")
     parser.add_argument("--certificate", type=Path)
     parser.add_argument("--private-key", type=Path)
+    parser.add_argument("--previous-certificate-sha256")
+    parser.add_argument("--previous-key-sha256")
     args = parser.parse_args(argv)
     try:
         if args.check_binding:
             if (
                 not args.release_id
                 or not args.configuration_id
-                or any((args.source, args.receipt_sha256, args.certificate, args.private_key))
+                or any(
+                    (
+                        args.source,
+                        args.receipt_sha256,
+                        args.certificate,
+                        args.private_key,
+                        args.previous_certificate_sha256,
+                        args.previous_key_sha256,
+                    )
+                )
             ):
                 raise DeliveryError("explicit binding-only arguments required")
             result = check_binding(args.release_id, args.configuration_id)
@@ -474,8 +536,20 @@ def main(argv: list[str] | None = None) -> None:
                 (args.source, args.receipt_sha256, args.certificate, args.private_key)
             ):
                 raise DeliveryError("explicit installation inputs required")
+            if (args.previous_certificate_sha256 is None) != (args.previous_key_sha256 is None):
+                raise DeliveryError("paired previous TLS fingerprints required for rotation")
+            rotation = {}
+            if args.previous_certificate_sha256 is not None:
+                rotation = {
+                    "previous_certificate_sha256": args.previous_certificate_sha256,
+                    "previous_key_sha256": args.previous_key_sha256,
+                }
             result = install_proxy(
-                args.source, args.receipt_sha256, args.certificate, args.private_key
+                args.source,
+                args.receipt_sha256,
+                args.certificate,
+                args.private_key,
+                **rotation,
             )
         print(json.dumps(result, sort_keys=True))
     except (OSError, ValueError, subprocess.SubprocessError) as error:

scripts/release/tls_rotation.py

@@ -0,0 +1,286 @@
+"""Explicit TLS replacement while stopped, with bounded retained prior material."""
+
+import fcntl
+import hashlib
+import os
+import re
+import stat
+from pathlib import Path
+
+from . import configuration, install, preflight, proxy_service
+from .policy import DeliveryError, canonical, fields, read_json, sha256
+
+MAX_ROTATIONS = 16
+MAX_MATERIAL_BYTES = 65536
+
+
+def _hash(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _material(path: Path, mode: int, gid: int) -> bytes:
+    preflight.protected(path)
+    before = path.lstat()
+    if (
+        not stat.S_ISREG(before.st_mode)
+        or before.st_nlink != 1
+        or (before.st_uid, before.st_gid, stat.S_IMODE(before.st_mode)) != (0, gid, mode)
+        or not 1 <= before.st_size <= MAX_MATERIAL_BYTES
+    ):
+        raise DeliveryError("protected bounded TLS material required")
+    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
+    with os.fdopen(fd, "rb") as stream:
+        opened = os.fstat(stream.fileno())
+        data = stream.read(MAX_MATERIAL_BYTES + 1)
+        after = os.fstat(stream.fileno())
+    keys = (
+        "st_dev",
+        "st_ino",
+        "st_mode",
+        "st_nlink",
+        "st_uid",
+        "st_gid",
+        "st_size",
+        "st_mtime_ns",
+        "st_ctime_ns",
+    )
+    if len(data) != before.st_size or any(
+        getattr(before, key) != getattr(opened, key) or getattr(opened, key) != getattr(after, key)
+        for key in keys
+    ):
+        raise DeliveryError("TLS material changed while observed")
+    return data
+
+
+def _pair(directory: Path) -> tuple[str, str]:
+    preflight.protected(directory, directory=True)
+    info = directory.lstat()
+    if (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (0, 0, 0o711):
+        raise DeliveryError("protected TLS directory differs")
+    entries = []
+    for entry in directory.iterdir():
+        entries.append(entry.name)
+        if len(entries) > 2:
+            raise DeliveryError("unknown TLS directory member")
+    if set(entries) != {"fullchain.pem", "key.pem"}:
+        raise DeliveryError("complete TLS pair required")
+    return (
+        _hash(_material(directory / "fullchain.pem", 0o644, 0)),
+        _hash(_material(directory / "key.pem", 0o640, proxy_service.UID)),
+    )
+
+
+def _stage_file(path: Path, data: bytes, mode: int, gid: int) -> None:
+    """Finish only an exact prefix in this operation's protected staging directory."""
+    preflight.protected(path.parent, directory=True)
+    fd = os.open(path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC, mode)
+    try:
+        info = os.fstat(fd)
+        if (
+            not stat.S_ISREG(info.st_mode)
+            or info.st_nlink != 1
+            or info.st_uid != 0
+            or info.st_gid not in {0, gid}
+            or (
+                stat.S_IMODE(info.st_mode) != mode
+                and not (
+                    info.st_size == 0
+                    and info.st_gid == 0
+                    and not stat.S_IMODE(info.st_mode) & ~mode
+                )
+            )
+            or not 0 <= info.st_size <= len(data) <= MAX_MATERIAL_BYTES
+        ):
+            raise DeliveryError("unknown TLS staging file")
+        prefix = os.read(fd, len(data) + 1)
+        if not data.startswith(prefix) or len(prefix) != info.st_size:
+            raise DeliveryError("TLS staging bytes differ")
+        os.fchmod(fd, mode)
+        if len(prefix) != len(data):
+            with os.fdopen(os.dup(fd), "wb", closefd=True) as stream:
+                stream.write(data[len(prefix) :])
+                stream.flush()
+        os.fchown(fd, 0, gid)
+        os.fchmod(fd, mode)
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+    if _material(path, mode, gid) != data:
+        raise DeliveryError("TLS staging readback differs")
+
+
+def _previous_binding(binding: dict, previous: tuple[str, str], known_snapshot) -> None:
+    directory = proxy_service.UNITS / proxy_service.batch(binding).name
+    if canonical(known_snapshot(directory)) != canonical(binding):
+        raise DeliveryError("previous TLS unit binding differs")
+    path = proxy_service.batch(binding) / "launch.json"
+    preflight.protected(path)
+    value = fields(
+        read_json(path, 4096),
+        {
+            "schema_version",
+            "release_id",
+            "configuration_id",
+            "origin",
+            "configuration_sha256",
+            "certificate_sha256",
+            "key_sha256",
+        },
+        "previous proxy launch spec",
+    )
+    if (
+        type(value["schema_version"]) is not int
+        or value["schema_version"] != 1
+        or (value["release_id"], value["configuration_id"])
+        != (binding["release_id"], binding["configuration_id"])
+        or (value["certificate_sha256"], value["key_sha256"]) != previous
+    ):
+        raise DeliveryError("previous TLS fingerprints do not match the installed unit")
+
+
+def rotate(
+    selected: dict,
+    spec: dict,
+    certificate: Path,
+    key: Path,
+    previous_certificate: str,
+    previous_key: str,
+    old_directory: Path | None,
+    known_snapshot,
+) -> dict:
+    if (os.getuid(), os.geteuid(), os.getgid()) != (0, 0, 0):
+        raise DeliveryError("root TLS rotation operator required")
+    previous = (sha256(previous_certificate), sha256(previous_key))
+    desired = (sha256(spec["certificate_sha256"]), sha256(spec["key_sha256"]))
+    new_certificate = _material(certificate, 0o600, 0)
+    new_key = _material(key, 0o600, 0)
+    if (_hash(new_certificate), _hash(new_key)) != desired:
+        raise DeliveryError("new TLS inputs changed")
+    header = {
+        "schema_version": 1,
+        "selected": selected,
+        "configuration_sha256": spec["configuration_sha256"],
+        "origin": spec["origin"],
+        "previous_certificate_sha256": previous[0],
+        "previous_key_sha256": previous[1],
+        "certificate_sha256": desired[0],
+        "key_sha256": desired[1],
+    }
+    identifier = "tls-" + _hash(canonical(header))
+    current = proxy_service.ROOT / "tls"
+    history = proxy_service.ROOT / "tls-history"
+    operation = history / identifier
+    pending, archived = operation / "pending", operation / "previous"
+    plan_path = operation / "plan.json"
+    # Serialize with the existing lifecycle owner; its empty lock is not a log.
+    state = Path("/var/lib/kf-onlinejudge/operations")
+    install.directory(state, 0o700)
+    lock = state / "maintenance.lock"
+    install.exclusive_file(lock, b"", 0o600)
+    configuration.private(lock)
+    fd = os.open(lock, os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        install.require_stopped()
+        if install.selection() != selected:
+            raise DeliveryError("TLS rotation selection changed")
+        if previous == desired:
+            if _pair(current) != desired:
+                raise DeliveryError("current TLS pair differs from explicit fingerprints")
+            return {"rotation_id": None, "changed": False, "services_started": False}
+        install.directory(history, 0o700)
+        if history.stat().st_dev != proxy_service.ROOT.stat().st_dev or (
+            current.exists() and current.stat().st_dev != history.stat().st_dev
+        ):
+            raise DeliveryError("TLS rotation requires one protected filesystem")
+        entries = []
+        for entry in history.iterdir():
+            entries.append(entry)
+            if (
+                len(entries) > MAX_ROTATIONS
+                or re.fullmatch(r"tls-[0-9a-f]{64}", entry.name) is None
+            ):
+                raise DeliveryError("bounded TLS history capacity or membership differs")
+            preflight.protected(entry, directory=True)
+        if not operation.exists() and len(entries) >= MAX_ROTATIONS:
+            raise DeliveryError(
+                "TLS history is full; explicit verified archive maintenance required"
+            )
+        install.directory(operation, 0o700)
+        if not {p.name for p in operation.iterdir()} <= {
+            "pending",
+            "previous",
+            "plan.pending",
+            "plan.json",
+        }:
+            raise DeliveryError("unknown TLS rotation member")
+        if plan_path.exists() or plan_path.is_symlink():
+            configuration.private(plan_path)
+            saved = fields(
+                read_json(plan_path, 16384),
+                set(header) | {"rotation_id", "previous_binding"},
+                "TLS rotation plan",
+            )
+            if (
+                canonical({k: saved[k] for k in header}) != canonical(header)
+                or saved["rotation_id"] != identifier
+            ):
+                raise DeliveryError("TLS rotation plan differs")
+            binding = saved["previous_binding"]
+            _previous_binding(binding, previous, known_snapshot)
+        else:
+            if (
+                archived.exists()
+                or archived.is_symlink()
+                or old_directory is None
+                or _pair(current) != previous
+            ):
+                raise DeliveryError("unrecorded TLS rotation state")
+            binding = known_snapshot(old_directory)
+            _previous_binding(binding, previous, known_snapshot)
+            install.directory(pending, 0o711)
+            _stage_file(pending / "fullchain.pem", new_certificate, 0o644, 0)
+            _stage_file(pending / "key.pem", new_key, 0o640, proxy_service.UID)
+            if _pair(pending) != desired:
+                raise DeliveryError("new TLS pair readback differs")
+            value = canonical({**header, "rotation_id": identifier, "previous_binding": binding})
+            _stage_file(operation / "plan.pending", value, 0o600, 0)
+            os.rename(operation / "plan.pending", plan_path)
+            install.sync_directory(operation)
+        changed = False
+        if current.exists() or current.is_symlink():
+            observed = _pair(current)
+            if observed == desired:
+                if pending.exists() or pending.is_symlink() or _pair(archived) != previous:
+                    raise DeliveryError("completed TLS rotation archive differs")
+            elif observed == previous and not archived.exists() and not archived.is_symlink():
+                if _pair(pending) != desired:
+                    raise DeliveryError("prepared TLS pair differs")
+                os.rename(current, archived)
+                install.sync_directory(proxy_service.ROOT)
+                install.sync_directory(operation)
+                changed = True
+            else:
+                raise DeliveryError("unknown current TLS rotation state")
+        if not current.exists() and not current.is_symlink():
+            if _pair(archived) != previous or _pair(pending) != desired:
+                raise DeliveryError("interrupted TLS rotation state differs")
+            os.rename(pending, current)
+            install.sync_directory(proxy_service.ROOT)
+            install.sync_directory(operation)
+            changed = True
+        if _pair(current) != desired or _pair(archived) != previous:
+            raise DeliveryError("TLS rotation final readback differs")
+        install.require_stopped()
+        if install.selection() != selected:
+            raise DeliveryError("TLS rotation selection changed after publication")
+        return {
+            "rotation_id": identifier,
+            "changed": changed,
+            "previous_pair_retained": True,
+            "tls_pair_published": True,
+            "tls_verified": False,
+            "services_started": False,
+        }
+    finally:
+        os.close(fd)