From fed119a17cf673096a708618bb5098a96dc5d227 Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 19 Oct 2026 15:09:27 +0200 Subject: [PATCH] feat: prepare pinned Custodian worker bundle and required input admission Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181 --- Dockerfile | 9 +- SCOPE.md | 9 + docs/custodian-bounded-operations.md | 8 + docs/custodian-runtime-binding-proposal.md | 6 + k8s/custodian-release/README.md | 90 +++++ k8s/custodian-release/commands/manifest.json | 10 + .../commands/scripts/security_flow_daily.py | 79 +++++ .../commands/scripts/security_flow_report.py | 116 +++++++ .../scripts/sync_resolved_decisions.py | 58 ++++ .../scripts/weekly_allocation_report.py | 321 ++++++++++++++++++ k8s/custodian-release/input-policy.json | 29 ++ k8s/custodian-release/worker-patch.yaml | 40 +++ scripts/verify_custodian_bundle.py | 34 ++ src/activity_core/custodian_inputs.py | 81 +++++ src/activity_core/custodian_operations.py | 4 + tests/test_custodian_bundle.py | 31 ++ tests/test_custodian_inputs.py | 93 +++++ tests/test_custodian_operations.py | 42 +++ 18 files changed, 1058 insertions(+), 2 deletions(-) create mode 100644 k8s/custodian-release/README.md create mode 100644 k8s/custodian-release/commands/manifest.json create mode 100644 k8s/custodian-release/commands/scripts/security_flow_daily.py create mode 100644 k8s/custodian-release/commands/scripts/security_flow_report.py create mode 100644 k8s/custodian-release/commands/scripts/sync_resolved_decisions.py create mode 100644 k8s/custodian-release/commands/scripts/weekly_allocation_report.py create mode 100644 k8s/custodian-release/input-policy.json create mode 100644 k8s/custodian-release/worker-patch.yaml create mode 100644 scripts/verify_custodian_bundle.py create mode 100644 src/activity_core/custodian_inputs.py create mode 100644 tests/test_custodian_bundle.py create mode 100644 tests/test_custodian_inputs.py diff --git a/Dockerfile b/Dockerfile index e43201e..79be240 100644 --- a/Dockerfile +++ b/Dockerfile @@ -28,8 +28,9 @@ COPY Dockerfile ./Dockerfile COPY .forgejo/workflows/image.yaml ./.forgejo/workflows/image.yaml COPY schemas/ ./schemas/ COPY k8s/ ./k8s/ -COPY scripts/render_gitops.py scripts/check_gitops_promotion.py ./scripts/ -RUN uv sync --frozen --extra dev && uv run --no-sync python scripts/render_gitops.py --check --verify-frontend --verify-platform && uv run --no-sync pytest -q -p no:cacheprovider tests/test_frontend_patterns.py tests/test_frontend_monthly_review.py tests/test_pqrst_monthly.py tests/test_repository_grant.py tests/test_instruction_evaluation.py tests/test_admin_sync_api.py tests/test_gitops_release.py tests/test_release_broker.py tests/test_release_transport.py tests/test_release_dispatch.py +COPY scripts/render_gitops.py scripts/check_gitops_promotion.py scripts/verify_custodian_bundle.py ./scripts/ +COPY activity-definitions/ ./activity-definitions/ +RUN uv sync --frozen --extra dev && uv run --no-sync python scripts/render_gitops.py --check --verify-frontend --verify-platform && uv run --no-sync pytest -q -p no:cacheprovider tests/test_frontend_patterns.py tests/test_frontend_monthly_review.py tests/test_pqrst_monthly.py tests/test_repository_grant.py tests/test_instruction_evaluation.py tests/test_admin_sync_api.py tests/test_gitops_release.py tests/test_release_broker.py tests/test_release_transport.py tests/test_release_dispatch.py tests/test_custodian_bundle.py tests/test_custodian_inputs.py tests/test_custodian_operations.py # Stage 2 — runtime image FROM python:3.12-slim@sha256:78387bc3881b8273120a12ebe6c1ab22b018ccc2c9adf565ae1ac9b536e184ea AS runtime @@ -40,6 +41,10 @@ COPY --from=builder /opt/repo-manager/.venv /opt/repo-manager/.venv COPY alembic.ini ./ COPY migrations/ ./migrations/ COPY scripts/ ./scripts/ +# Pinned owner commands; local hash verification performs no network or execution. +COPY k8s/custodian-release/commands/ /opt/custodian-commands/ +COPY k8s/custodian-release/input-policy.json /opt/custodian-input-policy.json +RUN python /app/scripts/verify_custodian_bundle.py --root /opt/custodian-commands COPY activity-definitions/ ./activity-definitions/ COPY event-types/ ./event-types/ COPY tasks/ ./tasks/ diff --git a/SCOPE.md b/SCOPE.md index 9fa9ac3..fe83a6e 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -381,3 +381,12 @@ placement, accepted mounts/network, delivery and first unattended proof remain unverified; see docs/custodian-bounded-operations.md. This adds no scheduler or generic shell. The generated .custodian-brief.md is now an ignored local orientation snapshot; its existing bytes remain local. + + +Custodian delivery preparation now includes an image-baked, hash-verified four-file +owner bundle, tunable input policy, and a required daily pre-execution gate validating +completion markers, exact snapshot hashes and independent per-repo Forge +freshness receipts. The reviewed candidate in `k8s/custodian-release/` is outside +active kustomization; placeholders prevent claiming an accepted image/snapshot. +Both definitions stay disabled. No runtime publication, retention deletion or +schedule activation is implemented by these source artifacts. diff --git a/docs/custodian-bounded-operations.md b/docs/custodian-bounded-operations.md index 7170d9c..c5258b8 100644 --- a/docs/custodian-bounded-operations.md +++ b/docs/custodian-bounded-operations.md @@ -74,3 +74,11 @@ dispatch. It never invokes the real security command, Hub or schedule activation Concrete delivery options and unresolved owner choices are recorded in [custodian-runtime-binding-proposal.md](custodian-runtime-binding-proposal.md). + +Daily execution requires `ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY` and `ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS`, a protected JSON array of +existing Repo Manager receipts. Admission verifies exact complete snapshot +contents and fresh per-repo Forge observations before starting subprocesses. +The existing-worker candidate configures this gate; missing policy or receipts +refuse execution before subprocess. Native freshness noops rely on protected +Repo Manager issuer custody: the issuer checks HEAD against its last successful +commit before emitting `forge_head_matches_last_success`. diff --git a/docs/custodian-runtime-binding-proposal.md b/docs/custodian-runtime-binding-proposal.md index 286d168..8262061 100644 --- a/docs/custodian-runtime-binding-proposal.md +++ b/docs/custodian-runtime-binding-proposal.md @@ -1,5 +1,11 @@ # Custodian runtime binding proposal +Update: user selected the existing worker. The concrete pinned image package, +private state/fleet mount candidate and tunable policy now live in +[`k8s/custodian-release/README.md`](../k8s/custodian-release/README.md). +That packet supersedes the placement alternatives below. It remains outside +active kustomization; schedules are disabled. + Review artifact, 2026-10-19. Not a deployment or activation approval. Both native ActivityDefinitions remain `enabled: false`. No manifests, images or schedules were changed for this proposal. Read with `custodian-bounded-operations.md`. diff --git a/k8s/custodian-release/README.md b/k8s/custodian-release/README.md new file mode 100644 index 0000000..44d18cb --- /dev/null +++ b/k8s/custodian-release/README.md @@ -0,0 +1,90 @@ +# Existing-worker Custodian delivery candidate + +User selected the existing Activity Core worker. This directory prepares that +choice; it is not referenced by active `k8s/gitops/kustomization.yaml`. Neither +schedule is enabled. The patch adds no CPU/memory request or separate process. +The source Dockerfile now embeds the four fixed owner scripts and checks their +actual bytes during build. Source pin is in `commands/manifest.json`; do not +infer deployed bytes from Git HEAD. A new image digest is required. + +`worker-patch.yaml` is a strategic-merge review artifact preserving existing env, +volumes and resources. It reuses `actcore-ops-artefacts` with two narrowly scoped +subPaths: private state and a selected immutable snapshot. Explicit image and +snapshot placeholders must be replaced with real accepted evidence. Do not apply +this file directly. Review its merged diff and render through owner GitOps after +source/image acceptance. No Kubernetes resources were read or changed here. + +`input-policy.json` holds tunable defaults: 2-hour Forge observation age, 5-minute +clock tolerance, exact 15-repo set, 64MiB snapshot bound, recommended 35-day daily +and 13-bundle weekly retention. Retention settings are policy for owner review, +not an installed pruning job. Referenced evidence must survive pruning. Consumer +admission enforces observation age, clock tolerance, scope, hashes and size; +it ignores the exporter's unverified observation claim and instead requires +independent existing Repo Manager receipts for each pinned commit. A recent +`forge_head_matches_last_success` noop is valid re-observation; an old commit can +remain current. Applied/reconciled rows must bind expected/derived/projected +commits. Receipt files are trusted protected owner evidence, not signed proofs. + +## Concrete publisher contract + +Repo Manager commit `3e2726f` adds offline `python -m repo_manager.fleet_snapshot`; +its `docs/custodian-fleet-input-contract.md` defines input request, publication and +recommended ownership. It does not fetch objects or establish remote freshness. +Export selected verified Forge commits from the existing approved repository +object stores, retain the complete matching receipt rows as a JSON array, and +copy the completed immutable tree to `custodian/snapshots/` on the claim. +Only workplans/decisions are admitted. Do not fetch credentials into the worker. + +The candidate uses a fixed snapshot mount deliberately. It can prove one bounded +run after acceptance; it is not a recurring publication handoff. Before enabling +the daily schedule, add the owned pre-run publication/handoff within native +Activity Core orchestration, or a reviewed serialized mount update with consumers +quiescent. A static stale snapshot fails the freshness gate; never refresh its +timestamp. The worker image's packaged Repo Manager version is still the earlier +freshness CLI pin; the exporter is a publisher-side prerequisite, not falsely +claimed installed there. + +## Required preflight and verification + +1. Review/build the new Activity Core image and verify actual installed package + and scripts in it. Run `python /app/scripts/verify_custodian_bundle.py --root + /opt/custodian-commands --expected-revision + 60d893af1bf77f457504b64ff1c9ffb5c6ae6579`. Expected manifest SHA-256: + `8fddc3e90fe90992de2369cb4354ee43d9d25352e958ee371a702dc974f1df1f`. + Inspect the installed `activity_core.custodian_inputs` and operations module + hashes against this reviewed build too; old editable installs are insufficient. +2. Verify current shared-claim capacity, backups and ownership before populating + any input/state. Precreate `custodian/state` and required children owned by + UID1000, mode0700; never change the entire claim. Measure actual snapshot + footprint against retention before accepting this existing claim. Insufficient + capacity is a source/storage prerequisite; do not silently increase spending. +3. Freeze a complete snapshot plus its independent fresh receipt JSON and the + weekly owner request/real collector inputs. Mount the snapshot read-only; + protect state inputs from concurrent writers. Verify every manifest hash and + owner receipt at runtime. Owner-approved source publication expands readable + work-record scope to the explicit repos only; it grants no secret/credential + access and should be reviewed as that exact data-access delta. +4. Confirm existing direct Hub admission for decision GET/resolve, message POST + and progress sink. Source URL alone proves neither authorization nor delivery. + Keep configured recipients and owner accountability. No edge-relay queued + result can substitute for a central message ID. +5. Review resulting Deployment diff, current CPU/headroom and rollout policy. + No added requests are proposed. Existing rolling surge still needs current + headroom; verify available capacity rather than assuming this source's empty + requests justify extra capacity. Keep schedules disabled/paused throughout. +6. Offline fixture tests precede any separately reviewed bounded real run. Verify + missing/stale snapshot rejection before subprocess and collect operation, + report, sink and owner acknowledgment evidence. First unattended proof remains + an acceptance task, not a source-test result. + +Existing standing authorization can cover non-destructive owner deployment when +the reviewed diff, dry-run, rollback, abort owner and receipt are concrete. This +packet does not invent a blanket new founder gate. Changes to access scope, +credentials, public publication or spending limits retain their own authority +requirements; root reviews this exact candidate before deployment. + +Rollback restores prior image and mount/env source through GitOps, keeps schedules +paused, and preserves state/checkpoints/old bundles and snapshots. Daily Hub writes +cannot be undone by rollback. Inspect delivery checkpoints before retries. The +Activity Core runtime owner is the proposed abort/recovery owner; identify the +actual operator in the deployment receipt before applying. diff --git a/k8s/custodian-release/commands/manifest.json b/k8s/custodian-release/commands/manifest.json new file mode 100644 index 0000000..0836b5b --- /dev/null +++ b/k8s/custodian-release/commands/manifest.json @@ -0,0 +1,10 @@ +{ + "schema": "activity-core.custodian-command-bundle.v1", + "source_revision": "60d893af1bf77f457504b64ff1c9ffb5c6ae6579", + "files": { + "scripts/security_flow_daily.py": "4a4e84702b3fb38d086695bb6d601430d8914bd79286b4f683794206f110c353", + "scripts/security_flow_report.py": "eac9fdcc56d7f3ff613e6848c7523fb9a421d82a3b564a51c36f8c2b842eef02", + "scripts/sync_resolved_decisions.py": "182433ca7cd31e2b2773246ec1be02569ba6f089ac009add6543e5e48777b5ec", + "scripts/weekly_allocation_report.py": "93f546243331308d7e60f8db36e988d9847537a2fe46cce49ee2fffba1f7f247" + } +} diff --git a/k8s/custodian-release/commands/scripts/security_flow_daily.py b/k8s/custodian-release/commands/scripts/security_flow_daily.py new file mode 100644 index 0000000..1d18054 --- /dev/null +++ b/k8s/custodian-release/commands/scripts/security_flow_daily.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Daily decision sync and security report, with per-recipient delivery checkpoints. + +Dry-run by default. --apply is the scheduled mode. No credentials are stored. +""" +import argparse +import fcntl +import json +from pathlib import Path +import subprocess +import sys +from urllib.request import Request, urlopen + +from security_flow_report import DEFAULT_REPOS + +SCRIPTS = Path(__file__).resolve().parent + + +def notifications(report, state, agent, founder): + pending = [] + ready = {key: report[key] for key in ("READY", "UNBLOCKED")} + if ready != state.get("agent"): + pending.append(("agent", agent, ready)) + # ATTENDED is the daily operator reminder; empty lists never message the founder. + if report["ATTENDED"]: + pending.append(("founder", founder, {"ATTENDED": report["ATTENDED"]})) + return pending + + +def post(hub, recipient, payload): + body = {"from_agent": "the-custodian", "to_agent": recipient, + "subject": "Daily security flow", "body": json.dumps(payload, sort_keys=True)} + request = Request(hub.rstrip("/") + "/messages/", data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, method="POST") + with urlopen(request, timeout=30) as response: + return json.load(response)["id"] + + +def save(path, data): + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(json.dumps(data, indent=2) + "\n") + temporary.replace(path) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", default=str(Path.home())) + parser.add_argument("--repos", default=",".join(dict.fromkeys(["the-custodian", *DEFAULT_REPOS]))) + parser.add_argument("--hub", default="http://127.0.0.1:8000") + parser.add_argument("--agent", default="the-custodian") + parser.add_argument("--founder", default="human") + parser.add_argument("--state-dir", type=Path, default=Path.home() / ".local/state/custodian-security-flow") + parser.add_argument("--apply", action="store_true") + args = parser.parse_args() + args.state_dir.mkdir(parents=True, exist_ok=True) + with (args.state_dir / "run.lock").open("w") as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + command = [sys.executable, str(SCRIPTS / "sync_resolved_decisions.py"), + "--root", args.root, "--hub", args.hub] + subprocess.run(command + (["--apply"] if args.apply else []), check=True) + raw = subprocess.check_output([sys.executable, str(SCRIPTS / "security_flow_report.py"), + "--root", args.root, "--repos", args.repos, "--json"], text=True) + report = json.loads(raw) + state_path = args.state_dir / "delivery.json" + state = json.loads(state_path.read_text()) if state_path.exists() else {} + save(args.state_dir / "report.json", report) + for key, recipient, payload in notifications(report, state, args.agent, args.founder): + if not args.apply: + print(json.dumps({"dry_run": True, "to_agent": recipient, "report": payload})) + continue + message_id = post(args.hub, recipient, payload) + state[key] = payload + state[key + "_message_id"] = message_id + save(state_path, state) # A later recipient failure does not replay earlier sends. + print(raw) + + +if __name__ == "__main__": + main() diff --git a/k8s/custodian-release/commands/scripts/security_flow_report.py b/k8s/custodian-release/commands/scripts/security_flow_report.py new file mode 100644 index 0000000..d30ef9d --- /dev/null +++ b/k8s/custodian-release/commands/scripts/security_flow_report.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +"""Deterministic flow report for the NetKingdom security estate. + +Replaces agent sessions that poll wait tasks. Reads workplan files only (no hub, +no network) and prints five lists: + + READY todo/progress tasks with no unresolved explicit dependency or owner/human gate + UNBLOCKED wait tasks whose every depends_on target is done/cancel: requalify to todo + ATTENDED needs_human wait tasks: the attended-window checklist for the operator/founder + STALLED wait tasks qualified only by blocked_on message-from (owner reply), oldest first by file + DEPENDENT todo/progress tasks that still carry an unresolved dependency or owner gate + +Usage: security_flow_report.py [--repos a,b,c] [--json] +""" +import argparse, glob, json, os, re + +DEFAULT_REPOS = ("the-custodian net-kingdom railiance-platform activity-core access-engine secrets-engine ops-mason " + "ops-warden key-cape user-engine tenant-engine gate-house kings-guard " + "policy-nexus rapp-policy-nexus").split() +TERMINAL = {"done", "cancel"} + + +def field(block, key): + m = re.search(rf"^{key}:\s*(.*)$", block, re.M) + return m.group(1).strip().strip('"') if m else "" + + +def dependencies(block): + # Accept the convention's inline lists and YAML's indented multiline lists. + m = re.search(r"^depends_on:[ \t]*(.*(?:\n[ \t]+[^\n]*)*)", block, re.M) + return re.findall(r"[A-Z][A-Z0-9-]+-T\d+|[A-Z][A-Z0-9-]+-WP-\d+", m[1]) if m else [] + + +def load(root, repos): + tasks, wps = {}, {} + for r in repos: + for f in glob.glob(os.path.join(root, r, "workplans", "**", "*.md"), recursive=True): + text = open(f, encoding="utf-8", errors="ignore").read() + fm = text.split("---")[1] if text.startswith("---") else "" + wp = {"repo": r, "file": os.path.relpath(f, root), + "status": field(fm, "status"), "flavor": field(fm, "flavor"), + "title": field(fm, "title")} + for b in re.findall(r"```task\n(.*?)```", text, re.S): + tid = field(b, "id") + if not tid: + continue + deps = list(dict.fromkeys(dependencies(fm) + dependencies(b))) + tasks[tid] = {"id": tid, "status": field(b, "status"), "deps": deps, + "human": field(b, "needs_human") == "true", + "blocked_on": field(b, "blocked_on"), + "reason": field(b, "blocking_reason") or field(b, "intervention_note"), + "priority": field(b, "priority"), "wp": wp} + return tasks + + +def classify(tasks): + def dep_done(d): + if d in tasks: + return tasks[d]["status"] in TERMINAL + # workplan-level dependency: every known task of that workplan is terminal + own = [t for t in tasks.values() if t["id"].startswith(d + "-T")] + return bool(own) and all(t["status"] in TERMINAL for t in own) + + out = {"READY": [], "UNBLOCKED": [], "ATTENDED": [], "STALLED": [], "DEPENDENT": []} + for t in tasks.values(): + wp = t["wp"] + if wp["status"] in ("finished", "archived") or wp["flavor"] == "residual": + continue + missing = [d for d in t["deps"] if not dep_done(d)] + if t["status"] in ("todo", "progress"): + if t["human"]: + out["ATTENDED"].append(t | {"pending_dependencies": missing}) + elif missing or t["blocked_on"]: + out["DEPENDENT"].append(t | {"pending_dependencies": missing}) + else: + out["READY"].append(t) + elif t["status"] == "wait": + if t["human"]: + out["ATTENDED"].append(t | {"pending_dependencies": missing}) + elif t["deps"] and all(dep_done(d) for d in t["deps"]): + out["UNBLOCKED"].append(t) + elif not t["deps"] and t["blocked_on"].startswith("message-from"): + out["STALLED"].append(t) + prio = {"high": 0, "medium": 1, "low": 2} + for k in out: + out[k].sort(key=lambda t: (prio.get(t["priority"], 3), t["id"])) + return out + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--root", default=os.path.expanduser("~")) + ap.add_argument("--repos", default=",".join(DEFAULT_REPOS)) + ap.add_argument("--json", action="store_true") + a = ap.parse_args() + res = classify(load(a.root, a.repos.split(","))) + if a.json: + print(json.dumps({k: [{x: t[x] for x in ("id", "status", "priority", "reason", "blocked_on")} + | {"repo": t["wp"]["repo"]} + | ({"pending_dependencies": t["pending_dependencies"]} if "pending_dependencies" in t else {}) + for t in v] for k, v in res.items()}, indent=1)) + return + for k, v in res.items(): + print(f"\n## {k} ({len(v)})") + for t in v: + extra = t["blocked_on"] if k == "STALLED" else t["reason"] + if k == "DEPENDENT": + extra = ", ".join(t["pending_dependencies"]) or t["blocked_on"] + elif k == "ATTENDED" and t["pending_dependencies"]: + extra = "After " + ", ".join(t["pending_dependencies"]) + ": " + extra + print(f"- {t['id']} [{t['status']}/{t['priority'] or '-'}] {t['wp']['repo']}" + + (f" — {extra[:140]}" if extra else "")) + + +if __name__ == "__main__": + main() diff --git a/k8s/custodian-release/commands/scripts/sync_resolved_decisions.py b/k8s/custodian-release/commands/scripts/sync_resolved_decisions.py new file mode 100644 index 0000000..5202b8c --- /dev/null +++ b/k8s/custodian-release/commands/scripts/sync_resolved_decisions.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""Resolve open hub decisions that a repo decision log already records as resolved. + +Repo files are authoritative (ADR-001); the hub is a read model. This scans +~/*/decisions/*.md for fenced yaml blocks with `id`, `title`, `status: resolved` +and resolves the open hub decision whose title matches exactly (whitespace- +normalized, case-insensitive). Uses the sanctioned POST /decisions/{id}/resolve. + +Usage: sync_resolved_decisions.py [--apply] [--hub http://127.0.0.1:8000] +""" +import argparse, glob, json, os, re, urllib.request + + +def norm(s): + return re.sub(r"\s+", " ", s.strip().strip("'\"")).lower() + + +def repo_resolved(root): + out = {} + for f in glob.glob(os.path.join(root, "*", "decisions", "*.md")): + text = open(f, encoding="utf-8", errors="ignore").read() + for b in re.findall(r"```yaml\n(.*?)```", text, re.S): + g = lambda k: (re.search(rf"^{k}:\s*(.*)$", b, re.M) or [None, ""])[1] + if g("status").strip() in ("resolved", "accepted") and g("title"): + out[norm(g("title"))] = (g("id").strip(), os.path.relpath(f, root)) + return out + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--hub", default="http://127.0.0.1:8000") + ap.add_argument("--root", default=os.path.expanduser("~")) + ap.add_argument("--apply", action="store_true") + a = ap.parse_args() + known = repo_resolved(a.root) + data = json.load(urllib.request.urlopen(f"{a.hub}/decisions/?status=open&limit=500")) + data = data.get("items", data) if isinstance(data, dict) else data + hits = 0 + for d in data: + t = norm(d["title"]) + m = known.get(t) or next((v for k, v in known.items() + if min(len(k), len(t)) >= 40 and (k.startswith(t) or t.startswith(k))), None) + if not m: + continue + hits += 1 + rid, path = m + print(f"{d['id'][:8]} {rid} {path} | {d['title'][:70]}") + if a.apply: + body = json.dumps({"rationale": f"Resolved in repo decision log {path} ({rid}); hub synced to repo state.", + "decided_by": "bernd"}).encode() + req = urllib.request.Request(f"{a.hub}/decisions/{d['id']}/resolve", data=body, + headers={"Content-Type": "application/json"}, method="POST") + urllib.request.urlopen(req).read() + print(f"{hits} open hub decisions match resolved repo records" + ("" if a.apply else " (dry run)")) + + +if __name__ == "__main__": + main() diff --git a/k8s/custodian-release/commands/scripts/weekly_allocation_report.py b/k8s/custodian-release/commands/scripts/weekly_allocation_report.py new file mode 100644 index 0000000..4917a23 --- /dev/null +++ b/k8s/custodian-release/commands/scripts/weekly_allocation_report.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +"""Retain CUST-WP-0071 sizing evidence; no network, resizing or task emission.""" +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone +import hashlib +import json +import math +import os +from pathlib import Path +import re +import tempfile + +SCHEMA = "custodian.weekly-allocation.v1" +METRICS = ( + "cpu_p95_m", "cpu_peak_m", "memory_p95_bytes", "memory_peak_bytes", + "cpu_throttle_ratio", "latency_p95_seconds", "errors", "request_count", + "restarts", "oom_events", "cpu_sample_count", "memory_sample_count", +) +REQUIRED = tuple(m for m in METRICS if not m.endswith("sample_count")) + + +def timestamp(value): + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + if result.tzinfo is None: + raise ValueError("timestamps must include timezone") + return result.astimezone(timezone.utc) + + +def duration(value): + match = re.fullmatch(r"([1-9][0-9]*)([smhdw])", value) + if not match: + raise ValueError(f"unsupported duration: {value}") + return int(match[1]) * {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}[match[2]] + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, allow_nan=False).encode()).hexdigest() + + +def finite(value): + number = float(value) + if not math.isfinite(number) or number < 0: + raise ValueError("metric must be finite and nonnegative") + return number + + +def metric_series(telemetry): + """Do not distribute namespace aggregates over workloads, or sum percentiles.""" + series, queries = {}, {} + seconds = duration(telemetry["window"]) + for name, query in sorted(telemetry.get("queries", {}).items()): + response = query.get("response", {}) + data = response.get("data", {}) + queries[name] = { + "promql": query.get("promql"), "response_status": response.get("status"), + "result_type": data.get("resultType"), "warnings": response.get("warnings", []), + "error": response.get("error"), "response_sha256": digest(response), + "evaluation_times": [], + } + if name not in METRICS or response.get("status") != "success" or data.get("resultType") != "vector": + continue + for sample in data.get("result", []): + labels = sample.get("metric", {}) + namespace, workload = labels.get("namespace"), labels.get("workload") + # Additional dimensions are ambiguous: do not silently collapse pod/job series. + if not namespace or set(labels) - {"namespace", "workload", "__name__"}: + queries[name].setdefault("rejected", []).append("unaggregated or absent identity labels") + continue + key = (namespace, workload) + values = series.setdefault(key, {}) + if name in values: + values[name] = {"value": None, "reason": "ambiguous duplicate series"} + continue + try: + epoch, raw = sample["value"] + end = datetime.fromtimestamp(finite(epoch), timezone.utc) + number = finite(raw) + except (KeyError, ValueError, TypeError, OverflowError): + values[name] = {"value": None, "reason": "invalid sample"} + continue + queries[name]["evaluation_times"].append(end.isoformat()) + values[name] = { + "value": number, "window_start": (end - timedelta(seconds=seconds)).isoformat(), + "window_end": end.isoformat(), "query": name, + "capture_alignment_seconds": (end - timestamp(telemetry["captured_at"])).total_seconds(), + } + return series, queries + + +def recommendation_id(resource, namespace, workload): + return "allocation-" + digest([resource, namespace, workload, "sizing-review"])[:24] + + +def build_report(cluster, allocation, telemetry, inventory, provenance, *, as_of, + previous=None, reviews=None, max_age_hours=48): + now = timestamp(as_of) + if cluster.get("schema_version") != "railiance.cluster-resource-observation.v1": + raise ValueError("unsupported cluster schema") + if allocation.get("schema") != "custodian.allocation-reconcile.v1": + raise ValueError("unsupported allocation schema") + resource = cluster["resource_id"] + if inventory.get("id") != resource or allocation.get("host", {}).get("cluster_resource_id") != resource: + raise ValueError("resource identity mismatch") + if allocation.get("count_host_and_cluster_cpus_once") is not True: + raise ValueError("allocation must deduplicate host and cluster CPUs") + if cluster["captured_at"] != allocation["captured_at"]: + raise ValueError("cluster and reconcile snapshots differ") + if provenance.get("source_observation") != cluster["captured_at"]: + raise ValueError("provenance does not identify cluster snapshot") + seconds = duration(telemetry["window"]) + if seconds != 604800: + raise ValueError("weekly report requires a seven-day telemetry window") + step = duration(telemetry["step"]) + series, queries = metric_series(telemetry) + captured = {"cluster": cluster["captured_at"], "telemetry": telemetry["captured_at"]} + freshness = {} + for name, value in captured.items(): + age = (now - timestamp(value)).total_seconds() + freshness[name] = {"age_seconds": age, "status": "future" if age < 0 else "stale" if age > max_age_hours * 3600 else "fresh"} + query_signature = digest({name: q["promql"] for name, q in queries.items()}) + end = timestamp(telemetry["captured_at"]) + comparison_reason = "no previous retained report" + compatible = False + if previous: + prev_end = timestamp(previous["period"]["capture_end"]) + compatible = ( + previous.get("schema") == SCHEMA and previous.get("resource_id") == resource + and previous["period"].get("seconds") == seconds + and previous["period"].get("step_seconds") == step + and previous.get("query_signature") == query_signature + and abs((end - prev_end).total_seconds() - seconds) <= step + ) + comparison_reason = "comparable consecutive capture windows" if compatible else "nonconsecutive windows or changed query/resource/step contract" + previous_rows = {(r["namespace"], r["workload"]): r for r in (previous or {}).get("workloads", [])} + owners = {(r["namespace"], r["workload"]): r for r in allocation.get("workloads", [])} + if len(owners) != len(allocation.get("workloads", [])): + raise ValueError("duplicate allocation workload identity") + reviews = reviews or {} + rows, recommendations = [], [] + seen = set() + for workload in sorted(cluster.get("utilization", {}).get("workloads", []), key=lambda x: (x["namespace"], x["workload"])): + key = workload["namespace"], workload["workload"] + if key in seen: + raise ValueError("duplicate cluster workload identity") + seen.add(key) + owner = owners.get(key, {}) + measurements = {m: series.get(key, {}).get(m, {"value": None, "reason": "no workload series"}) for m in METRICS} + gaps = [m for m in REQUIRED if measurements[m]["value"] is None] + metric_stale = [m for m, v in measurements.items() if v["value"] is not None and not 0 <= (now - timestamp(v["window_end"])).total_seconds() <= max_age_hours * 3600] + metric_unaligned = [m for m, v in measurements.items() if v["value"] is not None and abs(v["capture_alignment_seconds"]) > step] + rid = recommendation_id(resource, *key) + reasons = [] + if gaps: + reasons.append("missing workload metrics: " + ", ".join(gaps)) + if metric_stale or any(v["status"] != "fresh" for v in freshness.values()): + reasons.append("stale or future observation") + if metric_unaligned: + reasons.append("metric evaluation windows differ from capture by more than one step") + if owner.get("owner", "unknown") == "unknown": + reasons.append("unknown owner") + if workload.get("requests", {}).get("cpu_millicores") == 0: + reasons.append("zero CPU request requires owner review") + review = reviews.get(rid) + decision, confidence, evidence = "investigate", "low", [] + if review: + if review.get("decision") not in {"keep", "increase", "decrease", "investigate"}: + raise ValueError("invalid owner review decision") + required = ("owner", "evidence_refs", "receiving_task", "configuration", "rollback_ref", "reviewed_at", "period_end", "coverage_ref", "service_acceptance_ref", "margin_ref") + if any(not review.get(k) for k in required): + raise ValueError("owner review missing evidence, receiving task or configuration/rollback") + if review["owner"] != owner.get("owner") or review["period_end"] != telemetry["captured_at"]: + reasons.append("owner review does not match this owner/window") + elif timestamp(review["reviewed_at"]) > now: + reasons.append("owner review is future-dated") + elif not reasons: + decision, confidence, evidence = review["decision"], "owner-reviewed", review["evidence_refs"] + else: + reasons.append("owner proposal retained; incomplete evidence prevents accepted sizing recommendation") + else: + reasons.append("no owner-reviewed sizing decision") + row = { + "namespace": key[0], "workload": key[1], "owner": owner.get("owner", "unknown"), + "service": owner.get("service", "unknown"), "tenant": owner.get("tenant", "unknown"), + "pods": workload.get("pods"), "requests": workload.get("requests"), + "limits": workload.get("limits"), "instantaneous_observed": workload.get("observed"), + "metrics": measurements, "missing_metrics": gaps, + "namespace_context_ref": key[0] if (key[0], None) in series else None, + "raw_scrape_coverage": "unknown; evaluation counts do not prove source coverage", + "recommendation_id": rid, "week_on_week": {}, "comparison_reasons": {}, + } + prior = previous_rows.get(key) + for metric in METRICS: + current = measurements[metric]["value"] + old_measurement = (prior or {}).get("metrics", {}).get(metric, {}) + old = old_measurement.get("value") + metric_compatible = False + reason = comparison_reason + if compatible and current is not None and old is not None: + try: + old_end, old_start = timestamp(old_measurement["window_end"]), timestamp(old_measurement["window_start"]) + new_end, new_start = timestamp(measurements[metric]["window_end"]), timestamp(measurements[metric]["window_start"]) + metric_compatible = ( + (old_end - old_start).total_seconds() == seconds + and (new_end - new_start).total_seconds() == seconds + and abs((new_end - old_end).total_seconds() - seconds) <= step + and abs((new_start - old_end).total_seconds()) <= step + and abs(measurements[metric]["capture_alignment_seconds"]) <= step + and abs((old_end - timestamp(previous["period"]["capture_end"])).total_seconds()) <= step + ) + reason = "comparable consecutive metric windows" if metric_compatible else "metric evaluation windows are not consecutive or capture-aligned" + except (KeyError, ValueError, TypeError): + reason = "missing or invalid previous metric window" + elif compatible: + reason = "missing current or previous metric" + row["week_on_week"][metric] = current - old if metric_compatible else None + row["comparison_reasons"][metric] = reason + rows.append(row) + prior_recommendation = next((r for r in (previous or {}).get("recommendations", []) if r["id"] == rid), {}) + recommendations.append({ + "id": rid, "decision": decision, "confidence": confidence, "owner": row["owner"], + "reasons": reasons, "evidence_refs": evidence, "owner_proposal": review, + "receiving_task": review.get("receiving_task") if review else prior_recommendation.get("receiving_task"), + "previous_decision": prior_recommendation.get("decision"), + "owner_outcome": review.get("outcome") if review else None, + "delivery": "not emitted", "acknowledgment": None, + }) + if seen != set(owners): + raise ValueError("cluster and allocation workload inventories differ") + return { + "schema": SCHEMA, "workplan": "CUST-WP-0071-T05", "as_of": as_of, + "adapter_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + "resource_id": resource, "period": {"capture_end": telemetry["captured_at"], "seconds": seconds, "step_seconds": step, + "note": "Actual per-query windows are retained on samples; capture time is not a common query evaluation time."}, + "freshness": freshness, "max_age_hours": max_age_hours, "query_signature": query_signature, + "queries": queries, "collector_provenance": provenance, + "source_sha256": {"cluster": digest(cluster), "allocation": digest(allocation), "telemetry": digest(telemetry), "inventory": digest(inventory)}, + "node": {"capacity": cluster.get("capacity"), "allocation": cluster.get("allocation"), + "scheduled_request_cpu_m": allocation.get("scheduled_request_cpu_m"), + "pending_unscheduled_cpu_m": allocation.get("pending_unscheduled_cpu_m"), + "reservation_residual_cpu_m": allocation.get("residual_cpu_m"), + "release_preflight": allocation.get("state_hub_preflight_observation"), + "operating_margin": None, "release_margin_acceptance": "not established by reservation residual", + "host_cluster_counted_once": True}, + "portfolio": {k: inventory.get(k) for k in ("id", "schema_version", "ownership", "capacity", "relationships", "evidence")}, + "workloads": rows, + "namespace_context": {ns: values for (ns, workload), values in sorted(series.items(), key=lambda x: str(x[0])) if workload is None}, + "comparison": {"compatible": compatible, "reason": comparison_reason, + "previous_report_sha256": digest(previous) if previous else None, + "disappeared_workloads": [list(key) for key in sorted(set(previous_rows) - seen)]}, + "measurement_gaps": cluster.get("measurement_gaps", []) + allocation.get("measurement_gaps", []), + "recommendations": recommendations, + "limitations": ["No representative pilot acceptance inferred from telemetry.", + "Namespace percentiles and peaks are not workload metrics or additive node peaks.", + "Portfolio capacity is dated evidence, not additional host capacity.", + "No procurement, sizing changes, owner-task creation, delivery or acknowledgment performed.", + "Installed executor, scoped reads, independent missed-run handling and first unattended receipt remain T05 gates."], + } + + +def retain(path, report): + """Never overwrite a retained week; identical retries succeed.""" + encoded = json.dumps(report, indent=2, sort_keys=True, allow_nan=False) + "\n" + path.parent.mkdir(parents=True, exist_ok=True) + temporary = None + try: + with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as stream: + temporary = Path(stream.name) + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + try: + os.link(temporary, path) + except FileExistsError: + if path.is_symlink() or path.read_text() != encoded: + raise ValueError("retained report differs; choose a new explicit revision path") + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--request", type=Path, help="Executor-owned JSON request; paths relative to its directory") + for name in ("cluster", "allocation", "telemetry", "inventory", "provenance"): + parser.add_argument("--" + name, type=Path) + parser.add_argument("--previous", type=Path) + parser.add_argument("--reviews", type=Path) + parser.add_argument("--as-of") + parser.add_argument("--max-age-hours", type=float, default=48) + parser.add_argument("--output", type=Path) + args = parser.parse_args() + try: + if args.request: + request = json.loads(args.request.read_text()) + if request.get("schema") != "custodian.weekly-allocation-request.v1": + raise ValueError("unsupported executor request schema") + if any(getattr(args, key) for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "previous", "reviews", "as_of", "output")): + raise ValueError("request cannot be mixed with input/output overrides") + for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "previous", "reviews", "output"): + value = request.get(key) + setattr(args, key, args.request.parent / value if value else None) + args.as_of = request.get("as_of") + if any(not getattr(args, key) for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "as_of", "output")): + raise ValueError("executor input paths, as_of and retention output must be explicitly bound") + if not math.isfinite(args.max_age_hours) or args.max_age_hours <= 0: + raise ValueError("max-age-hours must be positive and finite") + inputs = {name: json.loads(getattr(args, name).read_text()) for name in ("cluster", "allocation", "telemetry", "inventory", "provenance")} + report = build_report(**inputs, as_of=args.as_of, max_age_hours=args.max_age_hours, + previous=json.loads(args.previous.read_text()) if args.previous else None, + reviews=json.loads(args.reviews.read_text()) if args.reviews else None) + retain(args.output, report) + except (OSError, ValueError, KeyError, TypeError) as exc: + parser.exit(2, f"weekly allocation report refused: {exc}\n") + print(json.dumps({"status": "retained", "path": str(args.output), "sha256": digest(report), + "workloads": len(report["workloads"]), "delivery": "not emitted"})) + + +if __name__ == "__main__": + main() diff --git a/k8s/custodian-release/input-policy.json b/k8s/custodian-release/input-policy.json new file mode 100644 index 0000000..27fd431 --- /dev/null +++ b/k8s/custodian-release/input-policy.json @@ -0,0 +1,29 @@ +{ + "schema": "activity-core.custodian-input-policy.v1", + "publisher_owner": "repo-manager", + "consumer_owner": "activity-core", + "max_observation_age_seconds": 7200, + "max_future_skew_seconds": 300, + "daily_retention_days": 35, + "weekly_retention_count": 13, + "preserve_referenced_evidence": true, + "max_snapshot_bytes": 67108864, + "repositories": [ + "the-custodian", + "net-kingdom", + "railiance-platform", + "activity-core", + "access-engine", + "secrets-engine", + "ops-mason", + "ops-warden", + "key-cape", + "user-engine", + "tenant-engine", + "gate-house", + "kings-guard", + "policy-nexus", + "rapp-policy-nexus" + ], + "expected_hub_instance": "railiance01" +} diff --git a/k8s/custodian-release/worker-patch.yaml b/k8s/custodian-release/worker-patch.yaml new file mode 100644 index 0000000..6a47259 --- /dev/null +++ b/k8s/custodian-release/worker-patch.yaml @@ -0,0 +1,40 @@ +# REVIEW ONLY: not referenced by k8s/gitops/kustomization.yaml. +# Replace both explicit placeholders only after image/input evidence is accepted. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: actcore-worker + namespace: activity-core +spec: + template: + spec: + containers: + - name: worker + image: forgejo.coulomb.social/coulomb/activity-core@sha256:REQUIRES_REVIEWED_IMAGE_DIGEST + env: + - name: ACTIVITY_CORE_CUSTODIAN_ROOT + value: /opt/custodian-commands + - name: ACTIVITY_CORE_CUSTODIAN_FLEET_ROOT + value: /var/lib/custodian-inputs/fleet + - name: ACTIVITY_CORE_CUSTODIAN_STATE_ROOT + value: /var/lib/activity-core/custodian + - name: ACTIVITY_CORE_CUSTODIAN_REPOS + value: the-custodian,net-kingdom,railiance-platform,activity-core,access-engine,secrets-engine,ops-mason,ops-warden,key-cape,user-engine,tenant-engine,gate-house,kings-guard,policy-nexus,rapp-policy-nexus + - name: ACTIVITY_CORE_CUSTODIAN_HUB + value: http://state-hub.state-hub.svc.cluster.local:8000 + - name: ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY + value: /opt/custodian-input-policy.json + - name: ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS + value: /var/lib/activity-core/custodian/inputs/freshness-receipts.json + - name: ACTIVITY_CORE_CUSTODIAN_ALLOCATION_REQUEST + value: /var/lib/activity-core/custodian/requests/current.json + volumeMounts: + - name: ops-artefacts + mountPath: /var/lib/activity-core/custodian + subPath: custodian/state + - name: ops-artefacts + mountPath: /var/lib/custodian-inputs/fleet + subPath: custodian/snapshots/REQUIRES_ACCEPTED_SNAPSHOT_ID + readOnly: true +# Existing ops-artefacts volume/claim, UID1000 and resources are preserved. +# Owner precreates only these private subdirectories; no whole-claim chmod/chown. diff --git a/scripts/verify_custodian_bundle.py b/scripts/verify_custodian_bundle.py new file mode 100644 index 0000000..1d45487 --- /dev/null +++ b/scripts/verify_custodian_bundle.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +"""Verify actual installed bytes of the fixed Custodian command bundle; no execution.""" +import argparse +import hashlib +import json +from pathlib import Path + +FILES = {"scripts/security_flow_daily.py", "scripts/security_flow_report.py", + "scripts/sync_resolved_decisions.py", "scripts/weekly_allocation_report.py"} + + +def verify(root, expected_revision=None): + root = Path(root) + manifest = json.loads((root / "manifest.json").read_text()) + if manifest.get("schema") != "activity-core.custodian-command-bundle.v1" or set(manifest.get("files", {})) != FILES: + raise ValueError("unexpected bundle contract") + if expected_revision and manifest.get("source_revision") != expected_revision: + raise ValueError("unexpected source revision") + for relative, expected in manifest["files"].items(): + path = root / relative + if any(p.is_symlink() for p in [path, *path.parents]) or not path.is_file(): + raise ValueError("bundle path is not a regular installed file") + if hashlib.sha256(path.read_bytes()).hexdigest() != expected: + raise ValueError("installed bundle hash mismatch") + return {"source_revision": manifest["source_revision"], "verified_files": len(FILES), + "manifest_sha256": hashlib.sha256((root / "manifest.json").read_bytes()).hexdigest()} + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", type=Path, required=True) + parser.add_argument("--expected-revision") + args = parser.parse_args() + print(json.dumps(verify(args.root, args.expected_revision), sort_keys=True)) diff --git a/src/activity_core/custodian_inputs.py b/src/activity_core/custodian_inputs.py new file mode 100644 index 0000000..11cbf46 --- /dev/null +++ b/src/activity_core/custodian_inputs.py @@ -0,0 +1,81 @@ +"""Fail-closed admission for trusted-owner fleet snapshots and freshness receipts.""" +import hashlib +import json +import stat +from datetime import UTC, datetime +from pathlib import Path, PurePosixPath + + +def _read(path): + path = Path(path) + if any(p.is_symlink() for p in [path, *path.parents]) or not stat.S_ISREG(path.stat().st_mode) or path.stat().st_size > 8 * 1024 * 1024: + raise ValueError("unsafe fleet evidence path or size") + return path.read_bytes() + + +def admit(fleet, policy_path, receipts_path, repos, now=None): + """Files/receipts are protected owner inputs, not signatures or remote verification.""" + fleet = Path(fleet) + policy = json.loads(_read(policy_path)) + if policy.get("schema") != "activity-core.custodian-input-policy.v1" or set(policy["repositories"]) != set(repos): + raise ValueError("fleet policy repository mismatch") + raw = _read(fleet / "manifest.json") + if _read(fleet / "COMPLETE").decode().strip() != hashlib.sha256(raw).hexdigest(): + raise ValueError("incomplete fleet snapshot") + manifest = json.loads(raw) + records = manifest.get("repositories", []) + if manifest.get("schema") != "repo-manager.fleet-input.v1" or len(records) != len(repos) or {r["slug"] for r in records} != set(repos): + raise ValueError("fleet snapshot repository mismatch") + if {p.name for p in fleet.iterdir()} != set(repos) | {"manifest.json", "COMPLETE"}: + raise ValueError("snapshot contains unexpected repository or file") + receipts = json.loads(_read(receipts_path)) + now = now or datetime.now(UTC) + expected_files = set() + total = 0 + for record in records: + valid = False + for receipt in receipts: + if receipt.get("schema") != "repo-manager.freshness-receipt.v1" or receipt.get("repo_slug") != record["slug"] or receipt.get("forge_commit") != record["commit"]: + continue + hub = receipt.get("hub", {}) + if hub.get("verified_primary") is not True or hub.get("instance_role") != "primary" or hub.get("instance_label") != policy["expected_hub_instance"]: + continue + if receipt.get("outcome") not in {"applied", "noop"}: + continue + noop = receipt.get("outcome") == "noop" and receipt.get("reason") == "forge_head_matches_last_success" + bound = all(receipt.get(key) == record["commit"] for key in ("expected_commit", "derived_commit", "projected_commit")) + # Native issuer emits this reason only after HEAD == last successful commit. + # Receipts are protected owner evidence; this is not signature verification. + bound = bound or noop + if not bound: + continue + observed = datetime.fromisoformat(receipt["finished_at"]) + if observed.tzinfo is None: + continue + age = (now - observed).total_seconds() + if -policy["max_future_skew_seconds"] <= age <= policy["max_observation_age_seconds"]: + valid = True + if not valid: + raise ValueError("missing accepted fresh Forge receipt") + for entry in record["files"]: + relative = PurePosixPath(entry["path"]) + if relative.is_absolute() or ".." in relative.parts or relative.parts[0] not in {"workplans", "decisions"} or relative.suffix != ".md": + raise ValueError("unsafe snapshot record") + path = fleet / record["slug"] / relative + data = _read(path) + total += len(data) + if total > policy["max_snapshot_bytes"]: + raise ValueError("snapshot size policy exceeded") + if len(data) != entry["bytes"] or hashlib.sha256(data).hexdigest() != entry["sha256"]: + raise ValueError("snapshot record hash mismatch") + expected_files.add(path) + actual_files = set() + for slug in repos: + for path in (fleet / slug).rglob("*"): + if path.is_symlink(): + raise ValueError("snapshot symlink refused") + if path.is_file(): + actual_files.add(path) + if actual_files != expected_files or total > policy["max_snapshot_bytes"]: + raise ValueError("snapshot content outside manifest or size policy") + return hashlib.sha256(raw).hexdigest() diff --git a/src/activity_core/custodian_operations.py b/src/activity_core/custodian_operations.py index de7a67c..265829c 100644 --- a/src/activity_core/custodian_operations.py +++ b/src/activity_core/custodian_operations.py @@ -89,6 +89,10 @@ def build_command(operation: str, params: dict) -> tuple[list[str], Path, Path, raise ValueError("explicit canonical security-flow repository list required") for repo in repos: _path(str(fleet / repo / "workplans"), directory=True) + from activity_core.custodian_inputs import admit + policy_path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY")) + receipts_path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS")) + admit(fleet, policy_path, receipts_path, repos) 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 {"", "/"}: diff --git a/tests/test_custodian_bundle.py b/tests/test_custodian_bundle.py new file mode 100644 index 0000000..67dcce1 --- /dev/null +++ b/tests/test_custodian_bundle.py @@ -0,0 +1,31 @@ +import importlib.util +import shutil +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +spec = importlib.util.spec_from_file_location("bundle_verifier", ROOT / "scripts/verify_custodian_bundle.py") +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) + + +def test_reviewed_bundle_bytes(): + result = module.verify(ROOT / "k8s/custodian-release/commands", "60d893af1bf77f457504b64ff1c9ffb5c6ae6579") + assert result["verified_files"] == 4 + + +def test_tampered_installed_file_refused(tmp_path): + shutil.copytree(ROOT / "k8s/custodian-release/commands", tmp_path / "bundle") + (tmp_path / "bundle/scripts/security_flow_daily.py").write_text("changed") + with pytest.raises(ValueError, match="hash mismatch"): + module.verify(tmp_path / "bundle") + + +def test_installed_symlink_refused(tmp_path): + shutil.copytree(ROOT / "k8s/custodian-release/commands", tmp_path / "bundle") + path = tmp_path / "bundle/scripts/security_flow_daily.py" + path.unlink() + path.symlink_to(ROOT / "k8s/custodian-release/commands/scripts/security_flow_daily.py") + with pytest.raises(ValueError, match="regular installed"): + module.verify(tmp_path / "bundle") diff --git a/tests/test_custodian_inputs.py b/tests/test_custodian_inputs.py new file mode 100644 index 0000000..ed671b3 --- /dev/null +++ b/tests/test_custodian_inputs.py @@ -0,0 +1,93 @@ +import hashlib +import json +from datetime import UTC, datetime + +import pytest + +from activity_core.custodian_inputs import admit + + +@pytest.fixture +def evidence(tmp_path): + fleet = tmp_path / "fleet" + file = fleet / "example/workplans/task.md" + file.parent.mkdir(parents=True) + file.write_text("task") + manifest = {"schema": "repo-manager.fleet-input.v1", "freshness_verified": False, + "repositories": [{"slug": "example", "commit": "a" * 40, "files": [ + {"path": "workplans/task.md", "bytes": 4, "sha256": hashlib.sha256(b"task").hexdigest()}]}]} + raw = json.dumps(manifest).encode() + (fleet / "manifest.json").write_bytes(raw) + (fleet / "COMPLETE").write_text(hashlib.sha256(raw).hexdigest()) + policy = tmp_path / "policy.json" + policy.write_text(json.dumps({"schema": "activity-core.custodian-input-policy.v1", "repositories": ["example"], + "max_observation_age_seconds": 7200, "max_future_skew_seconds": 300, + "max_snapshot_bytes": 1024, "expected_hub_instance": "railiance01"})) + receipts = tmp_path / "receipts.json" + receipt = {"schema": "repo-manager.freshness-receipt.v1", "repo_slug": "example", "forge_commit": "a" * 40, + "outcome": "noop", "reason": "forge_head_matches_last_success", "finished_at": "2026-10-19T10:00:00Z", + "hub": {"verified_primary": True, "instance_role": "primary", "instance_label": "railiance01"}} + prior = {**receipt, "outcome": "applied", "reason": None, + "finished_at": "2026-10-01T00:00:00Z", + **{key: "a" * 40 for key in ("expected_commit", "derived_commit", "projected_commit")}} + receipts.write_text(json.dumps([receipt, prior])) + return fleet, policy, receipts + + +def check(evidence): + return admit(*evidence, ["example"], now=datetime(2026, 10, 19, 11, tzinfo=UTC)) + + +def test_accepts_independent_recent_receipt(evidence): + assert len(check(evidence)) == 64 + + +@pytest.mark.parametrize("case", ["missing", "stale", "future", "wrong_commit", "wrong_hub", "failed"]) +def test_rejects_invalid_freshness(evidence, case): + path = evidence[2] + rows = json.loads(path.read_text()) + if case == "missing": + rows = [] + elif case == "stale": + rows[0]["finished_at"] = "2026-10-18T10:00:00Z" + elif case == "future": + rows[0]["finished_at"] = "2026-10-20T10:00:00Z" + elif case == "wrong_commit": + rows[0]["forge_commit"] = "b" * 40 + elif case == "wrong_hub": + rows[0]["hub"]["instance_label"] = "workstation" + else: + rows[0]["outcome"] = "failed" + path.write_text(json.dumps(rows)) + with pytest.raises(ValueError, match="fresh Forge receipt"): + check(evidence) + + +@pytest.mark.parametrize("case", ["tamper", "extra", "marker", "extra_repo"]) +def test_rejects_changed_snapshot(evidence, case): + fleet = evidence[0] + if case == "tamper": + (fleet / "example/workplans/task.md").write_text("changed") + elif case == "extra": + (fleet / "example/workplans/extra.md").write_text("extra") + elif case == "extra_repo": + (fleet / "outside").mkdir() + else: + (fleet / "COMPLETE").write_text("incorrect") + with pytest.raises(ValueError): + check(evidence) + + +def test_native_noop_uses_protected_issuer_contract(evidence): + path = evidence[2] + path.write_text(json.dumps(json.loads(path.read_text())[:1])) + assert len(check(evidence)) == 64 + + +def test_fifo_refused_without_open(evidence): + import os + path = evidence[2] + path.unlink() + os.mkfifo(path) + with pytest.raises(ValueError, match="unsafe"): + check(evidence) diff --git a/tests/test_custodian_operations.py b/tests/test_custodian_operations.py index 06d826c..77770d7 100644 --- a/tests/test_custodian_operations.py +++ b/tests/test_custodian_operations.py @@ -1,4 +1,6 @@ import json +import hashlib +from datetime import datetime, timezone from pathlib import Path from unittest.mock import patch @@ -21,6 +23,26 @@ def runtime(tmp_path, monkeypatch): monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_HUB", "http://127.0.0.1:8000") monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_REPOS", "the-custodian") (fleet / "the-custodian/workplans").mkdir(parents=True) + (fleet / "the-custodian/workplans/task.md").write_text("task") + manifest = {"schema": "repo-manager.fleet-input.v1", "repositories": [{ + "slug": "the-custodian", "commit": "a" * 40, "files": [{"path": "workplans/task.md", + "bytes": 4, "sha256": hashlib.sha256(b"task").hexdigest()}]}]} + raw = json.dumps(manifest).encode() + (fleet / "manifest.json").write_bytes(raw) + (fleet / "COMPLETE").write_text(hashlib.sha256(raw).hexdigest()) + policy = state / "policy.json" + policy.write_text(json.dumps({"schema": "activity-core.custodian-input-policy.v1", + "repositories": ["the-custodian"], "max_observation_age_seconds": 7200, + "max_future_skew_seconds": 300, "max_snapshot_bytes": 1024, + "expected_hub_instance": "railiance01"})) + receipts = state / "receipts.json" + receipts.write_text(json.dumps([{"schema": "repo-manager.freshness-receipt.v1", + "repo_slug": "the-custodian", "outcome": "applied", + "finished_at": datetime.now(timezone.utc).isoformat(), + "hub": {"verified_primary": True, "instance_role": "primary", "instance_label": "railiance01"}, + **{key: "a" * 40 for key in ("forge_commit", "expected_commit", "derived_commit", "projected_commit")}}])) + monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY", str(policy)) + monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS", str(receipts)) for name in ("security_flow_daily.py", "weekly_allocation_report.py"): (root / "scripts" / name).write_text("# owner fixture\n") return root, fleet, state @@ -168,3 +190,23 @@ def test_subprocess_timeout_kills_group(): with pytest.raises(RuntimeError, match='timed out'): runner._execute(['fixed-owner-command'], Path('/tmp'), 1) kill.assert_called_once_with(12345, runner.signal.SIGKILL) + + +def test_configured_freshness_admission_refuses_before_subprocess(runtime, monkeypatch): + root, fleet, state = runtime + policy = state / "policy.json" + receipts = state / "receipts.json" + policy.write_text("{}") + receipts.write_text("[]") + monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY", str(policy)) + monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS", str(receipts)) + with patch.object(runner, "_execute") as execute, pytest.raises(ValueError): + runner.run_custodian_operation("custodian_security_flow", {"apply": True}) + execute.assert_not_called() + + +def test_missing_policy_refuses_before_subprocess(runtime, monkeypatch): + monkeypatch.delenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY") + with patch.object(runner, "_execute") as execute, pytest.raises(ValueError): + runner.run_custodian_operation("custodian_security_flow", {"apply": True}) + execute.assert_not_called()