whitehat-security/src/whitehat_security/targets.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from .engagement import AuthorizationError

REQUIRED = {
    "schema_version", "target_id", "posture_claim", "attacker_model",
    "applicability", "applicability_reason", "approval_classes", "routes",
    "identities", "abort_telemetry",
}
APPLICABLE_REQUIRED = {
    "adapter", "probe_pack", "known_bad_calibration", "fixture_lifecycle", "egress",
}
APPLICABILITY = {"applicable", "not_applicable", "pending"}


def load_registration(path: str | Path) -> dict[str, Any]:
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    missing = sorted(REQUIRED - data.keys())
    if missing:
        raise AuthorizationError(f"{path}: missing {', '.join(missing)}")
    if data["schema_version"] != "whitehat-target/v1":
        raise AuthorizationError(f"{path}: unsupported schema_version")
    if data["applicability"] not in APPLICABILITY:
        raise AuthorizationError(f"{path}: invalid applicability")
    if not data["applicability_reason"]:
        raise AuthorizationError(f"{path}: applicability_reason is required")
    if not data["approval_classes"]:
        raise AuthorizationError(f"{path}: at least one approval class is required")
    if data["applicability"] == "applicable":
        missing_live = sorted(APPLICABLE_REQUIRED - data.keys())
        if missing_live:
            raise AuthorizationError(f"{path}: applicable target missing {', '.join(missing_live)}")
        if not data["routes"]:
            raise AuthorizationError(f"{path}: applicable target must register routes")
        identities = data["identities"]
        count = identities.get("count")
        classes = set(data["approval_classes"])
        if classes & {"fixture-e2", "live-e2"} and count != 2:
            raise AuthorizationError(f"{path}: E2 registration must project two identities")
        if classes & {"fixture-e3", "e3"} and count != 1:
            raise AuthorizationError(f"{path}: E3 registration must project one runtime identity")
        if classes & {"fixture-capacity", "capacity"} and count not in {0, 1}:
            raise AuthorizationError(f"{path}: capacity registration projects at most one aggressor identity")
        if count and identities.get("ttl_seconds", 0) > 900:
            raise AuthorizationError(f"{path}: identity TTL must be at most 900 seconds")
        if data.get("known_bad_calibration") in {None, "", "pending"}:
            raise AuthorizationError(f"{path}: applicable target needs completed known-bad calibration")
    return data


def load_catalog(directory: str | Path) -> dict[str, dict[str, Any]]:
    root = Path(directory)
    paths = sorted(root.glob("*.json"))
    if not paths:
        raise AuthorizationError(f"{root}: no target registrations found")
    catalog: dict[str, dict[str, Any]] = {}
    for path in paths:
        registration = load_registration(path)
        target_id = registration["target_id"]
        if target_id in catalog:
            raise AuthorizationError(f"duplicate target_id: {target_id}")
        catalog[target_id] = registration
    return catalog