tegwick
eebcc39a12
WHITEHAT-WP-0012 records that identity and authorization evidence belongs in the declaring service's CI, and adds an in-process V2/V3/V4 observation judge that injects no failure. Live E4, R4, and V stay wait on an unnamed owner or operator, so the plan is blocked. No packet is authorized. Assistant: grok Assistant-Session: 01a10450-dda4-7253-91a9-b8dccb2f0173
79 lines
3.8 KiB
Python
79 lines
3.8 KiB
Python
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 classes & {"fixture-e4", "fixture-r4", "fixture-v"} and count != 0:
|
|
raise AuthorizationError(f"{path}: E4, R4, and V fixtures project no credential")
|
|
if (classes & {"fixture-e4", "fixture-r4", "fixture-v"}) and (
|
|
classes & {"live-e2", "e3", "capacity", "asm"}
|
|
):
|
|
raise AuthorizationError(
|
|
f"{path}: E4, R4, and V fixtures cannot carry a live approval class"
|
|
)
|
|
if classes & {"fixture-asm", "asm"} and count not in {0, 1, 2}:
|
|
raise AuthorizationError(f"{path}: ASM registration projects at most two identities")
|
|
if classes & {"fixture-asm", "asm"} and not data.get("test_id"):
|
|
raise AuthorizationError(f"{path}: ASM registration must preserve a Canon test_id")
|
|
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
|