tegwick
abd22fa0a6
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a025c2-407a-7a32-b40a-f37a52f03f62
281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""Sender identities and what they are permitted to claim.
|
|
|
|
AUDIT-WP-0004-T03. WP-0003 recorded tenant isolation as delivered, but the
|
|
receiver accepted whatever ``tenant`` and ``source`` a caller sent as long as
|
|
it held the one shared token. This module binds a credential to the sources and
|
|
tenants it may write for, and drives that binding from configuration rather
|
|
than literals.
|
|
|
|
Each identity carries a *list* of tokens so a credential can be rotated without
|
|
a delivery gap: publish the replacement alongside the incumbent, move the
|
|
sender, then drop the old one.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
from audit_core.redaction import POLICIES, POLICY_REDACT
|
|
|
|
WILDCARD = "*"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SenderIdentity:
|
|
"""A credential holder and the claims it is allowed to make."""
|
|
|
|
name: str
|
|
tokens: tuple[str, ...]
|
|
sources: frozenset[str]
|
|
tenants: frozenset[str] = field(default_factory=lambda: frozenset({WILDCARD}))
|
|
may_write: bool = True
|
|
may_read: bool = False
|
|
# Optional absolute expiry for deliberately temporary credentials. A token
|
|
# remains unusable after this instant even if a stale Secret value is still
|
|
# present in the process environment.
|
|
expires_at: datetime | None = None
|
|
# Secret-shaped field handling. Defaults to redact-and-accept, so a
|
|
# legitimate event is not lost over one field; a higher-assurance channel
|
|
# can be set to reject instead (AUDIT-WP-0004-T04).
|
|
secret_policy: str = POLICY_REDACT
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.name:
|
|
raise ValueError("sender identity needs a name")
|
|
if self.secret_policy not in POLICIES:
|
|
raise ValueError(
|
|
f"sender {self.name!r}: secret_policy must be one of {POLICIES}, "
|
|
f"got {self.secret_policy!r}"
|
|
)
|
|
if not self.tokens or any(not t for t in self.tokens):
|
|
raise ValueError(f"sender {self.name!r} needs at least one non-empty token")
|
|
if not self.sources:
|
|
raise ValueError(f"sender {self.name!r} needs at least one permitted source")
|
|
if self.expires_at is not None and self.expires_at.tzinfo is None:
|
|
raise ValueError(f"sender {self.name!r}: expires_at must include a timezone")
|
|
|
|
def permits_source(self, source: str) -> bool:
|
|
return WILDCARD in self.sources or source in self.sources
|
|
|
|
def is_active(self, now: datetime) -> bool:
|
|
"""Whether the credential is inside its configured lifetime."""
|
|
|
|
if now.tzinfo is None:
|
|
raise ValueError("authentication time must include a timezone")
|
|
return self.expires_at is None or now < self.expires_at
|
|
|
|
def has_full_tenant_scope(self) -> bool:
|
|
"""Whether this identity is unrestricted across tenants.
|
|
|
|
The read surfaces that are not tenant-keyed — chain verification,
|
|
counters, dead letters, secret findings — cannot be filtered, so they
|
|
are reserved for an identity that could read every tenant anyway.
|
|
Anything else would hand a scoped reader instance-wide facts.
|
|
"""
|
|
return WILDCARD in self.tenants
|
|
|
|
def permits_tenant(self, tenant: str) -> bool:
|
|
"""Whether this identity may write for ``tenant``.
|
|
|
|
Tenant identifiers are opaque here. Their shape
|
|
(``tenant:<grouping>:<name>``) is owned by the IAM Profile and
|
|
tenant-engine; matching is exact, never parsed or pattern-derived.
|
|
"""
|
|
return WILDCARD in self.tenants or tenant in self.tenants
|
|
|
|
|
|
class SenderRegistry:
|
|
"""Resolves a credential to the identity that holds it."""
|
|
|
|
def __init__(self, identities: Iterable[SenderIdentity]) -> None:
|
|
self.identities = tuple(identities)
|
|
if not self.identities:
|
|
raise ValueError("at least one sender identity is required")
|
|
|
|
def authenticate(
|
|
self,
|
|
authorization_header: str | None,
|
|
*,
|
|
now: datetime | None = None,
|
|
) -> SenderIdentity | None:
|
|
"""Return the matching identity, or None.
|
|
|
|
Every candidate token is compared even after a match is found, so the
|
|
work done does not depend on which credential was supplied or how many
|
|
identities precede it.
|
|
"""
|
|
supplied = str(authorization_header or "")
|
|
checked_at = now or datetime.now(timezone.utc)
|
|
matched: SenderIdentity | None = None
|
|
for identity in self.identities:
|
|
for token in identity.tokens:
|
|
try:
|
|
hit = hmac.compare_digest(supplied, f"Bearer {token}")
|
|
except TypeError:
|
|
# compare_digest rejects non-ASCII str; such a header is
|
|
# simply not a valid credential.
|
|
hit = False
|
|
if hit and identity.is_active(checked_at) and matched is None:
|
|
matched = identity
|
|
return matched
|
|
|
|
@classmethod
|
|
def from_env(cls, env: dict[str, str] | None = None) -> "SenderRegistry":
|
|
"""Build a registry from ``AUDIT_CORE_SENDERS`` (JSON).
|
|
|
|
Shape::
|
|
|
|
[{"name": "user-engine",
|
|
"tokens": ["current", "next"],
|
|
"sources": ["user-engine"],
|
|
"tenants": ["tenant:friendly:binky"],
|
|
"may_read": false}]
|
|
|
|
``tenants`` defaults to ``["*"]``. Omitting it is a deliberate choice to
|
|
accept any tenant from that sender and should be justified per sender,
|
|
not adopted by default.
|
|
"""
|
|
env = env if env is not None else dict(os.environ)
|
|
raw = env.get("AUDIT_CORE_SENDERS")
|
|
if raw:
|
|
return cls(_apply_scope_overlay(_parse_identities(raw), env))
|
|
|
|
legacy = (env.get("AUDIT_CORE_INGEST_TOKEN") or "").strip()
|
|
if not legacy:
|
|
raise ValueError(
|
|
"no sender configuration: set AUDIT_CORE_SENDERS (preferred) "
|
|
"or AUDIT_CORE_INGEST_TOKEN"
|
|
)
|
|
# Legacy single-token form. Kept so an existing deployment keeps
|
|
# working, but it grants every tenant — which is what T03 exists to
|
|
# stop, so it is deliberately noisy about what it is.
|
|
return cls([
|
|
SenderIdentity(
|
|
name="legacy-ingest-token",
|
|
tokens=(legacy,),
|
|
sources=frozenset({"user-engine"}),
|
|
tenants=frozenset({WILDCARD}),
|
|
)
|
|
])
|
|
|
|
|
|
def _load_scope_overlay(env: dict[str, str]) -> list[dict[str, Any]]:
|
|
"""Non-secret sender policy. Tokens never come from here.
|
|
|
|
``AUDIT_CORE_SENDERS_SCOPE`` is inline JSON (tests).
|
|
``AUDIT_CORE_SENDERS_SCOPE_PATH`` is a file (production ConfigMap).
|
|
"""
|
|
inline = env.get("AUDIT_CORE_SENDERS_SCOPE")
|
|
if inline:
|
|
payload = json.loads(inline)
|
|
else:
|
|
path = env.get("AUDIT_CORE_SENDERS_SCOPE_PATH")
|
|
if not path:
|
|
return []
|
|
payload = json.loads(Path(path).read_text())
|
|
if not isinstance(payload, list):
|
|
raise ValueError("sender scope overlay must be a JSON list")
|
|
return [entry for entry in payload if isinstance(entry, dict) and entry.get("name")]
|
|
|
|
|
|
def _apply_scope_overlay(
|
|
identities: list[SenderIdentity], env: dict[str, str]
|
|
) -> list[SenderIdentity]:
|
|
"""Overlay non-secret fields from Git/ConfigMap onto Secret-backed tokens.
|
|
|
|
ExternalSecret refresh cannot shrink ``user-engine`` tenants below the
|
|
declared scope (AUDIT-WP-0006-T05). Tokens are never taken from the overlay.
|
|
"""
|
|
overlay = {entry["name"]: entry for entry in _load_scope_overlay(env)}
|
|
if not overlay:
|
|
return identities
|
|
merged: list[SenderIdentity] = []
|
|
for identity in identities:
|
|
extra = overlay.get(identity.name)
|
|
if extra is None:
|
|
merged.append(identity)
|
|
continue
|
|
tenants = extra.get("tenants")
|
|
sources = extra.get("sources")
|
|
merged.append(
|
|
SenderIdentity(
|
|
name=identity.name,
|
|
tokens=identity.tokens,
|
|
sources=(
|
|
frozenset(str(s) for s in sources) if sources else identity.sources
|
|
),
|
|
tenants=(
|
|
frozenset(str(t) for t in tenants) if tenants else identity.tenants
|
|
),
|
|
may_write=(
|
|
bool(extra["may_write"]) if "may_write" in extra else identity.may_write
|
|
),
|
|
may_read=(
|
|
bool(extra["may_read"]) if "may_read" in extra else identity.may_read
|
|
),
|
|
secret_policy=identity.secret_policy,
|
|
expires_at=identity.expires_at,
|
|
)
|
|
)
|
|
return merged
|
|
|
|
|
|
def _parse_identities(raw: str) -> list[SenderIdentity]:
|
|
try:
|
|
entries = json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(f"AUDIT_CORE_SENDERS is not valid JSON: {exc}") from exc
|
|
if not isinstance(entries, list) or not entries:
|
|
raise ValueError("AUDIT_CORE_SENDERS must be a non-empty JSON list")
|
|
return [_identity_from(entry) for entry in entries]
|
|
|
|
|
|
def _identity_from(entry: Any) -> SenderIdentity:
|
|
if not isinstance(entry, dict):
|
|
raise ValueError("each sender entry must be an object")
|
|
tokens = entry.get("tokens") or ([entry["token"]] if entry.get("token") else [])
|
|
return SenderIdentity(
|
|
name=str(entry.get("name") or ""),
|
|
tokens=tuple(str(t) for t in tokens),
|
|
sources=frozenset(str(s) for s in (entry.get("sources") or [])),
|
|
tenants=frozenset(str(t) for t in (entry.get("tenants") or [WILDCARD])),
|
|
may_write=bool(entry.get("may_write", True)),
|
|
may_read=bool(entry.get("may_read", False)),
|
|
secret_policy=str(entry.get("secret_policy", POLICY_REDACT)),
|
|
expires_at=_parse_expiry(entry.get("expires_at")),
|
|
)
|
|
|
|
|
|
def _parse_expiry(value: Any) -> datetime | None:
|
|
if value in (None, ""):
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
except (TypeError, ValueError):
|
|
raise ValueError("sender expires_at must be an RFC3339 timestamp") from None
|
|
if parsed.tzinfo is None:
|
|
raise ValueError("sender expires_at must include a timezone")
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def development_registry(token: str) -> SenderRegistry:
|
|
"""A single permissive sender, for tests and local development.
|
|
|
|
Accepts any tenant. Not a production configuration — production binds each
|
|
sender to the tenants it may write for via ``AUDIT_CORE_SENDERS``.
|
|
"""
|
|
return SenderRegistry([
|
|
SenderIdentity(
|
|
name="development",
|
|
tokens=(token,),
|
|
sources=frozenset({"user-engine"}),
|
|
tenants=frozenset({WILDCARD}),
|
|
may_read=True,
|
|
)
|
|
])
|