feat(release): bind services to isolated journal readiness

2026-10-13 06:09:49 +08:00
parent 57f5819dfd
commit 83e235567f
7 changed files with 101 additions and 17 deletions
@@ -974,6 +974,21 @@ def test_journal_profile_never_substitutes_rotation_settings_for_a_dedicated_har
        journal.validate_storage(observed, path)


+def test_read_only_observer_mount_is_not_daemon_writability_proof():
+    path = Path("/var/log/journal/" + "b" * 32 + ".kf-onlinejudge")
+    observed = {
+        "path": str(path),
+        "root": "/",
+        "filesystem": "ext4",
+        "options": ["ro", "nosuid", "nodev", "noexec"],
+        "shared": False,
+        "capacity_bytes": 64 * 1024**2,
+    }
+    journal.validate_storage(observed, path, allow_read_only=True)
+    with pytest.raises(policy.DeliveryError):
+        journal.validate_storage(observed, path)
+
+
@pytest.mark.parametrize("failure", ["publication", "receipt", "readback"])
 def test_maintenance_does_not_leave_admission_open_when_final_confirmation_fails(
    monkeypatch, failure
@@ -316,13 +316,24 @@ The profile fixes 32-MiB persistent/8-MiB runtime rotation budgets, bounded file
 lines and seven-day retention. Its namespaced daemon override limits CPU, memory,
 swap and PIDs. Rotation settings are not a hard filesystem guarantee; the dedicated
 volume must also be observed. The `check_ready` interface reads actual namespace
-service/process/cgroup state. Connecting service templates to this profile,
-redaction and real retention/capacity behavior remain C+D/F installation acceptance
-work; installing the profile alone does not route existing service logs or verify
-their privacy. The profile relies on the pinned systemd 262
+service/process/cgroup state. Ordinary application, DB/MQ and Root-factory
+templates now select this namespace. Their protected pre-start binding checks
+require its recorded configuration and physical hard volume, while retaining
+ordinary private PID views and the Root factory's existing permission context.
+Installing the profile does not start or replace running services. Real routing,
+privacy and retention/capacity behavior remain C+D/F VM acceptance work.
+The profile relies on the pinned systemd 262
 [namespace interface](https://github.com/systemd/systemd/blob/v262/man/systemd.exec.xml)
 and [journal configuration](https://github.com/systemd/systemd/blob/v262/man/journald.conf.xml).

+The read-only status observer may see a `ProtectSystem` mount as read-only. That
+does not prove the logger can write: readiness separately reads the fixed known
+daemon's kernel mountinfo, matches its mount device to the protected volume and
+requires both mount and superblock writability. Main PID, invocation and process
+start time are checked again to reject a changed process. No writable exception,
+ptrace/setns capability or wider Python PID visibility is added to observation.
+Underlying read interface: [Linux proc mountinfo implementation](https://github.com/torvalds/linux/blob/v6.18/fs/proc_namespace.c).
+
 Production API logging uses a closed event vocabulary and omits arbitrary library
 messages, arguments, paths and exception text. Uvicorn's separate error handlers
 are redirected through that formatter; installed API launchers disable access
@@ -332,6 +332,7 @@ def render(release: str, batch: str) -> dict[str, bytes]:
            "StandardError=journal",
            "LogRateLimitIntervalSec=30s",
            "LogRateLimitBurst=100",
+            "LogNamespace=kf-onlinejudge",
            "ExecStartPre=+/usr/bin/env -i PATH=/usr/bin:/bin LC_ALL=C "
            + python_environment(payload)
            + f" {payload}/python/bin/python3.13"
@@ -152,6 +152,7 @@ def render(receipt: dict) -> bytes:
        "StandardError=journal",
        "LogRateLimitIntervalSec=30s",
        "LogRateLimitBurst=100",
+        "LogNamespace=kf-onlinejudge",
        "[Install]",
        "WantedBy=multi-user.target",
    ]
@@ -81,7 +81,7 @@ def bounded_text(path: Path, maximum: int) -> str:
    return value


-def validate_storage(value: dict, path: Path) -> None:
+def validate_storage(value: dict, path: Path, *, allow_read_only: bool = False) -> None:
    fields(
        value,
        {"path", "root", "filesystem", "options", "shared", "capacity_bytes"},
@@ -94,24 +94,38 @@ def validate_storage(value: dict, path: Path) -> None:
        or value["shared"] is not False
        or not isinstance(value["options"], list)
        or not all(isinstance(option, str) for option in value["options"])
-        or not {"rw", "nosuid", "nodev", "noexec"} <= set(value["options"])
+        or not {"nosuid", "nodev", "noexec"} <= set(value["options"])
+        or set(value["options"]) & {"ro", "rw"}
+        not in (({"rw"}, {"ro"}) if allow_read_only else ({"rw"},))
        or type(value["capacity_bytes"]) is not int
        or not CAPACITY * 9 // 10 <= value["capacity_bytes"] <= CAPACITY
    ):
        raise DeliveryError("dedicated journal hard storage budget differs")


-def storage() -> dict:
+def storage(*, allow_read_only: bool = False, observed_pid: int | None = None) -> dict:
    path = Path("/var/log/journal") / (machine_id() + "." + NAMESPACE)
    preflight.protected(path, directory=True)
-    rows = [
-        line.split() for line in bounded_text(Path("/proc/self/mountinfo"), 1024**2).splitlines()
-    ]
+    if observed_pid is not None and (type(observed_pid) is not int or not 1 < observed_pid < 2**31):
+        raise DeliveryError("known journal process required")
+    mountinfo = (
+        Path("/proc/self/mountinfo")
+        if observed_pid is None
+        else Path(f"/proc/{observed_pid}/mountinfo")
+    )
+    rows = [line.split() for line in bounded_text(mountinfo, 1024**2).splitlines()]
    selected = [row for row in rows if len(row) > 6 and row[4] == str(path)]
    if len(selected) != 1:
        raise DeliveryError("dedicated journal mount required; never created by installer")
    row = selected[0]
    separator = row.index("-")
+    if not allow_read_only and (
+        "rw" not in row[separator + 3].split(",") or "ro" in row[separator + 3].split(",")
+    ):
+        raise DeliveryError("journal filesystem superblock is not writable")
+    device = path.stat().st_dev
+    if row[2] != f"{os.major(device)}:{os.minor(device)}":
+        raise DeliveryError("journal daemon sees another storage device")
    info = os.statvfs(path)
    value = {
        "path": str(path),
@@ -121,7 +135,7 @@ def storage() -> dict:
        "shared": any(other[2] == row[2] and other != row for other in rows),
        "capacity_bytes": info.f_frsize * info.f_blocks,
    }
-    validate_storage(value, path)
+    validate_storage(value, path, allow_read_only=allow_read_only)
    return value


@@ -156,7 +170,7 @@ def validate_receipt(value: dict) -> None:
        raise DeliveryError("journal receipt differs from fixed profile")


-def check_profile() -> dict:
+def check_profile(*, allow_read_only: bool = False) -> dict:
    preflight.protected(RECEIPT)
    configuration.private(RECEIPT)
    receipt = read_json(RECEIPT, 4096)
@@ -177,7 +191,7 @@ def check_profile() -> dict:
    if {p.name for p in DROPIN.parent.iterdir()} != {DROPIN.name}:
        raise DeliveryError("unknown journal instance override")
    check_configuration_overrides()
-    return storage()
+    return storage(allow_read_only=allow_read_only)


 def check_configuration_overrides() -> None:
@@ -190,13 +204,15 @@ def check_configuration_overrides() -> None:


 def check_ready() -> dict:
-    value = check_profile()
+    # A cap-zero observer's ProtectSystem view is intentionally read-only.
+    # That view establishes physical identity/capacity, not daemon writability.
+    check_profile(allow_read_only=True)
    props = dict(
        line.split("=", 1)
        for line in manager(
            "show",
            UNIT,
-            "--property=ActiveState,SubState,MainPID,ControlGroup,FragmentPath,DropInPaths",
+            "--property=ActiveState,SubState,MainPID,ControlGroup,FragmentPath,DropInPaths,InvocationID,RootDirectory,RootImage",
        ).splitlines()
        if "=" in line
    )
@@ -213,9 +229,16 @@ def check_ready() -> dict:
        "DropInPaths", ""
    ).split() != [str(DROPIN)]:
        raise DeliveryError("unrecorded journal manager executable or override")
+    if (
+        props.get("RootDirectory")
+        or props.get("RootImage")
+        or not re.fullmatch(r"[0-9a-f]{32}", props.get("InvocationID", ""))
+    ):
+        raise DeliveryError("unknown journal process/root binding")
    pid = int(props.get("MainPID", "0"))
    if not 1 < pid < 2**31:
        raise DeliveryError("journal process identity unavailable")
+    before = bounded_text(Path(f"/proc/{pid}/stat"), 16384).rsplit(")", 1)[1].split()[19]
    status = dict(
        line.split(":", 1)
        for line in bounded_text(Path(f"/proc/{pid}/status"), 16384).splitlines()
@@ -235,7 +258,31 @@ def check_ready() -> dict:
        "cpu.max": "25000 100000",
    }:
        raise DeliveryError("effective journal budgets differ")
+    # Kernel mountinfo is read-only and does not enter the daemon namespace or
+    # require ptrace/setns. Match its mount device to this protected fixed path,
+    # and require the daemon's actual view to be writable. No extra capability
+    # or writable-path exception is given to the observer or Python services.
+    value = storage(observed_pid=pid)
+    current = dict(
+        line.split("=", 1)
+        for line in manager(
+            "show", UNIT, "--property=ActiveState,MainPID,InvocationID"
+        ).splitlines()
+        if "=" in line
+    )
+    after = bounded_text(Path(f"/proc/{pid}/stat"), 16384).rsplit(")", 1)[1].split()[19]
+    if (
+        current.get("ActiveState") != "active"
+        or current.get("MainPID") != str(pid)
+        or current.get("InvocationID") != props["InvocationID"]
+        or before != after
+    ):
+        raise DeliveryError("journal process changed during observation")
    value["effective_limits"] = limits
+    value["memory_current"] = int(bounded_text(directory / "memory.current", 64))
+    value["pids_current"] = int(bounded_text(directory / "pids.current", 64))
+    fs = os.statvfs(Path(value["path"]))
+    value["available_bytes"] = fs.f_frsize * fs.f_bavail
    return value


@@ -18,6 +18,7 @@ from . import (
    dependency_install,
    factory_service,
    install,
+    journal,
    operator_units,
    preflight,
    service_install,
@@ -66,6 +67,7 @@ def binding() -> tuple[dict, dict, Path]:
        "import_health.py",
        "operator_units.py",
        "outbox_health.py",
+        "journal.py",
    ):
        shipped = payload / "delivery/scripts/release" / name
        preflight.protected(shipped)
@@ -136,7 +138,7 @@ def inspect_unit(name: str, selected: dict) -> dict:
        for line in service_install.systemctl(
            "show",
            unit,
-            "--property=LoadState,ActiveState,SubState,MainPID,FragmentPath,ControlGroup",
+            "--property=LoadState,ActiveState,SubState,MainPID,FragmentPath,ControlGroup,LogNamespace",
        ).splitlines()
        if "=" in line
    )
@@ -146,6 +148,7 @@ def inspect_unit(name: str, selected: dict) -> dict:
        or value.get("SubState") != "running"
        or value.get("FragmentPath") not in {str(link), str(directory / unit)}
        or value.get("ControlGroup") != "/kfonlinejudge.slice/" + unit
+        or value.get("LogNamespace") != journal.NAMESPACE
    ):
        raise DeliveryError("selected service is not running")
    pid = int(value.get("MainPID", "0"))
