feat: preserve immutable configuration batches for controlled rotation

scripts/release/install.py

@@ -9,11 +9,21 @@ import subprocess
 from pathlib import Path
 
 from . import manifest, preflight
-from .policy import DeliveryError, canonical, digest, read_json, sha256
+from .policy import DeliveryError, canonical, digest, fields, read_json, sha256
 
 RELEASES = Path("/opt/kf-onlinejudge/releases")
 CURRENT = Path("/opt/kf-onlinejudge/current")
 IMAGES = Path("/var/lib/kf-judge/images")
+ACTIVE = Path("/etc/kf-onlinejudge/active.json")
+APPLICATION_UNITS = (
+    "kf-api.service",
+    "kf-coordinator.service",
+    "kf-outbox.service",
+    "kf-problem-import.service",
+    "kf-worker.service",
+    "kf-sandbox-factory.service",
+    "kf-proxy.service",
+)
 LEGACY_HEADER = {"image_kind": "linux_gnu_cxx_x86_64_v1", "architecture": "x86_64", "revision": 1}
 
 
@@ -281,26 +291,102 @@ def files(bundle: Path, expected: str) -> dict:
     return receipt
 
 
-def activate(release_id: str, expected: str) -> dict:
+def selection(*, check_configuration: bool = True) -> dict:
+    """One bounded protected binding; no process or health claim."""
+    from .configuration import check_installed, installed_receipt_path, private, validate_receipt
+
+    preflight.protected(ACTIVE)
+    private(ACTIVE)
+    value = fields(
+        read_json(ACTIVE, 4096),
+        {"schema_version", "release_id", "manifest_sha256", "configuration_id", "receipt_sha256"},
+        "active binding",
+    )
+    if (
+        type(value["schema_version"]) is not int
+        or value["schema_version"] != 1
+        or not isinstance(value["release_id"], str)
+        or not re.fullmatch(r"sd-[0-9a-f]{12}", value["release_id"])
+        or not isinstance(value["configuration_id"], str)
+        or not re.fullmatch(r"cfg-[0-9a-f]{24}", value["configuration_id"])
+    ):
+        raise DeliveryError("invalid active release/configuration binding")
+    sha256(value["manifest_sha256"])
+    sha256(value["receipt_sha256"])
+    receipt = installed_receipt_path(value["release_id"], value["configuration_id"])
+    preflight.protected(receipt)
+    private(receipt)
+    if digest(receipt) != value["receipt_sha256"]:
+        raise DeliveryError("active configuration receipt differs")
+    binding = validate_receipt(read_json(receipt, 16384), installed=True)
+    if any(
+        binding[key] != value[key] for key in ("release_id", "configuration_id", "manifest_sha256")
+    ):
+        raise DeliveryError("active configuration identity differs")
+    target = RELEASES / value["release_id"] / "payload"
+    info = CURRENT.lstat()
+    if not stat.S_ISLNK(info.st_mode) or info.st_uid != 0 or os.readlink(CURRENT) != str(target):
+        raise DeliveryError("active code pointer differs from configuration binding")
+    preflight.protected(target, directory=True)
+    if check_configuration:
+        check_installed(receipt)
+    return value
+
+
+def require_stopped() -> None:
+    """Do not select different private identities while any app unit is alive."""
+    for unit in APPLICATION_UNITS:
+        result = subprocess.run(
+            ["/usr/bin/systemctl", "show", unit, "--property=LoadState,ActiveState,MainPID"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+            check=False,
+            env={"PATH": "/usr/bin", "LC_ALL": "C"},
+        )
+        if len(result.stdout) > 4096:
+            raise DeliveryError("application stop observation exceeds limit")
+        values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)
+        if (
+            result.returncode
+            or values.get("LoadState") not in {"loaded", "not-found"}
+            or values.get("ActiveState") != "inactive"
+            or values.get("MainPID") != "0"
+        ):
+            raise DeliveryError("all application units must be stopped before selection")
+    delegated = Path("/sys/fs/cgroup/kfonlinejudge.slice/kf-judge-tasks")
+    if delegated.exists():
+        if delegated.is_symlink() or (delegated / "cgroup.procs").read_text().strip():
+            raise DeliveryError("delegated execution processes remain")
+        if any(path.is_dir() for path in delegated.iterdir()):
+            raise DeliveryError("delegated task groups remain; cleanup proof required")
+
+
+def activate(release_id: str, expected: str, batch: str) -> dict:
     """Only files + separately installed exact configuration may be selected.
 
     This command starts no service, migrates no table, and removes no old version.
     Runtime admission/isolation/health remains a separate bounded operation.
     """
-    if os.geteuid() != 0 or not re.fullmatch(r"sd-[0-9a-f]{12}", release_id):
+    if (
+        os.geteuid() != 0
+        or not re.fullmatch(r"sd-[0-9a-f]{12}", release_id)
+        or not re.fullmatch(r"cfg-[0-9a-f]{24}", batch)
+    ):
         raise DeliveryError("invalid activation identity or privilege")
     release = RELEASES / release_id
     preflight.protected(release, directory=True)
     data = manifest.verify(release, expected)
-    config_receipt = Path("/etc/kf-onlinejudge") / (release_id + ".configuration.json")
-    from .configuration import check_installed
+    from .configuration import check_installed, installed_receipt_path
 
+    config_receipt = installed_receipt_path(release_id, batch)
     binding = check_installed(config_receipt)
     if (
         binding.get("manifest_sha256") != expected
         or binding.get("source_sha") != data["source_sha"]
     ):
         raise DeliveryError("configuration not bound to selected release")
+    require_stopped()
     for port in (8000, 8100):
         import socket
 
@@ -317,20 +403,31 @@ def activate(release_id: str, expected: str) -> dict:
             preflight.protected(path, directory=True)
             if next(path.iterdir(), None) is not None:
                 raise DeliveryError("execution leases or task resources remain; activation refused")
-    if CURRENT.exists() or CURRENT.is_symlink():
-        info = CURRENT.lstat()
-        if not stat.S_ISLNK(info.st_mode) or info.st_uid != 0:
-            raise DeliveryError("unknown current entry refused")
-        old = Path(os.readlink(CURRENT))
-        if old.parent.parent != RELEASES or old.name != "payload":
-            raise DeliveryError("current link outside release layout")
-        preflight.protected(old, directory=True)
+    if ACTIVE.exists() or ACTIVE.is_symlink() or CURRENT.exists() or CURRENT.is_symlink():
+        # Retiring a byte-bound old batch is allowed after a kernel change.
+        # It is never accepted for execution under that changed environment.
+        selection(check_configuration=False)
+    active = {
+        "schema_version": 1,
+        "release_id": release_id,
+        "manifest_sha256": expected,
+        "configuration_id": batch,
+        "receipt_sha256": digest(config_receipt),
+    }
+    incoming = ACTIVE.parent / (".active-" + secrets.token_hex(16))
+    exclusive_file(incoming, canonical(active), 0o600)
     temporary = CURRENT.parent / (".current-" + secrets.token_hex(16))
     temporary.symlink_to(release / "payload")
     os.replace(temporary, CURRENT)
     sync_directory(CURRENT.parent)
+    os.replace(incoming, ACTIVE)
+    sync_directory(ACTIVE.parent)
+    # A power loss between the two atomic writes leaves a detectable binding
+    # mismatch. No service is started and check-active refuses that state.
+    selection()
     return {
         "selected_source_sha": data["source_sha"],
+        "configuration_id": batch,
         "activated": True,
         "services_started": False,
         "execution_verified": False,