SCP — the only thing allowed to mutate switch state (CUT-SS-8).
This class is the operator-facing vocabulary: create a bundle, commit a
vector, roll back, freeze, revoke, break glass. Every one of those verbs is
the same mutation mechanism (:class:~.mutation.SwitchMutator) with different
arguments, which is the point — there is no second write path, and no verb
that skips the audit chain or the constraint check on its way to the store.
SwitchControlPlane
SwitchControlPlane(*, store: InMemorySwitchStore, constraints: ConstraintSet, audit: HashChainAuditLog, authorizer: SwitchAuthorizer, validated_vectors: Sequence[GenerationVector] = (), evidence: RetirementEvidenceStore | None = None, clock: Callable[[], float] = time.time)
The authorized, audited switch-state service.
Source code in src/symfonic/services/switching/control_plane.py
| def __init__(
self,
*,
store: InMemorySwitchStore,
constraints: ConstraintSet,
audit: HashChainAuditLog,
authorizer: SwitchAuthorizer,
validated_vectors: Sequence[GenerationVector] = (),
evidence: RetirementEvidenceStore | None = None,
clock: Callable[[], float] = time.time,
) -> None:
self._store = store
self._evidence = evidence
self._clock = clock
self._mutator = SwitchMutator(
store=store,
constraints=constraints,
audit=audit,
authorizer=authorizer,
validated_vectors=validated_vectors,
clock=clock,
)
self._break_glass: dict[str, BreakGlassCredential] = {}
self._consumed: set[str] = set()
|
break_glass_revert
async
break_glass_revert(credential: BreakGlassCredential, bundle_id: str, *, expected_epoch: int | None = None, reason: str) -> BundleRecord
SCP-BG — revert-only, single-use, always audited, never a freeze verb.
Source code in src/symfonic/services/switching/control_plane.py
| async def break_glass_revert(
self,
credential: BreakGlassCredential,
bundle_id: str,
*,
expected_epoch: int | None = None,
reason: str,
) -> BundleRecord:
"""SCP-BG — revert-only, single-use, always audited, never a freeze verb."""
provisioned = self._break_glass.get(credential.credential_id)
if provisioned is None or provisioned.bundle_id != bundle_id:
raise SwitchAuthorizationError(
f"break-glass credential {credential.credential_id!r} is not "
f"provisioned for bundle {bundle_id!r}."
)
if credential.credential_id in self._consumed:
raise SwitchAuthorizationError(
f"break-glass credential {credential.credential_id!r} was already "
"consumed; each credential is single-use and re-provisioning runs "
"through the normal authenticated path."
)
record = await self._store.read(bundle_id)
stored = await self._mutator.swap(
Actor(identity=provisioned.identity, role=Role.BREAK_GLASS),
Action.BREAK_GLASS_REVERT,
bundle_id,
vector=record.rollback_vector or record.generation_vector,
expected_epoch=expected_epoch,
reason=reason,
)
self._consumed.add(credential.credential_id)
return stored
|
commit_vector
async
commit_vector(actor: Actor, bundle_id: str, vector: GenerationVector, *, expected_epoch: int | None = None, reason: str = '') -> BundleRecord
CUT-RB-2/CUT-SS-2 — one bundle, one validated vector, one CAS.
Source code in src/symfonic/services/switching/control_plane.py
| async def commit_vector(
self,
actor: Actor,
bundle_id: str,
vector: GenerationVector,
*,
expected_epoch: int | None = None,
reason: str = "",
) -> BundleRecord:
"""CUT-RB-2/CUT-SS-2 — one bundle, one validated vector, one CAS."""
return await self._mutator.swap(
actor,
Action.COMMIT_VECTOR,
bundle_id,
vector=vector,
expected_epoch=expected_epoch,
reason=reason,
)
|
freeze
async
freeze(actor: Actor, bundle_id: str, *, approver: Actor, expected_epoch: int | None = None, retiring: Sequence[str] | None = None, reason: str = 'freeze') -> BundleRecord
SCP-FRZ-1 — commit a reversible retirement-freeze epoch object.
Source code in src/symfonic/services/switching/control_plane.py
| async def freeze(
self,
actor: Actor,
bundle_id: str,
*,
approver: Actor,
expected_epoch: int | None = None,
retiring: Sequence[str] | None = None,
reason: str = "freeze",
) -> BundleRecord:
"""SCP-FRZ-1 — commit a reversible retirement-freeze epoch object."""
record = await self._store.read(bundle_id)
legacy = record.rollback_vector or record.generation_vector
state = FreezeState(
freeze_epoch_id=f"frz-{uuid.uuid4().hex[:12]}",
frozen_legacy_vector=legacy,
retiring=(
frozenset(retiring)
if retiring is not None
else self._default_retiring(record.generation_vector, legacy)
),
created_by=actor.identity,
approved_by=approver.identity,
created_at=self._clock(),
)
return await self._mutator.swap(
actor,
Action.FREEZE,
bundle_id,
vector=record.generation_vector,
expected_epoch=expected_epoch,
reason=reason,
approver=approver,
freeze_state=state,
)
|
retirement_ready
retirement_ready(bundle_id: str, freeze_epoch_id: str, *, registry: object | None = None, barrier: object | None = None, epoch: int | None = None) -> bool
SCP-FRZ-3 — every leg of the freeze-and-drain gate, or False.
Deliberately returns a boolean rather than raising: this is a gate a
runbook polls, and "not yet" is its normal answer, not an exception.
Source code in src/symfonic/services/switching/control_plane.py
| def retirement_ready(
self,
bundle_id: str,
freeze_epoch_id: str,
*,
registry: object | None = None,
barrier: object | None = None,
epoch: int | None = None,
) -> bool:
"""SCP-FRZ-3 — every leg of the freeze-and-drain gate, or ``False``.
Deliberately returns a boolean rather than raising: this is a gate a
runbook polls, and "not yet" is its normal answer, not an exception.
"""
if not self.audit.verify():
return False
if self._evidence is None or not self._evidence.covers(freeze_epoch_id):
return False
if barrier is not None and not barrier.acknowledged(freeze_epoch_id): # type: ignore[attr-defined]
return False
if registry is not None and epoch is not None:
return bool(registry.quiescent_below(bundle_id, epoch)) # type: ignore[attr-defined]
return True
|
revoke_freeze
async
revoke_freeze(actor: Actor, bundle_id: str, *, expected_epoch: int | None = None, reason: str) -> BundleRecord
SCP-REV — release-owner-only, audited, evidence-invalidating.
Source code in src/symfonic/services/switching/control_plane.py
| async def revoke_freeze(
self,
actor: Actor,
bundle_id: str,
*,
expected_epoch: int | None = None,
reason: str,
) -> BundleRecord:
"""SCP-REV — release-owner-only, audited, evidence-invalidating."""
record = await self._store.read(bundle_id)
state = record.freeze_state
if not state.active or state.frozen_legacy_vector is None:
raise FreezeViolationError(
f"bundle {bundle_id!r} carries no active freeze epoch to revoke."
)
stored = await self._mutator.swap(
actor,
Action.REVOKE_FREEZE,
bundle_id,
vector=state.frozen_legacy_vector,
expected_epoch=expected_epoch,
reason=reason,
freeze_state=NO_FREEZE,
bypass_freeze=True,
)
if self._evidence is not None:
self._evidence.invalidate_epoch(
state.freeze_epoch_id or "", reason=f"freeze revoked: {reason}"
)
return stored
|
rollback
async
rollback(actor: Actor, bundle_id: str, *, expected_epoch: int | None = None, reason: str = 'rollback') -> BundleRecord
CUT-RB-6 — rollback is a CAS to the named vector, never an edit.
Source code in src/symfonic/services/switching/control_plane.py
| async def rollback(
self,
actor: Actor,
bundle_id: str,
*,
expected_epoch: int | None = None,
reason: str = "rollback",
) -> BundleRecord:
"""CUT-RB-6 — rollback is a CAS to the *named* vector, never an edit."""
record = await self._store.read(bundle_id)
return await self._mutator.swap(
actor,
Action.ROLLBACK,
bundle_id,
vector=record.rollback_vector or record.generation_vector,
expected_epoch=expected_epoch,
reason=reason,
)
|