diff --git a/bounded-executor/cli.py b/bounded-executor/cli.py index b87a061..a81877b 100644 --- a/bounded-executor/cli.py +++ b/bounded-executor/cli.py @@ -17,6 +17,11 @@ The config (JSON, non-secret) names endpoints, token files and the pilot's targe Exit codes: 0 completed, 3 reverted, 4 refused before any effect, 5 stopped / unknown / revert failed / budget, 6 consumed but not in custody. The result printed is value-free. +Kill switch (NK-WP-0056-T07): SIGTERM or SIGINT, e.g. the founder's `kubectl -n sso delete job `, sets the stop +flag. Before consume the run refuses and spends nothing; after it, the runner finishes the step in flight, then +stops and reverts what it changed (BEA stop_and_revert). The Job's terminationGracePeriodSeconds must cover a +revert. + python3 bounded-executor/cli.py --journey-only --config FILE [--result FILE] python3 bounded-executor/cli.py --request --admission FILE --config FILE [--result FILE] @@ -46,7 +51,9 @@ import argparse import importlib.util import json import pathlib +import signal import sys +import threading HERE = pathlib.Path(__file__).resolve().parent @@ -63,6 +70,14 @@ X = _load("executor") OPS = _load("operations") C = _load("clients") EXIT = {"completed": 0, "reverted": 3, "refused": 4} +STOP = threading.Event() # the kill switch; set by SIGTERM/SIGINT, read between steps + + +def install_kill_switch() -> None: + def stop(_signum, _frame): + STOP.set() # never raises into the step in flight: the runner stops at the next boundary and reverts + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) def approval_carrier(cfg: dict, logins: list): @@ -125,7 +140,7 @@ def build(cfg: dict, logins: list): operations=operations, checks=checks, trust=lambda state: state in trusted, audit=C.AuditCore(cfg["audit_core"]["url"], cfg["audit_core"]["token_file"], cfg["audit_core"].get("ca_file")), - alert=alert) + alert=alert, should_stop=STOP.is_set) config = X.Config(executor_identity=cfg["executor_identity"], runtime_principal=cfg["runtime_principal"], tenant=cfg["tenant"], environment=cfg["environment"], approver_subjects=dict(cfg["approver_subjects"]), @@ -316,6 +331,7 @@ def main(argv=None) -> int: result, code = run_journey_only(cfg) _emit(result, args.result) return code + install_kill_switch() admission = X.bea.load(pathlib.Path(args.admission)) logins: list = [] try: diff --git a/bounded-executor/executor.py b/bounded-executor/executor.py index 6e7e6e5..d320ff2 100644 --- a/bounded-executor/executor.py +++ b/bounded-executor/executor.py @@ -354,6 +354,8 @@ class Run: self.preflight() except Refused as refusal: return {"outcome": "refused", "reason": str(refusal), "run_id": None, "effects": False} + if self.d.should_stop(): # the founder's stop before the one irreversible act: nothing is spent + return {"outcome": "refused", "reason": "stopped_before_consume", "run_id": None, "effects": False} self.start_wall, self.start_mono = self.d.wall(), self.d.monotonic() try: self.consume() diff --git a/bounded-executor/tests/test_approval_request.py b/bounded-executor/tests/test_approval_request.py index c688791..22778a6 100644 --- a/bounded-executor/tests/test_approval_request.py +++ b/bounded-executor/tests/test_approval_request.py @@ -336,5 +336,20 @@ class RequestPhase(Base): self.assertNotIn(SECRET, text) + + +class KillSwitch(unittest.TestCase): + def test_sigterm_sets_the_stop_flag_the_runner_reads(self): + import os + import signal + before = {sig: signal.getsignal(sig) for sig in (signal.SIGTERM, signal.SIGINT)} + self.addCleanup(lambda: [signal.signal(sig, h) for sig, h in before.items()]) + self.addCleanup(CLI.STOP.clear) + CLI.STOP.clear() + CLI.install_kill_switch() + os.kill(os.getpid(), signal.SIGTERM) + self.assertTrue(CLI.STOP.wait(2)) + + if __name__ == "__main__": unittest.main() diff --git a/bounded-executor/tests/test_executor.py b/bounded-executor/tests/test_executor.py index d74c83a..7cb5fe2 100644 --- a/bounded-executor/tests/test_executor.py +++ b/bounded-executor/tests/test_executor.py @@ -402,6 +402,12 @@ class FailureTests(Base): self.assertIn("restore-kv", self.ops()) self.assertNotIn("deliver", self.ops()) + def test_stop_before_consume_spends_nothing(self): + self.w.stop_after = 0 + result = self.run_it() + self.assertEqual((result["outcome"], result["reason"]), ("refused", "stopped_before_consume")) + self.assertEqual((self.w.carrier.consumes, self.w.events, self.ops()), (0, [], [])) + def test_settle_time_is_waited_and_counted(self): self.doc = synthetic_admission(self.root, lambda s: s["checks"][3].update(settle_seconds=125)) self.w = World(self.doc, self.clock) diff --git a/canon/schemas/posture-feedback-report_v0.1.schema.json b/canon/schemas/posture-feedback-report_v0.1.schema.json index 7cb7078..aadc8dd 100644 --- a/canon/schemas/posture-feedback-report_v0.1.schema.json +++ b/canon/schemas/posture-feedback-report_v0.1.schema.json @@ -112,7 +112,12 @@ "evidence-expired", "evidence-due-soon", "implemented-not-evidenced", - "declared-gap" + "declared-gap", + "execution-receipt-missing", + "execution-not-completed", + "execution-reverted", + "execution-budget-pressure", + "execution-recurring-revert" ] }, "severity": { diff --git a/canon/standards/posture-feedback_v0.1.md b/canon/standards/posture-feedback_v0.1.md index 0b72654..1a7e465 100644 --- a/canon/standards/posture-feedback_v0.1.md +++ b/canon/standards/posture-feedback_v0.1.md @@ -48,6 +48,8 @@ The evaluator routes only from authoritative declaration fields: - posture review, gaps, and implemented controls: `responsible_repo`; - evidence replacement: `evidence_freshness..responsible_repo`; - security-zone review: `zones.responsible_party`. +- execution findings (§4a): `responsible_repo` for the receipt's `executing_workload` in + `tools/posture-feedback/execution-owners.yaml`. If the relevant field is absent, owner resolution is `unknown`. The evaluator must not infer ownership from the service name, repository path, Git remote, @@ -75,6 +77,26 @@ Current adversarial levels are `E2`, `R4`, and `V2`–`V4`. This vocabulary come from Tenancy Posture §13. Mechanical evidence is evaluated for expiry only when its declaration explicitly supplies `valid_until`. +## 4a. Execution findings (added 2026-11-04, NK-WP-0056-T07) + +With `--receipts`, the evaluator also reads bounded-executor events: `execution.run.started` and +`execution.attribution.receipt`. Each event must pass the Execution Attribution Receipt v0.1 validator +(`tools/execution-attribution-receipt/`). An invalid event, or a started/receipt pair that does not match (EAR-10), +is a validation error, never a finding. Runs are joined by `run_id`. A finding's control is +`admission:#run:`. + +| Finding | Trigger | Severity | +| --- | --- | --- | +| `execution-receipt-missing` | a started event with no receipt among the inputs | high | +| `execution-not-completed` | outcome `unknown`, `revert_failed`, `stopped` or `budget` | high | +| `execution-reverted` | outcome `reverted` | medium | +| `execution-budget-pressure` | wall time used is at least 80 % of the admitted limit, or more than one attempt | low | +| `execution-recurring-revert` | two or more reverted admissions for the same executor and consumer set | high | + +`execution-recurring-revert` is the aggregate review of recurring admissions. Its control is +`consumers:` and its reason lists the admissions. The evaluator reads no wall clock here +either. Receipts carry their own times, and the report's `as_of` stays explicit. + ## 5. Proposal and safety boundary Every finding receives a stable id derived from its source declaration, diff --git a/tools/posture-feedback/execution-owners.yaml b/tools/posture-feedback/execution-owners.yaml new file mode 100644 index 0000000..add3c8c --- /dev/null +++ b/tools/posture-feedback/execution-owners.yaml @@ -0,0 +1,6 @@ +# Authoritative owner routing for execution findings (posture-feedback v0.1 section 3, NK-WP-0056-T07). +# executing_workload -> the repository responsible for the executor's behavior. The evaluator never infers +# an owner; a workload not listed here is routed to `unknown`. +executors: + - executing_workload: service:nk-bind-executor + responsible_repo: net-kingdom # founder decision 6d1adce9: NetKingdom writes the executor diff --git a/tools/posture-feedback/posture_feedback.py b/tools/posture-feedback/posture_feedback.py index f7bdc89..ef78f1a 100644 --- a/tools/posture-feedback/posture_feedback.py +++ b/tools/posture-feedback/posture_feedback.py @@ -24,6 +24,10 @@ ROOT = pathlib.Path(__file__).resolve().parents[2] TENANCY_SCHEMA = ROOT / "canon/schemas/tenancy-posture_v0.1.schema.json" REPORT_SCHEMA = ROOT / "canon/schemas/posture-feedback-report_v0.1.schema.json" TENANCY_VALIDATOR_PATH = ROOT / "tools/tenancy-posture/validate.py" +RECEIPT_VALIDATOR_PATH = ROOT / "tools/execution-attribution-receipt/validate.py" +EXECUTION_OWNERS = ROOT / "tools/posture-feedback/execution-owners.yaml" +NOT_COMPLETED = {"unknown", "revert_failed", "stopped", "budget"} +BUDGET_PRESSURE = 0.8 # share of the admitted wall time API_VERSION = "netkingdom.io/posture-feedback/v0.1" KIND = "PostureFeedbackReport" ADVERSARIAL_LEVELS = {"E2", "R4", "V2", "V3", "V4"} @@ -35,6 +39,10 @@ SPEC = importlib.util.spec_from_file_location( assert SPEC and SPEC.loader TENANCY_VALIDATOR = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(TENANCY_VALIDATOR) +_RSPEC = importlib.util.spec_from_file_location("receipt_validator_for_feedback", RECEIPT_VALIDATOR_PATH) +assert _RSPEC and _RSPEC.loader +RECEIPT_VALIDATOR = importlib.util.module_from_spec(_RSPEC) +_RSPEC.loader.exec_module(RECEIPT_VALIDATOR) def parse_timestamp(value: str) -> dt.datetime: @@ -321,6 +329,95 @@ def evaluate_entry( return results +def execution_owners(path: pathlib.Path = EXECUTION_OWNERS) -> dict[str, str]: + """executing_workload -> responsible_repo, from the authoritative declaration (§3). Nothing is inferred.""" + document = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + return {e["executing_workload"]: e["responsible_repo"] for e in document.get("executors", [])} + + +def execution_findings(paths: list[pathlib.Path], owners: dict[str, str]) -> tuple[list[dict], list[str]]: + """Proposal-only findings from bounded-executor events (§4a). Each file holds one event or a list of events. + Invalid events are validation errors, never findings. Runs are joined by run_id.""" + errors: list[str] = [] + runs: dict[str, dict[str, Any]] = {} + invalid: set[str] = set() # a run with an invalid event is reported once, as a validation error + for path in paths: + source = source_identifier(path) + try: + loaded = json.loads(path.read_text(encoding="utf-8")) + except Exception as exc: + errors.append(f"{path}: {exc}") + continue + for event in loaded if isinstance(loaded, list) else [loaded]: + report = RECEIPT_VALIDATOR.evaluate(event) + if not report["valid"]: + errors.extend(f"{path}: {f['rule']} at {f['at']}" for f in report["findings"]) + if isinstance(event, dict) and isinstance(event.get("correlation_id"), str): + invalid.add(event["correlation_id"]) + continue + runs.setdefault(event["correlation_id"], {"source": source})[report["kind"]] = event + + found: list[dict[str, Any]] = [] + reverted_by_scope: dict[tuple[str, str], list[str]] = {} + for run_id, run in sorted(runs.items()): + if run_id in invalid: + continue + if "started" in run and "receipt" in run: + pair = RECEIPT_VALIDATOR.evaluate_pair(run["started"], run["receipt"]) + if not pair["valid"]: + errors.extend(f"run {run_id}: {f['rule']} at {f['at']}" for f in pair["findings"]) + continue + event = run.get("receipt") or run["started"] + data = event["data"] + service = data["executing_workload"] + owner = owners.get(service, "unknown") + subject = event["subject"] + control = f"{subject}#run:{run_id}" + + def add(finding_class: str, severity: str, reason: str, action: str, state: str = "not-applicable"): + found.append(finding(source=run["source"], service=service, finding_class=finding_class, + severity=severity, control=control, owner=owner, evidence_state=state, + reason=reason, recommended_action=action)) + + if "receipt" not in run: + add("execution-receipt-missing", "high", + f"Run {run_id} for {subject} started (approval consumed) but no receipt is among the inputs.", + "Check audit-core's opening_without_closing finding and the Job's state; establish what the run did " + "before any further admission for this scope.", "unknown") + continue + consumers = sorted({s["consumer"] for s in data["steps"]}) + scope = ",".join(consumers) or "none" + outcome = data["outcome"] + if outcome in NOT_COMPLETED: + add("execution-not-completed", "high", + f"Run {run_id} for {subject} ended {outcome}; the target state is not established by the executor.", + "Inspect the receipt's steps and the target; restore a known state under a new admission or by hand.", + "unknown") + elif outcome == "reverted": + add("execution-reverted", "medium", + f"Run {run_id} for {subject} reverted after a failed check or definite failure.", + "Review the failing step and check before re-admitting the same plan.") + reverted_by_scope.setdefault((service, scope), []).append(event["subject"]) + budget = data["budget"] + if budget["used"]["wall_time_seconds"] >= BUDGET_PRESSURE * budget["limit"]["wall_time_seconds"] \ + or data["attempts"] > 1: + add("execution-budget-pressure", "low", + f"Run {run_id} for {subject} used {budget['used']['wall_time_seconds']:g}s of " + f"{budget['limit']['wall_time_seconds']}s and {data['attempts']} of {budget['limit']['attempts']} " + "attempt(s).", + "Review whether the admitted budget fits the plan before the next admission of this kind.") + for (service, scope), controls in sorted(reverted_by_scope.items()): + if len(controls) >= 2: + found.append(finding( + source="execution-aggregate", service=service, finding_class="execution-recurring-revert", + severity="high", control=f"consumers:{scope}", owner=owners.get(service, "unknown"), + evidence_state="not-applicable", + reason=f"{len(controls)} admissions for consumers {scope} reverted: {', '.join(sorted(controls))}.", + recommended_action="Review the recurring admission in aggregate: the plan, its checks or the target " + "drifted; stop admitting it until reviewed.")) + return found, errors + + def report_digest(report_without_digest: dict[str, Any]) -> str: encoded = json.dumps( report_without_digest, @@ -336,6 +433,8 @@ def build_report( as_of: dt.datetime, horizon_days: int, fail_on: str, + receipts: list[pathlib.Path] | None = None, + owners: dict[str, str] | None = None, ) -> dict[str, Any]: tenancy_schema = json.loads(TENANCY_SCHEMA.read_text(encoding="utf-8")) validation_errors: list[str] = [] @@ -354,6 +453,10 @@ def build_report( continue for entry in service_entries(document): findings.extend(evaluate_entry(source, entry, as_of, horizon_days)) + if receipts: + found, errors = execution_findings(receipts, execution_owners() if owners is None else owners) + findings.extend(found) + validation_errors.extend(errors) findings.sort( key=lambda item: ( @@ -408,7 +511,9 @@ def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Emit deterministic, proposal-only NetKingdom posture feedback." ) - parser.add_argument("declarations", nargs="+", type=pathlib.Path) + parser.add_argument("declarations", nargs="*", type=pathlib.Path) + parser.add_argument("--receipts", nargs="+", type=pathlib.Path, default=[], + help="bounded-executor started/receipt events (JSON), section 4a") parser.add_argument("--as-of", required=True, help="RFC 3339 timestamp with explicit offset") parser.add_argument("--horizon-days", type=int, default=30) parser.add_argument("--fail-on", choices=("none", "low", "medium", "high"), default="high") @@ -424,11 +529,14 @@ def main(argv: list[str] | None = None) -> int: except ValueError as exc: raise SystemExit(f"invalid --as-of: {exc}") from exc + if not args.declarations and not args.receipts: + raise SystemExit("give declarations, --receipts, or both") report = build_report( args.declarations, as_of=as_of, horizon_days=args.horizon_days, fail_on=args.fail_on, + receipts=args.receipts, ) print(json.dumps(report, indent=2, sort_keys=True)) if report["validation_errors"]: diff --git a/tools/posture-feedback/tests/test_execution_feedback.py b/tools/posture-feedback/tests/test_execution_feedback.py new file mode 100644 index 0000000..135b9ff --- /dev/null +++ b/tools/posture-feedback/tests/test_execution_feedback.py @@ -0,0 +1,126 @@ +"""Execution findings from bounded-executor receipts (posture-feedback v0.1 section 4a, NK-WP-0056-T07).""" +from __future__ import annotations + +import copy +import datetime as dt +import importlib.util +import json +import pathlib +import sys +import tempfile +import unittest +import uuid + +HERE = pathlib.Path(__file__).resolve().parent +ROOT = HERE.parents[2] +FIXTURES = ROOT / "tools/execution-attribution-receipt/fixtures/valid" +spec = importlib.util.spec_from_file_location("posture_feedback_exec", HERE.parent / "posture_feedback.py") +PF = importlib.util.module_from_spec(spec) +sys.modules["posture_feedback_exec"] = PF +spec.loader.exec_module(PF) +AS_OF = dt.datetime(2026, 11, 4, 12, 0, tzinfo=dt.UTC) + + +def events(name): + return [json.loads((FIXTURES / f"{name}-{k}.json").read_text()) for k in ("started", "receipt")] + + +def renamed(evs, admission): + """The same run shape as another run of another admission (fresh run id).""" + rid = str(uuid.uuid4()) + out = [] + for e in copy.deepcopy(evs): + kind = "started" if e["type"] == "execution.run.started" else "receipt" + e.update(id=f"run:{rid}:{kind}", correlation_id=rid, subject=f"admission:{admission}") + e["data"]["run_id"] = rid + if kind == "started": + e["data"]["admission"] = admission + else: + e["data"]["target"] = f"admission:{admission}" + e["data"]["decision_refs"][0]["ref"] = admission + out.append(e) + return out + + +class ExecutionFeedback(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.dir = pathlib.Path(tmp.name) + + def report(self, *runs, owners=None, fail_on="none"): + paths = [] + for i, evs in enumerate(runs): + path = self.dir / f"run-{i}.json" + path.write_text(json.dumps(evs)) + paths.append(path) + return PF.build_report([], as_of=AS_OF, horizon_days=14, fail_on=fail_on, receipts=paths, owners=owners) + + def classes(self, report): + return sorted((f["class"], f["severity"]) for f in report["findings"]) + + def test_a_clean_run_raises_nothing(self): + report = self.report(events("completed")) + self.assertEqual((report["findings"], report["validation_errors"], report["ok"]), ([], [], True)) + + def test_reverted_and_retry_findings_are_routed_to_the_declared_owner(self): + report = self.report(events("reverted"), events("completed-on-retry")) + self.assertEqual(self.classes(report), [("execution-budget-pressure", "low"), ("execution-reverted", "medium")]) + self.assertEqual({f["owner"] for f in report["findings"]}, {"net-kingdom"}) + self.assertEqual({f["service"] for f in report["findings"]}, {"service:nk-bind-executor"}) + self.assertEqual(report["automation"]["mode"], "proposal-only") + + def test_an_unmapped_executor_is_unknown_never_inferred(self): + report = self.report(events("reverted"), owners={}) + self.assertEqual({f["owner"] for f in report["findings"]}, {"unknown"}) + self.assertEqual(report["summary"]["unknown_owner"], 1) + + def test_not_completed_is_high(self): + evs = events("reverted") + evs[1]["data"].update(outcome="unknown", reverted=True) + report = self.report(evs, fail_on="high") + self.assertIn(("execution-not-completed", "high"), self.classes(report)) + self.assertFalse(report["ok"]) + + def test_a_start_without_receipt_is_high(self): + report = self.report(events("completed")[:1]) + self.assertEqual(self.classes(report), [("execution-receipt-missing", "high")]) + + def test_recurring_reverts_are_reviewed_in_aggregate(self): + base = events("reverted") + report = self.report(base, renamed(base, "BEA-2026-0002")) + aggregate = [f for f in report["findings"] if f["class"] == "execution-recurring-revert"] + self.assertEqual(len(aggregate), 1) + self.assertEqual((aggregate[0]["severity"], aggregate[0]["control"]), ("high", "consumers:authelia")) + self.assertIn("BEA-2026-0001", aggregate[0]["reason"]) + self.assertIn("BEA-2026-0002", aggregate[0]["reason"]) + + def test_invalid_events_are_validation_errors_not_findings(self): + evs = events("completed") + evs[1]["data"]["bind_password"] = "synthetic-value" + report = self.report(evs) + self.assertTrue(report["validation_errors"]) + self.assertEqual(report["findings"], []) + self.assertNotIn("synthetic-value", json.dumps(report)) + + def test_a_mismatched_pair_is_a_validation_error(self): + started, receipt = events("completed") + started["data"]["approval"] = "another-approval" + report = self.report([started, receipt]) + self.assertTrue(any("EAR-10" in e for e in report["validation_errors"])) + + def test_deterministic(self): + first = self.report(events("reverted"), events("completed-on-retry")) + second = self.report(events("reverted"), events("completed-on-retry")) + self.assertEqual(first["report_digest"], second["report_digest"]) + + def test_cli_receipts_only(self): + path = self.dir / "r.json" + path.write_text(json.dumps(events("reverted"))) + self.assertEqual(PF.main(["--as-of", "2026-11-04T12:00:00Z", "--fail-on", "medium", "--receipts", str(path)]), 1) + with self.assertRaises(SystemExit): + PF.main(["--as-of", "2026-11-04T12:00:00Z"]) + + +if __name__ == "__main__": + unittest.main()