feat: route Custodian routines through bounded owner operations

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181

src/activity_core/activities.py

@@ -11,6 +11,7 @@ activities that need DB access.
 
 from __future__ import annotations
 
+import asyncio
 import json
 import uuid
 from datetime import datetime, timezone
@@ -246,7 +247,11 @@ async def execute_bounded_operation(payload: dict[str, Any]) -> dict[str, Any]:
     if spec is None or source.get("operation") != spec.operation_id:
         raise ApplicationError("bounded operation is not registered", non_retryable=True)
     params = source.get("params") or {}
-    if spec.operation_id == "forgejo_package_prune":
+    if spec.operation_id in {"custodian_security_flow", "custodian_allocation_review"}:
+        from activity_core.custodian_operations import run_custodian_operation
+
+        result = await asyncio.to_thread(run_custodian_operation, spec.operation_id, params)
+    elif spec.operation_id == "forgejo_package_prune":
         from activity_core.context_resolvers.forgejo_prune import forgejo_package_prune
 
         result = forgejo_package_prune(params)

src/activity_core/bounded_operations.py

@@ -27,6 +27,20 @@ class BoundedOperationSpec:
 
 
 _SPECS = (
+    BoundedOperationSpec(
+        operation_id="custodian_security_flow", source_type="shell", query="custodian_security_flow",
+        resolve_before_execute=False, max_timeout_seconds=300, temporal_max_attempts=1,
+        idempotency="single Temporal attempt; owner report/checkpoint retry contract",
+        credential_route="none; trusted operator binds scoped owner runtime and Hub access",
+        evidence_mode="context-evidence-sink",
+    ),
+    BoundedOperationSpec(
+        operation_id="custodian_allocation_review", source_type="shell", query="custodian_allocation_review",
+        resolve_before_execute=False, max_timeout_seconds=300, temporal_max_attempts=1,
+        idempotency="single Temporal attempt; owner report/checkpoint retry contract",
+        credential_route="none; trusted operator binds scoped owner runtime and Hub access",
+        evidence_mode="context-evidence-sink",
+    ),
     BoundedOperationSpec(
         operation_id="sbom_nexus_ingest",
         source_type="sbom-nexus",
@@ -139,6 +153,9 @@ def pending_operation_value(spec: BoundedOperationSpec) -> dict[str, str]:
 def normalize_operation_result(operation_id: str, raw: Any) -> dict[str, Any]:
     """Project subprocess output to the operation's non-secret evidence shape."""
     result = raw if isinstance(raw, dict) else {}
+    if operation_id in {"custodian_security_flow", "custodian_allocation_review"}:
+        keys = ("kind", "status", "exit_code", "report_sha256", "counts", "apply", "owner_acknowledgment")
+        return {key: result.get(key) for key in keys}
     if operation_id == "forgejo_package_prune":
         return {
             "kind": operation_id,
@@ -248,7 +265,11 @@ def validate_bounded_operations(
             raise ValueError(f"bounded operation {spec.operation_id} params must be a mapping")
         _validate_timeout(spec, params)
 
-        if spec.operation_id == "sbom_nexus_ingest":
+        if spec.operation_id in {"custodian_security_flow", "custodian_allocation_review"}:
+            from activity_core.custodian_operations import validate_params
+            validate_params(spec.operation_id, params)
+            _require_context_evidence(params, spec.operation_id)
+        elif spec.operation_id == "sbom_nexus_ingest":
             _validate_sbom(params, instructions)
         elif spec.operation_id == "forgejo_package_prune":
             _validate_prune(params)

src/activity_core/custodian_operations.py

@@ -0,0 +1,177 @@
+"""Fixed Custodian owner commands for ACT-ADR-007; no definition-supplied argv.
+
+Operator environment supplies trusted deployment bindings. These checks prevent
+accidental redirection; they are not a credential-isolation sandbox.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import signal
+import stat
+import subprocess
+import sys
+import time
+import tempfile
+from urllib.parse import urlsplit
+
+OPERATIONS = {"custodian_security_flow", "custodian_allocation_review"}
+_MAX_JSON_BYTES = 8 * 1024 * 1024
+
+
+def validate_params(operation: str, params: dict) -> None:
+    if operation not in OPERATIONS:
+        raise ValueError("unknown Custodian operation")
+    allowed = {"timeout_seconds", "evidence_sinks"}
+    if operation == "custodian_security_flow":
+        allowed.add("apply")
+        if type(params.get("apply")) is not bool:
+            raise ValueError("security flow requires explicit boolean apply")
+    if set(params) - allowed:
+        raise ValueError("Custodian operation rejects command/path/unknown definition parameters")
+    timeout = params.get("timeout_seconds", 240)
+    if type(timeout) is not int or not 1 <= timeout <= 240:
+        raise ValueError("Custodian operation timeout must be an integer in 1..240")
+
+
+def _path(value: str | None, *, directory=False, existing=True) -> Path:
+    if not value:
+        raise ValueError("Custodian runtime binding missing")
+    path = Path(value)
+    if not path.is_absolute() or ".." in path.parts:
+        raise ValueError("Custodian runtime paths must be absolute without traversal")
+    if any(part.is_symlink() for part in [path, *path.parents]):
+        raise ValueError("Custodian runtime paths must not traverse symlinks")
+    if existing and (not path.is_dir() if directory else not path.is_file()):
+        raise ValueError("Custodian runtime path unavailable")
+    return path
+
+
+def _json(path: Path) -> tuple[dict, bytes]:
+    if path.stat().st_size > _MAX_JSON_BYTES:
+        raise ValueError("Custodian evidence exceeds size bound")
+    raw = path.read_bytes()
+    value = json.loads(raw)
+    if not isinstance(value, dict):
+        raise ValueError("Custodian evidence must be an object")
+    return value, raw
+
+
+def _private_state() -> Path:
+    path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_STATE_ROOT"), directory=True)
+    info = path.stat()
+    if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700:
+        raise ValueError("Custodian state root must be owned by executor with mode 0700")
+    return path
+
+
+def _within(path: Path, roots: list[Path]) -> bool:
+    return any(path == root or root in path.parents for root in roots)
+
+
+def build_command(operation: str, params: dict) -> tuple[list[str], Path, Path, dict | None]:
+    validate_params(operation, params)
+    root = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_ROOT"), directory=True)
+    fleet = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_FLEET_ROOT"), directory=True)
+    state = _private_state()
+    if _within(state, [root, fleet]):
+        raise ValueError("Custodian runtime state must be outside source/fleet checkouts")
+    script_name = "security_flow_daily.py" if operation == "custodian_security_flow" else "weekly_allocation_report.py"
+    script = _path(str(root / "scripts" / script_name))
+    argv = [sys.executable, str(script)]
+    request = None
+    if operation == "custodian_security_flow":
+        repos = os.environ.get("ACTIVITY_CORE_CUSTODIAN_REPOS", "").split(",")
+        if not repos or any(re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", repo) is None for repo in repos):
+            raise ValueError("explicit canonical security-flow repository list required")
+        for repo in repos:
+            _path(str(fleet / repo / "workplans"), directory=True)
+        hub = os.environ.get("ACTIVITY_CORE_CUSTODIAN_HUB", "")
+        parsed = urlsplit(hub)
+        if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
+            raise ValueError("explicit non-credential Custodian Hub base URL required")
+        state = _path(str(state / "security-flow"), directory=True)
+        for name in ("run.lock", "report.json", "delivery.json", "report.json.tmp", "delivery.json.tmp"):
+            _path(str(state / name), existing=False)
+        argv += ["--root", str(fleet), "--repos", ",".join(repos), "--hub", hub, "--state-dir", str(state)]
+        if params["apply"]:
+            argv.append("--apply")
+        output = state / "report.json"
+    else:
+        request_path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_ALLOCATION_REQUEST"))
+        if not _within(request_path, [state]):
+            raise ValueError("allocation request must be in accepted private runtime state")
+        request, _ = _json(request_path)
+        if request.get("schema") != "custodian.weekly-allocation-request.v1" or not request.get("as_of"):
+            raise ValueError("allocation request schema/as_of unbound")
+        paths = {}
+        for name in ("cluster", "allocation", "telemetry", "inventory", "provenance", "previous", "reviews", "output"):
+            value = request.get(name)
+            if not value and name in {"previous", "reviews"}:
+                continue
+            if not isinstance(value, str) or not value:
+                raise ValueError("allocation input/output unbound")
+            candidate = request_path.parent / value
+            paths[name] = _path(str(candidate), existing=name != "output")
+            if not _within(paths[name], [state] if name == "output" else [state, fleet, root]):
+                raise ValueError("allocation input/output outside accepted runtime roots")
+        output = paths["output"]
+        if output == request_path:
+            raise ValueError("allocation output aliases its request")
+        if output in paths.values() and any(output == path for key, path in paths.items() if key != "output"):
+            raise ValueError("allocation output aliases an input")
+        argv += ["--request", str(request_path)]
+    return argv, root, output, request
+
+
+def _execute(argv: list[str], cwd: Path, timeout: int) -> int:
+    # Fixed executable and argv, no shell; kill the group so child helper scripts
+    # cannot outlive a timed-out parent. Never persist child stdout/stderr.
+    with subprocess.Popen(argv, cwd=cwd, stdin=subprocess.DEVNULL,
+                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                          start_new_session=True) as process:
+        try:
+            return process.wait(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            os.killpg(process.pid, signal.SIGKILL)
+            process.wait()
+            raise RuntimeError("Custodian operation timed out") from None
+
+
+def run_custodian_operation(operation: str, params: dict) -> dict:
+    argv, cwd, output, request = build_command(operation, params)
+    snapshot = None
+    try:
+        if request is not None:
+            # Execute the validated request bytes, not a mutable producer pointer.
+            with tempfile.NamedTemporaryFile(mode="w", dir=Path(argv[-1]).parent,
+                                             suffix=".json", delete=False) as stream:
+                snapshot = Path(stream.name)
+                json.dump(request, stream)
+            argv[-1] = str(snapshot)
+        started = time.time_ns()
+        code = _execute(argv, cwd, params.get("timeout_seconds", 240))
+    finally:
+        if snapshot is not None:
+            snapshot.unlink(missing_ok=True)
+    if code != 0:
+        raise RuntimeError(f"Custodian operation failed with exit {code}; inspect owner state")
+    output = _path(str(output))
+    data, raw = _json(output)
+    if operation == "custodian_security_flow":
+        if output.stat().st_mtime_ns < started:
+            raise RuntimeError("Custodian report was not refreshed")
+        if not all(isinstance(data.get(key), list) for key in ("READY", "UNBLOCKED", "ATTENDED", "STALLED", "DEPENDENT")):
+            raise RuntimeError("Custodian security report shape invalid")
+        counts = {key.lower(): len(data[key]) for key in ("READY", "UNBLOCKED", "ATTENDED", "STALLED", "DEPENDENT")}
+    else:
+        if data.get("schema") != "custodian.weekly-allocation.v1" or not isinstance(data.get("recommendations"), list):
+            raise RuntimeError("Custodian allocation report shape invalid")
+        counts = {"recommendations": len(data["recommendations"])}
+    return {"kind": operation, "status": "completed", "exit_code": code,
+            "report_sha256": hashlib.sha256(raw).hexdigest(), "counts": counts,
+            "apply": params.get("apply", False),
+            "owner_acknowledgment": "not established by command completion"}