coulomb/whitehat-security
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bind WP-0025 receipts to a canonical projection contract

Browse source 
Railiance's WP-0025 review found the first adapter fail-open: --contract
was optional, receipt_id was not canonical, and target, identities,
resources, UIDs, broker digest and times were unbound. Require the
contract and refuse any receipt that is not the platform canonical form.

Assistant: grok
Assistant-Session: 01a02670-3345-76f2-a014-70fde8e2a2bb
This commit is contained in:
tegwick 2026-08-22 22:21:34 +02:00
parent da6f5fb3f8
commit 5a0eb6b343
6 changed files with 592 additions and 140 deletions
Download patch file Download diff file Expand all files Collapse all files

26
src/whitehat_security/plane.py
View file

@ -106,8 +106,8 @@ class IdentityHandle:
class CredentialBroker(Protocol):
def project(self, engagement: Engagement, registration: dict[str, Any]
) -> tuple[IdentityHandle, ...]:
def project(self, engagement: Engagement, registration: dict[str, Any],
now: datetime | None = None) -> tuple[IdentityHandle, ...]:
...
def revoke(self, lease_id: str) -> None:
@ -148,8 +148,8 @@ class ReceiptBroker:
def load(cls, path: str | Path) -> "ReceiptBroker":
return cls(json.loads(Path(path).read_text(encoding="utf-8")))
def project(self, engagement: Engagement, registration: dict[str, Any]
) -> tuple[IdentityHandle, ...]:
def project(self, engagement: Engagement, registration: dict[str, Any],
now: datetime | None = None) -> tuple[IdentityHandle, ...]:
if self.receipt.get("engagement_id") != engagement.raw["engagement_id"]:
raise AuthorizationError("projection receipt engagement_id does not match")
if not self.receipt.get("target_ready"):
@ -160,7 +160,8 @@ class ReceiptBroker:
expiry = datetime.fromisoformat(str(expires_at).replace("Z", "+00:00"))
if expiry.tzinfo is None:
raise AuthorizationError("projection receipt expiry must include a timezone")
if datetime.now(UTC) > expiry.astimezone(UTC):
clock = (now or datetime.now(UTC)).astimezone(UTC)
if clock > expiry.astimezone(UTC):
raise AuthorizationError("projection receipt has expired")
keys = list(self.receipt.get("mounted_keys") or [])
count = int(registration["identities"]["count"])
@ -193,8 +194,8 @@ class ReceiptBroker:
class UnconnectedCustodyBroker:
"""Fail-closed live broker. It never requests or returns credential values."""
def project(self, engagement: Engagement, registration: dict[str, Any]
) -> tuple[IdentityHandle, ...]:
def project(self, engagement: Engagement, registration: dict[str, Any],
now: datetime | None = None) -> tuple[IdentityHandle, ...]:
raise AuthorizationError(
"live custody broker is not connected; no credential was requested"
)
@ -212,14 +213,15 @@ class LocalBroker:
self._secrets: dict[str, bytes] = {}
self._leases: dict[str, tuple[str, str]] = {}
def project(self, engagement: Engagement, registration: dict[str, Any]
) -> tuple[IdentityHandle, ...]:
def project(self, engagement: Engagement, registration: dict[str, Any],
now: datetime | None = None) -> tuple[IdentityHandle, ...]:
if registration.get("applicability") != "applicable":
raise AuthorizationError("broker will not project identities for a non-applicable target")
count = int(registration["identities"]["count"])
lease_id = os.urandom(8).hex()
ttl = min(int(engagement.raw["credential_max_ttl_seconds"]), STANDING_E2_TTL)
expires_at = (datetime.now(UTC) + timedelta(seconds=ttl)).isoformat().replace("+00:00", "Z")
clock = (now or datetime.now(UTC)).astimezone(UTC)
expires_at = (clock + timedelta(seconds=ttl)).isoformat().replace("+00:00", "Z")
handles: list[IdentityHandle] = []
if count >= 1:
role = "owner" if count == 2 else "runtime"
@ -302,7 +304,7 @@ def admit(*, engagement: Engagement, registration: dict[str, Any],
selected = broker or default_broker(engagement)
count = int(registration["identities"]["count"])
if count:
identities = selected.project(engagement, registration)
identities = selected.project(engagement, registration, now=now)
if len(identities) != count:
raise AuthorizationError(f"broker must project exactly {count} identities")
expected = {2: {"owner", "attacker"}, 1: {"runtime"}}[count]
@ -313,7 +315,7 @@ def admit(*, engagement: Engagement, registration: dict[str, Any],
lease_id = identities[0].lease_id
else:
if approval_class not in FIXTURE_CLASSES:
selected.project(engagement, registration)
selected.project(engagement, registration, now=now)
identities = ()
lease_id = os.urandom(8).hex()
if hasattr(selected, "_leases"):