tegwick
95129d7a35
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
73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Callable
|
|
|
|
from .model import Observation, ProbeResult, Response, observe
|
|
from .plane import PlaneLease
|
|
|
|
Request = Callable[[], Response]
|
|
State = Callable[[], object]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DifferentialProbe:
|
|
probe_id: str
|
|
operation: str
|
|
owner_request: Request
|
|
attacker_request: Request
|
|
absent_request: Request
|
|
fixture_markers: tuple[str, ...]
|
|
state: State | None = None
|
|
hide_existence: bool = True
|
|
|
|
|
|
def execute(probe: DifferentialProbe, *, salt: bytes | None = None) -> ProbeResult:
|
|
"""Run one cold probe, retaining response bodies only in this stack frame."""
|
|
run_salt = salt or os.urandom(32)
|
|
before = probe.state() if probe.state else None
|
|
owner = observe(probe.owner_request(), salt=run_salt, fixture_markers=probe.fixture_markers)
|
|
absent = observe(probe.absent_request(), salt=run_salt, fixture_markers=probe.fixture_markers)
|
|
attacker = observe(probe.attacker_request(), salt=run_salt, fixture_markers=probe.fixture_markers)
|
|
after = probe.state() if probe.state else None
|
|
|
|
reasons: list[str] = []
|
|
if attacker.fixture_match_count:
|
|
reasons.append("attacker response contains a tenant-B fixture marker")
|
|
if before != after:
|
|
reasons.append("tenant-B fixture state changed")
|
|
if probe.hide_existence and not _equivalent_denial(attacker, absent):
|
|
reasons.append("attacker response differs from the absent-object reference")
|
|
|
|
return ProbeResult(
|
|
probe_id=probe.probe_id,
|
|
operation=probe.operation,
|
|
outcome="finding" if reasons else "pass",
|
|
reasons=reasons,
|
|
observations={"owner": owner, "attacker": attacker, "absent": absent},
|
|
)
|
|
|
|
|
|
def execute_authorized(probe: DifferentialProbe, *, lease: PlaneLease,
|
|
route: str, salt: bytes | None = None,
|
|
now: datetime | None = None) -> ProbeResult:
|
|
"""Live entry point: plane admission is rechecked immediately before calls."""
|
|
lease.engagement.validate(now=now)
|
|
lease.permits(technique="e2-differential", route=route)
|
|
lease.watcher.acquire()
|
|
try:
|
|
return execute(probe, salt=salt)
|
|
finally:
|
|
lease.watcher.release()
|
|
|
|
|
|
def _equivalent_denial(left: Observation, right: Observation) -> bool:
|
|
return (
|
|
left.status == right.status
|
|
and left.content_type == right.content_type
|
|
and left.count == right.count
|
|
and left.schema == right.schema
|
|
and left.run_digest == right.run_digest
|
|
)
|