aegis/engines/sanctum.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026 Juan Luna. All rights reserved.
# SPDX-License-Identifier: LicenseRef-Aegis-Commercial-1.0
# Licensed under the Aegis Commercial License 1.0; see LICENSE and NOTICE.
"""Sanctum engine — streaming de-identification and request scanning.
Wraps :class:`~aegis.core.streaming_deidentifier.StreamingDeidentifier` and
:class:`~aegis.proxy.waf.AegisWAF` so a pipeline that is not an Aegis proxy can
still redact across chunk boundaries and scan prompts:
engine = SanctumEngine()
for clean in engine.deidentify_stream(provider_chunks):
yield clean
verdict = engine.scan_prompt(user_text)
Why a holdback exists at all
----------------------------
A streaming redactor cannot release every byte it receives: an identifier can
straddle a chunk boundary, so a bounded tail is withheld until enough context
arrives to settle it. :meth:`deidentify_stream` therefore emits *less* than it
has consumed until :meth:`finalize` — a caller that forwards chunks without
finalising will truncate its own output.
Boundaries carried over unchanged
---------------------------------
- Redaction is **finite pattern matching**, not de-identification in the HIPAA
Safe Harbor or Expert Determination sense (``CLM-016``, ``CLM-057``). Do not
describe its output as de-identified data.
- Bounded quantifiers make the holdback a real bound, and the price is
one-directional: an evasion padding whitespace past the bound is not matched.
- The holdback is checked against the ceiling ``specs/aegis_stream_buffer.smt2``
declares, ``R_max = 4W + Q + E + P``, computed by
:class:`aegis.core.stream_bounds.StreamRetentionBounds`. In-process there is
no queue, no canonical SSE event and no evidence preview, so ``Q``, ``E`` and
``P`` are zero and the ceiling is ``4W``. This bounds **one engine's
holdback**, not process memory: nothing here limits how many engines a caller
runs concurrently.
- When a candidate cannot settle inside the window the stream **fails closed**
with ``StreamingDeidentificationError`` rather than emitting text it could
not finish inspecting.
- The WAF operates on the application-visible representation after parsing. It
establishes no guarantee against prompt injection (``CLM-024``), and its
measured corpus result is bounded to that corpus.
- Redaction protects the *evidence record*, not the upstream provider: the
request still reaches the provider as the caller sent it.
"""
from __future__ import annotations
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from aegis.core.stream_bounds import (
UTF8_MAX_BYTES_PER_CHAR,
StreamBoundsError,
StreamRetentionBounds,
require_window_in_domain,
)
from aegis.core.streaming_deidentifier import (
StreamingDeidentificationError,
StreamingDeidentifier,
)
from aegis.engines import require_module
from aegis.licensing.model import LicenseEntitlement
from aegis.proxy.waf import AegisWAF
_MODULE_NAME = "sanctum"
@dataclass(frozen=True, slots=True)
class ScanVerdict:
"""A WAF decision, flattened to what a caller acts on."""
allowed: bool
reason: str
score: float
shadow_blocked: bool
class SanctumEngine:
"""Streaming redaction and prompt scanning as a library."""
def __init__(
self,
*,
window_chars: int = 128,
enable_phi: bool = True,
enable_pci: bool = True,
strict_mode: bool = True,
shadow_mode: bool = False,
entitlement: LicenseEntitlement | None = None,
) -> None:
self._entitlement = require_module(_MODULE_NAME, entitlement=entitlement)
self._window_chars = require_window_in_domain(window_chars)
self._bounds = StreamRetentionBounds.in_process(self._window_chars)
self._enable_phi = enable_phi
self._enable_pci = enable_pci
self._waf = AegisWAF(strict_mode=strict_mode, shadow_mode=shadow_mode)
@property
def entitlement(self) -> LicenseEntitlement | None:
return self._entitlement
@property
def bounds(self) -> StreamRetentionBounds:
"""The declared retained-byte bounds this engine redacts under."""
return self._bounds
@property
def retained_bytes_ceiling(self) -> int:
"""``R_max`` for this engine, in bytes.
In-process there is no queue, no canonical event and no evidence
preview, so ``Q``, ``E`` and ``P`` are zero and the ceiling is ``4W``.
See :class:`aegis.core.stream_bounds.StreamRetentionBounds`.
"""
return self._bounds.max_retained_bytes
def _new_deidentifier(self) -> StreamingDeidentifier:
return StreamingDeidentifier(
window_chars=self._window_chars,
enable_phi=self._enable_phi,
enable_pci=self._enable_pci,
)
def _require_within_bounds(self, deidentifier: StreamingDeidentifier) -> None:
"""Fail closed if the holdback ever exceeds the declared ceiling.
The redactor already bounds its own holdback and raises when it cannot
settle a candidate. This is the independent check: it reads the same
retention the bound is written about and compares it against ``R_max``
computed from the spec's expression, so a future change that widens the
holdback stops the stream instead of quietly invalidating the bound.
"""
self._bounds.require_admits(
deidentifier.retained_chars * UTF8_MAX_BYTES_PER_CHAR,
what="sanctum de-identification holdback",
)
# ── streaming ───────────────────────────────────────────────────────
def deidentify_stream(self, chunks: Iterable[str]) -> Iterator[str]:
"""Redact across chunk boundaries, yielding only settled text.
The final flush is emitted when the source is exhausted, so consuming
this iterator to completion covers every byte fed in. Abandoning it
early discards whatever was still inside the holdback — which is the
safe direction, since that text was never inspected.
"""
deidentifier = self._new_deidentifier()
for chunk in chunks:
settled = deidentifier.feed(chunk)
self._require_within_bounds(deidentifier)
if settled:
yield settled
tail = deidentifier.flush()
if tail:
yield tail
def deidentify_text(self, raw_text: str) -> str:
"""Redact a complete string. Equivalent to feeding it as one chunk."""
deidentifier = self._new_deidentifier()
settled = deidentifier.feed(raw_text)
self._require_within_bounds(deidentifier)
return settled + deidentifier.flush()
# ── scanning ────────────────────────────────────────────────────────
def scan_prompt(self, prompt: str) -> ScanVerdict:
"""Run the request WAF over a prompt string."""
return self.scan_payload({"prompt": prompt})
def scan_payload(self, payload: object) -> ScanVerdict:
"""Run the request WAF over a parsed request body."""
result = self._waf.inspect_payload(payload)
return ScanVerdict(
allowed=bool(result.allowed),
reason=str(result.reason),
score=float(result.score),
shadow_blocked=bool(result.shadow_blocked),
)
__all__ = [
"SanctumEngine",
"ScanVerdict",
"StreamBoundsError",
"StreamRetentionBounds",
"StreamingDeidentificationError",
]