fix(samples): consume accepted Payment Mandates atomically

.github/workflows/linter.yaml

@@ -4,19 +4,25 @@ on:
   pull_request:
     branches: [main]
 
+permissions:
+  contents: read
+  statuses: write
+  pull-requests: write
+
 jobs:
   build:
     name: Lint Code Base
     runs-on: ubuntu-latest
 
     steps:
       - name: Checkout Code
-        uses: actions/checkout@v5
+        uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5.1.0
         with:
           fetch-depth: 0
+          persist-credentials: false
 
       - name: Lint Code Base
-        uses: super-linter/super-linter/slim@v8
+        uses: super-linter/super-linter/slim@4ce20838b8ab83717e78138c5b3a1407148e0918 # v8.7.0
         env:
           DEFAULT_BRANCH: main
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
@@ -25,6 +31,7 @@ jobs:
           VALIDATE_ALL_CODEBASE: false
           FILTER_REGEX_EXCLUDE: "^(\\.github/|\\.vscode/|code/samples/).*|CODE_OF_CONDUCT.md|CHANGELOG.md"
           VALIDATE_BIOME_FORMAT: false
+          VALIDATE_BIOME_LINT: false
           VALIDATE_PYTHON_BLACK: false
           VALIDATE_PYTHON_FLAKE8: false
           VALIDATE_PYTHON_ISORT: false

code/samples/python/src/common/durable_reservation_store.py

@@ -0,0 +1,92 @@
+"""Durable, single-host consume-once reservations."""
+
+import sqlite3
+
+from pathlib import Path
+
+
+class ReservationStoreError(RuntimeError):
+    """Raised when replay state cannot be read or durably updated."""
+
+
+def reserve_once(
+    database_path: Path,
+    reservation_key: str,
+    *,
+    timeout_seconds: float = 5.0,
+) -> bool:
+    """Atomically reserve ``reservation_key`` in durable local state.
+
+    A separate SQLite connection is used for every call so independent threads
+    and processes on one host share the same transaction boundary.
+    ``BEGIN IMMEDIATE`` serializes contenders before the existence check, and
+    ``synchronous=FULL`` asks SQLite to synchronize the commit before a
+    successful return.
+
+    Args:
+      database_path: SQLite database holding permanent reservations.
+      reservation_key: Presenter-invariant mandate or transaction identifier.
+      timeout_seconds: Maximum time to wait for another writer.
+
+    Returns:
+      ``True`` only for the caller that created the reservation, or ``False``
+      when the key was already reserved.
+
+    Raises:
+      ValueError: If the key or timeout is invalid.
+      ReservationStoreError: If state is corrupt, unreadable, locked beyond the
+        timeout, or cannot be committed. Callers must treat this as a refusal.
+    """
+    if not reservation_key:
+        raise ValueError('reservation_key must not be empty')
+    if timeout_seconds < 0:
+        raise ValueError('timeout_seconds must not be negative')
+
+    connection: sqlite3.Connection | None = None
+    try:
+        database_path.parent.mkdir(parents=True, exist_ok=True)
+        connection = sqlite3.connect(
+            database_path,
+            timeout=timeout_seconds,
+            isolation_level=None,
+        )
+        connection.execute('PRAGMA synchronous = FULL')
+        connection.execute('BEGIN IMMEDIATE')
+
+        integrity_result = connection.execute('PRAGMA quick_check').fetchall()
+        if integrity_result != [('ok',)]:
+            raise ReservationStoreError(
+                'reservation state failed integrity check'
+            )
+
+        connection.execute(
+            """
+        CREATE TABLE IF NOT EXISTS reservations (
+          reservation_key TEXT PRIMARY KEY,
+          reserved_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+        ) WITHOUT ROWID
+        """
+        )
+        existing = connection.execute(
+            'SELECT 1 FROM reservations WHERE reservation_key = ?',
+            (reservation_key,),
+        ).fetchone()
+        if existing:
+            connection.rollback()
+            return False
+
+        connection.execute(
+            'INSERT INTO reservations (reservation_key) VALUES (?)',
+            (reservation_key,),
+        )
+        connection.commit()
+        return True
+    except ReservationStoreError:
+        raise
+    except (OSError, sqlite3.Error) as exc:
+        raise ReservationStoreError(
+            f'reservation state unavailable at {database_path}'
+        ) from exc
+    finally:
+        if connection is not None:
+            connection.close()

