tegwick
4349758608
Declare Engine/PIP machine-readably, publish a total fail-closed PEP stance map, stop minting local decision ids on engine-unavailable DENY, bind allows to a 30s request lifetime, confine the local authorization double, classify evidence and emit a denial/revocation heartbeat, and prove access-control facts remain claims. Assistant: grok Assistant-Session: 01a04cea-f0d6-7ab3-9ffd-881eb6bea6cb
88 lines
2.5 KiB
Python
88 lines
2.5 KiB
Python
"""Local evidence classification under security-layer-model §9.6.
|
|
|
|
An append-only local audit and outbox prove the records they hold were
|
|
not altered after arrival. They do not prove that an event never sent
|
|
did not happen. Completeness is not claimed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
from typing import Iterable
|
|
|
|
from user_engine.domain import AuditRecord, OutboxEvent
|
|
|
|
HEARTBEAT_EVENT_TYPE = "user_engine.evidence.heartbeat"
|
|
HEARTBEAT_INTERVAL = timedelta(hours=1)
|
|
|
|
# Low-volume classes whose absence can be mistaken for "it was allowed"
|
|
# or "the membership still holds". Cadence is a heartbeat, not a rate.
|
|
LOAD_BEARING_OUTBOX_TYPES = frozenset(
|
|
{
|
|
"account.status_changed",
|
|
"tenant_account.status_changed",
|
|
"prepared_account.revoked",
|
|
"family_invitation.revoked",
|
|
}
|
|
)
|
|
|
|
LOAD_BEARING_AUDIT_SUMMARIES = frozenset(
|
|
{
|
|
"authorization denied",
|
|
"authorization denied (stance fail_closed)",
|
|
}
|
|
)
|
|
|
|
ATTRIBUTIVE_OUTBOX_TYPES = frozenset(
|
|
{
|
|
"user.created",
|
|
"user.self_service_profile_updated",
|
|
"identity.linked",
|
|
"membership.added",
|
|
"profile.value_set",
|
|
"application.registered",
|
|
"application.bound",
|
|
"catalog.published",
|
|
"registration.started",
|
|
"registration.factor_verified",
|
|
"registration.completed",
|
|
"access_profile.registered",
|
|
"active_access_context.selected",
|
|
}
|
|
)
|
|
|
|
|
|
def classify_outbox(event_type: str) -> str:
|
|
if event_type == HEARTBEAT_EVENT_TYPE:
|
|
return "heartbeat"
|
|
if event_type in LOAD_BEARING_OUTBOX_TYPES:
|
|
return "load-bearing"
|
|
return "attributive"
|
|
|
|
|
|
def classify_audit(summary: str | None) -> str:
|
|
if summary in LOAD_BEARING_AUDIT_SUMMARIES or (
|
|
summary is not None and summary.startswith("authorization denied")
|
|
):
|
|
return "load-bearing"
|
|
return "attributive"
|
|
|
|
|
|
def load_bearing_counts(
|
|
audit_records: Iterable[AuditRecord],
|
|
outbox_events: Iterable[OutboxEvent],
|
|
) -> dict[str, int]:
|
|
counts: dict[str, int] = {
|
|
"authorization_denied": 0,
|
|
"account.status_changed": 0,
|
|
"tenant_account.status_changed": 0,
|
|
"prepared_account.revoked": 0,
|
|
"family_invitation.revoked": 0,
|
|
}
|
|
for record in audit_records:
|
|
if classify_audit(record.summary) == "load-bearing":
|
|
counts["authorization_denied"] += 1
|
|
for event in outbox_events:
|
|
if event.event_type in counts:
|
|
counts[event.event_type] += 1
|
|
return counts
|