Add governed test plane and close T04/T08
Encode fail-closed admission, target registrations, and a credential broker that never returns secret values. Calibrate audit-core shaped probes in-process. Send no packets and request no live credentials. Assistant: grok Assistant-Session: 01a02670-3345-76f2-a014-70fde8e2a2bb
This commit is contained in:
parent
0aab0cb4c6
commit
95129d7a35
35 changed files with 1599 additions and 103 deletions
61
src/whitehat_security/targets.py
Normal file
61
src/whitehat_security/targets.py
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
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"]
|
||||
if identities.get("count") != 2:
|
||||
raise AuthorizationError(f"{path}: E2 registration must project two identities")
|
||||
if 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
|
||||
Loading…
Add table
Add a link
Reference in a new issue