Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 5 additions & 15 deletions src/mpp/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
5 changes: 3 additions & 2 deletions src/mpp/_body_digest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -21,7 +22,7 @@ def compute(body: str | bytes | dict[str, Any]) -> str:
Digest in the format ``sha-256=<base64>``.
"""
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()
Expand Down
13 changes: 12 additions & 1 deletion src/mpp/_parsing.py
Original file line number Diff line number Diff line change
Expand Up @@ -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("=")

Expand Down
10 changes: 5 additions & 5 deletions src/mpp/extensions/mcp/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
58 changes: 58 additions & 0 deletions tests/test_challenge_id.py
Original file line number Diff line number Diff line change
Expand Up @@ -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