Skip to content

symfonic.services.shadow.store

store

The governed recording store.

Aggregate classification (threat-model §6.2): governed tenant data — tenant-scoped, credential-scrubbed at capture, access-controlled, retention-bounded, encrypted at rest, erasable. Each of those is a property of this class rather than of an operational runbook:

  • every row is filed under a tenant and no read crosses tenants (TM-29e), and a recording id already held for one tenant is refused for another rather than overwritten — ids derive from adopter invocation ids, which are unique only within a tenant;
  • every read is authorized against a grant and appended to an access audit (TM-29a) — including the reads that were refused;
  • every row carries an expiry and purge_expired enforces it (TM-29c);
  • rows are stored only as SealedPayload; the plaintext never lives here;
  • PRODUCTION writes are refused until the privacy-deletion wiring has been attested by an actual erase round-trip (SEC-PRIV-5).

AccessAuditEntry dataclass 

AccessAuditEntry(at: datetime, actor: str, action: str, recording_id: str, tenant_id: str, allowed: bool, reason: str = '')

Append-only: every read attempt, allowed or not.

AccessGrant dataclass 

AccessGrant(actor: str, tenants: frozenset[str], purpose: str)

Who may read recordings, for which tenants, and why.

RecordingStore 

RecordingStore(*, cipher: EncryptionPort | None = None, retention: timedelta = timedelta(days=7), grants: tuple[AccessGrant, ...] = (), deletion_wiring: DeletionWiringAttestation | None = None, clock: Callable[[], datetime] | None = None)

An in-memory governed store. The seams are the cipher and the clock.

Source code in src/symfonic/services/shadow/store.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def __init__(
    self,
    *,
    cipher: EncryptionPort | None = None,
    retention: timedelta = timedelta(days=7),
    grants: tuple[AccessGrant, ...] = (),
    deletion_wiring: DeletionWiringAttestation | None = None,
    clock: Callable[[], datetime] | None = None,
) -> None:
    if retention <= timedelta(0):
        raise ValueError("the recording store must be retention-bounded")
    self._cipher = cipher or RefusingCipher()
    self._retention = retention
    self._grants = {grant.actor: grant for grant in grants}
    self._wiring = deletion_wiring
    self._native_privacy_only = False
    self._erased_tenants: set[str] = set()
    self._lock = RLock()
    self._clock = clock or (lambda: datetime.now(UTC))
    self._rows: dict[str, StoredRecording] = {}
    self._audit: list[AccessAuditEntry] = []

attest_deletion_wiring 

attest_deletion_wiring(attestation: DeletionWiringAttestation) -> None

Accept an attestation. Refuses one that does not prove erasure.

Source code in src/symfonic/services/shadow/store.py 
114
115
116
117
def attest_deletion_wiring(self, attestation: DeletionWiringAttestation) -> None:
    """Accept an attestation. Refuses one that does not prove erasure."""
    with self._lock:
        self._attest_deletion_wiring_locked(attestation)

mark_native_privacy_only 

mark_native_privacy_only() -> None

Join native erasure only while this synchronous store is synthetic-only.

Source code in src/symfonic/services/shadow/store.py 
104
105
106
107
108
109
110
111
112
def mark_native_privacy_only(self) -> None:
    """Join native erasure only while this synchronous store is synthetic-only."""
    with self._lock:
        if self._wiring is not None:
            raise RecordingStoreError(
                "native registration refuses production-capable recording stores: "
                "synchronous put has no atomic native write fence"
            )
        self._native_privacy_only = True

StoredRecording dataclass 

StoredRecording(recording_id: str, tenant_id: str, mode: RecordingMode, captured_at: datetime, expires_at: datetime, digest: str, sealed: SealedPayload)

What the store actually holds: metadata plus ciphertext.