code/samples/python/src/roles/credentials_provider_mcp/server.py

@@ -9,6 +9,8 @@
 inside ``issue_payment_credential`` before a payment token is issued.
 """
 
+# cspell:ignore fastmcp levelname sdjwt
+
 import json
 import logging
 import os
@@ -28,6 +30,10 @@
   MERCHANT_PAYMENT_PROCESSOR_PUB_PATH,
   TEMP_DB,
 )
+from common.durable_reservation_store import (
+  ReservationStoreError,
+  reserve_once,
+)
 from fastmcp import FastMCP
 from fastmcp.server.middleware.logging import LoggingMiddleware
 from jwcrypto.jwk import JWK
@@ -63,6 +69,13 @@
     )
 )
 
+_CONSUMED_MANDATES_PATH = Path(
+    os.environ.get(
+        "AP2_CONSUMED_MANDATES_PATH",
+        str(TEMP_DB / "ap2_consumed_mandates.sqlite3"),
+    )
+)
+
 _TOKEN_EXPIRY_SECONDS = 300
 
 
@@ -77,13 +90,26 @@ def _load_token_store() -> dict[str, Any]:
   return {}
 
 
-def _save_token_store(store: dict[str, Any]) -> None:
+def _load_token_store_for_update() -> dict[str, Any]:
+  """Load credential state without treating unreadable state as empty."""
+  try:
+    with open(_TOKEN_STORE_PATH) as store_file:
+      return json.load(store_file)
+  except FileNotFoundError:
+    return {}
+  except (json.JSONDecodeError, OSError) as exc:
+    raise RuntimeError("credential store is unavailable") from exc
+
+
+def _save_token_store(store: dict[str, Any]) -> bool:
   try:
     _TOKEN_STORE_PATH.parent.mkdir(parents=True, exist_ok=True)
     with open(_TOKEN_STORE_PATH, "w") as f:
       json.dump(store, f, indent=2)
+    return True
   except OSError:
     _logger.exception("token_store save failed")
+    return False
 
 
 def _get_agent_provider_public_key() -> JWK | None:
@@ -214,10 +240,52 @@ def issue_payment_credential(
           "message": "; ".join(violations),
       }
 
-    token = "tok_" + str(uuid.uuid4()).replace("-", "")
     reference = compute_sha256_b64url(
         MandateClient().get_closed_mandate_jwt(payment_mandate_chain)
     )
+    try:
+      store = _load_token_store_for_update()
+    except RuntimeError:
+      _logger.exception("credential state is unavailable")
+      return {
+          "error": "credential_state_unavailable",
+          "message": (
+              "payment credential state could not be read; refusing to issue"
+              " a payment credential"
+          ),
+      }
+    if reference in store:
+      return {
+          "error": "mandate_already_used",
+          "message": (
+              "this closed Payment Mandate was already accepted by this"
+              " Credential Provider; present a new mandate for a new payment"
+          ),
+      }
+    # Consume-once for this sample: a closed Payment Mandate already accepted
+    # by this Credential Provider is not accepted again. The hash of the closed
+    # mandate (the same value used as the receipt `reference`) is the
+    # presenter-invariant key; a fresh token per presentation is not.
+    try:
+      reserved = reserve_once(_CONSUMED_MANDATES_PATH, reference)
+    except ReservationStoreError:
+      _logger.exception("closed-mandate replay state is unavailable")
+      return {
+          "error": "replay_state_unavailable",
+          "message": (
+              "closed Payment Mandate consumption state could not be"
+              " verified; refusing to issue a payment credential"
+          ),
+      }
+    if not reserved:
+      return {
+          "error": "mandate_already_used",
+          "message": (
+              "this closed Payment Mandate was already accepted by this"
+              " Credential Provider; present a new mandate for a new payment"
+          ),
+      }
+    token = "tok_" + str(uuid.uuid4()).replace("-", "")
     expires_at = int(time.time()) + _TOKEN_EXPIRY_SECONDS
     token_data = {
         "token": token,
@@ -229,7 +297,6 @@ def issue_payment_credential(
     }
 
     # Store the token data with both the token and the reference as keys.
-    store = _load_token_store()
     store.update(
         dict.fromkeys(
             [
@@ -239,7 +306,14 @@ def issue_payment_credential(
             token_data,
         )
     )
-    _save_token_store(store)
+    if not _save_token_store(store):
+      return {
+          "error": "credential_state_unavailable",
+          "message": (
+              "payment credential state could not be saved; the mandate"
+              " remains consumed"
+          ),
+      }
 
     _logger.info("issue_payment_credential result: token=%s...", token[:16])
     return {
@@ -265,7 +339,8 @@ def revoke_payment_credential(payment_token: str) -> Mapping[str, Any]:
   store = _load_token_store()
   if payment_token in store:
     del store[payment_token]
-    _save_token_store(store)
+    if not _save_token_store(store):
+      return {"revoked": False, "error": "credential_state_unavailable"}
     _logger.info("revoke_payment_credential result: revoked=True")
     return {"revoked": True}
   _logger.warning("revoke_payment_credential result: token_not_found")

code/samples/python/src/roles/merchant_payment_processor_mcp/server.py

@@ -9,6 +9,8 @@
 Checkout JWTs are properly ES256-signed instead of using stubs.
 """
 
