"""Isolated public signing-key registry and guarded rotation operations.""" from __future__ import annotations import base64 from datetime import datetime, timedelta, timezone import hashlib import hmac import re from typing import Callable, Mapping, Optional, Sequence from sqlalchemy import select from sqlalchemy.orm import Session from licensing_shared.canonical_json import canonicalize_json from licensing_shared.constants import validate_identifier from licensing_shared.models import format_rfc3339 from licensing_shared.models import SignedLicenseDocument from licensing_shared.verifier import ED25519_PUBLIC_KEY_BYTES from .constants import ( DEFAULT_IDEMPOTENCY_DAYS, MAX_IDEMPOTENCY_KEY_CHARACTERS, MAX_REASON_CHARACTERS, SIGNING_KEY_CHANGE_REASONS, SIGNING_KEY_COMPROMISE_REASONS, ) from .errors import ServerErrorCode, ServerLicensingError from .models import ( AuditEvent, IdempotencyRecord, SigningKeyMetadata, User, ) from .security import SnapshotSigner, new_identifier SIGNING_KEY_PURPOSE = "license_snapshot" SIGNING_KEY_REGISTRATION_SCOPE = "admin.signing_key.register" SIGNING_KEY_ACTIVATION_SCOPE = "admin.signing_key.activate" SIGNING_KEY_COMPROMISE_SCOPE = "admin.signing_key.compromise" SIGNING_KEY_STATUSES = frozenset(("staged", "active", "retired", "compromised")) SIGNING_KEY_STATE_DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$") def _utc(value: datetime) -> datetime: if not isinstance(value, datetime): raise TypeError("value must be a datetime") if value.tzinfo is None or value.utcoffset() is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def _optional_utc(value: Optional[datetime], field_name: str) -> Optional[datetime]: if value is None: return None if not isinstance(field_name, str) or not field_name: raise ValueError("field_name must be a non-empty string") try: return _utc(value) except TypeError as exc: raise TypeError(f"{field_name} must be a datetime or None") from exc def _bounded_text(value: str, field_name: str, maximum: int) -> str: if not isinstance(value, str): raise TypeError(f"{field_name} must be a string") if isinstance(maximum, bool) or not isinstance(maximum, int) or maximum < 1: raise ValueError("maximum must be a positive integer") normalized = value.strip() if not normalized or len(normalized) > maximum: raise ValueError(f"{field_name} is empty or too long") return normalized def _optional_note(value: Optional[str]) -> Optional[str]: if value is None: return None return _bounded_text(value, "note", MAX_REASON_CHARACTERS) def _reason_code(value: str) -> str: normalized = validate_identifier(value, "reason_code") if normalized not in SIGNING_KEY_CHANGE_REASONS: raise ValueError("reason_code is not allowed for signing-key administration") return normalized def _compromise_reason_code(value: str) -> str: normalized = validate_identifier(value, "reason_code") if normalized not in SIGNING_KEY_COMPROMISE_REASONS: raise ValueError("reason_code is not allowed for signing-key compromise") return normalized def _signer_material(signer: SnapshotSigner) -> tuple[str, bytes, str]: if ( not all( callable(getattr(signer, member, None)) for member in ("public_key_bytes", "sign_payload") ) or not isinstance(getattr(signer, "key_id", None), str) ): raise TypeError("signer does not implement SnapshotSigner") key_id = validate_identifier(signer.key_id, "signer key_id") public_key = signer.public_key_bytes() if not isinstance(public_key, bytes): raise TypeError("signer public key must be bytes") if len(public_key) != ED25519_PUBLIC_KEY_BYTES: raise ValueError("signer public key is not a raw Ed25519 public key") return key_id, public_key, hashlib.sha256(public_key).hexdigest() def _time_mapping(value: Optional[datetime], field_name: str) -> Optional[str]: if value is None: return None return format_rfc3339(_utc(value), field_name) def _key_mapping(row: SigningKeyMetadata) -> dict[str, object]: if not isinstance(row, SigningKeyMetadata): raise TypeError("row must be SigningKeyMetadata") public_key = bytes(row.public_key_bytes) return { "keyId": row.key_id, "purpose": row.purpose, "publicKey": base64.b64encode(public_key).decode("ascii"), "publicKeySha256": hashlib.sha256(public_key).hexdigest(), "status": row.status, "notBefore": _time_mapping(row.not_before, "not_before"), "expiresAt": _time_mapping(row.expires_at, "expires_at"), "activatedAt": _time_mapping(row.activated_at, "activated_at"), "retiredAt": _time_mapping(row.retired_at, "retired_at"), } def _registry_status_from_rows( rows: Sequence[SigningKeyMetadata], signer: SnapshotSigner, now: datetime, ) -> dict[str, object]: if isinstance(rows, (str, bytes)) or not isinstance(rows, Sequence): raise TypeError("rows must be a sequence of SigningKeyMetadata values") normalized_now = _utc(now) signer_key_id, signer_public_key, signer_fingerprint = _signer_material(signer) entries: list[dict[str, object]] = [] issues: list[str] = [] current_row: Optional[SigningKeyMetadata] = None active_rows: list[SigningKeyMetadata] = [] seen_key_ids: set[str] = set() for row in rows: if not isinstance(row, SigningKeyMetadata): raise TypeError("rows must contain SigningKeyMetadata values") if row.key_id in seen_key_ids: issues.append("duplicate_key_id") seen_key_ids.add(row.key_id) entries.append(_key_mapping(row)) if row.purpose != SIGNING_KEY_PURPOSE: issues.append("unexpected_key_purpose") if row.status not in SIGNING_KEY_STATUSES: issues.append("invalid_key_status") public_key = bytes(row.public_key_bytes) if len(public_key) != ED25519_PUBLIC_KEY_BYTES: issues.append("invalid_public_key_length") not_before = _utc(row.not_before) expires_at = _optional_utc(row.expires_at, "expires_at") activated_at = _optional_utc(row.activated_at, "activated_at") retired_at = _optional_utc(row.retired_at, "retired_at") if ( row.status != "compromised" and expires_at is not None and expires_at <= not_before ): issues.append("invalid_key_validity_window") if row.status == "staged" and ( activated_at is not None or retired_at is not None ): issues.append("invalid_staged_lifecycle") elif row.status == "active" and ( activated_at is None or retired_at is not None ): issues.append("invalid_active_lifecycle") elif row.status in ("retired", "compromised") and ( retired_at is None or expires_at is None or expires_at > retired_at ): issues.append("invalid_retired_lifecycle") if row.status == "active": active_rows.append(row) if normalized_now < not_before: issues.append("active_key_not_yet_valid") if expires_at is not None and normalized_now >= expires_at: issues.append("active_key_expired") if row.key_id == signer_key_id: current_row = row if len(active_rows) != 1: issues.append("active_key_count_invalid") if current_row is None: issues.append("configured_key_not_registered") else: if not hmac.compare_digest(bytes(current_row.public_key_bytes), signer_public_key): issues.append("configured_key_material_mismatch") if current_row.status != "active": issues.append("configured_key_not_active") if normalized_now < _utc(current_row.not_before): issues.append("configured_key_not_yet_valid") current_expiry = _optional_utc(current_row.expires_at, "expires_at") if current_expiry is not None and normalized_now >= current_expiry: issues.append("configured_key_expired") if active_rows and active_rows[0].key_id != signer_key_id: issues.append("active_key_does_not_match_signer") issues = sorted(set(issues)) can_activate = False if current_row is not None and current_row.status == "staged": current_expiry = _optional_utc(current_row.expires_at, "expires_at") can_activate = ( len(active_rows) <= 1 and hmac.compare_digest(bytes(current_row.public_key_bytes), signer_public_key) and _utc(current_row.not_before) <= normalized_now and (current_expiry is None or normalized_now < current_expiry) and all( issue in ( "active_key_count_invalid", "configured_key_not_active", "active_key_does_not_match_signer", ) for issue in issues ) ) state = { "purpose": SIGNING_KEY_PURPOSE, "configuredKeyId": signer_key_id, "configuredPublicKeySha256": signer_fingerprint, "keys": entries, } return { **state, "activeKeyId": active_rows[0].key_id if len(active_rows) == 1 else None, "configuredKeyStatus": None if current_row is None else current_row.status, "issuanceReady": not issues, "canActivateConfiguredKey": can_activate, "issues": issues, "stateDigest": hashlib.sha256(canonicalize_json(state)).hexdigest(), } def inspect_signing_key_registry( session: Session, signer: SnapshotSigner, now: datetime, ) -> dict[str, object]: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") normalized_now = _utc(now) rows = tuple( session.scalars( select(SigningKeyMetadata) .where(SigningKeyMetadata.purpose == SIGNING_KEY_PURPOSE) .order_by(SigningKeyMetadata.key_id) ).all() ) return _registry_status_from_rows(rows, signer, normalized_now) class RegistryBoundSnapshotSigner: """Fail closed at the signing call if registry state no longer matches.""" def __init__( self, session: Session, signer: SnapshotSigner, *, now_factory: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") _signer_material(signer) if not callable(now_factory): raise TypeError("now_factory must be callable") self._session = session self._signer = signer self._now_factory = now_factory @property def key_id(self) -> str: return self._signer.key_id def public_key_bytes(self) -> bytes: return self._signer.public_key_bytes() def sign_payload(self, payload: Mapping[str, object]) -> SignedLicenseDocument: if not isinstance(payload, Mapping): raise TypeError("payload must be a mapping") status = inspect_signing_key_registry( self._session, self._signer, _utc(self._now_factory()), ) if status["issuanceReady"] is not True: raise ServerLicensingError( ServerErrorCode.SIGNING_UNAVAILABLE, "configured signing key is not active in the public metadata registry", status_code=503, retryable=True, ) return self._signer.sign_payload(payload) class SigningKeyAdministrationService: def __init__( self, session: Session, signer: SnapshotSigner, *, now_factory: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") _signer_material(signer) if not callable(now_factory): raise TypeError("now_factory must be callable") self._session = session self._signer = signer self._now_factory = now_factory def _now(self) -> datetime: return _utc(self._now_factory()) def status(self) -> dict[str, object]: with self._session.begin(): return inspect_signing_key_registry( self._session, self._signer, self._now(), ) def register_current_key( self, not_before: datetime, actor_user_id: str, reason_code: str, idempotency_key: str, correlation_id: str, *, expires_at: Optional[datetime] = None, note: Optional[str] = None, ) -> dict[str, object]: normalized_not_before = _utc(not_before) normalized_expiry = _optional_utc(expires_at, "expires_at") if normalized_expiry is not None and normalized_expiry <= normalized_not_before: raise ValueError("expires_at must be later than not_before") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _reason_code(reason_code) normalized_note = _optional_note(note) normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") signer_key_id, signer_public_key, signer_fingerprint = _signer_material( self._signer ) request = { "keyId": signer_key_id, "publicKeySha256": signer_fingerprint, "notBefore": _time_mapping(normalized_not_before, "not_before"), "expiresAt": _time_mapping(normalized_expiry, "expires_at"), "actorUserId": normalized_actor, "reasonCode": normalized_reason, "note": normalized_note, "correlationId": normalized_correlation, } request_digest = hashlib.sha256(canonicalize_json(request)).digest() now = self._now() with self._session.begin(): actor = self._require_server_admin(normalized_actor) existing = self._session.scalar( select(SigningKeyMetadata) .where(SigningKeyMetadata.key_id == signer_key_id) .with_for_update() ) replay = self._load_idempotent_response( SIGNING_KEY_REGISTRATION_SCOPE, signer_key_id, normalized_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result created = existing is None if existing is None: existing = SigningKeyMetadata( id=new_identifier("signing_key"), key_id=signer_key_id, purpose=SIGNING_KEY_PURPOSE, public_key_bytes=signer_public_key, status="staged", not_before=normalized_not_before, expires_at=normalized_expiry, activated_at=None, retired_at=None, ) self._session.add(existing) self._session.flush() elif not ( existing.purpose == SIGNING_KEY_PURPOSE and existing.status == "staged" and hmac.compare_digest( bytes(existing.public_key_bytes), signer_public_key, ) and _utc(existing.not_before) == normalized_not_before and _optional_utc(existing.expires_at, "expires_at") == normalized_expiry and existing.activated_at is None and existing.retired_at is None ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "signing key ID already has different public metadata or lifecycle state", status_code=409, ) result = { **_key_mapping(existing), "registered": created, "idempotentReplay": False, "correlationId": normalized_correlation, } self._store_idempotent_response( SIGNING_KEY_REGISTRATION_SCOPE, signer_key_id, normalized_key, request_digest, result, now, ) if created: self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=actor.id, action="signing_key.public_metadata_registered", target_type="signing_key", target_id=signer_key_id, reason=normalized_note or normalized_reason, correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "expiresAt": result["expiresAt"], "notBefore": result["notBefore"], "publicKeySha256": signer_fingerprint, "purpose": SIGNING_KEY_PURPOSE, "reasonCode": normalized_reason, "status": "staged", }, ) ) return result def preview_current_key_activation( self, actor_user_id: str, reason_code: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _reason_code(reason_code) _optional_note(note) normalized_correlation = validate_identifier(correlation_id, "correlation_id") with self._session.begin(): self._require_server_admin(normalized_actor) result = inspect_signing_key_registry( self._session, self._signer, self._now(), ) result["reasonCode"] = normalized_reason result["correlationId"] = normalized_correlation return result def activate_current_key( self, actor_user_id: str, reason_code: str, expected_state_digest: str, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _reason_code(reason_code) normalized_note = _optional_note(note) normalized_state_digest = _bounded_text( expected_state_digest, "expected_state_digest", 64, ) if SIGNING_KEY_STATE_DIGEST_PATTERN.fullmatch(normalized_state_digest) is None: raise ValueError( "expected_state_digest must be 64 lowercase hexadecimal characters" ) normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") signer_key_id, _signer_public_key, signer_fingerprint = _signer_material( self._signer ) request_digest = hashlib.sha256( canonicalize_json( { "actorUserId": normalized_actor, "reasonCode": normalized_reason, "note": normalized_note, "expectedStateDigest": normalized_state_digest, "configuredKeyId": signer_key_id, "configuredPublicKeySha256": signer_fingerprint, "correlationId": normalized_correlation, } ) ).digest() now = self._now() with self._session.begin(): actor = self._require_server_admin(normalized_actor) rows = self._rows(lock=True) replay = self._load_idempotent_response( SIGNING_KEY_ACTIVATION_SCOPE, SIGNING_KEY_PURPOSE, normalized_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result preview = _registry_status_from_rows(rows, self._signer, now) if not hmac.compare_digest( str(preview["stateDigest"]), normalized_state_digest, ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "signing-key registry changed after preview; preview again", status_code=409, ) if preview["canActivateConfiguredKey"] is not True: raise ServerLicensingError( ServerErrorCode.CONFLICT, "configured signing key is not a valid staged activation candidate", status_code=409, ) current_row = next(row for row in rows if row.key_id == signer_key_id) active_row = next((row for row in rows if row.status == "active"), None) previous_active_key_id = None if active_row is None else active_row.key_id if active_row is not None: active_row.status = "retired" active_row.expires_at = now active_row.retired_at = now self._session.flush() current_row.status = "active" current_row.activated_at = now current_row.retired_at = None self._session.flush() status = _registry_status_from_rows(rows, self._signer, now) if status["issuanceReady"] is not True: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "signing-key activation did not produce a ready registry", status_code=500, ) result = { **status, "previousActiveKeyId": previous_active_key_id, "activatedKeyId": signer_key_id, "previousStateDigest": normalized_state_digest, "reasonCode": normalized_reason, "correlationId": normalized_correlation, "executed": True, "idempotentReplay": False, } self._store_idempotent_response( SIGNING_KEY_ACTIVATION_SCOPE, SIGNING_KEY_PURPOSE, normalized_key, request_digest, result, now, ) self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=actor.id, action="signing_key.activated", target_type="signing_key", target_id=signer_key_id, reason=normalized_note or normalized_reason, correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "previousActiveKeyId": previous_active_key_id, "publicKeySha256": signer_fingerprint, "purpose": SIGNING_KEY_PURPOSE, "reasonCode": normalized_reason, "stateDigest": status["stateDigest"], }, ) ) return result def preview_key_compromise( self, key_id: str, actor_user_id: str, reason_code: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_key_id = validate_identifier(key_id, "key_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _compromise_reason_code(reason_code) _optional_note(note) normalized_correlation = validate_identifier(correlation_id, "correlation_id") with self._session.begin(): self._require_server_admin(normalized_actor) rows = self._rows(lock=False) result = self._compromise_preview( rows, normalized_key_id, self._now(), ) result["reasonCode"] = normalized_reason result["correlationId"] = normalized_correlation return result def compromise_key( self, key_id: str, actor_user_id: str, reason_code: str, expected_state_digest: str, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_key_id = validate_identifier(key_id, "key_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _compromise_reason_code(reason_code) normalized_note = _optional_note(note) normalized_state_digest = _bounded_text( expected_state_digest, "expected_state_digest", 64, ) if SIGNING_KEY_STATE_DIGEST_PATTERN.fullmatch(normalized_state_digest) is None: raise ValueError( "expected_state_digest must be 64 lowercase hexadecimal characters" ) normalized_idempotency_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = hashlib.sha256( canonicalize_json( { "keyId": normalized_key_id, "actorUserId": normalized_actor, "reasonCode": normalized_reason, "note": normalized_note, "expectedStateDigest": normalized_state_digest, "correlationId": normalized_correlation, } ) ).digest() now = self._now() with self._session.begin(): actor = self._require_server_admin(normalized_actor) rows = self._rows(lock=True) replay = self._load_idempotent_response( SIGNING_KEY_COMPROMISE_SCOPE, normalized_key_id, normalized_idempotency_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result preview = self._compromise_preview(rows, normalized_key_id, now) if not hmac.compare_digest( str(preview["stateDigest"]), normalized_state_digest, ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "signing-key registry changed after preview; preview again", status_code=409, ) if preview["canCompromise"] is not True: raise ServerLicensingError( ServerErrorCode.CONFLICT, "signing key is already marked compromised", status_code=409, ) target = next(row for row in rows if row.key_id == normalized_key_id) previous_status = target.status issuance_stopped = previous_status == "active" previous_active_count = sum(row.status == "active" for row in rows) target.status = "compromised" if previous_status in ("staged", "active"): target.expires_at = now target.retired_at = now self._session.flush() status = _registry_status_from_rows(rows, self._signer, now) active_count = sum(row.status == "active" for row in rows) if active_count != previous_active_count - int(issuance_stopped): raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "signing-key compromise produced an invalid active-key state", status_code=500, ) result = { **status, "targetKeyId": normalized_key_id, "previousStatus": previous_status, "compromisedStatus": target.status, "issuanceStopped": issuance_stopped, "previousStateDigest": normalized_state_digest, "reasonCode": normalized_reason, "correlationId": normalized_correlation, "executed": True, "idempotentReplay": False, } self._store_idempotent_response( SIGNING_KEY_COMPROMISE_SCOPE, normalized_key_id, normalized_idempotency_key, request_digest, result, now, ) self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=actor.id, action="signing_key.compromised", target_type="signing_key", target_id=normalized_key_id, reason=normalized_note or normalized_reason, correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "previousStatus": previous_status, "purpose": SIGNING_KEY_PURPOSE, "reasonCode": normalized_reason, "issuanceStopped": issuance_stopped, "stateDigest": status["stateDigest"], }, ) ) return result def _compromise_preview( self, rows: Sequence[SigningKeyMetadata], key_id: str, now: datetime, ) -> dict[str, object]: if isinstance(rows, (str, bytes)) or not isinstance(rows, Sequence): raise TypeError("rows must be a sequence of SigningKeyMetadata values") normalized_key_id = validate_identifier(key_id, "key_id") normalized_now = _utc(now) status = _registry_status_from_rows(rows, self._signer, normalized_now) target = next((row for row in rows if row.key_id == normalized_key_id), None) if target is None: raise ServerLicensingError( ServerErrorCode.CONFLICT, "signing key was not found in the public metadata registry", status_code=409, ) return { **status, "targetKeyId": normalized_key_id, "targetStatus": target.status, "canCompromise": target.status != "compromised", "wouldStopIssuance": target.status == "active", } def _require_server_admin(self, actor_user_id: str) -> User: actor = self._session.get(User, actor_user_id) if actor is None or actor.status != "active" or not actor.is_server_admin: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "signing-key administration requires an active administrator", status_code=403, ) return actor def _rows(self, *, lock: bool) -> tuple[SigningKeyMetadata, ...]: if not isinstance(lock, bool): raise TypeError("lock must be a Boolean") query = ( select(SigningKeyMetadata) .where(SigningKeyMetadata.purpose == SIGNING_KEY_PURPOSE) .order_by(SigningKeyMetadata.key_id) ) if lock: query = query.with_for_update() return tuple(self._session.scalars(query).all()) def _load_idempotent_response( self, scope: str, subject_key: str, idempotency_key: str, request_digest: bytes, now: datetime, ) -> Optional[dict[str, object]]: normalized_scope = validate_identifier(scope, "scope") normalized_subject = validate_identifier(subject_key, "subject_key") if not isinstance(request_digest, bytes) or len(request_digest) != 32: raise ValueError("request_digest must contain 32 bytes") normalized_now = _utc(now) record = self._session.scalar( select(IdempotencyRecord) .where( IdempotencyRecord.scope == normalized_scope, IdempotencyRecord.subject_key == normalized_subject, IdempotencyRecord.idempotency_key == idempotency_key, ) .with_for_update() ) if record is None: return None if _utc(record.expires_at) <= normalized_now: self._session.delete(record) self._session.flush() return None if not hmac.compare_digest(record.request_digest, request_digest): raise ServerLicensingError( ServerErrorCode.IDEMPOTENCY_CONFLICT, "idempotency key was already used for a different request", status_code=409, ) return record.response_json def _store_idempotent_response( self, scope: str, subject_key: str, idempotency_key: str, request_digest: bytes, response: Mapping[str, object], now: datetime, ) -> None: normalized_scope = validate_identifier(scope, "scope") normalized_subject = validate_identifier(subject_key, "subject_key") if not isinstance(request_digest, bytes) or len(request_digest) != 32: raise ValueError("request_digest must contain 32 bytes") if not isinstance(response, Mapping): raise TypeError("response must be a mapping") normalized_now = _utc(now) self._session.add( IdempotencyRecord( id=new_identifier("idempotency"), scope=normalized_scope, subject_key=normalized_subject, idempotency_key=idempotency_key, request_digest=request_digest, response_status=200, response_json=dict(response), expires_at=normalized_now + timedelta(days=DEFAULT_IDEMPOTENCY_DAYS), ) ) __all__ = [ "RegistryBoundSnapshotSigner", "SIGNING_KEY_ACTIVATION_SCOPE", "SIGNING_KEY_COMPROMISE_SCOPE", "SIGNING_KEY_PURPOSE", "SIGNING_KEY_REGISTRATION_SCOPE", "SigningKeyAdministrationService", "inspect_signing_key_registry", ]