The framework's pause-token signer, named on the public platform surface.
human_interaction(signer=...) needs a signer, and every pause binding
carries a pin. The framework already implements both in
symfonic.services.switching, which is not a public package, so an app on
public imports could only hand-write an EnvelopeSignerPort. This module
re-exports exactly the five existing classes an app composes them from:
- :class:
Keyset and :class:KeyState -- provision the signing key;
- :class:
EnvelopeSigner -- the signer human_interaction takes;
- :class:
InvocationPin -- the pin a PauseBinding carries;
- :class:
PinEnvelope -- the signed envelope a token carries, so an app's
encode_token/decode_token can serialize it and build it back.
Constructing one grants nothing: EnvelopeSigner.verify checks the
signature over every field and the payload before a pause is honoured.
They are the same class objects, not wrappers: no behaviour, key, default
secret or policy is added here. platform -> runtime-service is an allowed
edge of the dependency matrix; symfonic.platform loads this module only
when one of the five names is first read.
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
|
continuation_attestor() -> object
Declared port: the attestor over this signer's keyset (CORE-194).
It offers no public MAC; the kernel seals only for a live memory-free
root run and admits only a row it verifies. None over a Keyset
subclass, whose internals this package cannot vouch for: such a
deployment keeps working and gets no durable NoMemory resume, as does a
signer without this method.
Source code in src/symfonic/services/switching/envelope.py
| def continuation_attestor(self) -> object:
"""Declared port: the attestor over this signer's keyset (CORE-194).
It offers no public MAC; the kernel seals only for a live memory-free
root run and admits only a row it verifies. ``None`` over a ``Keyset``
subclass, whose internals this package cannot vouch for: such a
deployment keeps working and gets no durable NoMemory resume, as does a
signer without this method.
"""
from symfonic.services.switching.attestation import KeysetContinuationAttestor
if type(self._keyset) is not Keyset:
return None
return KeysetContinuationAttestor(self._keyset)
|
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(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
|
InvocationPin(bundle_id: str, epoch: int, vector_hash: str, source: str, stale_binding: bool = False, freeze_epoch_id: str | None = None)
The travelling half of a binding: what a checkpoint or token carries.
A pin names an epoch and a vector hash, never the vector's contents. A
resuming worker resolves the contents from its own binding source and
refuses if they disagree, so a pin cannot smuggle an unvalidated vector
across a process boundary.
The compact form recorded on a RequestContext (RCX-5).
Source code in src/symfonic/services/switching/values.py
| def token(self) -> str:
"""The compact form recorded on a ``RequestContext`` (RCX-5)."""
return f"{self.bundle_id}@{self.epoch}:{self.vector_hash}"
|
key_id → (material, state), with exactly one active key.
Key material never leaves this object: there is no accessor that returns
it, __repr__ names only ids and states, and signing happens here rather
than in the caller. A getter would be convenient exactly once and then live
forever in a log line.
Source code in src/symfonic/services/switching/keys.py
| def __init__(self) -> None:
self._keys: dict[str, tuple[bytes, KeyState]] = {}
self._available = True
|
retire(key_id: str) -> None
KEY-4 — after the overlap window; its envelopes now verify-fail.
Source code in src/symfonic/services/switching/keys.py
| def retire(self, key_id: str) -> None:
"""KEY-4 — after the overlap window; its envelopes now verify-fail."""
material, _ = self._require(key_id)
self._keys[key_id] = (material, KeyState.RETIRED)
|
rotate(key_id: str, *, secret: bytes) -> None
KEY-3 — the new key becomes active; the previous one verifies only.
Source code in src/symfonic/services/switching/keys.py
| def rotate(self, key_id: str, *, secret: bytes) -> None:
"""KEY-3 — the new key becomes active; the previous one verifies only."""
self.provision(key_id, secret=secret, state=KeyState.ACTIVE)
|
set_available(available: bool) -> None
KEY-5 — model secret-manager/keyset unavailability.
Source code in src/symfonic/services/switching/keys.py
| def set_available(self, available: bool) -> None:
"""KEY-5 — model secret-manager/keyset unavailability."""
self._available = available
|
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.
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()}"
)
|
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
|