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 CapacitySample, characterize 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 from .reporting import 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 capacity_fixture() -> dict: baseline = [ CapacitySample("aggressor", 10, 0, 100), CapacitySample("neighbour", 12, 0, 80), ] loaded = [ CapacitySample("aggressor", 25, 0.01, 120), CapacitySample("neighbour", 18, 0.02, 60), ] return asdict(characterize( baseline=baseline, loaded=loaded, governor_bound=True, aggressor_peak=10, aggressor_ceiling=10, )) 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 bound to the receipt") 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") commands.add_parser("capacity-fixture") message = commands.add_parser("risk-message") message.add_argument("report") 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) 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 == "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": print(json.dumps(capacity_fixture(), indent=2, sort_keys=True)) return if args.command == "risk-message": report = RunReport(**json.loads(Path(args.report).read_text(encoding="utf-8"))) print(risk_nexus_message(report), end="") return raise SystemExit(2) if __name__ == "__main__": main(sys.argv[1:])