"""Immutable catalog candidate registration and guarded publication.""" from __future__ import annotations 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.catalog import LicensingCatalog, SkuDefinition from licensing_shared.constants import MAX_CATALOG_DOCUMENT_BYTES, PRODUCT_ID, validate_identifier from licensing_shared.models import SignedLicenseDocument, format_rfc3339 from .constants import ( CATALOG_RELEASE_REASONS, DEFAULT_IDEMPOTENCY_DAYS, MAX_IDEMPOTENCY_KEY_CHARACTERS, MAX_REASON_CHARACTERS, ) from .errors import ServerErrorCode, ServerLicensingError from .models import AuditEvent, CatalogRelease, IdempotencyRecord, User from .security import SnapshotSigner, new_identifier CATALOG_STAGE_SCOPE = "admin.catalog.stage" CATALOG_PUBLICATION_SCOPE = "admin.catalog.publish" CATALOG_RELEASE_STATUSES = frozenset(("staged", "active", "retired")) CATALOG_STATE_DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$") MAXIMUM_CATALOG_CHANGE_IDENTIFIERS = 256 MAXIMUM_CATALOG_RELEASE_HISTORY = 512 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 CATALOG_RELEASE_REASONS: raise ValueError("reason_code is not allowed for catalog administration") return normalized def _catalog_candidate( document: Mapping[str, object], ) -> tuple[LicensingCatalog, dict[str, object], bytes]: if not isinstance(document, Mapping): raise TypeError("catalog document must be a mapping") catalog = LicensingCatalog.from_mapping(document) if catalog.product_id != PRODUCT_ID: raise ValueError("catalog product ID does not match the application product") normalized = catalog.to_mapping() body = canonicalize_json(normalized) if len(body) > MAX_CATALOG_DOCUMENT_BYTES: raise ValueError("catalog document exceeds the maximum size") return catalog, normalized, hashlib.sha256(body).digest() 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 _release_mapping(row: CatalogRelease) -> dict[str, object]: if not isinstance(row, CatalogRelease): raise TypeError("row must be CatalogRelease") return { "releaseId": row.id, "productId": row.product_id, "revision": row.revision, "catalogSha256": bytes(row.document_digest).hex(), "status": row.status, "stagedAt": _time_mapping(row.staged_at, "staged_at"), "publishedAt": _time_mapping(row.published_at, "published_at"), "retiredAt": _time_mapping(row.retired_at, "retired_at"), } def _change_values(values: set[str]) -> dict[str, object]: identifiers = sorted(values) returned = identifiers[:MAXIMUM_CATALOG_CHANGE_IDENTIFIERS] return { "count": len(identifiers), "ids": returned, "truncated": len(returned) != len(identifiers), } def _sku_contract(definition: SkuDefinition) -> dict[str, object]: if not isinstance(definition, SkuDefinition): raise TypeError("definition must be SkuDefinition") mapping = definition.to_mapping() mapping.pop("active", None) mapping.pop("label", None) return mapping def _device_policy_contract_is_compatible( active: LicensingCatalog, candidate: LicensingCatalog, identifier: str, ) -> bool: if not isinstance(active, LicensingCatalog): raise TypeError("active must be LicensingCatalog") if not isinstance(candidate, LicensingCatalog): raise TypeError("candidate must be LicensingCatalog") normalized_identifier = validate_identifier(identifier, "identifier") active_mapping = active.device_policies[normalized_identifier].to_mapping() candidate_mapping = candidate.device_policies[normalized_identifier].to_mapping() if active.schema_version < candidate.schema_version: candidate_mapping.pop("connectedRefreshHours", None) candidate_mapping.pop("offlineRefreshReminderDays", None) return active_mapping == candidate_mapping def _catalog_changes( active: Optional[LicensingCatalog], candidate: LicensingCatalog, ) -> dict[str, object]: if active is not None and not isinstance(active, LicensingCatalog): raise TypeError("active must be LicensingCatalog or None") if not isinstance(candidate, LicensingCatalog): raise TypeError("candidate must be LicensingCatalog") active_entitlements = set() if active is None else set(active.entitlements) active_policies = set() if active is None else set(active.device_policies) active_skus = set() if active is None else set(active.skus) candidate_entitlements = set(candidate.entitlements) candidate_policies = set(candidate.device_policies) candidate_skus = set(candidate.skus) changed_entitlement_metadata = set() changed_sku_labels = set() activated_skus = set() deactivated_skus = set() if active is not None: changed_entitlement_metadata = { identifier for identifier in active_entitlements.intersection(candidate_entitlements) if active.entitlements[identifier].to_mapping() != candidate.entitlements[identifier].to_mapping() } changed_sku_labels = { identifier for identifier in active_skus.intersection(candidate_skus) if active.skus[identifier].label != candidate.skus[identifier].label } activated_skus = { identifier for identifier in active_skus.intersection(candidate_skus) if not active.skus[identifier].active and candidate.skus[identifier].active } deactivated_skus = { identifier for identifier in active_skus.intersection(candidate_skus) if active.skus[identifier].active and not candidate.skus[identifier].active } return { "addedEntitlements": _change_values( candidate_entitlements.difference(active_entitlements) ), "removedEntitlements": _change_values( active_entitlements.difference(candidate_entitlements) ), "changedEntitlementMetadata": _change_values(changed_entitlement_metadata), "addedDevicePolicies": _change_values( candidate_policies.difference(active_policies) ), "removedDevicePolicies": _change_values( active_policies.difference(candidate_policies) ), "addedSkus": _change_values(candidate_skus.difference(active_skus)), "removedSkus": _change_values(active_skus.difference(candidate_skus)), "changedSkuLabels": _change_values(changed_sku_labels), "activatedSkus": _change_values(activated_skus), "deactivatedSkus": _change_values(deactivated_skus), } def _compatibility_issues( active: Optional[LicensingCatalog], candidate: LicensingCatalog, ) -> list[str]: if active is None: return [] issues = [] if candidate.product_id != active.product_id: issues.append("product_id_changed") if candidate.revision != active.revision + 1: issues.append("revision_not_next") if candidate.schema_version < active.schema_version: issues.append("schema_version_rollback") elif candidate.schema_version > active.schema_version + 1: issues.append("schema_version_skipped") if not set(active.entitlements).issubset(candidate.entitlements): issues.append("stable_entitlement_removed") if not set(active.device_policies).issubset(candidate.device_policies): issues.append("stable_device_policy_removed") if not set(active.skus).issubset(candidate.skus): issues.append("stable_sku_removed") for identifier in set(active.device_policies).intersection(candidate.device_policies): if not _device_policy_contract_is_compatible(active, candidate, identifier): issues.append("existing_device_policy_changed") break for identifier in set(active.skus).intersection(candidate.skus): if _sku_contract(active.skus[identifier]) != _sku_contract( candidate.skus[identifier] ): issues.append("existing_sku_contract_changed") break return sorted(set(issues)) def _validated_release_catalog(row: CatalogRelease) -> LicensingCatalog: if row.product_id != PRODUCT_ID: raise ValueError("catalog release product ID is invalid") if row.status not in CATALOG_RELEASE_STATUSES: raise ValueError("catalog release status is invalid") if not isinstance(row.revision, int) or isinstance(row.revision, bool) or row.revision < 1: raise ValueError("catalog release revision is invalid") catalog, normalized, digest = _catalog_candidate(row.document_json) if catalog.product_id != row.product_id or catalog.revision != row.revision: raise ValueError("catalog release identity does not match its document") if normalized != row.document_json: raise ValueError("catalog release document is not normalized") if not hmac.compare_digest(bytes(row.document_digest), digest): raise ValueError("catalog release digest does not match its document") published_at = _optional_utc(row.published_at, "published_at") retired_at = _optional_utc(row.retired_at, "retired_at") if row.status == "staged" and any( value is not None for value in (row.published_by_user_id, published_at, retired_at) ): raise ValueError("staged catalog release has publication metadata") if row.status == "active" and ( row.published_by_user_id is None or published_at is None or retired_at is not None ): raise ValueError("active catalog release lifecycle is invalid") if row.status == "retired" and ( row.published_by_user_id is None or published_at is None or retired_at is None ): raise ValueError("retired catalog release lifecycle is invalid") return catalog def _registry_state( rows: Sequence[CatalogRelease], runtime_catalog: LicensingCatalog, ) -> tuple[dict[str, object], list[str], Optional[CatalogRelease]]: if isinstance(rows, (str, bytes)) or not isinstance(rows, Sequence): raise TypeError("rows must be a sequence of CatalogRelease values") if not isinstance(runtime_catalog, LicensingCatalog): raise TypeError("runtime_catalog must be LicensingCatalog") if len(rows) > MAXIMUM_CATALOG_RELEASE_HISTORY: raise ValueError("catalog release history exceeds the supported bound") issues = [] active_rows = [] releases = [] seen_revisions = set() for row in rows: if not isinstance(row, CatalogRelease): raise TypeError("rows must contain CatalogRelease values") try: _validated_release_catalog(row) except (TypeError, ValueError): issues.append("invalid_catalog_release_record") if row.revision in seen_revisions: issues.append("duplicate_catalog_revision") seen_revisions.add(row.revision) releases.append(_release_mapping(row)) if row.status == "active": active_rows.append(row) if len(active_rows) != 1: issues.append("active_catalog_count_invalid") runtime_digest = runtime_catalog.sha256() active_row = active_rows[0] if len(active_rows) == 1 else None if active_row is not None: if active_row.revision != runtime_catalog.revision: issues.append("active_catalog_revision_mismatch") if not hmac.compare_digest(bytes(active_row.document_digest).hex(), runtime_digest): issues.append("active_catalog_digest_mismatch") state = { "productId": runtime_catalog.product_id, "runtimeRevision": runtime_catalog.revision, "runtimeCatalogSha256": runtime_digest, "releases": releases, } return state, sorted(set(issues)), active_row def inspect_catalog_registry( session: Session, runtime_catalog: LicensingCatalog, ) -> dict[str, object]: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") if not isinstance(runtime_catalog, LicensingCatalog): raise TypeError("runtime_catalog must be LicensingCatalog") rows = tuple( session.scalars( select(CatalogRelease) .where(CatalogRelease.product_id == runtime_catalog.product_id) .order_by(CatalogRelease.revision) ).all() ) state, issues, active_row = _registry_state(rows, runtime_catalog) return { **state, "activeRevision": None if active_row is None else active_row.revision, "activeCatalogSha256": ( None if active_row is None else bytes(active_row.document_digest).hex() ), "catalogReady": not issues, "issues": issues, "stateDigest": hashlib.sha256(canonicalize_json(state)).hexdigest(), } class CatalogBoundSnapshotSigner: """Refuse issuance if the runtime catalog is not the reviewed publication.""" def __init__( self, session: Session, catalog: LicensingCatalog, signer: SnapshotSigner, ) -> None: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") if not isinstance(catalog, LicensingCatalog): raise TypeError("catalog must be LicensingCatalog") 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") self._session = session self._catalog = catalog self._signer = signer @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_catalog_registry(self._session, self._catalog) if status["catalogReady"] is not True: raise ServerLicensingError( ServerErrorCode.CATALOG_UNAVAILABLE, "runtime catalog is not the active reviewed catalog publication", status_code=503, retryable=True, ) return self._signer.sign_payload(payload) class CatalogAdministrationService: def __init__( self, session: Session, runtime_catalog: LicensingCatalog, *, now_factory: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") if not isinstance(runtime_catalog, LicensingCatalog): raise TypeError("runtime_catalog must be LicensingCatalog") if not callable(now_factory): raise TypeError("now_factory must be callable") self._session = session self._runtime_catalog = runtime_catalog 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_catalog_registry(self._session, self._runtime_catalog) def stage_candidate( self, document: Mapping[str, object], actor_user_id: str, reason_code: str, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: candidate, normalized_document, candidate_digest = _catalog_candidate(document) 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") request_digest = hashlib.sha256( canonicalize_json( { "actorUserId": normalized_actor, "catalogSha256": candidate_digest.hex(), "correlationId": normalized_correlation, "note": normalized_note, "productId": candidate.product_id, "reasonCode": normalized_reason, "revision": candidate.revision, } ) ).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( CATALOG_STAGE_SCOPE, f"{candidate.product_id}.{candidate.revision}", normalized_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result existing = next( (row for row in rows if row.revision == candidate.revision), None, ) active_row = next((row for row in rows if row.status == "active"), None) active_catalog = ( None if active_row is None else _validated_release_catalog(active_row) ) issues = _compatibility_issues(active_catalog, candidate) if issues: raise ServerLicensingError( ServerErrorCode.CONFLICT, "catalog candidate is not append-only compatible: " + ", ".join(issues), status_code=409, ) created = existing is None if existing is None: existing = CatalogRelease( id=new_identifier("catalog_release"), product_id=candidate.product_id, revision=candidate.revision, document_json=normalized_document, document_digest=candidate_digest, status="staged", staged_by_user_id=actor.id, staged_at=now, published_by_user_id=None, published_at=None, retired_at=None, ) self._session.add(existing) self._session.flush() elif not ( existing.status == "staged" and existing.product_id == candidate.product_id and hmac.compare_digest(bytes(existing.document_digest), candidate_digest) and existing.document_json == normalized_document ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "catalog revision already has different content or lifecycle state", status_code=409, ) result = { **_release_mapping(existing), "changes": _catalog_changes(active_catalog, candidate), "registered": created, "idempotentReplay": False, "correlationId": normalized_correlation, } self._store_idempotent_response( CATALOG_STAGE_SCOPE, f"{candidate.product_id}.{candidate.revision}", normalized_key, request_digest, result, now, ) if created: self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=actor.id, action="catalog.release_staged", target_type="catalog_release", target_id=existing.id, reason=normalized_note or normalized_reason, correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "catalogRevision": candidate.revision, "catalogSha256": candidate_digest.hex(), "entitlementCount": len(candidate.entitlements), "reasonCode": normalized_reason, "skuCount": len(candidate.skus), }, ) ) return result def preview_publication( self, revision: int, actor_user_id: str, reason_code: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_revision = self._revision(revision) 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 = self._publication_preview( self._rows(lock=False), normalized_revision, ) result["reasonCode"] = normalized_reason result["correlationId"] = normalized_correlation return result def publish_candidate( self, revision: int, 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_revision = self._revision(revision) 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 CATALOG_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") request_digest = hashlib.sha256( canonicalize_json( { "actorUserId": normalized_actor, "correlationId": normalized_correlation, "expectedStateDigest": normalized_state_digest, "note": normalized_note, "productId": self._runtime_catalog.product_id, "reasonCode": normalized_reason, "revision": normalized_revision, } ) ).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( CATALOG_PUBLICATION_SCOPE, self._runtime_catalog.product_id, normalized_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result preview = self._publication_preview(rows, normalized_revision) if not hmac.compare_digest( str(preview["stateDigest"]), normalized_state_digest, ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "catalog registry changed after preview; preview again", status_code=409, ) if preview["canPublish"] is not True: raise ServerLicensingError( ServerErrorCode.CONFLICT, "catalog candidate is not publishable: " + ", ".join(str(value) for value in preview["issues"]), status_code=409, ) candidate_row = next(row for row in rows if row.revision == normalized_revision) active_row = next((row for row in rows if row.status == "active"), None) previous_revision = None if active_row is None else active_row.revision if active_row is not None: active_row.status = "retired" active_row.retired_at = now self._session.flush() candidate_row.status = "active" candidate_row.published_by_user_id = actor.id candidate_row.published_at = now candidate_row.retired_at = None self._session.flush() status = inspect_catalog_registry(self._session, self._runtime_catalog) if status["catalogReady"] is not True: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "catalog publication did not produce a ready registry", status_code=500, ) result = { **status, "previousActiveRevision": previous_revision, "publishedRevision": normalized_revision, "previousStateDigest": normalized_state_digest, "reasonCode": normalized_reason, "correlationId": normalized_correlation, "executed": True, "idempotentReplay": False, } self._store_idempotent_response( CATALOG_PUBLICATION_SCOPE, self._runtime_catalog.product_id, normalized_key, request_digest, result, now, ) self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=actor.id, action="catalog.release_published", target_type="catalog_release", target_id=candidate_row.id, reason=normalized_note or normalized_reason, correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "catalogRevision": normalized_revision, "catalogSha256": bytes(candidate_row.document_digest).hex(), "previousCatalogRevision": previous_revision, "reasonCode": normalized_reason, "stateDigest": status["stateDigest"], }, ) ) return result def _publication_preview( self, rows: Sequence[CatalogRelease], revision: int, ) -> dict[str, object]: state, registry_issues, active_row = _registry_state( rows, self._runtime_catalog, ) candidate_row = next((row for row in rows if row.revision == revision), None) if candidate_row is None: raise ServerLicensingError( ServerErrorCode.CONFLICT, "catalog candidate revision was not found", status_code=404, ) candidate = _validated_release_catalog(candidate_row) active_catalog = ( None if active_row is None else _validated_release_catalog(active_row) ) issues = [ issue for issue in registry_issues if issue not in ( "active_catalog_digest_mismatch", "active_catalog_revision_mismatch", ) ] active_count = sum(row.status == "active" for row in rows) if active_count == 0: issues = [ issue for issue in issues if issue != "active_catalog_count_invalid" ] if candidate_row.status != "staged": issues.append("candidate_not_staged") if candidate.revision != self._runtime_catalog.revision: issues.append("candidate_runtime_revision_mismatch") if not hmac.compare_digest( bytes(candidate_row.document_digest).hex(), self._runtime_catalog.sha256(), ): issues.append("candidate_runtime_digest_mismatch") issues.extend(_compatibility_issues(active_catalog, candidate)) preview_state = { **state, "candidateRevision": candidate.revision, "candidateCatalogSha256": bytes(candidate_row.document_digest).hex(), "activeRevision": None if active_row is None else active_row.revision, } unique_issues = sorted(set(issues)) return { **preview_state, "changes": _catalog_changes(active_catalog, candidate), "issues": unique_issues, "canPublish": not unique_issues, "stateDigest": hashlib.sha256( canonicalize_json(preview_state) ).hexdigest(), } 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, "catalog administration requires an active administrator", status_code=403, ) return actor def _rows(self, *, lock: bool) -> tuple[CatalogRelease, ...]: if not isinstance(lock, bool): raise TypeError("lock must be a Boolean") query = ( select(CatalogRelease) .where(CatalogRelease.product_id == self._runtime_catalog.product_id) .order_by(CatalogRelease.revision) ) if lock: query = query.with_for_update() return tuple(self._session.scalars(query).all()) @staticmethod def _revision(value: int) -> int: if isinstance(value, bool) or not isinstance(value, int): raise TypeError("revision must be an integer") if value < 1: raise ValueError("revision must be at least one") return value 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__ = [ "CATALOG_PUBLICATION_SCOPE", "CATALOG_STAGE_SCOPE", "CatalogAdministrationService", "CatalogBoundSnapshotSigner", "inspect_catalog_registry", ]