#!/usr/bin/env python3 # /// script # dependencies = ["jsonschema>=4.23,<5", "PyYAML>=6,<7"] # /// """Emit deterministic, proposal-only feedback for tenancy posture declarations.""" from __future__ import annotations import argparse import datetime as dt import hashlib import importlib.util import json import pathlib import sys from collections import Counter from typing import Any import jsonschema import yaml 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"} SEVERITY_RANK = {"low": 1, "medium": 2, "high": 3} SPEC = importlib.util.spec_from_file_location( "tenancy_posture_validator_for_feedback", TENANCY_VALIDATOR_PATH ) 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: parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None or parsed.utcoffset() is None: raise ValueError("timestamp must include an explicit UTC offset") return parsed.astimezone(dt.timezone.utc) def format_timestamp(value: dt.datetime) -> str: return value.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") def load_document(path: pathlib.Path) -> dict[str, Any]: document = yaml.safe_load(path.read_text(encoding="utf-8")) if not isinstance(document, dict): raise ValueError("document must be a YAML object") return document def service_entries(document: dict[str, Any]) -> list[dict[str, Any]]: if "services" in document: return document["services"] return [document] def source_identifier(path: pathlib.Path) -> str: """Return a portable declaration id without treating the path as ownership.""" resolved = path.resolve() try: return resolved.relative_to(ROOT.parent).as_posix() except ValueError: return resolved.as_posix() def finding_id( source: str, service: str, finding_class: str, control: str, due: str | None, ) -> str: identity = "\x1f".join((source, service, finding_class, control, due or "")) digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:16] return f"NKFB-{digest}" def finding( *, source: str, service: str, finding_class: str, severity: str, control: str, owner: str, evidence_state: str, reason: str, recommended_action: str, due: str | None = None, ) -> dict[str, Any]: result = { "id": finding_id(source, service, finding_class, control, due), "class": finding_class, "severity": severity, "source": source, "service": service, "control": control, "owner": owner, "evidence_state": evidence_state, "reason": reason, "recommended_action": recommended_action, } if due is not None: result["due"] = due return result def review_finding( *, source: str, service: str, owner: str, control: str, due: dt.date, as_of: dt.datetime, horizon_days: int, prefix: str, ) -> dict[str, Any] | None: due_text = due.isoformat() if as_of.date() > due: return finding( source=source, service=service, finding_class=f"{prefix}-review-overdue", severity="high", control=control, owner=owner, evidence_state="overdue", due=due_text, reason=f"{control} review was due on {due_text} and is overdue at {format_timestamp(as_of)}.", recommended_action=f"Review {control}, update its declaration and evidence, and retain the review record.", ) days = (due - as_of.date()).days if days <= horizon_days: return finding( source=source, service=service, finding_class=f"{prefix}-review-due-soon", severity="medium", control=control, owner=owner, evidence_state="due-soon", due=due_text, reason=f"{control} review is due in {days} day(s), within the {horizon_days}-day horizon.", recommended_action=f"Schedule and evidence the {control} review before {due_text}.", ) return None def current_evidence_keys(entry: dict[str, Any]) -> set[str]: current = entry["tenancy"]["current"] return { f"{axis}{level}" for axis, level in current.items() if isinstance(level, int) } def evidence_findings( *, source: str, entry: dict[str, Any], as_of: dt.datetime, horizon_days: int, ) -> list[dict[str, Any]]: service = entry["service"] freshness = entry.get("evidence_freshness", {}) current_keys = current_evidence_keys(entry) results: list[dict[str, Any]] = [] for key in sorted(ADVERSARIAL_LEVELS & current_keys): if key not in freshness: results.append( finding( source=source, service=service, finding_class="evidence-freshness-unknown", severity="high", control=key, owner="unknown", evidence_state="unknown", reason=f"Current adversarial claim {key} has no authoritative evidence_freshness entry.", recommended_action=( f"Declare the {key} observation, expiry, bounded scope, responsible repository, " "and replacement action; do not infer freshness from prose." ), ) ) for key in sorted(current_keys & set(freshness)): metadata = freshness[key] valid_until_raw = metadata.get("valid_until") if not valid_until_raw: continue valid_until = parse_timestamp(valid_until_raw) owner = metadata["responsible_repo"] if as_of > valid_until: results.append( finding( source=source, service=service, finding_class="evidence-expired", severity="high", control=key, owner=owner, evidence_state="expired", due=format_timestamp(valid_until), reason=f"Evidence for {key} expired before {format_timestamp(as_of)}.", recommended_action=metadata["remediation"], ) ) continue remaining = valid_until - as_of if remaining <= dt.timedelta(days=horizon_days): results.append( finding( source=source, service=service, finding_class="evidence-due-soon", severity="medium", control=key, owner=owner, evidence_state="due-soon", due=format_timestamp(valid_until), reason=( f"Evidence for {key} expires within the {horizon_days}-day horizon " f"at {format_timestamp(valid_until)}." ), recommended_action=metadata["remediation"], ) ) return results def evaluate_entry( source: str, entry: dict[str, Any], as_of: dt.datetime, horizon_days: int, ) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] service = entry["service"] posture = entry["tenancy"] posture_owner = entry.get("responsible_repo", "unknown") posture_review = review_finding( source=source, service=service, owner=posture_owner, control="tenancy-posture", due=dt.date.fromisoformat(posture["review_due"]), as_of=as_of, horizon_days=horizon_days, prefix="posture", ) if posture_review: results.append(posture_review) zones = entry.get("zones") if zones: zone_review = review_finding( source=source, service=service, owner=zones["responsible_party"], control="security-zones", due=dt.date.fromisoformat(zones["review_due"]), as_of=as_of, horizon_days=horizon_days, prefix="zone", ) if zone_review: results.append(zone_review) results.extend( evidence_findings( source=source, entry=entry, as_of=as_of, horizon_days=horizon_days, ) ) current = posture["current"] for axis, level in sorted(posture.get("implemented", {}).items()): current_level = current[axis] results.append( finding( source=source, service=service, finding_class="implemented-not-evidenced", severity="medium", control=f"{axis}{level}", owner=posture_owner, evidence_state="implemented", reason=f"{axis}{level} is implemented while the current evidenced level is {axis}{current_level}.", recommended_action=f"Produce the canon-required evidence for {axis}{level} or remove the implemented claim.", ) ) for axis, description in sorted(posture.get("gap", {}).items()): results.append( finding( source=source, service=service, finding_class="declared-gap", severity="low", control=axis, owner=posture_owner, evidence_state="gap", reason=description, recommended_action=f"Review the declared {axis}-axis gap and retain, schedule, or resolve it explicitly.", ) ) 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, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return f"sha256:{hashlib.sha256(encoded).hexdigest()}" def build_report( paths: list[pathlib.Path], *, 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] = [] findings: list[dict[str, Any]] = [] for path in paths: source = source_identifier(path) try: document = load_document(path) except Exception as exc: validation_errors.append(f"{path}: {exc}") continue errors = TENANCY_VALIDATOR.validate(path, tenancy_schema) if errors: validation_errors.extend(errors) 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: ( item["source"], item["service"], item["class"], item["control"], item.get("due", ""), ) ) counts = Counter(item["severity"] for item in findings) threshold_failed = False if fail_on != "none": threshold = SEVERITY_RANK[fail_on] threshold_failed = any( SEVERITY_RANK[item["severity"]] >= threshold for item in findings ) report: dict[str, Any] = { "apiVersion": API_VERSION, "kind": KIND, "ok": not validation_errors and not threshold_failed, "as_of": format_timestamp(as_of), "horizon_days": horizon_days, "fail_on": fail_on, "automation": { "mode": "proposal-only", "external_write_permitted": False, "policy_mutation_permitted": False, "declaration_mutation_permitted": False, }, "summary": { "total": len(findings), "high": counts["high"], "medium": counts["medium"], "low": counts["low"], "unknown_owner": sum(item["owner"] == "unknown" for item in findings), }, "validation_errors": sorted(validation_errors), "findings": findings, } report["report_digest"] = report_digest(report) report_schema = json.loads(REPORT_SCHEMA.read_text(encoding="utf-8")) jsonschema.Draft202012Validator( report_schema, format_checker=jsonschema.FormatChecker(), ).validate(report) return report 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("--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") return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) if args.horizon_days < 0: raise SystemExit("--horizon-days must be non-negative") try: as_of = parse_timestamp(args.as_of) 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"]: return 2 return 0 if report["ok"] else 1 if __name__ == "__main__": raise SystemExit(main())