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 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
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._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 
100
101
102
103
104
105
106
107
108
109
110
111
112
def attest_deletion_wiring(self, attestation: DeletionWiringAttestation) -> None:
    """Accept an attestation. Refuses one that does not prove erasure."""
    if attestation.participant_id != self.participant_id:
        raise RecordingStoreError(
            f"attestation names participant {attestation.participant_id!r}, "
            f"not {self.participant_id!r}"
        )
    if not attestation.verified:
        raise RecordingStoreError(
            "the deletion-wiring attestation did not prove erasure "
            f"(seeded={attestation.seeded}, residual={attestation.residual})"
        )
    self._wiring = attestation

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.