+# cspell:ignore fastmcp levelname pyca SECP
+
 import json
 import logging
 import os
@@ -30,6 +32,10 @@
   MERCHANT_PAYMENT_PROCESSOR_PUB_PATH,
   TEMP_DB,
 )
+from common.durable_reservation_store import (
+  ReservationStoreError,
+  reserve_once,
+)
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric import ec
 from fastmcp import FastMCP
@@ -112,6 +118,14 @@ def _get_merchant_payment_processor_signing_key(
 # ── Store helpers ───────────────────────────────────────────────────────
 
 
+_CONSUMED_TRANSACTIONS_PATH = Path(
+    os.environ.get(
+        "AP2_CONSUMED_TRANSACTIONS_PATH",
+        str(TEMP_DB / "ap2_consumed_transactions.sqlite3"),
+    )
+)
+
+
 def _load_token_store() -> dict[str, Any]:
   try:
     if _TOKEN_STORE_PATH.exists():
@@ -274,6 +288,29 @@ async def initiate_payment(
         "message": "; ".join(violations),
     }
 
+  # Consume-once at the processor sample: one closed Payment Mandate / Checkout
+  # reaches receipt creation once. The consumed key is the transaction_id
+  # (hash of the checkout JWT), which the presenter cannot vary; it is recorded
+  # before receipt creation so a concurrent or later replay is refused.
+  try:
+    reserved = reserve_once(_CONSUMED_TRANSACTIONS_PATH, checkout_jwt_hash)
+  except ReservationStoreError:
+    _logger.exception("transaction replay state is unavailable")
+    return {
+        "error": "replay_state_unavailable",
+        "message": (
+            "transaction consumption state could not be verified; refusing"
+            " to initiate payment"
+        ),
+    }
+  if not reserved:
+    return {
+        "error": "mandate_already_used",
+        "message": (
+            "a payment for this checkout was already processed under this"
+            " Payment Mandate"
+        ),
+    }
   _logger.info(
       "Creating payment receipt with order_id=%s",
       order_id,