diff --git a/.github/workflows/linter.yaml b/.github/workflows/linter.yaml index fea1b9c1..663a5b19 100644 --- a/.github/workflows/linter.yaml +++ b/.github/workflows/linter.yaml @@ -4,6 +4,11 @@ on: pull_request: branches: [main] +permissions: + contents: read + statuses: write + pull-requests: write + jobs: build: name: Lint Code Base @@ -11,12 +16,13 @@ jobs: 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 diff --git a/code/samples/python/src/common/durable_reservation_store.py b/code/samples/python/src/common/durable_reservation_store.py new file mode 100644 index 00000000..a124c452 --- /dev/null +++ b/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() diff --git a/code/samples/python/src/roles/credentials_provider_mcp/server.py b/code/samples/python/src/roles/credentials_provider_mcp/server.py index 3aa2c42e..5f8f7206 100644 --- a/code/samples/python/src/roles/credentials_provider_mcp/server.py +++ b/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") diff --git a/code/samples/python/src/roles/merchant_payment_processor_mcp/server.py b/code/samples/python/src/roles/merchant_payment_processor_mcp/server.py index e8aa5066..e2081a48 100644 --- a/code/samples/python/src/roles/merchant_payment_processor_mcp/server.py +++ b/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, diff --git a/code/samples/python/tests/durable_reservation_store_tests.py b/code/samples/python/tests/durable_reservation_store_tests.py new file mode 100644 index 00000000..8f3db074 --- /dev/null +++ b/code/samples/python/tests/durable_reservation_store_tests.py @@ -0,0 +1,82 @@ +"""Regression tests for durable consume-once reservations.""" + +import multiprocessing +import sqlite3 + +from pathlib import Path + +import pytest + +from common.durable_reservation_store import ( + ReservationStoreError, + reserve_once, +) + + +def _reserve_in_process(arguments: tuple[str, str]) -> bool: + """Reserve a key from an independent process.""" + database_path, reservation_key = arguments + return reserve_once(Path(database_path), reservation_key) + + +def test_sequential_replay_is_refused(tmp_path: Path): + database_path = tmp_path / 'consumed.sqlite3' + + assert reserve_once(database_path, 'closed-mandate-hash') + assert not reserve_once(database_path, 'closed-mandate-hash') + + +def test_concurrent_processes_allow_exactly_one_reservation(tmp_path: Path): + database_path = tmp_path / 'consumed.sqlite3' + arguments = [(str(database_path), 'shared-transaction') for _ in range(8)] + + context = multiprocessing.get_context('spawn') + with context.Pool(processes=8) as pool: + results = pool.map(_reserve_in_process, arguments) + + assert results.count(True) == 1 + assert results.count(False) == 7 + + +def test_reservation_survives_a_new_store_instance(tmp_path: Path): + database_path = tmp_path / 'consumed.sqlite3' + + assert reserve_once(database_path, 'checkout-hash') + + # A fresh call opens a fresh database connection, as a restarted service + # would, and must still observe the durable reservation. + assert not reserve_once(database_path, 'checkout-hash') + + +def test_corrupt_state_fails_closed(tmp_path: Path): + database_path = tmp_path / 'consumed.sqlite3' + database_path.write_bytes(b'not a sqlite database') + + with pytest.raises(ReservationStoreError): + reserve_once(database_path, 'closed-mandate-hash') + + +def test_unreadable_state_fails_closed(tmp_path: Path): + database_path = tmp_path / 'directory-instead-of-database' + database_path.mkdir() + + with pytest.raises(ReservationStoreError): + reserve_once(database_path, 'closed-mandate-hash') + + +def test_busy_state_fails_closed_when_lock_timeout_expires(tmp_path: Path): + database_path = tmp_path / 'consumed.sqlite3' + assert reserve_once(database_path, 'existing-reservation') + + locking_connection = sqlite3.connect(database_path, isolation_level=None) + locking_connection.execute('BEGIN IMMEDIATE') + try: + with pytest.raises(ReservationStoreError): + reserve_once( + database_path, + 'new-reservation', + timeout_seconds=0, + ) + finally: + locking_connection.rollback() + locking_connection.close() diff --git a/code/samples/python/tests/payment_mandate_consumption_tests.py b/code/samples/python/tests/payment_mandate_consumption_tests.py new file mode 100644 index 00000000..bb931a5a --- /dev/null +++ b/code/samples/python/tests/payment_mandate_consumption_tests.py @@ -0,0 +1,131 @@ +"""Credential-provider integration tests for consume-once mandates.""" + +import json + +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +from roles.credentials_provider_mcp import server + + +class _VerifiedChain: + @staticmethod + def verify(**_expected_bindings): + return [] + + +class _PaymentMandateChain: + @staticmethod + def parse(_payloads): + return _VerifiedChain() + + +class _MandateClient: + @staticmethod + def verify(**_verification_inputs): + return ['verified-payload'] + + @staticmethod + def get_closed_mandate_jwt(_payment_mandate_chain): + return 'closed-mandate-jwt' + + +def _agent_provider_public_key(): + return object() + + +def _configure_verified_mandate(monkeypatch, tmp_path: Path): + monkeypatch.setattr( + server, '_load_persisted_mandate', lambda _filename: 'mandate-chain' + ) + monkeypatch.setattr( + server, '_get_agent_provider_public_key', _agent_provider_public_key + ) + monkeypatch.setattr(server, 'MandateClient', _MandateClient) + monkeypatch.setattr(server, 'PaymentMandateChain', _PaymentMandateChain) + monkeypatch.setattr( + server, 'compute_sha256_b64url', lambda _value: 'closed-mandate-hash' + ) + monkeypatch.setattr(server, '_TOKEN_STORE_PATH', tmp_path / 'tokens.json') + monkeypatch.setattr( + server, + '_CONSUMED_MANDATES_PATH', + tmp_path / 'consumed-mandates.sqlite3', + ) + + +def _issue_payment_credential(): + return server.issue_payment_credential( + payment_mandate_chain_id='pay_123', + open_checkout_hash='open-checkout-hash', + checkout_jwt_hash='checkout-jwt-hash', + payment_nonce='payment-nonce', + ) + + +def _issue_payment_credential_for_worker(_worker_index): + return _issue_payment_credential() + + +def test_credential_provider_refuses_sequential_replay(monkeypatch, tmp_path): + _configure_verified_mandate(monkeypatch, tmp_path) + + first = _issue_payment_credential() + replay = _issue_payment_credential() + + assert 'payment_token' in first + assert replay['error'] == 'mandate_already_used' + + +def test_credential_provider_refuses_reference_from_legacy_store( + monkeypatch, tmp_path +): + _configure_verified_mandate(monkeypatch, tmp_path) + server._TOKEN_STORE_PATH.write_text( + json.dumps({'closed-mandate-hash': {'token': 'legacy-token'}}), + encoding='utf-8', + ) + + replay = _issue_payment_credential() + + assert replay['error'] == 'mandate_already_used' + assert not server._CONSUMED_MANDATES_PATH.exists() + + +def test_credential_provider_refuses_concurrent_replay(monkeypatch, tmp_path): + _configure_verified_mandate(monkeypatch, tmp_path) + + with ThreadPoolExecutor(max_workers=8) as executor: + results = list( + executor.map(_issue_payment_credential_for_worker, range(8)) + ) + + assert sum('payment_token' in result for result in results) == 1 + assert ( + sum(result.get('error') == 'mandate_already_used' for result in results) + == 7 + ) + + +def test_credential_provider_fails_closed_on_corrupt_replay_state( + monkeypatch, tmp_path +): + _configure_verified_mandate(monkeypatch, tmp_path) + server._CONSUMED_MANDATES_PATH.write_bytes(b'not a sqlite database') + + result = _issue_payment_credential() + + assert result['error'] == 'replay_state_unavailable' + assert 'payment_token' not in result + + +def test_credential_provider_fails_closed_on_unreadable_replay_state( + monkeypatch, tmp_path +): + _configure_verified_mandate(monkeypatch, tmp_path) + server._CONSUMED_MANDATES_PATH.mkdir() + + result = _issue_payment_credential() + + assert result['error'] == 'replay_state_unavailable' + assert 'payment_token' not in result diff --git a/code/samples/python/tests/payment_processor_consumption_tests.py b/code/samples/python/tests/payment_processor_consumption_tests.py new file mode 100644 index 00000000..1577326a --- /dev/null +++ b/code/samples/python/tests/payment_processor_consumption_tests.py @@ -0,0 +1,161 @@ +"""Payment-processor integration tests for consume-once transactions.""" + +import asyncio +import json + +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace + +from roles.merchant_payment_processor_mcp import server + + +class _VerifiedChain: + closed_mandate = object() + + @staticmethod + def verify(**_expected_bindings): + return [] + + +class _PaymentMandateChain: + @staticmethod + def parse(_payloads): + return _VerifiedChain() + + +class _MandateClient: + @staticmethod + def verify(**_verification_inputs): + return ['verified-payload'] + + @staticmethod + def get_closed_mandate_jwt(_payment_mandate_chain): + return 'closed-mandate-jwt' + + +class _ReceiptContent: + root = SimpleNamespace(payment_id='payment-123') + + @staticmethod + def model_dump(): + return {} + + @staticmethod + def model_dump_json(): + return '{}' + + +class _ReceiptClient: + @staticmethod + def create_payment_receipt(**_receipt_inputs): + return _ReceiptContent() + + +def _agent_provider_public_key(): + return object() + + +def _configure_verified_payment(monkeypatch, tmp_path: Path): + token_store_path = tmp_path / 'tokens.json' + token_store_path.write_text( + json.dumps( + { + 'token-123': { + 'used': True, + 'payment_mandate_chain': 'mandate-chain', + 'payment_nonce': 'payment-nonce', + 'order_id': 'order-123', + } + } + ), + encoding='utf-8', + ) + monkeypatch.setattr(server, '_TOKEN_STORE_PATH', token_store_path) + monkeypatch.setattr( + server, + '_CONSUMED_TRANSACTIONS_PATH', + tmp_path / 'consumed-transactions.sqlite3', + ) + monkeypatch.setattr( + server, '_get_agent_provider_public_key', _agent_provider_public_key + ) + monkeypatch.setattr(server, 'MandateClient', _MandateClient) + monkeypatch.setattr(server, 'PaymentMandateChain', _PaymentMandateChain) + monkeypatch.setattr(server, 'ReceiptClient', _ReceiptClient) + monkeypatch.setattr(server, 'create_jwt', lambda **_inputs: 'receipt-jwt') + monkeypatch.setattr( + server, + '_get_merchant_payment_processor_signing_key', + lambda **_inputs: SimpleNamespace(kid='payment-processor-key'), + ) + + async def _discard_receipt(_receipt): + return None + + monkeypatch.setattr( + server, + '_send_payment_receipt_to_credentials_provider', + _discard_receipt, + ) + + +def _initiate_payment(): + return asyncio.run( + server.initiate_payment( + payment_token='token-123', + checkout_jwt_hash='checkout-jwt-hash', + open_checkout_hash='open-checkout-hash', + ) + ) + + +def _initiate_payment_for_worker(_worker_index): + return _initiate_payment() + + +def test_payment_processor_refuses_sequential_replay(monkeypatch, tmp_path): + _configure_verified_payment(monkeypatch, tmp_path) + + first = _initiate_payment() + replay = _initiate_payment() + + assert first['status'] == 'success' + assert replay['error'] == 'mandate_already_used' + + +def test_payment_processor_refuses_concurrent_replay(monkeypatch, tmp_path): + _configure_verified_payment(monkeypatch, tmp_path) + + with ThreadPoolExecutor(max_workers=8) as executor: + results = list(executor.map(_initiate_payment_for_worker, range(8))) + + assert sum(result.get('status') == 'success' for result in results) == 1 + assert ( + sum(result.get('error') == 'mandate_already_used' for result in results) + == 7 + ) + + +def test_payment_processor_fails_closed_on_corrupt_replay_state( + monkeypatch, tmp_path +): + _configure_verified_payment(monkeypatch, tmp_path) + server._CONSUMED_TRANSACTIONS_PATH.write_bytes(b'not a sqlite database') + + result = _initiate_payment() + + assert result['error'] == 'replay_state_unavailable' + assert 'payment_receipt' not in result + + +def test_payment_processor_fails_closed_on_unreadable_replay_state( + monkeypatch, tmp_path +): + _configure_verified_payment(monkeypatch, tmp_path) + server._CONSUMED_TRANSACTIONS_PATH.mkdir() + + result = _initiate_payment() + + assert result['error'] == 'replay_state_unavailable' + assert 'payment_receipt' not in result