diff --git a/docs/legacy-purchase-continuations.md b/docs/legacy-purchase-continuations.md index f43b5bbc..418d8dc0 100644 --- a/docs/legacy-purchase-continuations.md +++ b/docs/legacy-purchase-continuations.md @@ -15,11 +15,19 @@ replay. The coordinator input must never be sent to an AdCP seller. ```python from adcp.compat import ( LegacyPurchaseCoordinator, + PendingTaskResolution, ReconciliationResult, SqliteCompatibilityContinuationStore, ) -store = SqliteCompatibilityContinuationStore("state/adcp-continuations.sqlite3") +store = SqliteCompatibilityContinuationStore( + "state/adcp-continuations.sqlite3", + max_records=20_000, + max_bytes=64 * 1024 * 1024, + max_payload_bytes=1024 * 1024, + max_records_per_principal=2_000, + max_bytes_per_principal=8 * 1024 * 1024, +) async def execute_legacy_purchase(execution): # Route using execution.target_binding to the same authenticated seller @@ -39,10 +47,21 @@ async def reconcile_legacy_purchase(execution, operation): return ReconciliationResult.not_applied() return ReconciliationResult.ambiguous() +async def poll_legacy_purchase(execution, operation): + # Read-only, idempotent polling only. Submit approval/input separately, + # then let this callback observe the original seller task's new state. + task_id = operation.result["task_id"] + return PendingTaskResolution(task_id, await legacy_client.get_task(task_id)) + coordinator = LegacyPurchaseCoordinator( store=store, executor=execute_legacy_purchase, reconciler=reconcile_legacy_purchase, + pending_poller=poll_legacy_purchase, + # Load from an application secret manager. Keep the same key across every + # process/restart; generate at least 256 secret bits and never store it in + # the continuation ledger. + token_derivation_key=continuation_token_key, ) ``` @@ -52,6 +71,9 @@ token in `purchase_continuation`: ```python token = await coordinator.issue_legacy_create_continuation( principal_id=authenticated_principal, + # Stable identity of this discovery/projection transaction. Exact retries + # must reuse it; a genuinely new discovery must use a new value. + issuance_idempotency_key=discovery_transaction_id, account=account, source_adcp_version="3.1.15", # exact negotiated patch release expires_at=expires_at, @@ -87,12 +109,34 @@ The SQLite ledger is created with mode `0600`; its direct parent must be owned by the current user and cannot be group/world writable. Existing database and sidecar files with group or other access are rejected before every pathname open. This is access control, not encryption; use an encrypted volume or an -application-owned encrypted store when payloads require encryption at rest. +application-owned encrypted store when non-secret payloads require encryption +at rest. The built-in stores reject credential-bearing fields, +`push_notification_config`, webhook/callback URLs, URL user information, and +presigned/signed query parameters rather than writing them to the ledger. Move +such values to an application secret store and resolve them only inside the +executor or pending poller. + `purge_resolved_before(cutoff)` removes only old succeeded/failed and never-claimed continuations. It deliberately retains claimed, `in_flight`, -`pending`, and `ambiguous` operations regardless of age. +`pending`, and `ambiguous` operations regardless of age. Configure global and +per-principal record/logical-byte limits plus a per-payload limit for the +deployment; quota exhaustion fails closed and rolls back the attempted state +change. Before entering `in_flight` or `ambiguous`, the SQLite store durably +records a full result-payload reservation against both byte quotas. Every +worker uses that stored amount even if its local quota configuration differs. +Pending and terminal writes consume the immutable reservation and therefore +cannot be starved by another principal's later ledger use—or by a lower runtime +payload setting—after the seller mutation starts. Logical quotas do not include +SQLite indexes, free pages, or transient WAL growth, so production must also +impose a filesystem/container volume quota and caller-level issuance/polling +rate limits. Ledgers created by the initial pre-release coordinator are migrated in place. +The migration audits existing payloads against the credential policy and fails +startup if operator remediation is required. Pre-fingerprint authorizations +also block equivalent new issuance until they are resolved or quarantined, so +an upgrade cannot silently create a second redeemable token. + The old ledger did not retain the buyer-visible pricing subset, so unresolved old rows are explicitly non-executable instead of exposing seller-only options. The first exact retry may atomically adopt the sanitized execution snapshot @@ -115,19 +159,25 @@ The SDK cannot infer security or commercial identity. The application must: and make it globally unambiguous by binding issuer, tenant, and subject; - preserve the original account and seller target/session, especially for 2.5, whose wire request has no account field; -- encrypt sensitive stored discovery payloads at rest, set a state-aware - retention policy, and restrict ledger access. Never purge unresolved - `in_flight` or `ambiguous` operations automatically; +- encrypt confidential non-secret discovery payloads at rest, set a state-aware + retention policy, and restrict ledger access. Secret-bearing payloads must + not enter this ledger at all. Never purge unresolved `in_flight`, `pending`, + or `ambiguous` operations automatically; - authorize the actual `create_media_buy` call and select its credentials; - implement authoritative reconciliation using a seller transaction identity; - keep the exact negotiated patch version and full observed product/pricing payload until expiry and reconciliation retention have elapsed. -The opaque token is generated with at least 128 bits of randomness and only its -SHA-256 hash is stored. A principal mismatch is reported as not found to avoid -cross-tenant token enumeration. Natural account comparison excludes mutable -display metadata such as `operator_unit.name` but includes the account's actual -natural key. +The opaque token is derived with HMAC-SHA-256 from the application-held key, +the principal-scoped issuance identity, and a canonical hash of every issuance +binding; only its SHA-256 hash, key fingerprint, and binding hash are stored. +The unique fingerprint makes exact projection retries return the same token, +while changed bindings produce a different token even after an old terminal +row has been purged. Reusing an unpurged issuance key with changed inputs fails +closed. A principal mismatch is reported as not found to avoid cross-tenant +token enumeration. Natural account comparison excludes mutable display +metadata such as `operator_unit.name` but includes the account's actual natural +key. ## Claim and crash behavior @@ -136,6 +186,7 @@ The durable operation ledger moves through: ```text claimed -> in_flight -> succeeded | failed | pending \-> ambiguous -> claimed (only after authoritative absence) +pending -> pending | succeeded | failed ``` The token is consumed when the first seller mutation is reserved. Exact @@ -149,8 +200,8 @@ An exception, timeout, or cancellation observed by the coordinator after executor. Look up a revision-bearing snapshot and use the fenced recovery API: ```python -operation = await coordinator.get_legacy_purchase_operation( - operation_id, +operation = await coordinator.get_legacy_purchase_operation_by_idempotency_key( + compatibility_input.idempotency_key, principal_id=authenticated_principal, ) @@ -163,6 +214,29 @@ result = await coordinator.recover_legacy_purchase( ) ``` +A submitted, working, or input-required response is durable but not terminal. +Look up its latest revision and poll the original task through the configured +poller: + +```python +operation = await coordinator.get_legacy_purchase_operation_by_idempotency_key( + compatibility_input.idempotency_key, + principal_id=authenticated_principal, +) +result = await coordinator.refresh_pending_legacy_purchase( + operation, + principal_id=authenticated_principal, + target_binding=stable_seller_session_id, +) +``` + +Every refresh requires `PendingTaskResolution`, binds both pending and terminal +results to the original `task_id`, validates the later envelope against the +exact legacy schema, and commits by revision CAS. A stale concurrent poll +cannot overwrite a newer task state. The poller must be idempotent and +read-only because concurrent workers can both perform the lookup before one +wins the durable CAS. + Recovery atomically fences `in_flight` to `ambiguous` with a revision CAS. A stale snapshot cannot recover or complete the operation. The SDK never reopens the token by elapsed time and never blindly resends the legacy request. A diff --git a/src/adcp/compat/__init__.py b/src/adcp/compat/__init__.py index 2645c31d..15d9993b 100644 --- a/src/adcp/compat/__init__.py +++ b/src/adcp/compat/__init__.py @@ -21,8 +21,10 @@ LegacyPurchaseCoordinator, LegacyPurchaseExecution, LegacyPurchaseExecutor, + LegacyPurchasePendingPoller, LegacyPurchaseReconciler, LegacyPurchaseResult, + PendingTaskResolution, ReconciliationResult, ReconciliationStatus, canonical_account_identity, @@ -40,8 +42,10 @@ "LegacyPurchaseCoordinator", "LegacyPurchaseExecution", "LegacyPurchaseExecutor", + "LegacyPurchasePendingPoller", "LegacyPurchaseReconciler", "LegacyPurchaseResult", + "PendingTaskResolution", "ReconciliationResult", "ReconciliationStatus", "SqliteCompatibilityContinuationStore", diff --git a/src/adcp/compat/purchase_continuation.py b/src/adcp/compat/purchase_continuation.py index 57b7ce31..8cacc080 100644 --- a/src/adcp/compat/purchase_continuation.py +++ b/src/adcp/compat/purchase_continuation.py @@ -14,8 +14,10 @@ from __future__ import annotations import asyncio +import base64 import copy import hashlib +import hmac import inspect import re import secrets @@ -24,6 +26,7 @@ from datetime import datetime, timezone from enum import Enum from typing import Any, ClassVar, Protocol, TypeAlias, runtime_checkable +from urllib.parse import unquote_plus, urlsplit import rfc8785 from pydantic import BaseModel, ValidationError @@ -45,11 +48,95 @@ ["LegacyPurchaseExecution", "CompatibilityPurchaseOperation"], "ReconciliationResult | Awaitable[ReconciliationResult]", ] +LegacyPurchasePendingPoller: TypeAlias = Callable[ + ["LegacyPurchaseExecution", "CompatibilityPurchaseOperation"], + "PendingTaskResolution | Awaitable[PendingTaskResolution]", +] _SOURCE_VERSION_RE = re.compile(r"^(?:2\.5|3\.[01])\.\d+$") _REQUIRED_LOSSES = frozenset({"feed_version_not_atomic", "pricing_version_not_atomic"}) _MUTATION_LOSS = "mutation_idempotency_not_guaranteed" _ALLOWED_LOSSES = _REQUIRED_LOSSES | {_MUTATION_LOSS} +_MIN_TOKEN_DERIVATION_KEY_BYTES = 32 +_FORBIDDEN_PERSISTED_KEYS = frozenset( + { + "access_token", + "access_key", + "access_key_id", + "api_key", + "api_token", + "auth_token", + "auth", + "authentication", + "authorization", + "authorization_code", + "bearer_token", + "client_secret", + "cookie", + "credential", + "credentials", + "id_token", + "jwt", + "key", + "password", + "passwd", + "private_key", + "proxy_authorization", + "push_notification_config", + "refresh_token", + "secret_key", + "secret", + "set_cookie", + "signing_secret", + "signature", + "webhook_secret", + "webhook_url", + "callback_url", + "token", + } +) +_SENSITIVE_URL_QUERY_KEYS = frozenset( + { + "access_token", + "api_key", + "credential", + "key", + "password", + "secret", + "sig", + "signature", + "token", + "x_amz_credential", + "x_amz_security_token", + "x_amz_signature", + "x_goog_credential", + "x_goog_signature", + } +) +_FORBIDDEN_COMPACT_KEYS = frozenset(key.replace("_", "") for key in _FORBIDDEN_PERSISTED_KEYS) +_FORBIDDEN_COMPACT_SUFFIXES = ( + "accesskey", + "accesskeyid", + "accesstoken", + "apikey", + "authtoken", + "authorizationcode", + "authorization", + "callbackurl", + "clientsecret", + "credential", + "credentials", + "idtoken", + "jwt", + "password", + "privatekey", + "refreshtoken", + "secret", + "signature", + "setcookie", + "token", + "webhookurl", +) class CompatibilityContinuationErrorCode(str, Enum): @@ -69,6 +156,9 @@ class CompatibilityContinuationErrorCode(str, Enum): AMBIGUOUS_MUTATION = "ambiguous_legacy_mutation" INVALID_LEGACY_RESPONSE = "invalid_legacy_create_response" STORE_CONFLICT = "continuation_store_conflict" + PERSISTENCE_POLICY = "continuation_persistence_policy" + STORE_QUOTA_EXCEEDED = "continuation_store_quota_exceeded" + PENDING_RESOLUTION_REQUIRED = "pending_legacy_resolution_required" class CompatibilityContinuationError(Exception): @@ -114,6 +204,8 @@ class LegacyPurchaseContinuation: """ token_hash: str + issuance_fingerprint: str | None + issuance_binding_hash: str | None principal_id: str account_identity: str source_adcp_version: str @@ -141,6 +233,7 @@ class CompatibilityPurchaseOperation: state: CompatibilityOperationState revision: int execution_input: JsonObject + reserved_result_bytes: int = 0 result: JsonObject | None = None @@ -161,6 +254,14 @@ class LegacyPurchaseExecution: listed_purchase_context: JsonObject | None +@dataclass(frozen=True) +class PendingTaskResolution: + """Task-bound result returned by a read-only pending-task poller.""" + + task_id: str + result: LegacyPurchaseResult + + class ReconciliationStatus(str, Enum): APPLIED = "authoritatively_applied" NOT_APPLIED = "authoritatively_not_applied" @@ -222,6 +323,10 @@ async def get_operation( self, operation_id: str, *, principal_id: str ) -> CompatibilityPurchaseOperation | None: ... + async def get_operation_by_idempotency_key( + self, idempotency_key: str, *, principal_id: str + ) -> CompatibilityPurchaseOperation | None: ... + async def mark_in_flight( self, operation: CompatibilityPurchaseOperation ) -> CompatibilityPurchaseOperation: ... @@ -256,22 +361,57 @@ class InMemoryCompatibilityContinuationStore: is_durable: ClassVar[bool] = False - def __init__(self) -> None: + def __init__(self, *, max_records: int = 20_000, max_bytes: int = 64 * 1024 * 1024) -> None: + if type(max_records) is not int or max_records <= 0: + raise ValueError("max_records must be a positive integer") + if type(max_bytes) is not int or max_bytes <= 0: + raise ValueError("max_bytes must be a positive integer") self._continuations: dict[str, LegacyPurchaseContinuation] = {} self._claimed_by: dict[str, str] = {} self._operations: dict[tuple[str, str], CompatibilityPurchaseOperation] = {} self._lock = asyncio.Lock() self._clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc) + self._max_records = max_records + self._max_bytes = max_bytes async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> None: async with self._lock: - if continuation.token_hash in self._continuations: + copied = _copy_continuation(continuation) + _validate_continuation_persistence(copied) + if any( + value.issuance_fingerprint is None + and value.principal_id == copied.principal_id + and value.observed_payload_hash == copied.observed_payload_hash + and value.account_identity == copied.account_identity + and value.target_binding == copied.target_binding + for value in self._continuations.values() + ): raise _error( CompatibilityContinuationErrorCode.STORE_CONFLICT, - "continuation token hash is already registered", - "Issue a fresh cryptographically random continuation token.", + "equivalent pre-migration authorization requires operator resolution", + "Resolve or quarantine the legacy continuation before reissuing.", ) - self._continuations[continuation.token_hash] = _copy_continuation(continuation) + existing = self._continuations.get(continuation.token_hash) + by_fingerprint = next( + ( + value + for value in self._continuations.values() + if value.issuance_fingerprint is not None + and value.issuance_fingerprint == continuation.issuance_fingerprint + and value.principal_id == continuation.principal_id + ), + None, + ) + if existing == copied and (by_fingerprint is None or by_fingerprint == existing): + return + if existing is not None or by_fingerprint is not None: + raise _error( + CompatibilityContinuationErrorCode.STORE_CONFLICT, + "continuation issuance fingerprint is already registered differently", + "Use the same token derivation key and exact issuance inputs.", + ) + self._check_quota(additional=copied) + self._continuations[continuation.token_hash] = copied async def get_continuation( self, token_hash: str, *, principal_id: str @@ -333,6 +473,8 @@ async def claim( revision=1, execution_input=_json_copy(execution_input), ) + _validate_persistable_payload(operation.execution_input, context="execution input") + self._check_quota(additional=operation) self._operations[key] = operation self._claimed_by[token_hash] = operation.operation_id return _copy_operation(operation) @@ -349,6 +491,13 @@ async def get_operation( return _copy_operation(operation) return None + async def get_operation_by_idempotency_key( + self, idempotency_key: str, *, principal_id: str + ) -> CompatibilityPurchaseOperation | None: + async with self._lock: + operation = self._operations.get((principal_id, idempotency_key)) + return _copy_operation(operation) if operation is not None else None + async def mark_in_flight( self, operation: CompatibilityPurchaseOperation ) -> CompatibilityPurchaseOperation: @@ -381,6 +530,7 @@ async def complete( }: raise ValueError("completed result requires pending, succeeded, or failed state") copied = _json_copy(result) + _validate_persistable_payload(copied, context="legacy result") return await self._transition( operation, allowed={ @@ -436,8 +586,30 @@ async def _transition( result=copy.deepcopy(result), ) self._operations[key] = updated + try: + self._check_current_quota() + except BaseException: + self._operations[key] = current + raise return _copy_operation(updated) + def _check_quota( + self, + *, + additional: LegacyPurchaseContinuation | CompatibilityPurchaseOperation, + ) -> None: + records = len(self._continuations) + len(self._operations) + 1 + values: list[Any] = [*self._continuations.values(), *self._operations.values(), additional] + logical_bytes = sum(len(repr(value).encode("utf-8")) for value in values) + if records > self._max_records or logical_bytes > self._max_bytes: + raise _quota_error(self._max_records, self._max_bytes) + + def _check_current_quota(self) -> None: + values: list[Any] = [*self._continuations.values(), *self._operations.values()] + logical_bytes = sum(len(repr(value).encode("utf-8")) for value in values) + if len(values) > self._max_records or logical_bytes > self._max_bytes: + raise _quota_error(self._max_records, self._max_bytes) + class LegacyPurchaseCoordinator: """Validate, atomically claim, execute, and replay a legacy purchase.""" @@ -448,6 +620,8 @@ def __init__( store: CompatibilityContinuationStore, executor: LegacyPurchaseExecutor, reconciler: LegacyPurchaseReconciler | None = None, + pending_poller: LegacyPurchasePendingPoller | None = None, + token_derivation_key: bytes | bytearray | memoryview | None = None, allow_non_durable_store: bool = False, clock: Callable[[], datetime] | None = None, ) -> None: @@ -461,12 +635,27 @@ def __init__( self.store = store self.executor = executor self.reconciler = reconciler + self.pending_poller = pending_poller + if token_derivation_key is None: + if store.is_durable: + raise ValueError( + "durable continuation coordination requires a stable " + "token_derivation_key of at least 32 bytes" + ) + token_derivation_key = secrets.token_bytes(_MIN_TOKEN_DERIVATION_KEY_BYTES) + if not isinstance(token_derivation_key, (bytes, bytearray, memoryview)): + raise TypeError("token_derivation_key must be bytes-like") + key = bytes(token_derivation_key) + if len(key) < _MIN_TOKEN_DERIVATION_KEY_BYTES or not any(key): + raise ValueError("token_derivation_key must be a high-entropy secret of 32+ bytes") + self._token_derivation_key = key self._clock = clock or (lambda: datetime.now(timezone.utc)) async def issue_legacy_create_continuation( self, *, principal_id: str, + issuance_idempotency_key: str, account: Mapping[str, Any] | Any, source_adcp_version: str, expires_at: datetime, @@ -482,6 +671,7 @@ async def issue_legacy_create_continuation( """Persist all projection bindings and return the opaque bearer token.""" _require_text(principal_id, "principal_id") + _require_text(issuance_idempotency_key, "issuance_idempotency_key") _require_text(target_binding, "target_binding") if type(mutation_idempotency_guaranteed) is not bool: raise _invalid("mutation_idempotency_guaranteed must be a boolean") @@ -517,12 +707,45 @@ async def issue_legacy_create_continuation( listed = ( _json_copy(listed_purchase_context) if listed_purchase_context is not None else None ) - - token = secrets.token_urlsafe(32) + for context, value in ( + ("observed request", observed_req), + ("observed response", observed_resp), + ("buyer-visible products", {"products": list(projected)}), + ("listed purchase context", listed), + ): + if value is not None: + _validate_persistable_payload(value, context=context) + + issuance_fingerprint = _full_hash( + { + "principal_id": principal_id, + "issuance_idempotency_key": issuance_idempotency_key, + } + ) + issuance_binding_hash = _full_hash( + { + "account_identity": account_identity, + "source_adcp_version": source_adcp_version, + "expires_at": expires_at.isoformat(), + "observed_request": observed_req, + "observed_response": observed_resp, + "product_ids": list(ids), + "projected_products": list(projected), + "losses": sorted(loss_set), + "mutation_idempotency_guaranteed": mutation_idempotency_guaranteed, + "target_binding": target_binding, + "listed_purchase_context": listed, + } + ) + token = _derive_token( + self._token_derivation_key, issuance_fingerprint, issuance_binding_hash + ) token_hash = _token_hash(token) observed_payload_hash = _full_hash({"request": observed_req, "response": observed_resp}) record = LegacyPurchaseContinuation( token_hash=token_hash, + issuance_fingerprint=issuance_fingerprint, + issuance_binding_hash=issuance_binding_hash, principal_id=principal_id, account_identity=account_identity, source_adcp_version=source_adcp_version, @@ -570,6 +793,68 @@ async def continue_legacy_purchase( ) return await self._drive(operation, record, target_binding=target_binding) + async def refresh_pending_legacy_purchase( + self, + operation: CompatibilityPurchaseOperation, + *, + principal_id: str, + target_binding: str, + ) -> JsonObject: + """Poll and CAS-advance a pending seller task through an application callback. + + The poller must be an idempotent, read-only lookup of the already-created + seller task. Input/approval submission happens outside this callback. + The returned task identity is checked for pending and terminal results. + """ + + _require_text(principal_id, "principal_id") + _require_text(target_binding, "target_binding") + if operation.principal_id != principal_id: + raise _not_found() + current = await self.store.get_operation(operation.operation_id, principal_id=principal_id) + if current is None: + raise _not_found() + if current.revision != operation.revision: + raise _store_state_error("operation revision changed before pending refresh") + if current.state != CompatibilityOperationState.PENDING or current.result is None: + raise _error( + CompatibilityContinuationErrorCode.PENDING_RESOLUTION_REQUIRED, + "operation is not a revision-bearing pending seller task", + "Look up a fresh pending operation snapshot before refreshing it.", + details={"operation_id": current.operation_id}, + ) + if self.pending_poller is None: + raise _error( + CompatibilityContinuationErrorCode.PENDING_RESOLUTION_REQUIRED, + "no pending seller task poller is configured", + "Configure pending_poller to read the original seller task state.", + details={"operation_id": current.operation_id}, + ) + record = await self.store.get_continuation(current.token_hash, principal_id=principal_id) + if record is None: + raise _not_found() + self._validate_bindings(current.execution_input, record, target_binding=target_binding) + execution = _execution_from(current, record, current.execution_input, target_binding) + resolution = await _call_callback( + self.pending_poller, _copy_execution(execution), _copy_operation(current) + ) + previous_task_id = current.result.get("task_id") + if not isinstance(resolution, PendingTaskResolution): + raise TypeError("pending_poller must return PendingTaskResolution") + if resolution.task_id != previous_task_id: + raise _invalid_legacy_response(record.source_adcp_version, []) + copied, state = _validated_result( + resolution.result, source_adcp_version=record.source_adcp_version + ) + if ( + state == CompatibilityOperationState.PENDING + and copied.get("task_id") != previous_task_id + ): + raise _invalid_legacy_response(record.source_adcp_version, []) + completed = await _shielded_transition(self.store.complete(current, copied, state=state)) + assert completed.result is not None + return copy.deepcopy(completed.result) + async def get_legacy_purchase_operation( self, operation_id: str, *, principal_id: str ) -> CompatibilityPurchaseOperation: @@ -582,6 +867,20 @@ async def get_legacy_purchase_operation( raise _not_found() return _copy_operation(operation) + async def get_legacy_purchase_operation_by_idempotency_key( + self, idempotency_key: str, *, principal_id: str + ) -> CompatibilityPurchaseOperation: + """Look up a principal-scoped operation when only the buyer key is known.""" + + _require_text(idempotency_key, "idempotency_key") + _require_text(principal_id, "principal_id") + operation = await self.store.get_operation_by_idempotency_key( + idempotency_key, principal_id=principal_id + ) + if operation is None: + raise _not_found() + return _copy_operation(operation) + async def recover_legacy_purchase( self, operation: CompatibilityPurchaseOperation, @@ -1239,6 +1538,130 @@ def _token_hash(token: str) -> str: return hashlib.sha256(token.encode("utf-8")).hexdigest() +def _derive_token(key: bytes, issuance_fingerprint: str, issuance_binding_hash: str) -> str: + message = f"{issuance_fingerprint}:{issuance_binding_hash}".encode("ascii") + digest = hmac.new(key, message, hashlib.sha256).digest() + return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") + + +def _validate_continuation_persistence(value: LegacyPurchaseContinuation) -> None: + for context, payload in ( + ("observed request", value.observed_request), + ("observed response", value.observed_response), + ( + "buyer-visible products", + ( + {"products": list(value.projected_products)} + if value.projected_products is not None + else None + ), + ), + ("listed purchase context", value.listed_purchase_context), + ): + if payload is not None: + _validate_persistable_payload(payload, context=context) + + +def _validate_persistable_payload(value: Any, *, context: str) -> None: + """Reject credentials and signed URLs before payloads cross the durable boundary.""" + + def walk(item: Any) -> None: + if isinstance(item, Mapping): + for raw_key, child in item.items(): + normalized = _normalize_sensitive_key(raw_key) + if _is_forbidden_persisted_key(normalized): + raise _persistence_policy_error(context, "credential-bearing field") + walk(child) + return + if isinstance(item, (list, tuple)): + for child in item: + walk(child) + return + if not isinstance(item, str): + return + if re.match( + r"^\s*(?:authorization|cookie|proxy-authorization|x-api-key)\s*:", + item, + flags=re.IGNORECASE, + ): + raise _persistence_policy_error(context, "credential-bearing header") + try: + parsed = urlsplit(item) + except ValueError: + return + if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc: + return + if parsed.username is not None or parsed.password is not None: + raise _persistence_policy_error(context, "URL user information") + decoded_query = unquote_plus(parsed.query) + query_keys = { + _normalize_sensitive_key(part.partition("=")[0]) + for part in re.split(r"[&;]", decoded_query) + if part + } + if query_keys & _SENSITIVE_URL_QUERY_KEYS or any( + key.startswith(("x_amz_", "x_goog_")) + or key.endswith(("access_key_id", "credential", "signature", "token")) + for key in query_keys + ): + raise _persistence_policy_error(context, "credential-bearing URL") + decoded_fragment = unquote_plus(parsed.fragment) + fragment_keys = { + _normalize_sensitive_key(part.partition("=")[0]) + for part in re.split(r"[&;]", decoded_fragment) + if part + } + if fragment_keys & _SENSITIVE_URL_QUERY_KEYS: + raise _persistence_policy_error(context, "credential-bearing URL fragment") + + walk(value) + + +def _normalize_sensitive_key(value: object) -> str: + text = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", str(value)) + text = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", text) + return re.sub(r"[^a-zA-Z0-9]+", "_", text).strip("_").lower() + + +def _is_forbidden_persisted_key(normalized: str) -> bool: + candidates = {normalized} + candidate = normalized + while True: + stripped = next( + ( + candidate[: -len(wrapper)] + for wrapper in ("_value", "_data", "_string", "_bytes") + if candidate.endswith(wrapper) + ), + None, + ) + if stripped is None: + break + candidates.add(stripped) + candidate = stripped + for candidate in candidates: + compact = candidate.replace("_", "") + if ( + candidate in _FORBIDDEN_PERSISTED_KEYS + or compact in _FORBIDDEN_COMPACT_KEYS + or compact.endswith(_FORBIDDEN_COMPACT_SUFFIXES) + or candidate.split("_")[-1] + in { + "authorization", + "cookie", + "credential", + "credentials", + "jwt", + "password", + "secret", + "signature", + "token", + } + ): + return True + return False + + def _json_copy(value: Mapping[str, Any]) -> JsonObject: payload = copy.deepcopy(dict(value)) # RFC 8785 both validates the JSON domain and gives deterministic handling @@ -1332,6 +1755,7 @@ def _validated_result( }.get(value.status) if expected is not None and state != expected: raise _invalid_legacy_response(source_adcp_version, []) + _validate_persistable_payload(payload, context="legacy result") return payload, state @@ -1428,6 +1852,25 @@ def _invalid(message: str) -> CompatibilityContinuationError: ) +def _persistence_policy_error(context: str, reason: str) -> CompatibilityContinuationError: + return _error( + CompatibilityContinuationErrorCode.PERSISTENCE_POLICY, + f"{context} contains a {reason} that cannot be persisted", + "Remove credentials, push notification configuration, and signed URLs before " + "issuing or advancing a durable continuation.", + details={"context": context}, + ) + + +def _quota_error(max_records: int, max_bytes: int) -> CompatibilityContinuationError: + return _error( + CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED, + "continuation ledger quota would be exceeded", + "Resolve retained operations or purge eligible terminal/expired rows before retrying.", + details={"max_records": max_records, "max_bytes": max_bytes}, + ) + + def _legacy_request_error(message: str) -> CompatibilityContinuationError: return _error( CompatibilityContinuationErrorCode.INVALID_LEGACY_REQUEST, @@ -1494,8 +1937,10 @@ def _ambiguous_error( "LegacyPurchaseCoordinator", "LegacyPurchaseExecution", "LegacyPurchaseExecutor", + "LegacyPurchasePendingPoller", "LegacyPurchaseReconciler", "LegacyPurchaseResult", + "PendingTaskResolution", "ReconciliationResult", "ReconciliationStatus", "canonical_account_identity", diff --git a/src/adcp/compat/sqlite_continuation_store.py b/src/adcp/compat/sqlite_continuation_store.py index 436d63e2..5922ac28 100644 --- a/src/adcp/compat/sqlite_continuation_store.py +++ b/src/adcp/compat/sqlite_continuation_store.py @@ -17,6 +17,7 @@ import secrets import sqlite3 import stat +import time from collections.abc import Callable, Mapping from contextlib import closing from datetime import datetime, timezone @@ -29,11 +30,16 @@ CompatibilityOperationState, CompatibilityPurchaseOperation, LegacyPurchaseContinuation, + _quota_error, + _validate_continuation_persistence, + _validate_persistable_payload, ) _SCHEMA = """ CREATE TABLE IF NOT EXISTS adcp_compat_continuations ( token_hash TEXT PRIMARY KEY, + issuance_fingerprint TEXT, + issuance_binding_hash TEXT, principal_id TEXT NOT NULL, account_identity TEXT NOT NULL, source_adcp_version TEXT NOT NULL, @@ -53,7 +59,6 @@ ); CREATE INDEX IF NOT EXISTS adcp_compat_continuations_principal_idx ON adcp_compat_continuations (principal_id, token_hash); - CREATE TABLE IF NOT EXISTS adcp_compat_operations ( operation_id TEXT PRIMARY KEY, principal_id TEXT NOT NULL, @@ -65,6 +70,7 @@ ), revision INTEGER NOT NULL DEFAULT 1, execution_input_json TEXT NOT NULL DEFAULT '{}', + reserved_result_bytes INTEGER NOT NULL DEFAULT 0 CHECK (reserved_result_bytes >= 0), result_json TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, @@ -73,6 +79,10 @@ FOREIGN KEY (token_hash) REFERENCES adcp_compat_continuations(token_hash) ); +CREATE TABLE IF NOT EXISTS adcp_compat_metadata ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); """ @@ -87,6 +97,11 @@ def __init__( *, timeout: float = 30.0, clock: Callable[[], datetime] | None = None, + max_records: int = 20_000, + max_bytes: int = 64 * 1024 * 1024, + max_payload_bytes: int = 1024 * 1024, + max_records_per_principal: int = 2_000, + max_bytes_per_principal: int = 8 * 1024 * 1024, ) -> None: raw = str(path) if raw == ":memory:" or raw.startswith("file::memory:"): @@ -97,6 +112,20 @@ def __init__( self._ensure_private_parent_directory() self.timeout = timeout self._clock = clock or (lambda: datetime.now(timezone.utc)) + for name, value in ( + ("max_records", max_records), + ("max_bytes", max_bytes), + ("max_payload_bytes", max_payload_bytes), + ("max_records_per_principal", max_records_per_principal), + ("max_bytes_per_principal", max_bytes_per_principal), + ): + if type(value) is not int or value <= 0: + raise ValueError(f"{name} must be a positive integer") + self.max_records = max_records + self.max_bytes = max_bytes + self.max_payload_bytes = max_payload_bytes + self.max_records_per_principal = min(max_records_per_principal, max_records) + self.max_bytes_per_principal = min(max_bytes_per_principal, max_bytes) self._ensure_private_database_file() with closing(self._connect()) as conn, conn: conn.executescript(_SCHEMA) @@ -124,8 +153,21 @@ def _ensure_timestamp_columns(self, conn: sqlite3.Connection) -> None: } if "projected_products_json" not in continuation_columns: conn.execute( - "ALTER TABLE adcp_compat_continuations " "ADD COLUMN projected_products_json TEXT" + "ALTER TABLE adcp_compat_continuations ADD COLUMN projected_products_json TEXT" + ) + if "issuance_fingerprint" not in continuation_columns: + conn.execute( + "ALTER TABLE adcp_compat_continuations ADD COLUMN issuance_fingerprint TEXT" ) + if "issuance_binding_hash" not in continuation_columns: + conn.execute( + "ALTER TABLE adcp_compat_continuations ADD COLUMN issuance_binding_hash TEXT" + ) + conn.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS adcp_compat_continuations_issuance_idx " + "ON adcp_compat_continuations (principal_id, issuance_fingerprint) " + "WHERE issuance_fingerprint IS NOT NULL" + ) if "mutation_idempotency_guaranteed" not in continuation_columns: conn.execute( "ALTER TABLE adcp_compat_continuations " @@ -137,8 +179,7 @@ def _ensure_timestamp_columns(self, conn: sqlite3.Connection) -> None: } if "revision" not in operation_columns: conn.execute( - "ALTER TABLE adcp_compat_operations " - "ADD COLUMN revision INTEGER NOT NULL DEFAULT 1" + "ALTER TABLE adcp_compat_operations ADD COLUMN revision INTEGER NOT NULL DEFAULT 1" ) if "execution_input_json" not in operation_columns: conn.execute( @@ -147,12 +188,51 @@ def _ensure_timestamp_columns(self, conn: sqlite3.Connection) -> None: ) operations_sql_row = conn.execute( - "SELECT sql FROM sqlite_master WHERE type = 'table' " - "AND name = 'adcp_compat_operations'" + "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'adcp_compat_operations'" ).fetchone() operations_sql = operations_sql_row["sql"] if operations_sql_row is not None else "" if "'pending'" not in operations_sql or "'failed'" not in operations_sql: self._rebuild_operations_table(conn) + operation_columns = { + row["name"] for row in conn.execute("PRAGMA table_info(adcp_compat_operations)") + } + if "reserved_result_bytes" not in operation_columns: + conn.execute( + "ALTER TABLE adcp_compat_operations " + "ADD COLUMN reserved_result_bytes INTEGER NOT NULL DEFAULT 0 " + "CHECK (reserved_result_bytes >= 0)" + ) + # Pre-reservation ledgers may already contain a seller mutation that + # requires a terminal write. Give each unresolved row a durable budget + # once, preserving any larger pending result already on disk. + conn.execute( + "UPDATE adcp_compat_operations " + "SET reserved_result_bytes = MAX(?, " + "length(CAST(COALESCE(result_json, '') AS BLOB))) " + "WHERE state IN ('in_flight', 'pending', 'ambiguous') " + "AND reserved_result_bytes = 0", + (self.max_payload_bytes,), + ) + self._audit_legacy_payloads(conn) + + @staticmethod + def _audit_legacy_payloads(conn: sqlite3.Connection) -> None: + audited = conn.execute( + "SELECT value FROM adcp_compat_metadata WHERE key = 'persistence_policy_version'" + ).fetchone() + if audited is not None and audited["value"] == "1": + return + for row in conn.execute("SELECT * FROM adcp_compat_continuations"): + _validate_continuation_persistence(_decode_continuation(row)) + for row in conn.execute("SELECT * FROM adcp_compat_operations"): + operation = _decode_operation(row) + _validate_persistable_payload(operation.execution_input, context="execution input") + if operation.result is not None: + _validate_persistable_payload(operation.result, context="legacy result") + conn.execute( + "INSERT OR REPLACE INTO adcp_compat_metadata (key, value) VALUES (?, ?)", + ("persistence_policy_version", "1"), + ) @staticmethod def _rebuild_operations_table(conn: sqlite3.Connection) -> None: @@ -209,7 +289,7 @@ def _ensure_private_parent_directory(self) -> None: try: directory_status = directory.lstat() except FileNotFoundError: - directory.mkdir(mode=0o700) + directory.mkdir(mode=0o700, exist_ok=True) directory_status = directory.lstat() except OSError as exc: raise PermissionError(f"SQLite directory {directory} is not accessible") from exc @@ -305,8 +385,7 @@ def _ensure_private_file(path: Path, description: str) -> None: mode = stat.S_IMODE(file_status.st_mode) if mode & 0o077: raise PermissionError( - f"{description} {path} has mode {mode:#o}; " - "restrict it to 0o600 before opening" + f"{description} {path} has mode {mode:#o}; restrict it to 0o600 before opening" ) finally: os.close(descriptor) @@ -321,7 +400,15 @@ def _connect(self) -> sqlite3.Connection: try: conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") - conn.execute("PRAGMA journal_mode = WAL") + deadline = time.monotonic() + self.timeout + while True: + try: + conn.execute("PRAGMA journal_mode = WAL") + break + except sqlite3.OperationalError as exc: + if "locked" not in str(exc).lower() or time.monotonic() >= deadline: + raise + time.sleep(min(0.01, max(0.0, deadline - time.monotonic()))) conn.execute("PRAGMA synchronous = FULL") return conn except BaseException: @@ -332,50 +419,96 @@ async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> No await asyncio.to_thread(self._put_continuation, copy.deepcopy(continuation)) def _put_continuation(self, value: LegacyPurchaseContinuation) -> None: + _validate_continuation_persistence(value) now = _format_datetime(self._clock()) if value.projected_products is None: raise ValueError("new continuations require buyer-visible product bindings") + payloads = ( + _dumps(value.observed_request), + _dumps(value.observed_response), + _dumps(list(value.product_ids)), + _dumps(list(value.projected_products)), + _dumps(sorted(value.losses)), + ( + _dumps(value.listed_purchase_context) + if value.listed_purchase_context is not None + else None + ), + ) + self._enforce_payload_quota(*(payload for payload in payloads if payload is not None)) try: with closing(self._connect()) as conn, conn: + conn.execute("BEGIN IMMEDIATE") + legacy = conn.execute( + """ + SELECT 1 FROM adcp_compat_continuations + WHERE issuance_fingerprint IS NULL + AND principal_id = ? + AND observed_payload_hash = ? + AND account_identity = ? + AND target_binding = ? + LIMIT 1 + """, + ( + value.principal_id, + value.observed_payload_hash, + value.account_identity, + value.target_binding, + ), + ).fetchone() + if legacy is not None: + raise _error( + CompatibilityContinuationErrorCode.STORE_CONFLICT, + "equivalent pre-migration authorization requires operator resolution", + "Resolve or quarantine the legacy continuation before reissuing.", + ) conn.execute( """ INSERT INTO adcp_compat_continuations ( - token_hash, principal_id, account_identity, + token_hash, issuance_fingerprint, issuance_binding_hash, + principal_id, account_identity, source_adcp_version, expires_at, observed_request_json, observed_response_json, observed_payload_hash, product_ids_json, projected_products_json, losses_json, mutation_idempotency_guaranteed, target_binding, listed_purchase_context_json, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( value.token_hash, + value.issuance_fingerprint, + value.issuance_binding_hash, value.principal_id, value.account_identity, value.source_adcp_version, _format_datetime(value.expires_at), - _dumps(value.observed_request), - _dumps(value.observed_response), + payloads[0], + payloads[1], value.observed_payload_hash, - _dumps(list(value.product_ids)), - _dumps(list(value.projected_products)), - _dumps(sorted(value.losses)), + payloads[2], + payloads[3], + payloads[4], int(value.mutation_idempotency_guaranteed), value.target_binding, - ( - _dumps(value.listed_purchase_context) - if value.listed_purchase_context is not None - else None - ), + payloads[5], now, now, ), ) + self._enforce_ledger_quota(conn, principal_id=value.principal_id) except sqlite3.IntegrityError as exc: + with closing(self._connect()) as conn: + row = conn.execute( + "SELECT * FROM adcp_compat_continuations " + "WHERE token_hash = ? OR (principal_id = ? AND issuance_fingerprint = ?)", + (value.token_hash, value.principal_id, value.issuance_fingerprint), + ).fetchone() + if row is not None and _decode_continuation(row) == value: + return raise _error( CompatibilityContinuationErrorCode.STORE_CONFLICT, - "continuation token hash is already registered", - "Issue a fresh cryptographically random continuation token.", + "continuation issuance fingerprint is already registered differently", + "Use the same token derivation key and exact issuance inputs.", ) from exc async def get_continuation( @@ -406,13 +539,15 @@ async def claim( execution_input: Mapping[str, Any], now: datetime, ) -> CompatibilityPurchaseOperation: + snapshot = copy.deepcopy(dict(execution_input)) + _validate_persistable_payload(snapshot, context="execution input") return await asyncio.to_thread( self._claim, token_hash, principal_id, idempotency_key, payload_hash, - copy.deepcopy(dict(execution_input)), + snapshot, now, ) @@ -425,6 +560,7 @@ def _claim( execution_input: dict[str, Any], now: datetime, ) -> CompatibilityPurchaseOperation: + execution_input_json = _dumps(execution_input) with closing(self._connect()) as conn, conn: conn.execute("BEGIN IMMEDIATE") claim_time = max(_as_utc(now), _as_utc(self._clock())) @@ -448,6 +584,7 @@ def _claim( # but not the sanitized execution snapshot. An exact retry # can adopt it atomically; the revision increment fences # every pre-migration operation object. + self._enforce_payload_quota(execution_input_json) updated_at = _format_datetime(self._clock()) adopted = conn.execute( "UPDATE adcp_compat_operations " @@ -455,7 +592,7 @@ def _claim( "WHERE operation_id = ? AND revision = ? " "AND execution_input_json = '{}'", ( - _dumps(execution_input), + execution_input_json, updated_at, operation.operation_id, operation.revision, @@ -463,6 +600,9 @@ def _claim( ) if adopted.rowcount != 1: raise _state_error("legacy execution input changed concurrently") + # This bounded, one-time migration write may be required to + # reconcile a seller mutation that already executed. Global + # fullness must not prevent recovery of that existing row. operation = CompatibilityPurchaseOperation( operation_id=operation.operation_id, principal_id=operation.principal_id, @@ -472,6 +612,7 @@ def _claim( state=operation.state, revision=operation.revision + 1, execution_input=copy.deepcopy(execution_input), + reserved_result_bytes=operation.reserved_result_bytes, result=copy.deepcopy(operation.result), ) elif operation.execution_input != execution_input: @@ -502,6 +643,7 @@ def _claim( "Replay the original idempotency key, or restart product discovery.", ) + self._enforce_payload_quota(execution_input_json) operation_id = secrets.token_urlsafe(24) updated = conn.execute( """ @@ -534,11 +676,12 @@ def _claim( payload_hash, CompatibilityOperationState.CLAIMED.value, 1, - _dumps(execution_input), + execution_input_json, _format_datetime(claim_time), _format_datetime(claim_time), ), ) + self._enforce_ledger_quota(conn, principal_id=principal_id) conn.commit() return CompatibilityPurchaseOperation( operation_id=operation_id, @@ -549,6 +692,7 @@ def _claim( state=CompatibilityOperationState.CLAIMED, revision=1, execution_input=copy.deepcopy(execution_input), + reserved_result_bytes=0, ) async def get_operation( @@ -561,12 +705,29 @@ def _get_operation( ) -> CompatibilityPurchaseOperation | None: with closing(self._connect()) as conn: row = conn.execute( - "SELECT * FROM adcp_compat_operations " - "WHERE operation_id = ? AND principal_id = ?", + "SELECT * FROM adcp_compat_operations WHERE operation_id = ? AND principal_id = ?", (operation_id, principal_id), ).fetchone() return _decode_operation(row) if row is not None else None + async def get_operation_by_idempotency_key( + self, idempotency_key: str, *, principal_id: str + ) -> CompatibilityPurchaseOperation | None: + return await asyncio.to_thread( + self._get_operation_by_idempotency_key, idempotency_key, principal_id + ) + + def _get_operation_by_idempotency_key( + self, idempotency_key: str, principal_id: str + ) -> CompatibilityPurchaseOperation | None: + with closing(self._connect()) as conn: + row = conn.execute( + "SELECT * FROM adcp_compat_operations " + "WHERE idempotency_key = ? AND principal_id = ?", + (idempotency_key, principal_id), + ).fetchone() + return _decode_operation(row) if row is not None else None + async def mark_in_flight( self, operation: CompatibilityPurchaseOperation ) -> CompatibilityPurchaseOperation: @@ -602,6 +763,8 @@ async def complete( CompatibilityOperationState.FAILED, }: raise ValueError("complete state must be pending, succeeded, or failed") + snapshot = copy.deepcopy(dict(result)) + _validate_persistable_payload(snapshot, context="legacy result") return await asyncio.to_thread( self._transition, copy.deepcopy(operation), @@ -611,7 +774,7 @@ async def complete( CompatibilityOperationState.PENDING, }, state, - copy.deepcopy(dict(result)), + snapshot, ) async def fence_in_flight( @@ -658,6 +821,7 @@ def _transition( or current.token_hash != operation.token_hash or current.payload_hash != operation.payload_hash or current.execution_input != operation.execution_input + or current.reserved_result_bytes != operation.reserved_result_bytes ): raise _state_error("operation binding changed in continuation store") if current.revision != operation.revision: @@ -667,16 +831,43 @@ def _transition( f"cannot transition operation from {current.state.value} to {target.value}" ) result_json = _dumps(result) if result is not None else None + if ( + result_json is not None + and len(result_json.encode("utf-8")) > current.reserved_result_bytes + ): + raise CompatibilityContinuationError( + CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED, + "operation result exceeds its reserved durable capacity", + recovery_guidance=( + "Reconcile using a result within the operation's " + "original durable reservation." + ), + details={"reserved_result_bytes": current.reserved_result_bytes}, + ) + reserved_result_bytes = current.reserved_result_bytes + if current.state is CompatibilityOperationState.CLAIMED and target in { + CompatibilityOperationState.IN_FLIGHT, + CompatibilityOperationState.AMBIGUOUS, + }: + reserved_result_bytes = self.max_payload_bytes + elif target in { + CompatibilityOperationState.CLAIMED, + CompatibilityOperationState.SUCCEEDED, + CompatibilityOperationState.FAILED, + }: + reserved_result_bytes = 0 updated_at = _format_datetime(self._clock()) updated = conn.execute( """ UPDATE adcp_compat_operations - SET state = ?, result_json = ?, revision = revision + 1, updated_at = ? + SET state = ?, result_json = ?, reserved_result_bytes = ?, + revision = revision + 1, updated_at = ? WHERE operation_id = ? AND state = ? AND revision = ? """, ( target.value, result_json, + reserved_result_bytes, updated_at, current.operation_id, current.state.value, @@ -685,6 +876,15 @@ def _transition( ) if updated.rowcount != 1: raise _state_error("operation state changed concurrently") + if current.state is CompatibilityOperationState.CLAIMED and target in { + CompatibilityOperationState.IN_FLIGHT, + CompatibilityOperationState.AMBIGUOUS, + }: + # Reserve one full result payload before the seller mutation can + # run. Later pending/terminal writes consume that reservation, + # so another principal cannot strand an executed mutation by + # filling the global quota between execution and completion. + self._enforce_ledger_quota(conn, principal_id=current.principal_id) conn.commit() return CompatibilityPurchaseOperation( operation_id=current.operation_id, @@ -695,9 +895,104 @@ def _transition( state=target, revision=current.revision + 1, execution_input=copy.deepcopy(current.execution_input), + reserved_result_bytes=reserved_result_bytes, result=copy.deepcopy(result), ) + def _enforce_payload_quota(self, *serialized_values: str) -> None: + payload_bytes = sum(len(value.encode("utf-8")) for value in serialized_values) + if payload_bytes > self.max_payload_bytes: + raise _quota_error(self.max_records, self.max_bytes) + + def _enforce_ledger_quota(self, conn: sqlite3.Connection, *, principal_id: str) -> None: + records = conn.execute( + "SELECT " + "(SELECT COUNT(*) FROM adcp_compat_continuations) + " + "(SELECT COUNT(*) FROM adcp_compat_operations)" + ).fetchone()[0] + continuation_bytes = conn.execute( + """ + SELECT COALESCE(SUM( + length(CAST(token_hash AS BLOB)) + + length(CAST(COALESCE(issuance_fingerprint, '') AS BLOB)) + + length(CAST(COALESCE(issuance_binding_hash, '') AS BLOB)) + + length(CAST(principal_id AS BLOB)) + + length(CAST(account_identity AS BLOB)) + + length(CAST(source_adcp_version AS BLOB)) + + length(CAST(expires_at AS BLOB)) + + length(CAST(observed_request_json AS BLOB)) + + length(CAST(observed_response_json AS BLOB)) + + length(CAST(observed_payload_hash AS BLOB)) + + length(CAST(product_ids_json AS BLOB)) + + length(CAST(COALESCE(projected_products_json, '') AS BLOB)) + + length(CAST(losses_json AS BLOB)) + + length(CAST(target_binding AS BLOB)) + + length(CAST(COALESCE(listed_purchase_context_json, '') AS BLOB)) + ), 0) FROM adcp_compat_continuations + """ + ).fetchone()[0] + operation_bytes = conn.execute( + """ + SELECT COALESCE(SUM( + length(CAST(operation_id AS BLOB)) + + length(CAST(principal_id AS BLOB)) + + length(CAST(idempotency_key AS BLOB)) + + length(CAST(token_hash AS BLOB)) + + length(CAST(payload_hash AS BLOB)) + + CASE + WHEN state = 'pending' THEN length(CAST('succeeded' AS BLOB)) + ELSE length(CAST(state AS BLOB)) + END + + length(CAST(execution_input_json AS BLOB)) + + CASE + WHEN state IN ('in_flight', 'pending', 'ambiguous') + THEN reserved_result_bytes + ELSE length(CAST(COALESCE(result_json, '') AS BLOB)) + END + ), 0) FROM adcp_compat_operations + """ + ).fetchone()[0] + if records > self.max_records or continuation_bytes + operation_bytes > self.max_bytes: + raise _quota_error(self.max_records, self.max_bytes) + principal_records = conn.execute( + "SELECT " + "(SELECT COUNT(*) FROM adcp_compat_continuations WHERE principal_id = ?) + " + "(SELECT COUNT(*) FROM adcp_compat_operations WHERE principal_id = ?)", + (principal_id, principal_id), + ).fetchone()[0] + principal_continuation_bytes = conn.execute( + """ + SELECT COALESCE(SUM( + length(CAST(observed_request_json AS BLOB)) + + length(CAST(observed_response_json AS BLOB)) + + length(CAST(product_ids_json AS BLOB)) + + length(CAST(COALESCE(projected_products_json, '') AS BLOB)) + + length(CAST(losses_json AS BLOB)) + + length(CAST(COALESCE(listed_purchase_context_json, '') AS BLOB)) + ), 0) FROM adcp_compat_continuations WHERE principal_id = ? + """, + (principal_id,), + ).fetchone()[0] + principal_operation_bytes = conn.execute( + """ + SELECT COALESCE(SUM( + length(CAST(execution_input_json AS BLOB)) + + CASE + WHEN state IN ('in_flight', 'pending', 'ambiguous') + THEN reserved_result_bytes + ELSE length(CAST(COALESCE(result_json, '') AS BLOB)) + END + ), 0) FROM adcp_compat_operations WHERE principal_id = ? + """, + (principal_id,), + ).fetchone()[0] + if ( + principal_records > self.max_records_per_principal + or principal_continuation_bytes + principal_operation_bytes + > self.max_bytes_per_principal + ): + raise _quota_error(self.max_records_per_principal, self.max_bytes_per_principal) + async def purge_resolved_before(self, cutoff: datetime) -> int: """Delete only old terminal or never-claimed continuations. @@ -822,6 +1117,8 @@ def _decode_continuation(row: sqlite3.Row) -> LegacyPurchaseContinuation: projected = _loads(row["projected_products_json"]) return LegacyPurchaseContinuation( token_hash=row["token_hash"], + issuance_fingerprint=row["issuance_fingerprint"], + issuance_binding_hash=row["issuance_binding_hash"], principal_id=row["principal_id"], account_identity=row["account_identity"], source_adcp_version=row["source_adcp_version"], @@ -853,6 +1150,7 @@ def _decode_operation(row: sqlite3.Row) -> CompatibilityPurchaseOperation: state=CompatibilityOperationState(row["state"]), revision=row["revision"], execution_input=_require_object(_loads(row["execution_input_json"])), + reserved_result_bytes=row["reserved_result_bytes"], result=_require_object(result) if result is not None else None, ) diff --git a/tests/test_purchase_continuation.py b/tests/test_purchase_continuation.py index 87d9ed73..50830f01 100644 --- a/tests/test_purchase_continuation.py +++ b/tests/test_purchase_continuation.py @@ -24,6 +24,7 @@ CompatibilityContinuationErrorCode, InMemoryCompatibilityContinuationStore, LegacyPurchaseCoordinator, + PendingTaskResolution, ReconciliationResult, SqliteCompatibilityContinuationStore, ) @@ -39,6 +40,7 @@ / "vectors.json" ) _NOW = datetime(2098, 1, 1, tzinfo=timezone.utc) +_TOKEN_KEY = b"test-only-continuation-token-key-32-bytes-minimum" def _cases() -> list[dict[str, Any]]: @@ -59,6 +61,7 @@ def _coordinator(store: Any, executor: Any, *, reconciler: Any = None) -> Legacy store=store, executor=executor, reconciler=reconciler, + token_derivation_key=_TOKEN_KEY, allow_non_durable_store=not store.is_durable, clock=lambda: _NOW, ) @@ -91,6 +94,7 @@ async def _issue( continuation = case["compact_projection"]["purchase_continuation"] token = await coordinator.issue_legacy_create_continuation( principal_id=principal, + issuance_idempotency_key=f"discovery-{case['source_version']}", account=case["continuation_input"]["account"], source_adcp_version=case["source_version"], expires_at=datetime.fromisoformat( @@ -619,6 +623,176 @@ def execute(_ctx: Any) -> Any: assert calls == 1 +@pytest.mark.asyncio +async def test_pending_result_can_be_polled_to_completion_with_revision_fencing() -> None: + case = _cases()[2] + store = InMemoryCompatibilityContinuationStore() + pending = {"status": "submitted", "task_id": "task-123"} + resolutions = iter( + [ + {"status": "working", "task_id": "task-123", "percentage": 75}, + _success_result(case["source_version"], "mb-pending-complete"), + ] + ) + + async def resolve(_ctx: Any, _operation: Any) -> PendingTaskResolution: + return PendingTaskResolution("task-123", next(resolutions)) + + coordinator = LegacyPurchaseCoordinator( + store=store, + executor=lambda _ctx: pending, + pending_poller=resolve, + token_derivation_key=_TOKEN_KEY, + allow_non_durable_store=True, + clock=lambda: _NOW, + ) + await _issue(coordinator, case) + assert ( + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + == pending + ) + first_snapshot = await coordinator.get_legacy_purchase_operation_by_idempotency_key( + case["continuation_input"]["idempotency_key"], principal_id="principal-acme" + ) + + working = await coordinator.refresh_pending_legacy_purchase( + first_snapshot, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert working["status"] == "working" + second_snapshot = await coordinator.get_legacy_purchase_operation( + first_snapshot.operation_id, principal_id="principal-acme" + ) + assert second_snapshot.revision == first_snapshot.revision + 1 + with pytest.raises(CompatibilityContinuationError) as stale: + await coordinator.refresh_pending_legacy_purchase( + first_snapshot, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert stale.value.code == CompatibilityContinuationErrorCode.STORE_CONFLICT + + completed = await coordinator.refresh_pending_legacy_purchase( + second_snapshot, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert completed == _success_result(case["source_version"], "mb-pending-complete") + assert ( + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + == completed + ) + + +@pytest.mark.asyncio +async def test_pending_refresh_rejects_task_identity_substitution() -> None: + case = _cases()[2] + store = InMemoryCompatibilityContinuationStore() + coordinator = LegacyPurchaseCoordinator( + store=store, + executor=lambda _ctx: {"status": "submitted", "task_id": "task-original"}, + pending_poller=lambda _ctx, _operation: PendingTaskResolution( + "task-substituted", + {"status": "working", "task_id": "task-substituted", "percentage": 10}, + ), + token_derivation_key=_TOKEN_KEY, + allow_non_durable_store=True, + clock=lambda: _NOW, + ) + await _issue(coordinator, case) + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + snapshot = await coordinator.get_legacy_purchase_operation_by_idempotency_key( + case["continuation_input"]["idempotency_key"], principal_id="principal-acme" + ) + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.refresh_pending_legacy_purchase( + snapshot, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_LEGACY_RESPONSE + current = await coordinator.get_legacy_purchase_operation( + snapshot.operation_id, principal_id="principal-acme" + ) + assert current == snapshot + + +@pytest.mark.asyncio +async def test_pending_refresh_binds_terminal_result_to_original_task() -> None: + case = _cases()[2] + store = InMemoryCompatibilityContinuationStore() + coordinator = LegacyPurchaseCoordinator( + store=store, + executor=lambda _ctx: {"status": "submitted", "task_id": "task-original"}, + pending_poller=lambda _ctx, _operation: PendingTaskResolution( + "task-other", _success_result(case["source_version"], "mb-wrong-task") + ), + token_derivation_key=_TOKEN_KEY, + allow_non_durable_store=True, + clock=lambda: _NOW, + ) + await _issue(coordinator, case) + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + snapshot = await coordinator.get_legacy_purchase_operation_by_idempotency_key( + case["continuation_input"]["idempotency_key"], principal_id="principal-acme" + ) + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.refresh_pending_legacy_purchase( + snapshot, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_LEGACY_RESPONSE + assert ( + await coordinator.get_legacy_purchase_operation( + snapshot.operation_id, principal_id="principal-acme" + ) + == snapshot + ) + + +@pytest.mark.asyncio +async def test_pending_refresh_requires_configured_poller() -> None: + case = _cases()[2] + store = InMemoryCompatibilityContinuationStore() + coordinator = _coordinator( + store, lambda _ctx: {"status": "submitted", "task_id": "task-unresolved"} + ) + await _issue(coordinator, case) + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + snapshot = await coordinator.get_legacy_purchase_operation_by_idempotency_key( + case["continuation_input"]["idempotency_key"], principal_id="principal-acme" + ) + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.refresh_pending_legacy_purchase( + snapshot, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.PENDING_RESOLUTION_REQUIRED + + @pytest.mark.asyncio @pytest.mark.parametrize("case", _cases(), ids=lambda c: c["source_version"]) @pytest.mark.parametrize("arm", ["submitted", "errors"]) @@ -693,6 +867,33 @@ async def test_submitted_task_result_requires_task_identity() -> None: assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_LEGACY_RESPONSE +@pytest.mark.asyncio +async def test_executor_result_rejects_signed_url_before_persistence() -> None: + case = _cases()[2] + store = InMemoryCompatibilityContinuationStore() + coordinator = _coordinator( + store, + lambda _ctx: { + "status": "submitted", + "task_id": "task-sensitive", + "message": "https://seller.example/task?token=secret", + }, + ) + await _issue(coordinator, case) + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.PERSISTENCE_POLICY + operation = await coordinator.get_legacy_purchase_operation_by_idempotency_key( + case["continuation_input"]["idempotency_key"], principal_id="principal-acme" + ) + assert operation.state.value == "ambiguous" + assert operation.result is None + + @pytest.mark.asyncio async def test_synchronous_executor_runs_off_event_loop_thread() -> None: case = _cases()[2] @@ -804,6 +1005,7 @@ async def test_31_without_replay_guarantee_requires_mutation_loss() -> None: with pytest.raises(CompatibilityContinuationError) as exc: await coordinator.issue_legacy_create_continuation( principal_id="principal-acme", + issuance_idempotency_key="invalid-projected-pricing", account=case["continuation_input"]["account"], source_adcp_version=case["source_version"], expires_at=datetime.fromisoformat( @@ -828,6 +1030,7 @@ async def test_replay_guarantee_requires_strict_boolean(invalid: Any) -> None: with pytest.raises(CompatibilityContinuationError) as exc: await coordinator.issue_legacy_create_continuation( principal_id="principal-acme", + issuance_idempotency_key="invalid-fixed-pricing", account=case["continuation_input"]["account"], source_adcp_version=case["source_version"], expires_at=datetime.fromisoformat( @@ -958,6 +1161,152 @@ async def test_issuance_rejects_source_patch_without_exact_bundled_schema() -> N assert "bundled source schema release" in str(exc.value) +@pytest.mark.asyncio +async def test_identical_issuance_is_idempotent_across_sqlite_restart(tmp_path: Path) -> None: + case = copy.deepcopy(_cases()[2]) + database = tmp_path / "continuations.sqlite3" + first = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + await _issue(first, case) + first_token = case["continuation_input"]["continuation_token"] + + restarted = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + await _issue(restarted, case) + assert case["continuation_input"]["continuation_token"] == first_token + with closing(sqlite3.connect(database)) as conn: + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 1 + + +@pytest.mark.asyncio +async def test_pre_fingerprint_authorization_blocks_duplicate_reissuance(tmp_path: Path) -> None: + case = copy.deepcopy(_cases()[2]) + database = tmp_path / "continuations.sqlite3" + first = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + await _issue(first, case) + with closing(sqlite3.connect(database)) as conn, conn: + conn.execute( + "UPDATE adcp_compat_continuations " + "SET token_hash = ?, issuance_fingerprint = NULL, issuance_binding_hash = NULL", + ("f" * 64,), + ) + + restarted = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(restarted, case) + assert exc.value.code == CompatibilityContinuationErrorCode.STORE_CONFLICT + with closing(sqlite3.connect(database)) as conn: + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 1 + + +@pytest.mark.asyncio +async def test_reused_issuance_key_with_changed_discovery_conflicts(tmp_path: Path) -> None: + case = copy.deepcopy(_cases()[2]) + database = tmp_path / "continuations.sqlite3" + coordinator = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + await _issue(coordinator, case) + changed = copy.deepcopy(case) + changed["legacy_request"]["brief"] = "A materially different brief." + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, changed) + assert exc.value.code == CompatibilityContinuationErrorCode.STORE_CONFLICT + with closing(sqlite3.connect(database)) as conn: + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 1 + + +@pytest.mark.asyncio +async def test_purged_issuance_key_with_changed_bindings_gets_different_token( + tmp_path: Path, +) -> None: + case = copy.deepcopy(_cases()[2]) + store = SqliteCompatibilityContinuationStore(tmp_path / "continuations.sqlite3") + coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-before-purge")) + await _issue(coordinator, case) + old_token = case["continuation_input"]["continuation_token"] + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert await store.purge_resolved_before(_NOW) == 1 + + changed = copy.deepcopy(case) + changed["legacy_request"]["brief"] = "A new discovery with changed authorization bindings." + await _issue(coordinator, changed) + assert changed["continuation_input"]["continuation_token"] != old_token + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "url", + [ + "https://creative.example/render?X-Amz-Signature=secret", + "https://creative.example/render?a=1;X-Amz-Signature=secret", + "https://creative.example/render?a=1%3BX-Amz-Signature=secret", + ], +) +async def test_issuance_rejects_presigned_url_before_persistence(tmp_path: Path, url: str) -> None: + case = copy.deepcopy(_cases()[2]) + case["legacy_response"]["products"][0]["format_ids"][0]["agent_url"] = url + coordinator = _coordinator( + SqliteCompatibilityContinuationStore(tmp_path / "continuations.sqlite3"), + lambda _ctx: {}, + ) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, case) + assert exc.value.code == CompatibilityContinuationErrorCode.PERSISTENCE_POLICY + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "field", + [ + "apiKey", + "clientSecret", + "auth_token", + "key", + "signature", + "accessKeyId", + "jwt", + "authorizationCode", + "APIKey", + "sellerAccessKEYID", + "credentialValue", + "accessTokenValue", + "accessTokenDataValue", + ], +) +async def test_issuance_rejects_credential_aliases_before_persistence( + tmp_path: Path, field: str +) -> None: + case = copy.deepcopy(_cases()[2]) + case["legacy_response"][field] = "must-not-persist" + database = tmp_path / "continuations.sqlite3" + coordinator = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, case) + assert exc.value.code == CompatibilityContinuationErrorCode.PERSISTENCE_POLICY + with closing(sqlite3.connect(database)) as conn: + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 0 + + +@pytest.mark.asyncio +async def test_redemption_rejects_push_notification_config_before_claim() -> None: + case = copy.deepcopy(_cases()[2]) + store = InMemoryCompatibilityContinuationStore() + coordinator = _coordinator(store, lambda _ctx: {}) + await _issue(coordinator, case) + case["continuation_input"]["legacy_create_request"]["push_notification_config"] = { + "url": "https://buyer.example/events", + } + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.PERSISTENCE_POLICY + assert not store._operations + + @pytest.mark.asyncio async def test_expiry_is_rejected_before_claim_or_call() -> None: case = _cases()[1] @@ -1058,11 +1407,230 @@ def test_production_default_rejects_in_memory_store() -> None: ) +def test_durable_coordinator_requires_stable_token_derivation_key(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="token_derivation_key"): + LegacyPurchaseCoordinator( + store=SqliteCompatibilityContinuationStore(tmp_path / "continuations.sqlite3"), + executor=lambda _ctx: {}, + ) + + with pytest.raises(TypeError, match="bytes-like"): + LegacyPurchaseCoordinator( + store=SqliteCompatibilityContinuationStore(tmp_path / "integer-key.sqlite3"), + executor=lambda _ctx: {}, + token_derivation_key=32, # type: ignore[arg-type] + ) + + with pytest.raises(ValueError, match="high-entropy"): + LegacyPurchaseCoordinator( + store=SqliteCompatibilityContinuationStore(tmp_path / "zero-key.sqlite3"), + executor=lambda _ctx: {}, + token_derivation_key=b"\x00" * 32, + ) + + def test_sqlite_store_rejects_memory_database() -> None: with pytest.raises(ValueError, match="file-backed"): SqliteCompatibilityContinuationStore(":memory:") +def test_sqlite_store_validates_positive_quotas(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="max_records"): + SqliteCompatibilityContinuationStore(tmp_path / "continuations.sqlite3", max_records=0) + + +@pytest.mark.asyncio +async def test_sqlite_record_quota_rolls_back_claim_and_can_be_retried(tmp_path: Path) -> None: + case = copy.deepcopy(_cases()[2]) + database = tmp_path / "continuations.sqlite3" + constrained = _coordinator( + SqliteCompatibilityContinuationStore(database, max_records=1), lambda _ctx: {} + ) + await _issue(constrained, case) + with pytest.raises(CompatibilityContinuationError) as exc: + await constrained.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED + + calls = 0 + + def execute(ctx: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + return _success_for(ctx, "mb-after-capacity") + + reopened = _coordinator(SqliteCompatibilityContinuationStore(database, max_records=2), execute) + result = await reopened.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert result == _success_result(case["source_version"], "mb-after-capacity") + assert calls == 1 + + +@pytest.mark.asyncio +async def test_sqlite_principal_quota_does_not_consume_other_principal_capacity( + tmp_path: Path, +) -> None: + store = SqliteCompatibilityContinuationStore( + tmp_path / "continuations.sqlite3", + max_records=10, + max_records_per_principal=1, + ) + coordinator = _coordinator(store, lambda _ctx: {}) + first = copy.deepcopy(_cases()[1]) + await _issue(coordinator, first, principal="principal-one") + + same_principal = copy.deepcopy(_cases()[2]) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, same_principal, principal="principal-one") + assert exc.value.code == CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED + + other_principal = copy.deepcopy(_cases()[2]) + await _issue(coordinator, other_principal, principal="principal-two") + with closing(sqlite3.connect(store.path)) as conn: + assert conn.execute("SELECT COUNT(*) FROM adcp_compat_continuations").fetchone()[0] == 2 + + +@pytest.mark.asyncio +async def test_sqlite_payload_quota_rejects_large_discovery(tmp_path: Path) -> None: + case = copy.deepcopy(_cases()[2]) + case["legacy_response"]["products"][0]["description"] = "x" * 5_000 + coordinator = _coordinator( + SqliteCompatibilityContinuationStore( + tmp_path / "continuations.sqlite3", max_payload_bytes=1_000 + ), + lambda _ctx: {}, + ) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, case) + assert exc.value.code == CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED + + +@pytest.mark.asyncio +async def test_sqlite_reserves_terminal_payload_before_executor_runs(tmp_path: Path) -> None: + case = copy.deepcopy(_cases()[2]) + calls = 0 + + def execute(_ctx: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + return {} + + coordinator = _coordinator( + SqliteCompatibilityContinuationStore( + tmp_path / "continuations.sqlite3", + max_bytes=100_000, + max_payload_bytes=1_000_000, + ), + execute, + ) + await _issue(coordinator, case) + + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + + assert exc.value.code == CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED + assert calls == 0 + + +@pytest.mark.asyncio +async def test_sqlite_quota_cannot_strand_terminal_write_after_execution( + tmp_path: Path, +) -> None: + case = copy.deepcopy(_cases()[2]) + entered = asyncio.Event() + release = asyncio.Event() + + async def execute(ctx: Any) -> dict[str, Any]: + entered.set() + await release.wait() + return _success_for(ctx, "mb-reserved") + + store = SqliteCompatibilityContinuationStore(tmp_path / "continuations.sqlite3") + coordinator = _coordinator(store, execute) + await _issue(coordinator, case) + purchase = asyncio.create_task( + coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + ) + await entered.wait() + + # Simulate ledger capacity being tightened or exhausted after the seller + # mutation starts. Its pre-reserved terminal write must still commit. + store.max_bytes = 1 + store.max_bytes_per_principal = 1 + store.max_payload_bytes = 1 + release.set() + + assert await purchase == _success_result(case["source_version"], "mb-reserved") + + +@pytest.mark.asyncio +async def test_sqlite_persists_reservation_across_worker_quota_configs( + tmp_path: Path, +) -> None: + case = copy.deepcopy(_cases()[2]) + database = tmp_path / "continuations.sqlite3" + + async def execute(_ctx: Any) -> dict[str, Any]: + raise TimeoutError + + initial = _coordinator(SqliteCompatibilityContinuationStore(database), execute) + await _issue(initial, case) + with pytest.raises(CompatibilityContinuationError) as ambiguous: + await initial.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert ambiguous.value.code == CompatibilityContinuationErrorCode.AMBIGUOUS_MUTATION + + async def reconcile(_ctx: Any, _operation: Any) -> ReconciliationResult: + return ReconciliationResult.applied( + _success_result(case["source_version"], "mb-cross-worker") + ) + + constrained_store = SqliteCompatibilityContinuationStore( + database, + max_bytes=100_000, + max_payload_bytes=10_000, + ) + recovered = _coordinator(constrained_store, execute, reconciler=reconcile) + + # The second worker must account for the first worker's durable 1 MiB + # reservation rather than substituting its own one-byte payload setting. + other = copy.deepcopy(_cases()[1]) + with pytest.raises(CompatibilityContinuationError) as quota: + await _issue( + recovered, + other, + principal="principal-other", + target="seller-session-other", + ) + assert quota.value.code == CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED + + # Its smaller current setting also cannot invalidate the existing + # operation's already-reserved terminal result. + result = await recovered.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert result == _success_result(case["source_version"], "mb-cross-worker") + + def test_sqlite_store_creates_private_ledger_and_rejects_loose_existing_file( tmp_path: Path, ) -> None: @@ -1076,6 +1644,19 @@ def test_sqlite_store_creates_private_ledger_and_rejects_loose_existing_file( SqliteCompatibilityContinuationStore(loose) +def test_concurrent_first_startup_safely_creates_missing_parent(tmp_path: Path) -> None: + database = tmp_path / "new" / "nested" / "continuations.sqlite3" + barrier = threading.Barrier(2) + + def construct() -> SqliteCompatibilityContinuationStore: + barrier.wait() + return SqliteCompatibilityContinuationStore(database) + + with ThreadPoolExecutor(max_workers=2) as executor: + stores = list(executor.map(lambda _index: construct(), range(2))) + assert len(stores) == 2 + + def test_sqlite_store_creates_private_parent_directory(tmp_path: Path) -> None: parent = tmp_path / "private-ledger" SqliteCompatibilityContinuationStore(parent / "continuations.sqlite3") @@ -1223,6 +1804,29 @@ def synchronized_migration( assert {"created_at", "updated_at"}.issubset(columns) +@pytest.mark.asyncio +async def test_migration_audits_and_rejects_existing_secret_payload(tmp_path: Path) -> None: + case = copy.deepcopy(_cases()[2]) + database = tmp_path / "continuations.sqlite3" + coordinator = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + await _issue(coordinator, case) + with closing(sqlite3.connect(database)) as conn, conn: + raw = conn.execute( + "SELECT observed_response_json FROM adcp_compat_continuations" + ).fetchone()[0] + payload = json.loads(raw) + payload["clientSecret"] = "legacy-secret" + conn.execute( + "UPDATE adcp_compat_continuations SET observed_response_json = ?", + (json.dumps(payload),), + ) + conn.execute("DELETE FROM adcp_compat_metadata WHERE key = 'persistence_policy_version'") + + with pytest.raises(CompatibilityContinuationError) as exc: + SqliteCompatibilityContinuationStore(database) + assert exc.value.code == CompatibilityContinuationErrorCode.PERSISTENCE_POLICY + + @pytest.mark.asyncio async def test_sqlite_migrates_origin_ledger_and_adopts_exact_retry_input( tmp_path: Path, @@ -1246,7 +1850,7 @@ async def uncertain(_ctx: Any) -> dict[str, Any]: with sqlite3.connect(database) as conn: conn.execute("ALTER TABLE adcp_compat_continuations DROP COLUMN projected_products_json") conn.execute( - "ALTER TABLE adcp_compat_continuations " "DROP COLUMN mutation_idempotency_guaranteed" + "ALTER TABLE adcp_compat_continuations DROP COLUMN mutation_idempotency_guaranteed" ) conn.execute("ALTER TABLE adcp_compat_operations DROP COLUMN revision") conn.execute("ALTER TABLE adcp_compat_operations DROP COLUMN execution_input_json") @@ -1325,7 +1929,7 @@ async def test_migrated_31_ledger_without_bound_replay_guarantee_fails_closed( coordinator = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) await _issue(coordinator, case) with sqlite3.connect(database) as conn: - conn.execute("UPDATE adcp_compat_continuations " "SET mutation_idempotency_guaranteed = 0") + conn.execute("UPDATE adcp_compat_continuations SET mutation_idempotency_guaranteed = 0") with pytest.raises(CompatibilityContinuationError) as exc: await coordinator.continue_legacy_purchase( case["continuation_input"], @@ -1358,7 +1962,7 @@ def execute(ctx: Any) -> dict[str, Any]: with sqlite3.connect(database) as conn: conn.execute("ALTER TABLE adcp_compat_continuations DROP COLUMN projected_products_json") conn.execute( - "ALTER TABLE adcp_compat_continuations " "DROP COLUMN mutation_idempotency_guaranteed" + "ALTER TABLE adcp_compat_continuations DROP COLUMN mutation_idempotency_guaranteed" ) restarted = _coordinator(SqliteCompatibilityContinuationStore(database), execute) replay = await restarted.continue_legacy_purchase(