symfonic.services.switching.envelope

envelope

ENV / CUT-PIN — the authenticated pin envelope for checkpoints and tokens.

One envelope shape wraps both durable artifact families, because both can outlive the process that wrote them and both must answer "which generation produced this?" before anything reads their payload. Verification happens before any field is used — including the version fields a reader would like to dispatch on — so a forged envelope cannot steer its own decoding.

EnvelopeSigner

EnvelopeSigner(keyset: Keyset, *, producer_package_version: str, clock: Callable[[], float] = time.time)

Mints and verifies pin envelopes against a keyset (ENV-2, ENV-6, KEY-5).

Source code in src/symfonic/services/switching/envelope.py

def __init__(
    self,
    keyset: Keyset,
    *,
    producer_package_version: str,
    clock: Callable[[], float] = time.time,
) -> None:
    self._keyset = keyset
    self._producer_version = producer_package_version
    self._clock = clock

mint

mint(*, pin: InvocationPin, payload: bytes, schema_id: str, envelope_version: int = 2) -> PinEnvelope

ENV-6 — refuse to mint when the active key is unavailable.

Source code in src/symfonic/services/switching/envelope.py

def mint(
    self,
    *,
    pin: InvocationPin,
    payload: bytes,
    schema_id: str,
    envelope_version: int = 2,
) -> PinEnvelope:
    """ENV-6 — refuse to mint when the active key is unavailable."""
    key_id = self._keyset.active_key_id()
    unsigned = PinEnvelope(
        envelope_version=envelope_version,
        generation_vector_hash=pin.vector_hash,
        schema_id=schema_id,
        producer_package_version=self._producer_version,
        created_at=self._clock(),
        key_id=key_id,
        pin=pin,
        payload=payload,
    )
    signature = self._keyset.sign(key_id, unsigned.signing_input())
    return PinEnvelope(
        envelope_version=unsigned.envelope_version,
        generation_vector_hash=unsigned.generation_vector_hash,
        schema_id=unsigned.schema_id,
        producer_package_version=unsigned.producer_package_version,
        created_at=unsigned.created_at,
        key_id=unsigned.key_id,
        pin=unsigned.pin,
        payload=unsigned.payload,
        signature=signature,
    )

verify

verify(envelope: PinEnvelope | None, *, pinless_policy: PinlessArtifactPolicy | None = None, schema_id: str = '') -> InvocationPin

Verify first, then dispatch on version. Never the other way round.

Source code in src/symfonic/services/switching/envelope.py

def verify(
    self,
    envelope: PinEnvelope | None,
    *,
    pinless_policy: PinlessArtifactPolicy | None = None,
    schema_id: str = "",
) -> InvocationPin:
    """Verify first, then dispatch on version. Never the other way round."""
    if envelope is None:
        policy = pinless_policy or PinlessArtifactPolicy()
        policy.resolve(schema_id=schema_id)
        raise PinlessArtifactError(
            "a pin-less artifact resolves to an attributed vector, not to an "
            "invocation pin; callers must handle it through the policy."
        )
    unsigned = PinEnvelope(
        envelope_version=envelope.envelope_version,
        generation_vector_hash=envelope.generation_vector_hash,
        schema_id=envelope.schema_id,
        producer_package_version=envelope.producer_package_version,
        created_at=envelope.created_at,
        key_id=envelope.key_id,
        pin=envelope.pin,
        payload=envelope.payload,
    )
    if not self._keyset.verifies(
        envelope.key_id, unsigned.signing_input(), envelope.signature
    ):
        raise EnvelopeVerificationError(
            f"envelope signed by key {envelope.key_id!r} does not verify; it is "
            "treated as tampered or corrupt and never partially honored."
        )
    if envelope.envelope_version not in SUPPORTED_ENVELOPE_VERSIONS:
        raise EnvelopeVersionError(
            f"envelope layout version {envelope.envelope_version} is outside "
            f"this build's supported window {sorted(SUPPORTED_ENVELOPE_VERSIONS)}; "
            "readers reject rather than guess at a layout."
        )
    return envelope.pin

PinEnvelope dataclass

PinEnvelope(envelope_version: int, generation_vector_hash: str, schema_id: str, producer_package_version: str, created_at: float, key_id: str, pin: InvocationPin, payload: bytes, signature: str = '')

ENV-1 — the authenticated wrapper around an opaque payload.

describe

describe() -> str

A diagnostic line: ids and hashes only (SEC-CRED-2, KEY-6).

Source code in src/symfonic/services/switching/envelope.py

def describe(self) -> str:
    """A diagnostic line: ids and hashes only (SEC-CRED-2, KEY-6)."""
    return (
        f"envelope v{self.envelope_version} schema={self.schema_id} "
        f"vector={self.generation_vector_hash} key={self.key_id} "
        f"pin={self.pin.token()}"
    )

signing_input

signing_input() -> bytes

Every envelope field plus the payload — ENV-2 covers both.

Source code in src/symfonic/services/switching/envelope.py

def signing_input(self) -> bytes:
    """Every envelope field plus the payload — ENV-2 covers both."""
    header = json.dumps(
        {
            "envelope_version": self.envelope_version,
            "generation_vector_hash": self.generation_vector_hash,
            "schema_id": self.schema_id,
            "producer_package_version": self.producer_package_version,
            "created_at": self.created_at,
            "key_id": self.key_id,
            "pin": [
                self.pin.bundle_id,
                self.pin.epoch,
                self.pin.vector_hash,
                self.pin.source,
                self.pin.stale_binding,
                self.pin.freeze_epoch_id,
            ],
        },
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")
    return header + b"\x00" + self.payload

PinlessArtifactPolicy dataclass

PinlessArtifactPolicy(accept: bool = False, attributed_vector_hash: str | None = None, reason: str = '')

CUT-PIN-1/2 — what happens to an artifact that carries no pin.

The default is refusal. Accepting one requires naming, at construction, the vector it should be attributed to and why — because the alternative is a baked-in default binding, which is the thing CUT-PIN-1 exists to forbid.