@@ -457,6 +460,10 @@ def observe(*, probe: bool) -> dict:
        "errors": [],
    }
    components, errors = result["components"], result["errors"]
+    try:
+        components["journal"] = journal.check_ready()
+    except (OSError, ValueError, subprocess.SubprocessError):
+        errors.append("bounded_journal_unavailable")
    for name in [*services.PROFILES, "postgres", "rabbitmq", "sandbox-factory"]:
        try:
            components[name] = inspect_unit(name, selected)
@@ -6,7 +6,7 @@ import json
 import os
 from pathlib import Path

-from . import configuration, install, manifest
+from . import configuration, install, journal, manifest
 from .policy import DeliveryError, canonical, digest

 PROFILES = {
@@ -87,6 +87,7 @@ def render(receipt: dict) -> dict[str, bytes]:
            "StandardError=journal",
            "LogRateLimitIntervalSec=30s",
            "LogRateLimitBurst=100",
+            "LogNamespace=" + journal.NAMESPACE,
        ]
        if environment:
            path = configuration.installed_path(release, batch, environment)
@@ -210,6 +211,7 @@ def main() -> None:
                args.configuration_id,
            ):
                raise DeliveryError("unit belongs to an unselected release/configuration")
+            journal.check_profile(allow_read_only=True)
            result = {"binding_valid": True, "execution_verified": False}
        else:
            result = generate(