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
83 lines
3 KiB
Python
83 lines
3 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from dataclasses import dataclass
|
|
|
|
from .differential import DifferentialProbe
|
|
from .model import Response
|
|
|
|
|
|
@dataclass
|
|
class AuditFixture:
|
|
"""In-process audit-core shape used to calibrate the E2 pack. No sockets."""
|
|
|
|
enforce_tenant: bool
|
|
|
|
def __post_init__(self) -> None:
|
|
self.events = {
|
|
"event-a": {
|
|
"id": "event-a", "tenant": "tenant-a",
|
|
"correlation_id": "corr-shared", "label": "fixture-a",
|
|
},
|
|
"event-b": {
|
|
"id": "event-b", "tenant": "tenant-b",
|
|
"correlation_id": "corr-shared", "label": "fixture-b",
|
|
},
|
|
}
|
|
|
|
def snapshot_events(self) -> object:
|
|
return deepcopy(self.events)
|
|
|
|
def _visible(self, actor_tenant: str, item: dict) -> bool:
|
|
return (not self.enforce_tenant) or item["tenant"] == actor_tenant
|
|
|
|
def get(self, actor_tenant: str, event_id: str) -> Response:
|
|
item = self.events.get(event_id)
|
|
if item is None or not self._visible(actor_tenant, item):
|
|
return Response(404, {"error": "not_found"})
|
|
return Response(200, deepcopy(item))
|
|
|
|
def by_correlation(self, actor_tenant: str, correlation_id: str) -> Response:
|
|
items = [
|
|
deepcopy(item) for item in self.events.values()
|
|
if item["correlation_id"] == correlation_id and self._visible(actor_tenant, item)
|
|
]
|
|
return Response(200, {"events": items})
|
|
|
|
def append(self, actor_tenant: str, event: dict) -> Response:
|
|
if self.enforce_tenant and event["tenant"] != actor_tenant:
|
|
return Response(403, {"error": "forbidden"})
|
|
self.events[event["id"]] = deepcopy(event)
|
|
return Response(202, {"id": event["id"], "tenant": event["tenant"]})
|
|
|
|
|
|
def audit_probe_suite(service: AuditFixture) -> list[DifferentialProbe]:
|
|
markers = ("event-b", "tenant-b", "fixture-b")
|
|
forged = {
|
|
"id": "event-forged-b", "tenant": "tenant-b",
|
|
"correlation_id": "corr-shared", "label": "fixture-forged",
|
|
}
|
|
return [
|
|
DifferentialProbe(
|
|
"audit-event-by-id", "read",
|
|
lambda: service.get("tenant-b", "event-b"),
|
|
lambda: service.get("tenant-a", "event-b"),
|
|
lambda: service.get("tenant-a", "absent"),
|
|
markers,
|
|
),
|
|
DifferentialProbe(
|
|
"audit-correlation-slice", "read",
|
|
lambda: service.by_correlation("tenant-b", "corr-shared"),
|
|
lambda: service.by_correlation("tenant-a", "corr-shared"),
|
|
lambda: service.by_correlation("tenant-a", "corr-absent"),
|
|
markers, hide_existence=False,
|
|
),
|
|
DifferentialProbe(
|
|
"audit-append-as-b", "create",
|
|
lambda: Response(202, {"id": "owner-control"}),
|
|
lambda: service.append("tenant-a", forged),
|
|
lambda: Response(403, {"error": "forbidden"}),
|
|
markers + ("event-forged-b",),
|
|
state=service.snapshot_events, hide_existence=False,
|
|
),
|
|
]
|