tegwick
45548e44a2
WH-ENG-20260822-AUDIT-E2-02 projected and then aborted: admit-plane had no receipt adapter, so the runner sent zero packets. Consume custody receipts as handles only, keep unconnected admission fail-closed, and retire -02. Assistant: grok Assistant-Session: 01a02670-3345-76f2-a014-70fde8e2a2bb
99 lines
4.6 KiB
Python
99 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
class AuthorizationError(ValueError):
|
|
pass
|
|
|
|
|
|
REQUIRED = {
|
|
"engagement_id", "authorization_id", "authorizer", "approved_at", "expires_at",
|
|
"target", "target_owner", "environment", "source", "routes", "fixture_ids",
|
|
"credential_lane", "credential_role", "credential_max_ttl_seconds", "techniques",
|
|
"prohibited_techniques", "rate_limit_per_minute", "max_concurrency", "window_start",
|
|
"window_end", "operator_contact", "abort_contact", "posture_claim", "attacker_model",
|
|
"finding_destination", "target_owner_acknowledged_at",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Engagement:
|
|
raw: dict[str, Any]
|
|
|
|
@classmethod
|
|
def load(cls, path: str | Path, *, now: datetime | None = None) -> "Engagement":
|
|
data = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
missing = sorted(REQUIRED - data.keys())
|
|
if missing:
|
|
raise AuthorizationError(f"incomplete engagement; missing: {', '.join(missing)}")
|
|
engagement = cls(data)
|
|
engagement.validate(now=now)
|
|
return engagement
|
|
|
|
def validate(self, *, now: datetime | None = None) -> None:
|
|
if self.raw.get("status") == "cancelled":
|
|
raise AuthorizationError("engagement is cancelled")
|
|
if self.raw.get("status") == "expired":
|
|
raise AuthorizationError(
|
|
"engagement window elapsed with no live run; identifier must not be reused"
|
|
)
|
|
if self.raw.get("status") == "aborted":
|
|
raise AuthorizationError(
|
|
"engagement aborted without target evidence; identifier must not be reused"
|
|
)
|
|
if self.raw.get("status") == "proposed":
|
|
raise AuthorizationError(
|
|
"engagement is proposed; operator approval and owner acknowledgement are pending"
|
|
)
|
|
current = now or datetime.now(UTC)
|
|
start = _timestamp(self.raw["window_start"])
|
|
end = _timestamp(self.raw["window_end"])
|
|
expiry = _timestamp(self.raw["expires_at"])
|
|
approved = _timestamp(self.raw["approved_at"])
|
|
if not self.raw["target_owner_acknowledged_at"]:
|
|
raise AuthorizationError("target-owner acknowledgement is pending")
|
|
acknowledged = _timestamp(self.raw["target_owner_acknowledged_at"])
|
|
if not approved <= current <= min(end, expiry):
|
|
raise AuthorizationError("engagement is outside its approved time/expiry window")
|
|
if current < start:
|
|
raise AuthorizationError("engagement window has not started")
|
|
if acknowledged < approved:
|
|
raise AuthorizationError("target-owner acknowledgement predates approval")
|
|
if acknowledged > current:
|
|
raise AuthorizationError("target-owner acknowledgement is in the future")
|
|
if start < approved:
|
|
raise AuthorizationError("engagement window starts before approval")
|
|
if self.raw["max_concurrency"] != 1:
|
|
raise AuthorizationError("v0.1 permits exactly one in-flight API/database operation")
|
|
if not 1 <= self.raw["rate_limit_per_minute"] <= 60:
|
|
raise AuthorizationError("API rate ceiling must be between 1 and 60 per minute")
|
|
if self.raw["environment"] == "production" and not self.raw.get("production_approval"):
|
|
raise AuthorizationError("production requires production-specific approval")
|
|
if self.raw["finding_destination"] != "risk-nexus":
|
|
raise AuthorizationError("finding destination must be risk-nexus")
|
|
parsed = urlparse(self.raw["target"])
|
|
if parsed.scheme and parsed.scheme not in {"http", "https", "postgresql"}:
|
|
raise AuthorizationError("unsupported target scheme")
|
|
if not self.raw["fixture_ids"]:
|
|
raise AuthorizationError("at least one disposable fixture id is required")
|
|
if not 1 <= self.raw["credential_max_ttl_seconds"] <= 3600:
|
|
raise AuthorizationError("credential TTL must be between 1 and 3600 seconds")
|
|
|
|
def permits(self, *, technique: str, route: str) -> None:
|
|
if technique not in self.raw["techniques"]:
|
|
raise AuthorizationError(f"technique not authorized: {technique}")
|
|
if route not in self.raw["routes"]:
|
|
raise AuthorizationError(f"route not authorized: {route}")
|
|
|
|
|
|
def _timestamp(value: str) -> datetime:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if parsed.tzinfo is None:
|
|
raise AuthorizationError("engagement timestamps must include a timezone")
|
|
return parsed.astimezone(UTC)
|