whitehat-security/src/whitehat_security/cli.py

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import asdict
from pathlib import Path

from .capacity import CapacitySample, characterize
from .differential import execute
from .e3 import CADENCE, PROBES
from .engagement import AuthorizationError, Engagement
from .fixtures import FixtureService, probe_suite
from .model import RunReport, utc_now
from .reporting import risk_nexus_message


def fixture_calibration() -> dict:
    started = utc_now()
    good = [execute(probe) for probe in probe_suite(FixtureService(enforce_tenant=True))]
    bad = [execute(probe) for probe in probe_suite(FixtureService(enforce_tenant=False))]
    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")
    commands.add_parser("e3-plan")
    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 == "e3-plan":
        print(json.dumps({"cadence": CADENCE, "probes": [asdict(probe) for probe in PROBES]},
                         indent=2, sort_keys=True))
        return
    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:])