tegwick
5384f051d2
Extend whitehat-target/v1 with fixture-asm/asm instead of mapping onto E2, E3, or capacity. Register all ten Canon tests as pending with named blockers, known-bad designs, and result routes. Add a value-safe conformance-message renderer. Authorizes no probe. Assistant: grok Assistant-Session: 01a05e32-c776-72a3-86ec-c490e027aca9
242 lines
10 KiB
Python
242 lines
10 KiB
Python
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:])
|