diff --git a/src/mpp/__init__.py b/src/mpp/__init__.py index 77d0545..914d4fa 100644 --- a/src/mpp/__init__.py +++ b/src/mpp/__init__.py @@ -8,13 +8,13 @@ import base64 import hashlib import hmac -import json from dataclasses import dataclass from datetime import UTC, datetime from typing import Any, Literal from mpp._parsing import ( ParseError, + canonical_json, format_authorization, format_payment_receipt, format_www_authenticate, @@ -105,12 +105,12 @@ def generate_challenge_id( request={"amount": "1000000", "currency": "0x...", "recipient": "0x..."}, ) """ - request_json = json.dumps(request, separators=(",", ":"), sort_keys=True, ensure_ascii=False) + request_json = canonical_json(request) request_b64 = _b64url_encode(request_json) opaque_b64 = "" if opaque is not None: - opaque_json = json.dumps(opaque, separators=(",", ":"), sort_keys=True, ensure_ascii=False) + opaque_json = canonical_json(opaque) opaque_b64 = _b64url_encode(opaque_json) hmac_input = "|".join( @@ -217,12 +217,7 @@ def create( digest=digest, opaque=meta, ) - request_json = json.dumps( - request, - separators=(",", ":"), - sort_keys=True, - ensure_ascii=False, - ) + request_json = canonical_json(request) request_b64 = _b64url_encode(request_json) return cls( id=challenge_id, @@ -281,12 +276,7 @@ def to_echo(self) -> ChallengeEcho: """ opaque_b64 = None if self.opaque is not None: - opaque_json = json.dumps( - self.opaque, - separators=(",", ":"), - sort_keys=True, - ensure_ascii=False, - ) + opaque_json = canonical_json(self.opaque) opaque_b64 = _b64url_encode(opaque_json) return ChallengeEcho( id=self.id, diff --git a/src/mpp/_body_digest.py b/src/mpp/_body_digest.py index 58957ad..82be254 100644 --- a/src/mpp/_body_digest.py +++ b/src/mpp/_body_digest.py @@ -7,9 +7,10 @@ import base64 import hashlib import hmac -import json from typing import Any +from mpp._parsing import canonical_json + def compute(body: str | bytes | dict[str, Any]) -> str: """Compute a SHA-256 digest of a request body. @@ -21,7 +22,7 @@ def compute(body: str | bytes | dict[str, Any]) -> str: Digest in the format ``sha-256=``. """ if isinstance(body, dict): - body = json.dumps(body, separators=(",", ":"), sort_keys=True, ensure_ascii=False) + body = canonical_json(body) if isinstance(body, str): body = body.encode("utf-8") digest = hashlib.sha256(body).digest() diff --git a/src/mpp/_parsing.py b/src/mpp/_parsing.py index 5d8d17e..5d161d1 100644 --- a/src/mpp/_parsing.py +++ b/src/mpp/_parsing.py @@ -38,9 +38,20 @@ class ParseError(Exception): """ +def canonical_json(data: Any) -> str: + """Serialize a value as canonical compact JSON. + + Sorted keys, no whitespace, and raw UTF-8 (never \\uXXXX escapes). The + challenge id HMAC binds the base64url encoding of this exact form, so every + site that produces or re-derives a wire value must go through here or the + id will not reproduce (draft-httpauth-payment-00 section 5.1.1). + """ + return json.dumps(data, separators=(",", ":"), sort_keys=True, ensure_ascii=False) + + def _b64_encode(data: dict[str, Any]) -> str: """Encode dict as URL-safe base64 JSON (compact, no padding).""" - compact_json = json.dumps(data, separators=(",", ":"), sort_keys=True) + compact_json = canonical_json(data) encoded = base64.urlsafe_b64encode(compact_json.encode()).decode() return encoded.rstrip("=") diff --git a/src/mpp/extensions/mcp/types.py b/src/mpp/extensions/mcp/types.py index 1196b4f..5c6d67d 100644 --- a/src/mpp/extensions/mcp/types.py +++ b/src/mpp/extensions/mcp/types.py @@ -15,6 +15,8 @@ from datetime import datetime from typing import TYPE_CHECKING, Any, Literal +from mpp._parsing import canonical_json + if TYPE_CHECKING: from mpp import Challenge, Credential, Receipt @@ -85,11 +87,10 @@ def from_dict(cls, data: dict[str, Any]) -> MCPChallenge: def to_core(self) -> Challenge: """Convert to core Challenge type, preserving MCP challenge metadata.""" import base64 - import json from mpp import Challenge - request_json = json.dumps(self.request, separators=(",", ":"), sort_keys=True) + request_json = canonical_json(self.request) request_b64 = base64.urlsafe_b64encode(request_json.encode()).decode().rstrip("=") return Challenge( @@ -183,16 +184,15 @@ def from_meta(cls, meta: dict[str, Any]) -> MCPCredential | None: def to_core(self) -> Credential: """Convert to core Credential type.""" import base64 - import json from mpp import ChallengeEcho, Credential - request_json = json.dumps(self.challenge.request, separators=(",", ":"), sort_keys=True) + request_json = canonical_json(self.challenge.request) request_b64 = base64.urlsafe_b64encode(request_json.encode()).decode().rstrip("=") opaque_b64 = None if self.challenge.opaque is not None: - opaque_json = json.dumps(self.challenge.opaque, separators=(",", ":"), sort_keys=True) + opaque_json = canonical_json(self.challenge.opaque) opaque_b64 = base64.urlsafe_b64encode(opaque_json.encode()).decode().rstrip("=") echo = ChallengeEcho( diff --git a/tests/test_challenge_id.py b/tests/test_challenge_id.py index 39ab5e0..c17d7da 100644 --- a/tests/test_challenge_id.py +++ b/tests/test_challenge_id.py @@ -504,3 +504,61 @@ def test_empty_meta_produces_opaque(self) -> None: meta={}, ) assert challenge.opaque == {} + + +class TestWireFormCanonicalization: + """The emitted `request` parameter must be the exact bytes the HMAC binds. + + Section 5.1.1 of draft-httpauth-payment-00 binds the challenge id to the + base64url-encoded request "as it appears on the wire". If the wire form and + the HMAC input disagree, the id is unreproducible by any conformant peer + even though this SDK still verifies its own challenges (verify decodes and + re-canonicalizes, hiding the mismatch). + """ + + NON_ASCII_REQUEST = { + "amount": "1000000", + "currency": "usd", + "description": "Payment for café ☕", + } + + def test_emitted_request_reproduces_the_challenge_id(self) -> None: + import base64 + import hashlib + import hmac + + from mpp._parsing import format_www_authenticate + + secret, realm = "server-secret", "api.example.com" + challenge = Challenge.create( + secret_key=secret, + realm=realm, + method="tempo", + intent="charge", + request=self.NON_ASCII_REQUEST, + ) + header = format_www_authenticate(challenge, realm) + wire_request = header.split('request="', 1)[1].split('"', 1)[0] + + # Recompute the id the way a peer does: HMAC over the wire string. + payload = "|".join([realm, "tempo", "charge", wire_request, "", "", ""]) + expected = ( + base64.urlsafe_b64encode( + hmac.new(secret.encode(), payload.encode(), hashlib.sha256).digest() + ) + .decode() + .rstrip("=") + ) + assert challenge.id == expected + + def test_emitted_request_is_raw_utf8(self) -> None: + from mpp._parsing import _b64_decode, _b64_encode + + encoded = _b64_encode(self.NON_ASCII_REQUEST) + import base64 + + decoded = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)).decode() + assert "café ☕" in decoded + assert "\\u00e9" not in decoded + assert "\\u2615" not in decoded + assert _b64_decode(encoded) == self.NON_ASCII_REQUEST