from __future__ import annotations import argparse import json import sys from dataclasses import asdict from pathlib import Path from .audit_fixtures import AuditFixture, audit_probe_suite from .capacity import capacity_calibration from .differential import execute from .e3 import CADENCE, PROBES, e3_calibration from .engagement import AuthorizationError, Engagement from .fixtures import FixtureService, probe_suite from .model import RunReport, utc_now from .plane import KillSwitch, admit, default_broker, retired_ids from .platform_custody import broker_from_receipt, finalize_run_report from .reporting import conformance_message, queue_risk_nexus, risk_nexus_message from .targets import load_catalog, load_registration def fixture_calibration() -> dict: started = utc_now() good_generic = probe_suite(FixtureService(enforce_tenant=True)) bad_generic = probe_suite(FixtureService(enforce_tenant=False)) good_audit = audit_probe_suite(AuditFixture(enforce_tenant=True)) bad_audit = audit_probe_suite(AuditFixture(enforce_tenant=False)) good = [execute(probe) for probe in (*good_generic, *good_audit)] bad = [execute(probe) for probe in (*bad_generic, *bad_audit)] detected = all(result.outcome == "finding" for result in bad) rejected = all(result.outcome == "pass" for result in good) return { "schema_version": "whitehat-calibration/v1", "evidence_class": "fixture", "run_id": f"fixture-calibration-{started}", "started_at": started, "ended_at": utc_now(), "outcome": "pass" if detected and rejected else "finding", "expected": {"known_good": "pass", "known_bad": "finding"}, "known_good": [asdict(result) for result in good], "known_bad": [asdict(result) for result in bad], "limitations": [ "Offline fixture evidence calibrates the harness; it is not target assurance.", "No network request, database connection, or live credential was used.", ], } def validate_pack(path: Path) -> None: data = json.loads(path.read_text(encoding="utf-8")) required = {"schema_version", "target", "posture_claim", "attacker_model", "probes"} missing = sorted(required - data.keys()) if missing: raise ValueError(f"{path}: missing {', '.join(missing)}") ids: set[str] = set() for probe in data["probes"]: for key in ("id", "operation", "route", "owner", "attacker", "absent"): if key not in probe: raise ValueError(f"{path}: probe missing {key}") if probe["id"] in ids: raise ValueError(f"{path}: duplicate probe id {probe['id']}") ids.add(probe["id"]) def main(argv: list[str] | None = None) -> None: parser = argparse.ArgumentParser(prog="whitehat") commands = parser.add_subparsers(dest="command", required=True) fixtures = commands.add_parser("fixtures", help="calibrate E2 probes offline") fixtures.add_argument("--output") engagement = commands.add_parser("validate-engagement") engagement.add_argument("path") packs = commands.add_parser("validate-packs") packs.add_argument("path") targets = commands.add_parser("validate-targets") targets.add_argument("path") admit_plane = commands.add_parser("admit-plane") admit_plane.add_argument("engagement") admit_plane.add_argument("registration") admit_plane.add_argument("--receipt", help="value-safe custody projection receipt") admit_plane.add_argument( "--contract", help="WP-0025 projection contract; required for railiance.custody-projection-receipt", ) admit_plane.add_argument( "--broker-receipt", help="WP-0025 broker-readiness receipt; required for railiance.custody-projection-receipt", ) finalize = commands.add_parser("finalize-report") finalize.add_argument("report") finalize.add_argument("--contract", required=True) finalize.add_argument("--receipt", required=True) finalize.add_argument("--cleanup-receipt", required=True) finalize.add_argument("--output", required=True) commands.add_parser("kill-switch") deliver = commands.add_parser("deliver") deliver.add_argument("report") deliver.add_argument("--outbox", default="outbox") commands.add_parser("e3-plan") e3_fix = commands.add_parser("e3-fixtures", help="calibrate E3 probes offline") e3_fix.add_argument("--output") capacity_fix = commands.add_parser( "capacity-fixture", help="calibrate P1/P2 evaluator offline" ) capacity_fix.add_argument("--output") message = commands.add_parser("risk-message") message.add_argument("report") conformance = commands.add_parser( "conformance-message", help="render a Gate House conformance envelope" ) conformance.add_argument("report") conformance.add_argument("--spec", required=True) conformance.add_argument("--test-id", required=True) conformance.add_argument("--component") conformance.add_argument("--invariants", default="") args = parser.parse_args(argv) if args.command == "fixtures": result = fixture_calibration() rendered = json.dumps(result, indent=2, sort_keys=True) + "\n" if args.output: Path(args.output).write_text(rendered, encoding="utf-8") else: print(rendered, end="") raise SystemExit(0 if result["outcome"] == "pass" else 1) if args.command == "validate-engagement": try: record = Engagement.load(args.path) except (AuthorizationError, OSError, ValueError, json.JSONDecodeError) as error: print(f"not authorized: {error}", file=sys.stderr) raise SystemExit(2) from None print(f"authorized: {record.raw['engagement_id']}") return if args.command == "validate-packs": paths = sorted(Path(args.path).glob("*.json")) if not paths: raise SystemExit("no probe packs found") for path in paths: validate_pack(path) print(f"validated {len(paths)} probe packs") return if args.command == "validate-targets": try: catalog = load_catalog(args.path) except (AuthorizationError, OSError, ValueError, json.JSONDecodeError) as error: print(f"not authorized: {error}", file=sys.stderr) raise SystemExit(2) from None print(f"validated {len(catalog)} target registrations") return if args.command == "admit-plane": try: record = Engagement.load(args.engagement) registration = load_registration(args.registration) broker = ( broker_from_receipt( args.receipt, contract_path=args.contract, broker_path=args.broker_receipt, ) if args.receipt else default_broker(record) ) lease = admit(engagement=record, registration=registration, broker=broker, kill_switch=KillSwitch(), retired=retired_ids()) except (AuthorizationError, OSError, ValueError, json.JSONDecodeError) as error: print(f"not authorized: {error}", file=sys.stderr) raise SystemExit(2) from None print(f"admitted: {lease.engagement.raw['engagement_id']} lease={lease.lease_id}") return if args.command == "kill-switch": switch = KillSwitch() if switch.engaged(): print(f"engaged: {switch.path}") raise SystemExit(1) print("clear") return if args.command == "finalize-report": try: report = json.loads(Path(args.report).read_text(encoding="utf-8")) contract = json.loads(Path(args.contract).read_text(encoding="utf-8")) projection = json.loads(Path(args.receipt).read_text(encoding="utf-8")) cleanup = json.loads(Path(args.cleanup_receipt).read_text(encoding="utf-8")) finalized = finalize_run_report( report, projection=projection, cleanup=cleanup, contract=contract ) RunReport(**finalized) output = Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) output.write_text( json.dumps(finalized, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) except (AuthorizationError, OSError, ValueError, json.JSONDecodeError, TypeError) as error: print(f"not authorized: {error}", file=sys.stderr) raise SystemExit(2) from None print(f"finalized: {output}") return if args.command == "deliver": try: report = RunReport(**json.loads(Path(args.report).read_text(encoding="utf-8"))) path = queue_risk_nexus(report, args.outbox) except (AuthorizationError, OSError, ValueError, json.JSONDecodeError, TypeError) as error: print(f"not authorized: {error}", file=sys.stderr) raise SystemExit(2) from None print(f"queued: {path}") return if args.command == "e3-plan": print(json.dumps({"cadence": CADENCE, "probes": [asdict(probe) for probe in PROBES]}, indent=2, sort_keys=True)) return if args.command == "e3-fixtures": result = e3_calibration() rendered = json.dumps(result, indent=2, sort_keys=True) + "\n" if args.output: Path(args.output).write_text(rendered, encoding="utf-8") else: print(rendered, end="") raise SystemExit(0 if result["outcome"] == "pass" else 1) if args.command == "capacity-fixture": result = capacity_calibration() rendered = json.dumps(result, indent=2, sort_keys=True) + "\n" if args.output: Path(args.output).write_text(rendered, encoding="utf-8") else: print(rendered, end="") raise SystemExit(0 if result["outcome"] == "pass" else 1) if args.command == "risk-message": report = RunReport(**json.loads(Path(args.report).read_text(encoding="utf-8"))) print(risk_nexus_message(report), end="") return if args.command == "conformance-message": report = RunReport(**json.loads(Path(args.report).read_text(encoding="utf-8"))) invariants = [item for item in args.invariants.split(",") if item] print(conformance_message( report, specification=args.spec, test_id=args.test_id, component=args.component, invariant_ids=invariants or None, ), end="") return raise SystemExit(2) if __name__ == "__main__": main(sys.argv[1:])