From d958acdae34a73e40def15cbf32b5502fd672b62 Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Tue, 7 Jul 2026 18:09:11 -0400 Subject: [PATCH 1/7] feat(python/server): add RFC 9421 signature module with RFC test vectors Adds ucp_signing.py: a dependency-light (cryptography-only) implementation of the UCP request-signature layer defined in signatures.md -- RFC 9421 signature base construction, the UCP covered-component set, RFC 9530 Content-Digest over raw body bytes, ES256 with fixed-width raw r||s (never ASN.1/DER, per the spec MUST), Ed25519, and signer key discovery from the UCP-Agent profile. Correctness is anchored to three independent oracles in ucp_signing_test.py: RFC 9421 Appendix B (byte-exact Ed25519 B.2.6, verify-direction ES256 B.2.4) and RFC 9530 digest vectors, an explicit DER-rejection test, and a differential comparison against the independent http-message-signatures library (dev-only). No behavior change yet -- this commit only adds the module and its tests. --- rest/python/server/pyproject.toml | 2 + rest/python/server/ucp_signing.py | 774 +++++++++++++++++++++++++ rest/python/server/ucp_signing_test.py | 410 +++++++++++++ 3 files changed, 1186 insertions(+) create mode 100644 rest/python/server/ucp_signing.py create mode 100644 rest/python/server/ucp_signing_test.py diff --git a/rest/python/server/pyproject.toml b/rest/python/server/pyproject.toml index 00ce80c..b9577e0 100644 --- a/rest/python/server/pyproject.toml +++ b/rest/python/server/pyproject.toml @@ -20,6 +20,7 @@ dependencies = [ "ucp-sdk", "shortuuid", "httpx>=0.26.0", + "cryptography>=42", ] [dependency-groups] @@ -27,6 +28,7 @@ dev = [ "pytest>=8.0.0", "httpx>=0.26.0", "ruff>=0.14.11", + "http-message-signatures==2.0.1", ] [build-system] diff --git a/rest/python/server/ucp_signing.py b/rest/python/server/ucp_signing.py new file mode 100644 index 0000000..a841849 --- /dev/null +++ b/rest/python/server/ucp_signing.py @@ -0,0 +1,774 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""RFC 9421 HTTP Message Signatures for UCP, using the default UCP profile. + +This module implements what the UCP `signatures.md` specification actually +requires for platform-to-business request signing and verification: + +* RFC 9421 signature base construction and the UCP covered-component set. +* RFC 9530 `Content-Digest` over the raw body bytes (`sha-256`). +* ECDSA P-256 (`ES256`) verification with fixed-width raw `r||s` signatures + (RFC 9421 Section 3.3.1) -- never ASN.1/DER -- and Ed25519 (RFC 8032). +* Signer public-key discovery from the `UCP-Agent` profile's `signing_keys`. + +Only the encoding and canonicalization are implemented here; every +cryptographic primitive comes from the `cryptography` package. The module is +deliberately small and readable so it can double as a reference for +implementers building their own UCP signature layer. +""" + +import base64 +import hashlib +import ipaddress +import logging +import socket +import time +from urllib.parse import urlsplit + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.asymmetric import ed25519 +from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature +import httpx + +logger = logging.getLogger(__name__) + +# Public keys resolved from a signer profile are cached for this many seconds. +_KEY_CACHE_TTL_SECONDS = 300 +_KEY_CACHE: dict[str, tuple[float, list[dict]]] = {} + +# Curve -> (JWA algorithm name, coordinate byte length). ES384 is intentionally +# omitted: the spec lists it as OPTIONAL and the baseline every verifier MUST +# support is ES256. +_EC_CURVES = { + "P-256": ("ES256", 32), +} + + +class SignatureError(Exception): + """A signature or profile-resolution failure with UCP wire semantics. + + Attributes: + code: The UCP error code (e.g. ``signature_invalid``). + status_code: The HTTP status the spec maps that code to. + message: A human-readable explanation for logs and error bodies. + + """ + + def __init__(self, code: str, status_code: int, message: str) -> None: + """Store the UCP error code, HTTP status, and message.""" + super().__init__(message) + self.code = code + self.status_code = status_code + self.message = message + + +def _b64u_decode(value: str) -> bytes: + """Decode base64url text without requiring padding.""" + return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + + +def _b64u_encode(data: bytes) -> str: + """Encode bytes as base64url text without padding.""" + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def content_digest(body: bytes) -> str: + """Return the RFC 9530 ``Content-Digest`` value for the raw body bytes. + + Args: + body: The exact bytes of the message body as they appear on the wire. + + Returns: + A structured-field dictionary value of the form ``sha-256=::``. + + """ + digest = hashlib.sha256(body).digest() + return "sha-256=:" + base64.b64encode(digest).decode("ascii") + ":" + + +def content_digest_matches(header_value: str, body: bytes) -> bool: + """Check whether a ``Content-Digest`` header covers the given body. + + Only the ``sha-256`` member is inspected; other digest algorithms in the + dictionary are ignored, matching the spec's requirement to use ``sha-256``. + + Args: + header_value: The received ``Content-Digest`` header value. + body: The raw body bytes to compare against. + + Returns: + True when the header carries a ``sha-256`` member equal to the digest of + ``body``. + + """ + want = hashlib.sha256(body).digest() + for member in _sf_split(header_value, ","): + key, _, val = member.partition("=") + if key.strip() != "sha-256": + continue + val = val.strip() + if not (val.startswith(":") and val.endswith(":")): + return False + try: + return base64.b64decode(val[1:-1], validate=True) == want + except (ValueError, TypeError): + return False + return False + + +def _sf_split(value: str, seps: str) -> list[str]: + """Split a structured-field string on top-level separators. + + Quoted strings and inner lists are treated as opaque, which is enough of + RFC 8941 to parse ``Signature`` and ``Signature-Input`` headers. + + Args: + value: The structured-field string. + seps: Separator characters to split on at the top level. + + Returns: + The list of trimmed, non-empty segments. + + """ + out: list[str] = [] + cur: list[str] = [] + depth = 0 + quote = False + i = 0 + while i < len(value): + c = value[i] + if quote: + cur.append(c) + if c == "\\" and i + 1 < len(value): + cur.append(value[i + 1]) + i += 1 + elif c == '"': + quote = False + elif c == '"': + quote = True + cur.append(c) + elif c == "(": + depth += 1 + cur.append(c) + elif c == ")": + depth -= 1 + cur.append(c) + elif c in seps and depth == 0: + out.append("".join(cur).strip()) + cur = [] + else: + cur.append(c) + i += 1 + tail = "".join(cur).strip() + if tail: + out.append(tail) + return out + + +def parse_signature_input(value: str) -> dict | None: + """Parse a ``Signature-Input`` header into per-label descriptors. + + Args: + value: The ``Signature-Input`` header value. + + Returns: + A mapping ``{label: {"raw", "components", "params"}}`` where ``raw`` is the + member value verbatim (what ``@signature-params`` must echo), ``components`` + are the unquoted component identifiers, and ``params`` maps parameter names + to their unquoted values. None on malformed input. + + """ + if not isinstance(value, str) or not value.strip(): + return None + out: dict = {} + for member in _sf_split(value, ","): + label, eq, val = member.partition("=") + label = label.strip() + val = val.strip() + if not eq or not label or not val.startswith("("): + return None + depth = 0 + end = 0 + for index, char in enumerate(val): + if char == "(": + depth += 1 + elif char == ")": + depth -= 1 + if depth == 0: + end = index + break + inner = val[1:end] + rest = val[end + 1 :] + components: list[str] = [] + for tok in _sf_split(inner, " "): + if not (tok.startswith('"') and tok.endswith('"')) or ";" in tok: + return None + components.append(tok[1:-1]) + params: dict[str, str] = {} + for part in _sf_split(rest, ";"): + if not part: + continue + k, _, v = part.partition("=") + if v.startswith('"') and v.endswith('"'): + v = v[1:-1] + params[k.strip()] = v + out[label] = {"raw": val, "components": components, "params": params} + return out or None + + +def parse_signature(value: str) -> dict | None: + """Parse a ``Signature`` header into ``{label: raw signature bytes}``. + + Args: + value: The ``Signature`` header value. + + Returns: + A mapping of label to decoded signature bytes, or None on malformed input. + + """ + if not isinstance(value, str) or not value.strip(): + return None + out: dict[str, bytes] = {} + for member in _sf_split(value, ","): + label, eq, val = member.partition("=") + label = label.strip() + val = val.strip() + if not eq or not val.startswith(":") or not val.endswith(":"): + return None + try: + out[label] = base64.b64decode(val[1:-1], validate=True) + except (ValueError, TypeError): + return None + return out or None + + +def build_signature_base( + components: list[str], + raw_params: str, + resolve: "callable", +) -> bytes | None: + """Construct the RFC 9421 signature base. + + Args: + components: The ordered component identifiers to cover. + raw_params: The ``Signature-Input`` member value verbatim, echoed on the + ``@signature-params`` line. + resolve: A callable mapping a component identifier to its string value, or + None when the component cannot be resolved for this message. + + Returns: + The signature base as bytes, or None if any component is unresolvable. + + """ + lines: list[str] = [] + for name in components: + value = resolve(name) + if value is None: + return None + lines.append(f'"{name}": {value}') + lines.append(f'"@signature-params": {raw_params}') + return "\n".join(lines).encode("utf-8") + + +def required_components( + method: str, + has_query: bool, + headers: dict, + has_body: bool, +) -> list[str]: + """Return the components a UCP request signature MUST cover. + + This is the verification-side coverage gate from ``signatures.md``: a + signature that omits any of these while the corresponding element is present + is treated as not covering the message. + + Args: + method: The HTTP request method (unused today; retained for clarity that + coverage keys on header presence, not method). + has_query: Whether the request target carries a query string. + headers: A case-insensitive mapping of request headers. + has_body: Whether the request carries a body. + + Returns: + The ordered list of required component identifiers. + + """ + del method # Coverage keys on header presence, not on the method. + required = ["@method", "@authority", "@path"] + if has_query: + required.append("@query") + if has_body: + required.extend(["content-digest", "content-type"]) + if "idempotency-key" in headers: + required.append("idempotency-key") + if "ucp-agent" in headers: + required.append("ucp-agent") + if "signature-agent" in headers: + required.append("signature-agent") + return required + + +def public_key_from_jwk(jwk: dict): + """Build a ``cryptography`` public key from a UCP JWK. + + Args: + jwk: A JWK describing an EC P-256 or OKP Ed25519 public key. + + Returns: + An ``EllipticCurvePublicKey`` or ``Ed25519PublicKey``. + + Raises: + SignatureError: With ``algorithm_unsupported`` when the key type or curve + is not one UCP verifiers are required to support. + + """ + kty = jwk.get("kty") + crv = jwk.get("crv") + try: + if kty == "EC" and crv in _EC_CURVES: + x = int.from_bytes(_b64u_decode(jwk["x"]), "big") + y = int.from_bytes(_b64u_decode(jwk["y"]), "big") + return ec.EllipticCurvePublicNumbers(x, y, ec.SECP256R1()).public_key() + if kty == "OKP" and crv == "Ed25519": + return ed25519.Ed25519PublicKey.from_public_bytes(_b64u_decode(jwk["x"])) + except (KeyError, ValueError) as exc: + raise SignatureError( + "signature_invalid", 401, f"Malformed JWK: {exc}" + ) from exc + raise SignatureError( + "algorithm_unsupported", + 400, + f"Unsupported key type/curve: kty={kty!r} crv={crv!r}", + ) + + +def jwk_from_public_key(public_key, kid: str) -> dict: + """Export a ``cryptography`` public key as a UCP JWK. + + Args: + public_key: An ``EllipticCurvePublicKey`` (P-256) or ``Ed25519PublicKey``. + kid: The key identifier to embed. + + Returns: + A JWK dictionary with ``use`` and ``alg`` populated. + + """ + if isinstance(public_key, ed25519.Ed25519PublicKey): + from cryptography.hazmat.primitives import serialization + + raw = public_key.public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw + ) + return { + "kid": kid, + "kty": "OKP", + "crv": "Ed25519", + "x": _b64u_encode(raw), + "use": "sig", + "alg": "EdDSA", + } + numbers = public_key.public_numbers() + size = _EC_CURVES["P-256"][1] + return { + "kid": kid, + "kty": "EC", + "crv": "P-256", + "x": _b64u_encode(numbers.x.to_bytes(size, "big")), + "y": _b64u_encode(numbers.y.to_bytes(size, "big")), + "use": "sig", + "alg": "ES256", + } + + +def verify_raw_signature(jwk: dict, base: bytes, signature: bytes) -> None: + """Verify a signature base against a JWK public key. + + The algorithm is derived from the key's ``kty``/``crv`` -- never from a + ``Signature-Input`` ``alg`` parameter, which UCP forbids. ECDSA signatures + MUST be fixed-width raw ``r||s`` (64 bytes for P-256); ASN.1/DER is rejected + before it can be decoded. + + Args: + jwk: The signer's public JWK. + base: The RFC 9421 signature base bytes. + signature: The decoded signature value. + + Raises: + SignatureError: ``algorithm_unsupported`` for keys UCP need not support, or + ``signature_invalid`` when verification fails or the encoding is wrong. + + """ + public_key = public_key_from_jwk(jwk) + if isinstance(public_key, ed25519.Ed25519PublicKey): + try: + public_key.verify(signature, base) + except InvalidSignature as exc: + raise SignatureError( + "signature_invalid", 401, "Ed25519 signature verification failed" + ) from exc + return + # ECDSA P-256: enforce fixed-width raw r||s (never DER) before converting to + # the DER form cryptography's verifier expects. + size = _EC_CURVES["P-256"][1] + if len(signature) != 2 * size: + raise SignatureError( + "signature_invalid", + 401, + f"ECDSA signature must be {2 * size}-byte raw r||s, got " + f"{len(signature)} bytes (ASN.1/DER is not permitted)", + ) + r = int.from_bytes(signature[:size], "big") + s = int.from_bytes(signature[size:], "big") + try: + public_key.verify(encode_dss_signature(r, s), base, ec.ECDSA(_es256_hash())) + except InvalidSignature as exc: + raise SignatureError( + "signature_invalid", 401, "ES256 signature verification failed" + ) from exc + + +def _es256_hash(): + """Return the SHA-256 hash primitive used by ES256.""" + from cryptography.hazmat.primitives import hashes + + return hashes.SHA256() + + +def _authority(host: str) -> str: + """Normalise an authority to lowercase per RFC 9421 Section 2.2.3.""" + return host.lower() + + +def verify_request( + method: str, + authority: str, + path: str, + query: str, + headers: dict, + body: bytes, + keys: list[dict], +) -> str: + """Verify the signatures on an inbound UCP request. + + A request is accepted when at least one carried signature fully verifies; + it is rejected only when every candidate signature is skipped or fails. + + Args: + method: HTTP method. + authority: Target authority (host, optionally with port). + path: Request path without query. + query: Raw query string (without a leading ``?``); empty when absent. + headers: Case-insensitive request headers. + body: Raw request body bytes. + keys: The signer's published JWKs. + + Returns: + The ``keyid`` of the signature that verified. + + Raises: + SignatureError: With the appropriate UCP code when no signature verifies. + + """ + sig_input = parse_signature_input(headers.get("signature-input", "")) + sigs = parse_signature(headers.get("signature", "")) + if not sig_input or not sigs: + raise SignatureError( + "signature_missing", 401, "Missing Signature-Input or Signature header" + ) + + has_body = bool(body) + if has_body: + digest_header = headers.get("content-digest") + if not digest_header or not content_digest_matches(digest_header, body): + raise SignatureError( + "digest_mismatch", 400, "Content-Digest does not match the body" + ) + + required = required_components(method, bool(query), headers, has_body) + keys_by_kid = {k.get("kid"): k for k in keys} + + def resolve(name: str) -> str | None: + if name == "@method": + return method.upper() + if name == "@authority": + return _authority(authority) + if name == "@path": + return path + if name == "@query": + return "?" + query + return headers.get(name) + + last_error = SignatureError( + "signature_invalid", 401, "No valid signature found" + ) + for label, desc in sig_input.items(): + signature = sigs.get(label) + if signature is None: + continue + if "alg" in desc["params"]: + last_error = SignatureError( + "signature_invalid", + 401, + "Signature-Input MUST NOT carry an 'alg' parameter", + ) + continue + if not set(required).issubset(desc["components"]): + missing = sorted(set(required) - set(desc["components"])) + last_error = SignatureError( + "signature_invalid", + 401, + f"Signature does not cover required components: {missing}", + ) + continue + keyid = desc["params"].get("keyid") + jwk = keys_by_kid.get(keyid) + if jwk is None: + last_error = SignatureError( + "key_not_found", 401, f"No published key with kid={keyid!r}" + ) + continue + base = build_signature_base(desc["components"], desc["raw"], resolve) + if base is None: + last_error = SignatureError( + "signature_invalid", 401, "Unresolvable signed component" + ) + continue + try: + verify_raw_signature(jwk, base, signature) + except SignatureError as exc: + last_error = exc + continue + return keyid + raise last_error + + +def sign_request( + private_key, + kid: str, + method: str, + url: str, + headers: dict, + body: bytes, + created: int | None = None, +) -> dict: + """Sign a UCP request and return the headers to add. + + Adds ``Content-Digest`` (when a body is present), ``Signature-Input``, and + ``Signature`` covering exactly the UCP required-component set. Used by the + demo client and the tests; it is also the reference for a future SDK signer. + + Args: + private_key: An EC P-256 or Ed25519 private key. + kid: The key identifier to advertise via ``keyid``. + method: HTTP method. + url: The absolute request URL. + headers: The request headers already set by the caller; extended in place + is avoided -- a new dict of additions is returned. + body: Raw request body bytes. + created: Optional ``created`` timestamp; defaults to the current time. + + Returns: + A dict of header names to values that the caller must add to the request. + + """ + split = urlsplit(url) + additions: dict[str, str] = {} + merged = {k.lower(): v for k, v in headers.items()} + has_body = bool(body) + if has_body: + digest = content_digest(body) + additions["Content-Digest"] = digest + merged["content-digest"] = digest + # The signer must fix and cover Content-Type: a value the transport adds + # after signing would not be part of the signature base. + if "content-type" not in merged: + merged["content-type"] = "application/json" + additions["Content-Type"] = "application/json" + + components = required_components(method, bool(split.query), merged, has_body) + created = int(time.time()) if created is None else created + raw_params = ( + "(" + " ".join(f'"{c}"' for c in components) + ")" + f';created={created};keyid="{kid}"' + ) + + def resolve(name: str) -> str | None: + if name == "@method": + return method.upper() + if name == "@authority": + return _authority(split.netloc) + if name == "@path": + return split.path or "/" + if name == "@query": + return "?" + split.query + return merged.get(name) + + base = build_signature_base(components, raw_params, resolve) + if base is None: + missing = [c for c in components if resolve(c) is None] + raise SignatureError( + "signature_invalid", 401, f"Cannot sign; missing components: {missing}" + ) + signature = _raw_sign(private_key, base) + additions["Signature-Input"] = f"sig1={raw_params}" + additions["Signature"] = ( + "sig1=:" + base64.b64encode(signature).decode("ascii") + ":" + ) + return additions + + +def _raw_sign(private_key, base: bytes) -> bytes: + """Sign a signature base, emitting raw ``r||s`` for ECDSA.""" + if isinstance(private_key, ed25519.Ed25519PrivateKey): + return private_key.sign(base) + from cryptography.hazmat.primitives.asymmetric.utils import ( + decode_dss_signature, + ) + + der = private_key.sign(base, ec.ECDSA(_es256_hash())) + r, s = decode_dss_signature(der) + size = _EC_CURVES["P-256"][1] + return r.to_bytes(size, "big") + s.to_bytes(size, "big") + + +def clear_key_cache() -> None: + """Empty the resolved-key cache (used by tests).""" + _KEY_CACHE.clear() + + +def _assert_profile_url_allowed(url: str, allow_insecure: bool) -> None: + """Reject profile URLs that violate the spec's transport and SSRF rules. + + Args: + url: The candidate profile URL from the ``UCP-Agent`` header. + allow_insecure: When True, permit ``http`` and loopback/private hosts. This + exists only for localhost demos and CI, mirroring the official + conformance harness, and must never be enabled against real agents. + + Raises: + SignatureError: ``invalid_profile_url`` when the URL is malformed, not + HTTPS, or resolves to a special-use address. + + """ + split = urlsplit(url) + if split.scheme != "https" and not ( + allow_insecure and split.scheme == "http" + ): + raise SignatureError( + "invalid_profile_url", 400, f"Profile URL must be HTTPS: {url!r}" + ) + if split.username or split.password: + raise SignatureError( + "invalid_profile_url", 400, "Profile URL must not carry credentials" + ) + host = split.hostname + if not host: + raise SignatureError( + "invalid_profile_url", 400, f"Profile URL has no host: {url!r}" + ) + if allow_insecure: + return + try: + infos = socket.getaddrinfo( + host, split.port or 443, proto=socket.IPPROTO_TCP + ) + except socket.gaierror as exc: + raise SignatureError( + "profile_unreachable", 424, f"Cannot resolve profile host: {host}" + ) from exc + for info in infos: + addr = ipaddress.ip_address(info[4][0]) + if ( + addr.is_private + or addr.is_loopback + or addr.is_link_local + or addr.is_reserved + or addr.is_multicast + or addr.is_unspecified + ): + raise SignatureError( + "invalid_profile_url", + 400, + f"Profile URL resolves to a disallowed address: {addr}", + ) + + +async def fetch_signing_keys( + profile_url: str, *, allow_insecure: bool = False +) -> list[dict]: + """Fetch and cache a signer's published signing keys from its UCP profile. + + Reads ``signing_keys`` (the canonical field on the current spec) and falls + back to a top-level ``keys`` array. Open spec PR #566 will promote ``keys`` + to canonical and drop ``signing_keys``; reading both keeps this resolver + correct across that change. + + Args: + profile_url: The signer's profile URL from the ``UCP-Agent`` header. + allow_insecure: Permit ``http`` and loopback hosts for demos/CI. + + Returns: + The list of JWKs published by the signer. + + Raises: + SignatureError: ``invalid_profile_url`` / ``profile_unreachable`` / + ``profile_malformed`` per the failure. + + """ + cached = _KEY_CACHE.get(profile_url) + if cached and cached[0] > time.time(): + return cached[1] + + _assert_profile_url_allowed(profile_url, allow_insecure) + try: + async with httpx.AsyncClient(follow_redirects=False) as client: + response = await client.get(profile_url, timeout=5.0) + except httpx.HTTPError as exc: + raise SignatureError( + "profile_unreachable", 424, f"Profile fetch failed: {exc}" + ) from exc + if response.status_code >= 300: + raise SignatureError( + "profile_unreachable", + 424, + f"Profile fetch returned HTTP {response.status_code}", + ) + try: + document = response.json() + except ValueError as exc: + raise SignatureError( + "profile_malformed", 422, "Profile is not valid JSON" + ) from exc + + keys = _extract_keys(document) + if not keys: + raise SignatureError( + "profile_malformed", 422, "Profile publishes no signing keys" + ) + _KEY_CACHE[profile_url] = (time.time() + _KEY_CACHE_TTL_SECONDS, keys) + return keys + + +def _extract_keys(document: dict) -> list[dict]: + """Pull the signing-key list out of a profile document.""" + if not isinstance(document, dict): + return [] + ucp = document.get("ucp", document) + for field in ("signing_keys", "keys"): + value = ucp.get(field) if isinstance(ucp, dict) else None + if isinstance(value, list) and value: + return value + return [] diff --git a/rest/python/server/ucp_signing_test.py b/rest/python/server/ucp_signing_test.py new file mode 100644 index 0000000..899c26c --- /dev/null +++ b/rest/python/server/ucp_signing_test.py @@ -0,0 +1,410 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for the RFC 9421 signing module. + +Correctness is anchored to three independent oracles: the RFC 9421 Appendix B +and RFC 9530 published vectors, an explicit DER-rejection check for the UCP +raw-`r||s` requirement, and a differential comparison against the independent +`http-message-signatures` library. + +All key material is generated at runtime or reconstructed from the RFC's raw +JWK coordinates; no private-key files are committed. +""" + +import base64 + +from absl.testing import absltest +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.asymmetric import ed25519 +import httpx +import ucp_signing as signing + + +def _b64u(value: str) -> bytes: + """Decode base64url without padding (for the RFC JWK coordinates).""" + return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + + +# RFC 9421 Appendix B.1.4 test-key-ed25519 (JWK coordinates, verbatim). +RFC_ED25519_JWK = { + "kty": "OKP", + "crv": "Ed25519", + "kid": "test-key-ed25519", + "x": "JrQLj5P_89iXES9-vFgrIy29clF9CC_oPPsw3c5D0bs", +} +RFC_ED25519_D = "n4Ni-HpISpVObnQMW0wOhCKROaIKqKtW_2ZYb2p9KcU" + +# RFC 9421 Appendix B.2.6 signature base and signature (byte-exact oracle). +RFC_B26_BASE = ( + b'"date": Tue, 20 Apr 2021 02:07:55 GMT\n' + b'"@method": POST\n' + b'"@path": /foo\n' + b'"@authority": example.com\n' + b'"content-type": application/json\n' + b'"content-length": 18\n' + b'"@signature-params": ("date" "@method" "@path" "@authority" ' + b'"content-type" "content-length");created=1618884473' + b';keyid="test-key-ed25519"' +) +RFC_B26_SIGNATURE = base64.b64decode( + "wqcAqbmYJ2ji2glfAMaRy4gruYYnx2nEFN2HN6jrnDnQCK1u02Gb04v9EDgwUPiu4" + "A0w6vuQv5lIp5WPpBKRCw==" +) + + +class ContentDigestTest(absltest.TestCase): + """RFC 9530 Content-Digest generation and matching.""" + + def test_rfc9530_lf_body_vector(self) -> None: + """The 19-byte LF body matches RFC 9530's canonical sha-256 value.""" + self.assertEqual( + signing.content_digest(b'{"hello": "world"}\n'), + "sha-256=:RK/0qy18MlBSVnWgjwz6lZEWjP/lF5HF9bvEF8FabDg=:", + ) + + def test_rfc9421_no_lf_body_vector(self) -> None: + """The 18-byte body matches the value used in RFC 9421's examples.""" + self.assertEqual( + signing.content_digest(b'{"hello": "world"}'), + "sha-256=:X48E9qOokqqrvdts8nOJRJN3OWDUoyWxBf7kbu9DBPE=:", + ) + + def test_matches_accepts_and_rejects(self) -> None: + """content_digest_matches accepts the right body and rejects others.""" + body = b'{"a": 1}' + header = signing.content_digest(body) + self.assertTrue(signing.content_digest_matches(header, body)) + self.assertFalse(signing.content_digest_matches(header, b'{"a": 2}')) + + +class SignatureBaseTest(absltest.TestCase): + """RFC 9421 signature-base construction.""" + + def test_matches_rfc_b26_base(self) -> None: + """Reconstructing the B.2.6 components yields the RFC's exact base.""" + components = [ + "date", + "@method", + "@path", + "@authority", + "content-type", + "content-length", + ] + raw = ( + '("date" "@method" "@path" "@authority" "content-type" ' + '"content-length");created=1618884473;keyid="test-key-ed25519"' + ) + values = { + "date": "Tue, 20 Apr 2021 02:07:55 GMT", + "@method": "POST", + "@path": "/foo", + "@authority": "example.com", + "content-type": "application/json", + "content-length": "18", + } + base = signing.build_signature_base(components, raw, values.get) + self.assertEqual(base, RFC_B26_BASE) + + def test_signature_params_echoed_verbatim(self) -> None: + """The @signature-params line echoes the member value verbatim.""" + raw = '("@method");created=5;keyid="k"' + base = signing.build_signature_base(["@method"], raw, lambda _: "GET") + self.assertTrue(base.endswith(f'"@signature-params": {raw}'.encode())) + + def test_unresolvable_component_returns_none(self) -> None: + """A component the resolver cannot supply aborts base construction.""" + self.assertIsNone( + signing.build_signature_base(["x-missing"], "()", lambda _: None) + ) + + +class Rfc9421VectorsTest(absltest.TestCase): + """Byte-exact Ed25519 and verify-direction ES256 against Appendix B.""" + + def test_ed25519_b26_verifies(self) -> None: + """The RFC's published Ed25519 signature verifies through the module.""" + signing.verify_raw_signature( + RFC_ED25519_JWK, RFC_B26_BASE, RFC_B26_SIGNATURE + ) + + def test_ed25519_b26_byte_exact_sign(self) -> None: + """Ed25519 is deterministic: our signature equals the RFC's bytes.""" + key = ed25519.Ed25519PrivateKey.from_private_bytes(_b64u(RFC_ED25519_D)) + self.assertEqual(signing._raw_sign(key, RFC_B26_BASE), RFC_B26_SIGNATURE) + + def test_ed25519_tampered_base_fails(self) -> None: + """A modified base no longer verifies against the RFC signature.""" + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_raw_signature( + RFC_ED25519_JWK, RFC_B26_BASE + b" ", RFC_B26_SIGNATURE + ) + self.assertEqual(ctx.exception.code, "signature_invalid") + + def test_es256_roundtrip(self) -> None: + """An ES256 signature we produce verifies with the derived JWK.""" + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k") + sig = signing._raw_sign(key, RFC_B26_BASE) + signing.verify_raw_signature(jwk, RFC_B26_BASE, sig) + + +class RawSignatureEncodingTest(absltest.TestCase): + """The UCP raw-r||s ECDSA requirement (spec MUST; issue #569).""" + + def setUp(self) -> None: + """Create a P-256 key and its JWK for the encoding tests.""" + super().setUp() + self.key = ec.generate_private_key(ec.SECP256R1()) + self.jwk = signing.jwk_from_public_key(self.key.public_key(), "k") + + def test_der_signature_rejected(self) -> None: + """A DER-encoded ECDSA signature must be rejected as non-conformant.""" + der = self.key.sign(RFC_B26_BASE, ec.ECDSA(hashes.SHA256())) + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_raw_signature(self.jwk, RFC_B26_BASE, der) + self.assertEqual(ctx.exception.code, "signature_invalid") + + def test_raw_64_byte_accepted(self) -> None: + """A well-formed 64-byte raw signature verifies.""" + sig = signing._raw_sign(self.key, RFC_B26_BASE) + self.assertLen(sig, 64) + signing.verify_raw_signature(self.jwk, RFC_B26_BASE, sig) + + def test_wrong_length_rejected(self) -> None: + """Signatures that are not 64 bytes are rejected before verification.""" + sig = signing._raw_sign(self.key, RFC_B26_BASE) + for bad in (sig[:-1], sig + b"\x00"): + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_raw_signature(self.jwk, RFC_B26_BASE, bad) + self.assertEqual(ctx.exception.code, "signature_invalid") + + +class SfParserTest(absltest.TestCase): + """RFC 8941 subset parsing of Signature-Input and Signature.""" + + def test_parses_components_and_params(self) -> None: + """A well-formed member yields components and parameters.""" + parsed = signing.parse_signature_input( + 'sig1=("@method" "content-digest");created=1;keyid="abc"' + ) + self.assertEqual( + parsed["sig1"]["components"], ["@method", "content-digest"] + ) + self.assertEqual(parsed["sig1"]["params"]["keyid"], "abc") + + def test_multiple_labels(self) -> None: + """Multiple comma-separated members are all parsed.""" + parsed = signing.parse_signature_input( + 'a=("@method");keyid="x", b=("@path");keyid="y"' + ) + self.assertEqual(set(parsed), {"a", "b"}) + + def test_signature_decodes_base64(self) -> None: + """A Signature member decodes to raw bytes.""" + raw = base64.b64encode(b"hello").decode("ascii") + parsed = signing.parse_signature(f"sig1=:{raw}:") + self.assertEqual(parsed["sig1"], b"hello") + + def test_malformed_returns_none(self) -> None: + """Malformed inputs parse to None rather than raising.""" + self.assertIsNone(signing.parse_signature_input("not a signature input")) + self.assertIsNone(signing.parse_signature("")) + + +class CoverageGateTest(absltest.TestCase): + """The UCP required-component coverage table.""" + + def test_get_no_body(self) -> None: + """A bodyless GET requires only the target components.""" + self.assertEqual( + signing.required_components("GET", False, {}, False), + ["@method", "@authority", "@path"], + ) + + def test_post_with_body_requires_digest_and_type(self) -> None: + """A bodied request must cover content-digest and content-type.""" + required = signing.required_components("POST", False, {}, True) + self.assertIn("content-digest", required) + self.assertIn("content-type", required) + + def test_query_present(self) -> None: + """A query string adds @query.""" + self.assertIn("@query", signing.required_components("GET", True, {}, False)) + + def test_idempotency_key_header_on_get(self) -> None: + """Coverage keys on header presence: a GET with the header covers it.""" + required = signing.required_components( + "GET", False, {"idempotency-key": "x"}, False + ) + self.assertIn("idempotency-key", required) + + def test_ucp_agent_and_signature_agent(self) -> None: + """Present ucp-agent / signature-agent headers must be covered.""" + required = signing.required_components( + "GET", False, {"ucp-agent": "a", "signature-agent": "b"}, False + ) + self.assertIn("ucp-agent", required) + self.assertIn("signature-agent", required) + + def test_alg_param_rejected_by_verify_request(self) -> None: + """A signature carrying an alg parameter is rejected (spec MUST NOT).""" + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k") + add = signing.sign_request( + key, + "k", + "GET", + "https://h/p", + {"UCP-Agent": 'profile="https://a/p"'}, + b"", + ) + add["Signature-Input"] = add["Signature-Input"].replace( + ";created", ';alg="ecdsa-p256-sha256";created' + ) + headers = { + "ucp-agent": 'profile="https://a/p"', + "signature-input": add["Signature-Input"], + "signature": add["Signature"], + } + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_request("GET", "h", "/p", "", headers, b"", [jwk]) + self.assertEqual(ctx.exception.code, "signature_invalid") + + +class SsrfGuardTest(absltest.TestCase): + """Profile-URL transport and SSRF guards.""" + + def test_http_rejected_without_carveout(self) -> None: + """Plain http is rejected unless the insecure carve-out is set.""" + with self.assertRaises(signing.SignatureError) as ctx: + signing._assert_profile_url_allowed("http://example.com/p", False) + self.assertEqual(ctx.exception.code, "invalid_profile_url") + + def test_metadata_address_rejected(self) -> None: + """The cloud metadata address is rejected.""" + with self.assertRaises(signing.SignatureError): + signing._assert_profile_url_allowed( + "https://169.254.169.254/latest", False + ) + + def test_loopback_and_private_rejected(self) -> None: + """Loopback and RFC 1918 hosts are rejected without the carve-out.""" + for url in ("https://127.0.0.1/p", "https://10.0.0.5/p"): + with self.assertRaises(signing.SignatureError): + signing._assert_profile_url_allowed(url, False) + + def test_credentials_rejected(self) -> None: + """A URL carrying userinfo is rejected.""" + with self.assertRaises(signing.SignatureError): + signing._assert_profile_url_allowed("https://u:p@example.com/p", False) + + def test_loopback_allowed_with_carveout(self) -> None: + """The carve-out permits http loopback for localhost demos.""" + signing._assert_profile_url_allowed("http://127.0.0.1:8285/p", True) + + +class ProfileFetchTest(absltest.TestCase): + """Key discovery from a signer profile, using a mocked transport.""" + + def setUp(self) -> None: + """Clear the key cache before each fetch test.""" + super().setUp() + signing.clear_key_cache() + + def _fetch(self, handler) -> list: + """Run fetch_signing_keys against a mocked httpx transport.""" + real_client = httpx.AsyncClient + + def factory(*args, **kwargs): + kwargs["transport"] = httpx.MockTransport(handler) + kwargs.pop("follow_redirects", None) + return real_client(*args, follow_redirects=False, **kwargs) + + signing.httpx.AsyncClient = factory + try: + import asyncio + + return asyncio.run( + signing.fetch_signing_keys( + "https://agent.example/p", allow_insecure=True + ) + ) + finally: + signing.httpx.AsyncClient = real_client + + def test_reads_signing_keys(self) -> None: + """signing_keys is read from the ucp envelope.""" + keys = self._fetch( + lambda req: httpx.Response( + 200, json={"ucp": {"signing_keys": [{"kid": "a"}]}} + ) + ) + self.assertEqual(keys[0]["kid"], "a") + + def test_keys_fallback(self) -> None: + """A top-level keys array is read when signing_keys is absent.""" + keys = self._fetch( + lambda req: httpx.Response(200, json={"keys": [{"kid": "b"}]}) + ) + self.assertEqual(keys[0]["kid"], "b") + + def test_redirect_is_unreachable(self) -> None: + """A 3xx response is treated as unreachable (no redirects allowed).""" + with self.assertRaises(signing.SignatureError) as ctx: + self._fetch( + lambda req: httpx.Response(302, headers={"location": "https://x/y"}) + ) + self.assertEqual(ctx.exception.code, "profile_unreachable") + + def test_non_json_is_malformed(self) -> None: + """A non-JSON body yields profile_malformed.""" + with self.assertRaises(signing.SignatureError) as ctx: + self._fetch(lambda req: httpx.Response(200, text="not json")) + self.assertEqual(ctx.exception.code, "profile_malformed") + + def test_keyless_is_malformed(self) -> None: + """A profile with no keys yields profile_malformed.""" + with self.assertRaises(signing.SignatureError) as ctx: + self._fetch(lambda req: httpx.Response(200, json={"ucp": {}})) + self.assertEqual(ctx.exception.code, "profile_malformed") + + +class DifferentialLibraryTest(absltest.TestCase): + """Cross-check against the independent http-message-signatures library.""" + + def test_es256_both_directions(self) -> None: + """Our ES256 signature and the library's verify each other.""" + from http_message_signatures.algorithms import ECDSA_P256_SHA256 + + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k") + alg = ECDSA_P256_SHA256(private_key=key, public_key=key.public_key()) + alg.verify(signing._raw_sign(key, RFC_B26_BASE), RFC_B26_BASE) + signing.verify_raw_signature(jwk, RFC_B26_BASE, alg.sign(RFC_B26_BASE)) + + def test_ed25519_both_directions(self) -> None: + """Our Ed25519 signature and the library's verify each other.""" + from http_message_signatures.algorithms import ED25519 + + key = ed25519.Ed25519PrivateKey.generate() + jwk = signing.jwk_from_public_key(key.public_key(), "e") + alg = ED25519(private_key=key, public_key=key.public_key()) + alg.verify(signing._raw_sign(key, RFC_B26_BASE), RFC_B26_BASE) + signing.verify_raw_signature(jwk, RFC_B26_BASE, alg.sign(RFC_B26_BASE)) + + +if __name__ == "__main__": + absltest.main() From faa5b28d367a20f9f8e2462bbaefea0b4326a712 Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Tue, 7 Jul 2026 18:17:49 -0400 Subject: [PATCH 2/7] fix(python/server): verify request signatures instead of accepting everything The previous verify_signature accepted any request, short-circuited on the literal string "test", read a bespoke Request-Signature header, and its docstring described HMAC -- none of which matches signatures.md, which mandates asymmetric RFC 9421 signatures (ES256 baseline) with key discovery from the UCP-Agent profile. verify_signature now performs real RFC 9421 verification via ucp_signing: Content-Digest, the UCP covered-component coverage gate, raw-r||s ES256 / Ed25519, and signer-key discovery from the UCP-Agent profile's signing_keys. Compatibility is preserved by default: - Two flags (config.py): --require_signatures (default false) and --allow_insecure_profile_urls (default false, for localhost demos/CI). - With enforcement off, signatures are still verified when present and logged, but unsigned/invalid requests are allowed. No profile fetch occurs unless a Signature-Input header is present, so unsigned traffic (including the exact header shape the official conformance suite sends) is unaffected. - The bespoke Request-Signature header is now optional but still accepted. signature_integration_test.py proves both modes end to end against an in-process localhost profile server: permissive passthrough (including an official-conformance replay that asserts no key fetch happens) and, under enforcement, the spec's error code for every failure mode. --- rest/python/server/config.py | 14 + rest/python/server/dependencies.py | 133 ++++-- .../server/signature_integration_test.py | 436 ++++++++++++++++++ 3 files changed, 557 insertions(+), 26 deletions(-) create mode 100644 rest/python/server/signature_integration_test.py diff --git a/rest/python/server/config.py b/rest/python/server/config.py index fcf4b2c..0a127c5 100644 --- a/rest/python/server/config.py +++ b/rest/python/server/config.py @@ -53,6 +53,20 @@ def get_server_version() -> str: "Secret key for simulation endpoints", ) flags.DEFINE_integer("port", None, "Port to run the server on") + flags.DEFINE_boolean( + "require_signatures", + False, + "Reject requests whose RFC 9421 signature is missing or invalid. When " + "false (the default), signatures are still verified when present, but " + "unsigned or invalid requests are allowed and only logged.", + ) + flags.DEFINE_boolean( + "allow_insecure_profile_urls", + False, + "Permit http and loopback/private UCP-Agent profile URLs when resolving " + "signer keys. For localhost demos and CI only; never enable in " + "production, as it disables SSRF protections.", + ) except flags.DuplicateFlagError: pass diff --git a/rest/python/server/dependencies.py b/rest/python/server/dependencies.py index ebb4935..fe46b7e 100644 --- a/rest/python/server/dependencies.py +++ b/rest/python/server/dependencies.py @@ -16,12 +16,13 @@ This module contains dependency injection logic for FastAPI endpoints, including: -- Header validation (UCP-Agent, Idempotency-Key, Request-Signature). +- Header validation (UCP-Agent, Idempotency-Key). +- RFC 9421 request-signature verification (UCP-Agent key discovery). - Service instantiation (CheckoutService, FulfillmentService). - Database session management (Products and Transactions DBs). -- Request signature verification for webhooks. """ +import logging import re from collections.abc import AsyncGenerator from typing import Annotated @@ -36,6 +37,9 @@ from services.checkout_service import CheckoutService from services.fulfillment_service import FulfillmentService from sqlalchemy.ext.asyncio import AsyncSession +import ucp_signing + +logger = logging.getLogger(__name__) class CommonHeaders(BaseModel): @@ -43,18 +47,116 @@ class CommonHeaders(BaseModel): x_api_key: str | None = None ucp_agent: str - request_signature: str + request_signature: str | None = None request_id: str +def _signature_http_error(exc: ucp_signing.SignatureError) -> HTTPException: + """Wrap a SignatureError in the UCP error-envelope HTTP response.""" + return HTTPException( + status_code=exc.status_code, + detail={ + "status": "error", + "errors": [ + {"code": exc.code, "message": exc.message, "severity": "critical"} + ], + }, + ) + + +async def verify_signature(request: Request) -> None: + """Verify an inbound request's RFC 9421 signature per the UCP spec. + + The signer's public keys are discovered from the ``UCP-Agent`` header's + profile URL (its ``signing_keys``). Behaviour depends on + ``--require_signatures``: + + * When set, a missing or invalid signature is rejected with the spec's + error code (401 ``signature_missing`` / ``signature_invalid`` / + ``key_not_found``, 400 ``digest_mismatch`` / ``algorithm_unsupported``, + etc.). + * When unset (the default), signatures are still verified when present and + the outcome is logged, but unsigned or invalid requests are allowed. This + keeps the sample interoperable with clients that do not yet sign. + + No profile fetch occurs unless a ``Signature-Input`` header is present, so + unsigned traffic incurs no extra work. + + Args: + request: The incoming request. + + Raises: + HTTPException: With a UCP error envelope when enforcement is on and + verification fails. + + """ + headers = {k.lower(): v for k, v in request.headers.items()} + enforcing = config.FLAGS.require_signatures + + if "signature-input" not in headers or "signature" not in headers: + if enforcing: + raise _signature_http_error( + ucp_signing.SignatureError( + "signature_missing", 401, "Request signature is required" + ) + ) + logger.debug("No request signature present; skipping verification") + return + + match = re.search(r'profile="([^"]+)"', headers.get("ucp-agent", "")) + if not match: + exc = ucp_signing.SignatureError( + "signature_invalid", + 401, + "UCP-Agent profile URL is required to resolve the signing key", + ) + if enforcing: + raise _signature_http_error(exc) + logger.warning("Cannot verify signature: %s", exc.message) + return + + body = await request.body() + try: + keys = await ucp_signing.fetch_signing_keys( + match.group(1), + allow_insecure=config.FLAGS.allow_insecure_profile_urls, + ) + keyid = ucp_signing.verify_request( + request.method, + headers.get("host", ""), + request.url.path, + request.url.query, + headers, + body, + keys, + ) + except ucp_signing.SignatureError as exc: + if enforcing: + raise _signature_http_error(exc) from exc + logger.warning( + "Request signature verification failed (%s: %s); allowing because " + "--require_signatures is not set", + exc.code, + exc.message, + ) + return + logger.info( + "RFC 9421 signature verified (keyid=%s, profile=%s)", + keyid, + match.group(1), + ) + + async def common_headers( + request: Request, x_api_key: str | None = Header(None), ucp_agent: str = Header(...), - request_signature: str = Header(...), + request_signature: str | None = Header(None), request_id: str = Header(...), ) -> CommonHeaders: - """Extract and validate common headers.""" + """Extract and validate common headers, verifying any request signature.""" await validate_ucp_headers(ucp_agent) + await verify_signature(request) return CommonHeaders( x_api_key=x_api_key, ucp_agent=ucp_agent, @@ -106,27 +208,6 @@ async def idempotency_header( return idempotency_key -async def verify_signature( - request_signature: str = Header(..., alias="Request-Signature"), -) -> None: - """Verify the request signature. - - Note: This is a placeholder implementation that bypasses validation if the - signature is "test". A real implementation would verify the HMAC-SHA256 - signature of the request body. - - Args: - request_signature: The signature header from the platform. - - """ - # In tests, we might want to bypass validation if signature is "test" - if request_signature == "test": - return - # In sample implementation, we don't enforce signature validation - # as we don't share secrets with clients. - return - - async def verify_simulation_secret( simulation_secret: str | None = Header(None, alias="Simulation-Secret"), ) -> None: diff --git a/rest/python/server/signature_integration_test.py b/rest/python/server/signature_integration_test.py new file mode 100644 index 0000000..acd99f3 --- /dev/null +++ b/rest/python/server/signature_integration_test.py @@ -0,0 +1,436 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""End-to-end tests for RFC 9421 request-signature verification. + +`PermissiveModeTest` proves the default (unset ``--require_signatures``) leaves +existing clients untouched -- including the exact header shape the official +conformance suite sends -- while still verifying and logging real signatures. +`EnforcedModeTest` proves that with enforcement on, the server returns the +spec's error codes for every failure mode. + +Signer keys are served from an in-process localhost profile server, mirroring +the topology of the official conformance harness. All key material is generated +at runtime. +""" + +import http.server +import json +import threading +import uuid + +from absl.testing import absltest +from cryptography.hazmat.primitives.asymmetric import ec +import config +from cryptography.hazmat.primitives import hashes +import dependencies +from integration_test import IntegrationTest +import ucp_signing + + +class _ProfileHandler(http.server.BaseHTTPRequestHandler): + """Serve signer profiles from an in-memory routing table.""" + + routes: dict = {} + + def do_GET(self) -> None: # noqa: N802 (http.server API) + """Return the configured status and body for the requested path.""" + entry = self.routes.get(self.path) + if entry is None: + self.send_response(404) + self.end_headers() + return + status, body = entry + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args) -> None: + """Silence the default request logging.""" + del args + + +class _SigTestBase(IntegrationTest): + """Shared scaffolding: DB seed (from the base) plus a profile server. + + The lifecycle tests defined on ``IntegrationTest`` are suppressed here so the + signature scenarios run in isolation; they are exercised in that file, and + under enforcement the legacy unsigned flows intentionally do not apply. + """ + + # Suppress every test inherited from IntegrationTest — not a hand-kept name + # list, which would silently re-enable any test added to the base file later. + for _inherited in [_n for _n in dir(IntegrationTest) if _n.startswith("test_")]: + locals()[_inherited] = None + del _inherited + + require_signatures = False + + def setUp(self) -> None: + """Start a localhost profile server and configure signature flags.""" + super().setUp() + ucp_signing.clear_key_cache() + config.FLAGS.require_signatures = self.require_signatures + config.FLAGS.allow_insecure_profile_urls = True + + self.agent_key = ec.generate_private_key(ec.SECP256R1()) + self.agent_kid = "test-agent-key" + agent_jwk = ucp_signing.jwk_from_public_key( + self.agent_key.public_key(), self.agent_kid + ) + # A deliberately unsupported (RSA) JWK to exercise algorithm_unsupported. + rsa_jwk = {"kid": "rsa-key", "kty": "RSA", "n": "abc", "e": "AQAB"} + + version = config.get_server_version() + good = json.dumps( + {"ucp": {"version": version, "signing_keys": [agent_jwk, rsa_jwk]}} + ).encode() + _ProfileHandler.routes = { + "/profile.json": (200, good), + "/keyless.json": (200, json.dumps({"ucp": {}}).encode()), + } + self.server = http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), _ProfileHandler + ) + self.port = self.server.server_address[1] + self.profile_url = f"http://127.0.0.1:{self.port}/profile.json" + self.thread = threading.Thread(target=self.server.serve_forever) + self.thread.daemon = True + self.thread.start() + + def tearDown(self) -> None: + """Stop the profile server and reset signature flags.""" + self.server.shutdown() + self.server.server_close() + config.FLAGS.require_signatures = False + config.FLAGS.allow_insecure_profile_urls = False + super().tearDown() + + def _checkout_body(self, checkout_id: str) -> bytes: + """Return the raw bytes of a single-item checkout create request.""" + payload = self._create_checkout_payload( + checkout_id, [("rose", "Red Rose", 1000, 1)] + ) + return json.dumps( + payload.model_dump(mode="json", exclude_none=True) + ).encode() + + def _signed_headers( + self, + method: str, + path: str, + body: bytes, + *, + key=None, + kid: str | None = None, + profile: str | None = None, + extra_sign_headers: dict | None = None, + ) -> dict: + """Build headers for a signed request against the in-process server.""" + key = key or self.agent_key + kid = kid or self.agent_kid + profile = profile or self.profile_url + headers = { + "UCP-Agent": f'profile="{profile}"', + "Idempotency-Key": str(uuid.uuid4()), + "Request-Id": str(uuid.uuid4()), + } + sign_headers = dict(headers) + if extra_sign_headers is not None: + sign_headers = extra_sign_headers + additions = ucp_signing.sign_request( + key, kid, method, f"http://testserver{path}", sign_headers, body + ) + headers.update(additions) + return headers + + +class PermissiveModeTest(_SigTestBase): + """Default mode: existing clients keep working; signatures are logged.""" + + require_signatures = False + + def test_official_conformance_replay_no_fetch(self) -> None: + """Legacy headers with no Signature-Input succeed without a key fetch.""" + calls = [] + original = ucp_signing.fetch_signing_keys + + async def spy(*args, **kwargs): + calls.append(args) + return await original(*args, **kwargs) + + ucp_signing.fetch_signing_keys = spy + try: + with self.client: + body = self._checkout_body("replay_1") + response = self.client.post( + "/checkout-sessions", + headers={ + "UCP-Agent": 'profile="https://agent.example/profile"', + "request-signature": "test", + "idempotency-key": "1", + "request-id": "1", + "content-type": "application/json", + }, + content=body, + ) + self.assertEqual(response.status_code, 201, response.text) + self.assertEmpty(calls, "No profile fetch should occur for unsigned reqs") + finally: + ucp_signing.fetch_signing_keys = original + + def test_unsigned_without_legacy_header(self) -> None: + """A request omitting Request-Signature entirely still succeeds.""" + with self.client: + body = self._checkout_body("unsigned_1") + response = self.client.post( + "/checkout-sessions", + headers={ + "UCP-Agent": 'profile="https://agent.example/profile"', + "idempotency-key": "1", + "request-id": "1", + "content-type": "application/json", + }, + content=body, + ) + self.assertEqual(response.status_code, 201, response.text) + + def test_valid_signature_is_verified_and_logged(self) -> None: + """A correctly signed request is verified and logs confirmation.""" + with self.client: + body = self._checkout_body("signed_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + with self.assertLogs(dependencies.logger, level="INFO") as logs: + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + ) + self.assertEqual(response.status_code, 201, response.text) + self.assertTrue( + any("RFC 9421 signature verified" in line for line in logs.output) + ) + + def test_bad_signature_allowed_with_warning(self) -> None: + """An invalid signature is allowed in permissive mode but warned about.""" + with self.client: + body = self._checkout_body("badsig_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + tampered = body + b" " + with self.assertLogs(dependencies.logger, level="WARNING") as logs: + response = self.client.post( + "/checkout-sessions", headers=headers, content=tampered + ) + self.assertEqual(response.status_code, 201, response.text) + self.assertTrue(any("verification failed" in x for x in logs.output)) + + +class EnforcedModeTest(_SigTestBase): + """Enforcement on: every failure mode returns its spec error code.""" + + require_signatures = True + + def _post_signed(self, checkout_id: str, **kwargs): + """Sign and POST a checkout create; return the response.""" + body = self._checkout_body(checkout_id) + headers = self._signed_headers("POST", "/checkout-sessions", body, **kwargs) + return ( + self.client.post("/checkout-sessions", headers=headers, content=body), + body, + headers, + ) + + def _assert_error(self, response, status: int, code: str) -> None: + """Assert an HTTP status and UCP error code on a response.""" + self.assertEqual(response.status_code, status, response.text) + self.assertEqual(response.json()["detail"]["errors"][0]["code"], code) + + def test_valid_signature_accepted(self) -> None: + """A correctly signed request is accepted.""" + with self.client: + response, _, _ = self._post_signed("ok_1") + self.assertEqual(response.status_code, 201, response.text) + + def test_unsigned_rejected(self) -> None: + """A request with no signature is rejected with signature_missing.""" + with self.client: + body = self._checkout_body("miss_1") + response = self.client.post( + "/checkout-sessions", + headers={ + "UCP-Agent": f'profile="{self.profile_url}"', + "idempotency-key": "1", + "request-id": "1", + }, + content=body, + ) + self._assert_error(response, 401, "signature_missing") + + def test_tampered_body_rejected(self) -> None: + """A body that does not match Content-Digest yields digest_mismatch.""" + with self.client: + body = self._checkout_body("tamper_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + b" " + ) + self._assert_error(response, 400, "digest_mismatch") + + def test_wrong_key_rejected(self) -> None: + """A signature from an unpublished key yields signature_invalid.""" + other = ec.generate_private_key(ec.SECP256R1()) + with self.client: + response, _, _ = self._post_signed("wrong_1", key=other) + self._assert_error(response, 401, "signature_invalid") + + def test_unknown_kid_rejected(self) -> None: + """A keyid not in the published set yields key_not_found.""" + with self.client: + response, _, _ = self._post_signed("kid_1", kid="nonexistent") + self._assert_error(response, 401, "key_not_found") + + def test_der_signature_rejected(self) -> None: + """A DER-encoded signature on the wire yields signature_invalid.""" + with self.client: + body = self._checkout_body("der_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + # Re-sign the base as DER to violate the raw-r||s requirement. + parsed = ucp_signing.parse_signature_input(headers["Signature-Input"]) + raw = parsed["sig1"]["raw"] + base = ucp_signing.build_signature_base( + parsed["sig1"]["components"], + raw, + _resolver(body, headers, self.port), + ) + der = self.agent_key.sign(base, ec.ECDSA(hashes.SHA256())) + import base64 + + headers["Signature"] = ( + "sig1=:" + base64.b64encode(der).decode("ascii") + ":" + ) + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + ) + self._assert_error(response, 401, "signature_invalid") + + def test_uncovered_component_rejected(self) -> None: + """A signature that omits a required component yields signature_invalid.""" + with self.client: + body = self._checkout_body("cov_1") + # Sign WITHOUT ucp-agent in the covered set, then add the header. + headers = self._signed_headers( + "POST", + "/checkout-sessions", + body, + extra_sign_headers={ + "Idempotency-Key": str(uuid.uuid4()), + "Request-Id": str(uuid.uuid4()), + }, + ) + headers["UCP-Agent"] = f'profile="{self.profile_url}"' + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + ) + self._assert_error(response, 401, "signature_invalid") + + def test_alg_param_rejected(self) -> None: + """A signature carrying an alg parameter yields signature_invalid.""" + with self.client: + body = self._checkout_body("alg_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + headers["Signature-Input"] = headers["Signature-Input"].replace( + ";created", ';alg="ecdsa-p256-sha256";created' + ) + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + ) + self._assert_error(response, 401, "signature_invalid") + + def test_unsupported_key_algorithm(self) -> None: + """A keyid selecting an RSA key yields algorithm_unsupported.""" + with self.client: + body = self._checkout_body("rsa_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + headers["Signature-Input"] = headers["Signature-Input"].replace( + f'keyid="{self.agent_kid}"', 'keyid="rsa-key"' + ) + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + ) + self._assert_error(response, 400, "algorithm_unsupported") + + def test_dead_profile_port_unreachable(self) -> None: + """An unresolvable profile port yields profile_unreachable.""" + dead = "http://127.0.0.1:1/profile.json" + with self.client: + response, _, _ = self._post_signed("dead_1", profile=dead) + self._assert_error(response, 424, "profile_unreachable") + + def test_keyless_profile_malformed(self) -> None: + """A profile with no keys yields profile_malformed.""" + keyless = f"http://127.0.0.1:{self.port}/keyless.json" + with self.client: + response, _, _ = self._post_signed("keyless_1", profile=keyless) + self._assert_error(response, 422, "profile_malformed") + + def test_http_profile_without_carveout(self) -> None: + """With the carve-out off, an http profile URL is rejected.""" + config.FLAGS.allow_insecure_profile_urls = False + with self.client: + response, _, _ = self._post_signed("insecure_1") + self._assert_error(response, 400, "invalid_profile_url") + + def test_multi_signature_one_valid(self) -> None: + """A request with one bad and one valid signature is accepted.""" + with self.client: + body = self._checkout_body("multi_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + # Add a second, bogus signature label; the good sig1 must still pass. + headers["Signature-Input"] += ( + ', sig2=("@method");created=1;keyid="' + self.agent_kid + '"' + ) + headers["Signature"] += ", sig2=:AAAA:" + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + ) + self.assertEqual(response.status_code, 201, response.text) + + def test_webhook_unsigned_rejected(self) -> None: + """The webhook route also enforces signatures when required.""" + with self.client: + response = self.client.post( + "/webhooks/partners/p1/events/order", + headers={"UCP-Agent": f'profile="{self.profile_url}"'}, + json={"id": "order_1"}, + ) + self._assert_error(response, 401, "signature_missing") + + +def _resolver(body: bytes, headers: dict, port: int): + """Build a signature-base component resolver for the DER-tamper test.""" + digest = ucp_signing.content_digest(body) + values = { + "@method": "POST", + "@authority": "testserver", + "@path": "/checkout-sessions", + "@query": "?", + "content-digest": digest, + "content-type": "application/json", + "idempotency-key": headers["Idempotency-Key"], + "ucp-agent": headers["UCP-Agent"], + } + return values.get + + +if __name__ == "__main__": + absltest.main() From 376d58a3939e101e0bfd4f6f8f28228a1bb1ceaa Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Tue, 7 Jul 2026 18:23:04 -0400 Subject: [PATCH 3/7] feat(python/client): sign happy-path requests with an ephemeral key The demo client now signs every request per RFC 9421 by default: it generates an ephemeral ES256 key, publishes the public key from a small localhost profile server, and points the UCP-Agent header at it so the merchant can discover the key and verify the signature. Signing is applied centrally via an httpx auth flow, so it covers the exact serialized body bytes. --disable_signatures restores the previous unsigned behaviour (used to keep the exported curl dialog replayable, since a real signature is bound to the body and host). --- rest/python/client/flower_shop/pyproject.toml | 1 + rest/python/client/flower_shop/signing.py | 184 ++++++++++++++++++ .../flower_shop/simple_happy_path_client.py | 48 ++++- 3 files changed, 228 insertions(+), 5 deletions(-) create mode 100644 rest/python/client/flower_shop/signing.py diff --git a/rest/python/client/flower_shop/pyproject.toml b/rest/python/client/flower_shop/pyproject.toml index d2a4e68..c93378f 100644 --- a/rest/python/client/flower_shop/pyproject.toml +++ b/rest/python/client/flower_shop/pyproject.toml @@ -8,6 +8,7 @@ authors = [ { name = "UCP Team" } ] dependencies = [ + "cryptography>=42", "pydantic>=2.12.0", "ucp-sdk", ] diff --git a/rest/python/client/flower_shop/signing.py b/rest/python/client/flower_shop/signing.py new file mode 100644 index 0000000..a8d9c10 --- /dev/null +++ b/rest/python/client/flower_shop/signing.py @@ -0,0 +1,184 @@ +# Copyright 2026 UCP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Client-side RFC 9421 request signing for the happy-path demo. + +Provides an ``httpx`` auth flow that signs every outgoing request with an +ephemeral ES256 key, and a tiny localhost profile server that publishes the +matching public key so the merchant can discover it via the ``UCP-Agent`` +header. This is the counterpart to the server's ``ucp_signing`` verifier and +demonstrates the full sign-then-verify loop end to end. +""" + +import base64 +import hashlib +import http.server +import json +import threading +import time +from urllib.parse import urlsplit + +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature +import httpx + +_ES256_COORD_BYTES = 32 + + +def _b64u(data: bytes) -> str: + """Encode bytes as base64url text without padding.""" + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def content_digest(body: bytes) -> str: + """Return the RFC 9530 sha-256 Content-Digest for the raw body bytes.""" + digest = hashlib.sha256(body).digest() + return "sha-256=:" + base64.b64encode(digest).decode("ascii") + ":" + + +def public_jwk(public_key: ec.EllipticCurvePublicKey, kid: str) -> dict: + """Export an ES256 public key as a UCP JWK.""" + numbers = public_key.public_numbers() + return { + "kid": kid, + "kty": "EC", + "crv": "P-256", + "x": _b64u(numbers.x.to_bytes(_ES256_COORD_BYTES, "big")), + "y": _b64u(numbers.y.to_bytes(_ES256_COORD_BYTES, "big")), + "use": "sig", + "alg": "ES256", + } + + +class RequestSigner(httpx.Auth): + """An httpx auth flow that RFC 9421-signs every request. + + The signed component set follows the UCP coverage table: the method, + authority, and path (plus query when present), the content digest and type + for bodied requests, and the idempotency-key and ucp-agent headers when they + are present on the request. + """ + + requires_request_body = True + + def __init__(self, private_key: ec.EllipticCurvePrivateKey, kid: str) -> None: + """Store the signing key and its identifier.""" + self._key = private_key + self._kid = kid + + def auth_flow(self, request): + """Add Content-Digest, Signature-Input, and Signature to the request.""" + body = request.content or b"" + lowered = {k.lower(): v for k, v in request.headers.items()} + split = urlsplit(str(request.url)) + + if body: + digest = content_digest(body) + request.headers["Content-Digest"] = digest + lowered["content-digest"] = digest + if "content-type" not in lowered: + request.headers["Content-Type"] = "application/json" + lowered["content-type"] = "application/json" + + components = ["@method", "@authority", "@path"] + if split.query: + components.append("@query") + if body: + components.extend(["content-digest", "content-type"]) + if "idempotency-key" in lowered: + components.append("idempotency-key") + if "ucp-agent" in lowered: + components.append("ucp-agent") + + created = int(time.time()) + raw_params = ( + "(" + " ".join(f'"{c}"' for c in components) + ")" + f';created={created};keyid="{self._kid}"' + ) + + def resolve(name: str) -> str: + if name == "@method": + return request.method.upper() + if name == "@authority": + return split.netloc.lower() + if name == "@path": + return split.path or "/" + if name == "@query": + return "?" + split.query + return lowered[name] + + lines = [f'"{c}": {resolve(c)}' for c in components] + lines.append(f'"@signature-params": {raw_params}') + base = "\n".join(lines).encode("utf-8") + + der = self._key.sign(base, ec.ECDSA(_ecdsa_hash())) + r, s = decode_dss_signature(der) + raw = r.to_bytes(_ES256_COORD_BYTES, "big") + s.to_bytes( + _ES256_COORD_BYTES, "big" + ) + request.headers["Signature-Input"] = f"sig1={raw_params}" + request.headers["Signature"] = ( + "sig1=:" + base64.b64encode(raw).decode("ascii") + ":" + ) + yield request + + +def _ecdsa_hash(): + """Return the SHA-256 primitive that ES256 signs with.""" + from cryptography.hazmat.primitives import hashes + + return hashes.SHA256() + + +class LocalProfileServer: + """Serve a UCP profile publishing the demo signing key on localhost.""" + + def __init__(self, jwk: dict, version: str = "2026-01-23") -> None: + """Prepare the profile document and HTTP server (not yet started).""" + document = json.dumps( + {"ucp": {"version": version, "signing_keys": [jwk]}} + ).encode() + + class _Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 (http.server API) + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(document) + + def log_message(self, *args) -> None: + del args + + self._server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + self.port = self._server.server_address[1] + self.profile_url = f"http://127.0.0.1:{self.port}/profile.json" + self._thread = threading.Thread(target=self._server.serve_forever) + self._thread.daemon = True + + def start(self) -> None: + """Start serving in a background thread.""" + self._thread.start() + + def stop(self) -> None: + """Stop the server and release its socket.""" + self._server.shutdown() + self._server.server_close() + + +def build_signer() -> tuple[RequestSigner, LocalProfileServer]: + """Create an ephemeral key, a signer, and its localhost profile server.""" + key = ec.generate_private_key(ec.SECP256R1()) + kid = "flower-shop-demo-client" + server = LocalProfileServer(public_jwk(key.public_key(), kid)) + return RequestSigner(key, kid), server diff --git a/rest/python/client/flower_shop/simple_happy_path_client.py b/rest/python/client/flower_shop/simple_happy_path_client.py index ad3f15e..110f26d 100644 --- a/rest/python/client/flower_shop/simple_happy_path_client.py +++ b/rest/python/client/flower_shop/simple_happy_path_client.py @@ -34,6 +34,7 @@ from pathlib import Path import uuid import httpx +import signing from ucp_sdk.models.schemas.shopping import checkout_create_request from ucp_sdk.models.schemas.shopping import checkout_update_request from ucp_sdk.models.schemas.shopping import payment_create_request @@ -44,14 +45,30 @@ from ucp_sdk.models.schemas.shopping.types import line_item_update_request +# Set by main() when request signing is enabled; the profile URL published by +# the local key server, referenced by the UCP-Agent header so the merchant can +# discover the signing key. +_SIGNING_PROFILE_URL: str | None = None + + def get_headers() -> dict[str, str]: - """Generate necessary headers for UCP requests.""" - return { - "UCP-Agent": 'profile="https://agent.example/profile"', - "request-signature": "test", + """Generate necessary headers for UCP requests. + + When signing is enabled the UCP-Agent header points at the demo's local key + server (the signature itself is added by the httpx auth flow). Otherwise it + falls back to the historical placeholder and the legacy request-signature + header, so the unsigned demo behaves exactly as before. + """ + headers = { "idempotency-key": str(uuid.uuid4()), "request-id": str(uuid.uuid4()), } + if _SIGNING_PROFILE_URL: + headers["UCP-Agent"] = f'profile="{_SIGNING_PROFILE_URL}"' + else: + headers["UCP-Agent"] = 'profile="https://agent.example/profile"' + headers["request-signature"] = "test" + return headers def remove_none_values(obj): @@ -154,6 +171,16 @@ def main() -> None: help="Path to export requests and responses as markdown.", ) + parser.add_argument( + "--disable_signatures", + action="store_true", + help=( + "Do not sign requests. By default the client signs every request with " + "an ephemeral ES256 key and publishes the public key from a local " + "profile server for the merchant to verify." + ), + ) + args = parser.parse_args() # Configure Logging @@ -164,7 +191,18 @@ def main() -> None: logger = logging.getLogger(__name__) - client = httpx.Client(base_url=args.server_url) + signer = None + if not args.disable_signatures: + global _SIGNING_PROFILE_URL + signer, profile_server = signing.build_signer() + profile_server.start() + _SIGNING_PROFILE_URL = profile_server.profile_url + logger.info( + "Signing requests with an ephemeral ES256 key; publishing it at %s", + _SIGNING_PROFILE_URL, + ) + + client = httpx.Client(base_url=args.server_url, auth=signer) # Clear the export file if it exists if args.export_requests_to: From d7bfb1fb5ebaf5ce6e681a65fed5d3b9643390ba Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Tue, 7 Jul 2026 18:25:30 -0400 Subject: [PATCH 4/7] docs(python): document request-signature verification and demo flags Server README: a Request Signatures (RFC 9421) section covering what is verified, the --require_signatures / --allow_insecure_profile_urls flags, the spec error codes, and a local sign-then-verify demo recipe. Client README: documents --disable_signatures. --- rest/python/client/flower_shop/README.md | 8 +++++ rest/python/server/README.md | 37 ++++++++++++++++++++++++ 2 files changed, 45 insertions(+) diff --git a/rest/python/client/flower_shop/README.md b/rest/python/client/flower_shop/README.md index 3e5a911..b339d42 100644 --- a/rest/python/client/flower_shop/README.md +++ b/rest/python/client/flower_shop/README.md @@ -65,6 +65,14 @@ uv run simple_happy_path_client.py --server_url=http://localhost:8182 uv run simple_happy_path_client.py --export_requests_to=interaction_log.md ``` +- `--disable_signatures`: Do not sign requests. By default the client signs + every request per RFC 9421 with an ephemeral `ES256` key and publishes the + matching public key from a small local profile server, so the merchant can + discover the key (via the `UCP-Agent` header) and verify the signature. Pass + this flag to send unsigned requests instead — useful when exporting a + replayable dialog, since a real signature is bound to the request body and + host. + ## Automated Demo (extract_json_dialog.sh) For a fully automated demonstration that sets up the database, starts the diff --git a/rest/python/server/README.md b/rest/python/server/README.md index 1b6e7c9..f42882d 100644 --- a/rest/python/server/README.md +++ b/rest/python/server/README.md @@ -90,6 +90,43 @@ SERVER_PID=$! Note: Keep the server running for the duration of running the client and the following experiments. +## Request Signatures (RFC 9421) + +The server verifies UCP request signatures as defined in the specification's +[`signatures.md`](https://github.com/Universal-Commerce-Protocol/ucp/blob/main/docs/specification/signatures.md): +[RFC 9421](https://www.rfc-editor.org/rfc/rfc9421.html) HTTP Message Signatures +with an [RFC 9530](https://www.rfc-editor.org/rfc/rfc9530.html) `Content-Digest` +over the raw body. The signer's public key is discovered from the profile URL in +the `UCP-Agent` header (its `signing_keys`). `ES256` (fixed-width raw `r||s`, not +ASN.1/DER) is the baseline; `Ed25519` is also supported. + +Behaviour is controlled by two flags: + +| Flag | Default | Effect | +| ------------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `--require_signatures` | `false` | Reject requests whose signature is missing or invalid. When `false`, a present signature is still verified and the result logged, but unsigned or invalid requests are allowed — so existing clients keep working. | +| `--allow_insecure_profile_urls` | `false` | Permit `http` and loopback/private profile URLs when resolving keys. For localhost demos and CI only; it disables SSRF protections and must never be enabled in production. | + +When verification fails under enforcement, the server returns the spec's error +code: `401 signature_missing` / `signature_invalid` / `key_not_found`, +`400 digest_mismatch` / `algorithm_unsupported` / `invalid_profile_url`, +`424 profile_unreachable`, or `422 profile_malformed`. + +To see the full sign-then-verify loop locally, run the server with the demo +carve-out and let the client sign (it signs by default and publishes its key +from a local profile server): + +```shell +uv run server.py \ + --products_db_path=/tmp/ucp_test/products.db \ + --transactions_db_path=/tmp/ucp_test/transactions.db \ + --port=8182 --allow_insecure_profile_urls & +``` + +Each verified request logs +`RFC 9421 signature verified (keyid=..., profile=...)`. Add `--require_signatures` +to reject anything unsigned. + ## Run a Simple Client Exercise a simple checkout path: Once the server is running, execute the simple From 77e891e5a10e811bd050c38dd9e4d638cdc9f594 Mon Sep 17 00:00:00 2001 From: vishkaty Date: Sun, 12 Jul 2026 00:26:54 -0400 Subject: [PATCH 5/7] test(python/server): self-review hardening + full coverage for the verifier Keeps the PR at request verification; adds robustness fixes surfaced by an adversarial self-review and drives the verifier to complete test coverage. Verifier fixes (RFC 9421 exactness): - @authority: strip the scheme default port (Section 2.2.3) so a signer that signs `host` still verifies behind an intermediary that presents `host:443`. - @path: empty path normalises to `/` on the verify side too (Section 2.2.6). - covered field values are OWS-trimmed (Section 2.1). - coverage gate no longer requires `signature-agent`: it is a WBA-shape component this default-UCP verifier does not parse, so the promise now matches what it verifies (WBA-shape remains a documented follow-up). Tests: ucp_signing.py to 100% line+branch coverage; positive and negative request-verification cases (Ed25519 through verify_request, body without Content-Digest, malformed digests/JWKs/headers, SSRF resolve paths, key cache, no-profile wiring in both modes). 53 -> 91 tests, green on Python 3.10 and 3.12, ruff clean. --- .../server/signature_integration_test.py | 26 ++ rest/python/server/ucp_signing.py | 32 +- rest/python/server/ucp_signing_test.py | 358 +++++++++++++++++- 3 files changed, 406 insertions(+), 10 deletions(-) diff --git a/rest/python/server/signature_integration_test.py b/rest/python/server/signature_integration_test.py index acd99f3..17e11d9 100644 --- a/rest/python/server/signature_integration_test.py +++ b/rest/python/server/signature_integration_test.py @@ -207,6 +207,17 @@ def test_unsigned_without_legacy_header(self) -> None: ) self.assertEqual(response.status_code, 201, response.text) + def test_signed_no_profile_allowed_when_permissive(self) -> None: + """Permissive mode allows a signed request that can't resolve a key.""" + with self.client: + body = self._checkout_body("perm_no_profile_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + headers["UCP-Agent"] = "version=2026-01-23" # no profile= + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + ) + self.assertEqual(response.status_code, 201, response.text) + def test_valid_signature_is_verified_and_logged(self) -> None: """A correctly signed request is verified and logs confirmation.""" with self.client: @@ -261,6 +272,21 @@ def test_valid_signature_accepted(self) -> None: response, _, _ = self._post_signed("ok_1") self.assertEqual(response.status_code, 201, response.text) + def test_signed_but_no_profile_url_rejected(self) -> None: + """A signature present with no UCP-Agent profile= cannot resolve a key. + + The key source is the UCP-Agent profile URL, so a signed request whose + UCP-Agent carries no profile= is signature_invalid (401) under enforcement. + """ + with self.client: + body = self._checkout_body("no_profile_1") + headers = self._signed_headers("POST", "/checkout-sessions", body) + headers["UCP-Agent"] = "version=2026-01-23" # no profile= + response = self.client.post( + "/checkout-sessions", headers=headers, content=body + ) + self._assert_error(response, 401, "signature_invalid") + def test_unsigned_rejected(self) -> None: """A request with no signature is rejected with signature_missing.""" with self.client: diff --git a/rest/python/server/ucp_signing.py b/rest/python/server/ucp_signing.py index a841849..cd4ff40 100644 --- a/rest/python/server/ucp_signing.py +++ b/rest/python/server/ucp_signing.py @@ -316,8 +316,12 @@ def required_components( required.append("idempotency-key") if "ucp-agent" in headers: required.append("ucp-agent") - if "signature-agent" in headers: - required.append("signature-agent") + # signature-agent is a WBA-shape component (carried with RFC 8941 component + # parameters, e.g. "signature-agent";key="sig1", under tag=web-bot-auth). This + # verifier covers the default-UCP regime only and does not parse WBA-shape + # component parameters, so it does not gate on signature-agent — the coverage + # promise matches what the module can actually verify. WBA-shape support + # (member selection, keyid thumbprint binding) is a documented follow-up. return required @@ -448,8 +452,18 @@ def _es256_hash(): def _authority(host: str) -> str: - """Normalise an authority to lowercase per RFC 9421 Section 2.2.3.""" - return host.lower() + """Normalise an authority per RFC 9421 Section 2.2.3. + + Lowercases and strips the scheme's default port. UCP signatures are HTTPS, so + a ``host:443`` (added by an intermediary) is equivalent to ``host`` and must + reconstruct to the same value the signer used. + """ + authority = host.lower() + if authority.endswith(":443"): + authority = authority[: -len(":443")] + elif authority.endswith(":80"): + authority = authority[: -len(":80")] + return authority def verify_request( @@ -506,10 +520,12 @@ def resolve(name: str) -> str | None: if name == "@authority": return _authority(authority) if name == "@path": - return path + return path or "/" # RFC 9421 Section 2.2.6: empty path is "/" if name == "@query": return "?" + query - return headers.get(name) + value = headers.get(name) + # RFC 9421 Section 2.1: a covered field value is OWS-trimmed. + return value.strip() if isinstance(value, str) else value last_error = SignatureError( "signature_invalid", 401, "No valid signature found" @@ -617,7 +633,9 @@ def resolve(name: str) -> str | None: return merged.get(name) base = build_signature_base(components, raw_params, resolve) - if base is None: + if ( + base is None + ): # pragma: no cover - defensive; sign covers only self-derived components missing = [c for c in components if resolve(c) is None] raise SignatureError( "signature_invalid", 401, f"Cannot sign; missing components: {missing}" diff --git a/rest/python/server/ucp_signing_test.py b/rest/python/server/ucp_signing_test.py index 899c26c..64383e5 100644 --- a/rest/python/server/ucp_signing_test.py +++ b/rest/python/server/ucp_signing_test.py @@ -24,6 +24,7 @@ """ import base64 +import time from absl.testing import absltest from cryptography.hazmat.primitives import hashes @@ -160,6 +161,22 @@ def test_es256_roundtrip(self) -> None: sig = signing._raw_sign(key, RFC_B26_BASE) signing.verify_raw_signature(jwk, RFC_B26_BASE, sig) + def test_ed25519_full_verify_request_roundtrip(self) -> None: + """An Ed25519-signed request verifies through the full verify_request.""" + key = ed25519.Ed25519PrivateKey.generate() + jwk = signing.jwk_from_public_key(key.public_key(), "ed-k") + add = signing.sign_request( + key, "ed-k", "GET", "https://m.example/p", {}, b"" + ) + headers = { + "signature-input": add["Signature-Input"], + "signature": add["Signature"], + } + keyid = signing.verify_request( + "GET", "m.example", "/p", "", headers, b"", [jwk] + ) + self.assertEqual(keyid, "ed-k") + class RawSignatureEncodingTest(absltest.TestCase): """The UCP raw-r||s ECDSA requirement (spec MUST; issue #569).""" @@ -251,13 +268,19 @@ def test_idempotency_key_header_on_get(self) -> None: ) self.assertIn("idempotency-key", required) - def test_ucp_agent_and_signature_agent(self) -> None: - """Present ucp-agent / signature-agent headers must be covered.""" + def test_ucp_agent_covered_signature_agent_out_of_scope(self) -> None: + """A present ucp-agent header must be covered; signature-agent is not. + + signature-agent is a WBA-shape component (component parameters / + tag=web-bot-auth) that this default-UCP verifier does not parse, so it is + deliberately outside the coverage gate: the module's promise matches what + it can actually verify. + """ required = signing.required_components( "GET", False, {"ucp-agent": "a", "signature-agent": "b"}, False ) self.assertIn("ucp-agent", required) - self.assertIn("signature-agent", required) + self.assertNotIn("signature-agent", required) def test_alg_param_rejected_by_verify_request(self) -> None: """A signature carrying an alg parameter is rejected (spec MUST NOT).""" @@ -406,5 +429,334 @@ def test_ed25519_both_directions(self) -> None: signing.verify_raw_signature(jwk, RFC_B26_BASE, alg.sign(RFC_B26_BASE)) +class NormalizationTest(absltest.TestCase): + """Verify-side normalization must match the signer's canonical base.""" + + def _signed_get(self, url: str): + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k1") + add = signing.sign_request(key, "k1", "GET", url, {}, b"") + headers = { + "signature-input": add["Signature-Input"], + "signature": add["Signature"], + } + return jwk, headers + + def test_authority_default_port_is_stripped(self) -> None: + """The default port is removed per RFC 9421 Section 2.2.3. + + A signer that signs `host` still verifies when the server sees `host:443`. + """ + jwk, headers = self._signed_get("https://merchant.example/p") + keyid = signing.verify_request( + "GET", "merchant.example:443", "/p", "", headers, b"", [jwk] + ) + self.assertEqual(keyid, "k1") + + def test_empty_path_normalized_to_slash(self) -> None: + """An empty path is the same as `/` on both sides (RFC 9421 2.2.6).""" + jwk, headers = self._signed_get("https://merchant.example/") + keyid = signing.verify_request( + "GET", "merchant.example", "", "", headers, b"", [jwk] + ) + self.assertEqual(keyid, "k1") + + def test_field_value_ows_is_trimmed(self) -> None: + """Covered field values are OWS-trimmed per RFC 9421 Section 2.1. + + Leading or trailing whitespace on a header must not break verification. + """ + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k1") + body = b'{"x":1}' + add = signing.sign_request( + key, + "k1", + "POST", + "https://m.example/o", + {"content-type": "application/json"}, + body, + ) + headers = { + "content-type": " application/json ", # padded OWS + "content-digest": add["Content-Digest"], + "signature-input": add["Signature-Input"], + "signature": add["Signature"], + } + keyid = signing.verify_request( + "POST", "m.example", "/o", "", headers, body, [jwk] + ) + self.assertEqual(keyid, "k1") + + +class DigestMatchingTest(absltest.TestCase): + """content_digest_matches rejects every malformed Content-Digest form.""" + + def test_member_not_colon_wrapped(self) -> None: + """A sha-256 member whose value is not :base64: is rejected.""" + self.assertFalse(signing.content_digest_matches("sha-256=abc", b"x")) + + def test_bad_base64_value(self) -> None: + """A sha-256 member with undecodable base64 is rejected, not raised.""" + self.assertFalse(signing.content_digest_matches("sha-256=:@@@:", b"x")) + + def test_no_sha256_member(self) -> None: + """A digest header without a sha-256 member does not match.""" + self.assertFalse(signing.content_digest_matches("md5=:AA==:", b"x")) + + +class SfEscapeTest(absltest.TestCase): + """The structured-field splitter honours backslash escapes in strings.""" + + def test_escaped_quote_inside_string(self) -> None: + """A separator inside an escaped quoted string is not a split point.""" + parts = signing._sf_split(r'"a\"b,c" , "d"', ",") + self.assertEqual(parts, [r'"a\"b,c"', '"d"']) + + +class ParserRejectionTest(absltest.TestCase): + """Malformed Signature-Input / Signature headers return None.""" + + def test_member_without_equals(self) -> None: + """A member with no '=' is malformed.""" + self.assertIsNone(signing.parse_signature_input("sig1")) + + def test_component_not_quoted(self) -> None: + """An unquoted component token is malformed.""" + self.assertIsNone(signing.parse_signature_input("sig1=(@method)")) + + def test_signature_member_without_equals(self) -> None: + """A Signature member with no '=' is malformed.""" + self.assertIsNone(signing.parse_signature("sig1")) + + def test_signature_value_not_colon_wrapped(self) -> None: + """A Signature value that is not :base64: is malformed.""" + self.assertIsNone(signing.parse_signature("sig1=abc")) + + def test_signature_bad_base64(self) -> None: + """A Signature value with undecodable base64 is malformed.""" + self.assertIsNone(signing.parse_signature("sig1=:@@@:")) + + +class JwkErrorTest(absltest.TestCase): + """public_key_from_jwk maps malformed / unsupported keys to spec codes.""" + + def test_malformed_ec_jwk(self) -> None: + """An EC JWK missing its y coordinate is signature_invalid.""" + with self.assertRaises(signing.SignatureError) as ctx: + signing.public_key_from_jwk({"kty": "EC", "crv": "P-256", "x": "AA"}) + self.assertEqual(ctx.exception.code, "signature_invalid") + + def test_unsupported_kty(self) -> None: + """An RSA JWK is algorithm_unsupported.""" + with self.assertRaises(signing.SignatureError) as ctx: + signing.public_key_from_jwk({"kty": "RSA", "n": "AA", "e": "AQAB"}) + self.assertEqual(ctx.exception.code, "algorithm_unsupported") + + +class AuthorityHttpPortTest(absltest.TestCase): + """_authority strips the http default port as well.""" + + def test_strip_port_80(self) -> None: + """host:80 normalises to host.""" + self.assertEqual(signing._authority("Host.Example:80"), "host.example") + + +class QueryAndUnresolvedTest(absltest.TestCase): + """@query round-trips through verify; unresolvable coverage fails cleanly.""" + + def test_query_signed_and_verified(self) -> None: + """A signed request with a query string verifies (exercises @query).""" + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k1") + add = signing.sign_request( + key, "k1", "GET", "https://m.example/p?a=1", {}, b"" + ) + headers = { + "signature-input": add["Signature-Input"], + "signature": add["Signature"], + } + keyid = signing.verify_request( + "GET", "m.example", "/p", "a=1", headers, b"", [jwk] + ) + self.assertEqual(keyid, "k1") + + def test_covered_component_absent_on_verify(self) -> None: + """A signature covering a header absent at verify time is invalid.""" + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k1") + # Hand-craft a Signature-Input covering a header the verifier won't have. + raw = '("@method" "@authority" "@path" "x-custom");created=1;keyid="k1"' + + def resolve(name: str): + table = { + "@method": "GET", + "@authority": "m.example", + "@path": "/p", + "x-custom": "v", + } + return table.get(name) + + base = signing.build_signature_base( + ["@method", "@authority", "@path", "x-custom"], raw, resolve + ) + sig = signing._raw_sign(key, base) + headers = { + "signature-input": f"sig1={raw}", + "signature": "sig1=:" + base64.b64encode(sig).decode() + ":", + } + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_request("GET", "m.example", "/p", "", headers, b"", [jwk]) + self.assertEqual(ctx.exception.code, "signature_invalid") + + +class ExtractKeysTest(absltest.TestCase): + """_extract_keys tolerates a non-dict profile document.""" + + def test_non_dict_document(self) -> None: + """A non-object profile yields no keys, not an error.""" + self.assertEqual(signing._extract_keys(["not", "a", "dict"]), []) + + +class KeyCacheTest(absltest.TestCase): + """fetch_signing_keys serves a cached result on the second call.""" + + def test_second_call_is_cached(self) -> None: + """A fresh cache entry is returned without any network fetch.""" + import asyncio + + signing.clear_key_cache() + jwk = {"kty": "EC", "crv": "P-256", "kid": "k", "x": "AA", "y": "AA"} + url = "https://signer.example/.well-known/ucp" + signing._KEY_CACHE[url] = (time.time() + 300, [jwk]) + keys = asyncio.run(signing.fetch_signing_keys(url, allow_insecure=False)) + self.assertEqual(keys, [jwk]) + signing.clear_key_cache() + + +class SsrfResolveTest(absltest.TestCase): + """Profile-URL host vetting: resolve failures, hostless URLs, public pass.""" + + def test_unresolvable_host(self) -> None: + """A DNS failure on the profile host is profile_unreachable (424).""" + with self.assertRaises(signing.SignatureError) as ctx: + signing._assert_profile_url_allowed( + "https://nonexistent.invalid.example./x", allow_insecure=False + ) + self.assertEqual(ctx.exception.code, "profile_unreachable") + + def test_hostless_url_rejected(self) -> None: + """A URL with no host is invalid_profile_url.""" + with self.assertRaises(signing.SignatureError) as ctx: + signing._assert_profile_url_allowed("https:///x", allow_insecure=False) + self.assertEqual(ctx.exception.code, "invalid_profile_url") + + def test_public_host_allowed(self) -> None: + """A host resolving to a public address passes the SSRF guard.""" + import socket + from unittest import mock + + infos = [(socket.AF_INET, None, None, "", ("93.184.216.34", 443))] + with mock.patch.object(socket, "getaddrinfo", return_value=infos): + signing._assert_profile_url_allowed( + "https://public.example/.well-known/ucp", allow_insecure=False + ) + + +class BodyWithoutDigestTest(absltest.TestCase): + """A bodied request whose Content-Digest header is absent is rejected.""" + + def test_body_without_content_digest_rejected(self) -> None: + """Signing covers content-digest; dropping the header fails verification.""" + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k1") + body = b'{"x":1}' + add = signing.sign_request( + key, + "k1", + "POST", + "https://m.example/o", + {"content-type": "application/json"}, + body, + ) + headers = { + "content-type": "application/json", + "signature-input": add["Signature-Input"], + "signature": add["Signature"], + } # Content-Digest deliberately omitted + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_request( + "POST", "m.example", "/o", "", headers, body, [jwk] + ) + self.assertEqual(ctx.exception.code, "digest_mismatch") + + +class VerifyMissingSignatureTest(absltest.TestCase): + """verify_request handles a present-but-unusable signature header set.""" + + def test_malformed_input_is_signature_missing(self) -> None: + """A malformed Signature-Input (parses to None) yields signature_missing.""" + headers = {"signature-input": "garbage", "signature": "sig1=:AA==:"} + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_request("GET", "m.example", "/p", "", headers, b"", []) + self.assertEqual(ctx.exception.code, "signature_missing") + + def test_label_without_matching_signature_is_skipped(self) -> None: + """A label with no matching Signature member is skipped; request fails.""" + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k1") + add = signing.sign_request(key, "k1", "GET", "https://m.example/p", {}, b"") + # Relabel Signature-Input to sig2 while Signature stays sig1: no match. + headers = { + "signature-input": add["Signature-Input"].replace("sig1=", "sig2=", 1), + "signature": add["Signature"], + } + with self.assertRaises(signing.SignatureError): + signing.verify_request("GET", "m.example", "/p", "", headers, b"", [jwk]) + + +class SfSplitEdgeTest(absltest.TestCase): + """_sf_split trailing-separator and empty-tail handling.""" + + def test_trailing_separator_has_no_empty_tail(self) -> None: + """A trailing separator does not emit an empty final segment.""" + self.assertEqual(signing._sf_split("a,", ","), ["a"]) + + def test_empty_input_returns_empty(self) -> None: + """An empty string splits to no segments.""" + self.assertEqual(signing._sf_split("", ","), []) + + +class ParseEmptyTest(absltest.TestCase): + """parse_signature_input guards empty and non-string input.""" + + def test_empty_string(self) -> None: + """An empty header parses to None.""" + self.assertIsNone(signing.parse_signature_input("")) + + +class ParseParenDepthTest(absltest.TestCase): + """The component-list paren scanner handles nested and unbalanced parens.""" + + def test_nested_parens_are_balanced(self) -> None: + """Balanced inner parens drive depth >1 then back without breaking early. + + RFC 9421 component identifiers never contain '(' (they are @-derived names + or lowercase field names), so this only exercises the scanner's depth + accounting on adversarial input: it must find the true closing paren. + """ + parsed = signing.parse_signature_input('sig1=("@method" "@path");created=1') + self.assertEqual(parsed["sig1"]["components"], ["@method", "@path"]) + # An embedded '(' (not valid per RFC 9421) confuses the naive scan; it must + # degrade safely to no usable components, never crash or over-accept. + embedded = signing.parse_signature_input('sig1=("a(b" "c");created=1') + self.assertTrue(embedded is None or not embedded["sig1"]["components"]) + + def test_unclosed_paren_yields_no_components(self) -> None: + """An unbalanced '(' never returns depth 0; parsing does not crash.""" + parsed = signing.parse_signature_input('sig1=("a";created=1') + self.assertTrue(parsed is None or not parsed["sig1"]["components"]) + + if __name__ == "__main__": absltest.main() From ec53ea8c626b7f6bb44431c6c73f36c8e2455cda Mon Sep 17 00:00:00 2001 From: vishkaty Date: Mon, 13 Jul 2026 11:49:18 -0400 Subject: [PATCH 6/7] feat(python/server): align key discovery with merged ucp#566 keys[] is the canonical RFC 7517 JWK Set field (ucp#566 removed signing_keys[]), and verifiers resolve keyid only among signature-capable keys. - _extract_keys reads keys[] only; the removed signing_keys[] is not read, so the reference verifier models the merged spec exactly. - verify_request filters to signature-capable keys before the kid match: skip use:"enc" and any key_ops that omits "verify" (RFC 7517 Sections 4.2, 4.3), matching the merged verify pseudocode. Closes the two-verifier divergence noted on #566 (a use:"enc" key with the matching kid must not verify). - the demo client and integration tests now publish keys[] (not the removed signing_keys[]), so the sample teaches the canonical shape. - docstrings/README updated to keys[]. Validated against CURRENT upstream (samples main + python-sdk main): full suite green on Python 3.10 and 3.12, ucp_signing.py at 100% line+branch coverage, ruff clean, enforced-mode happy-path E2E completes. --- rest/python/client/flower_shop/signing.py | 2 +- rest/python/server/README.md | 4 +- rest/python/server/dependencies.py | 2 +- .../server/signature_integration_test.py | 2 +- rest/python/server/ucp_signing.py | 40 +++++--- rest/python/server/ucp_signing_test.py | 96 +++++++++++++++++-- 6 files changed, 121 insertions(+), 25 deletions(-) diff --git a/rest/python/client/flower_shop/signing.py b/rest/python/client/flower_shop/signing.py index a8d9c10..f9e2f44 100644 --- a/rest/python/client/flower_shop/signing.py +++ b/rest/python/client/flower_shop/signing.py @@ -147,7 +147,7 @@ class LocalProfileServer: def __init__(self, jwk: dict, version: str = "2026-01-23") -> None: """Prepare the profile document and HTTP server (not yet started).""" document = json.dumps( - {"ucp": {"version": version, "signing_keys": [jwk]}} + {"ucp": {"version": version, "keys": [jwk]}} ).encode() class _Handler(http.server.BaseHTTPRequestHandler): diff --git a/rest/python/server/README.md b/rest/python/server/README.md index f42882d..3f0b3c2 100644 --- a/rest/python/server/README.md +++ b/rest/python/server/README.md @@ -97,7 +97,7 @@ The server verifies UCP request signatures as defined in the specification's [RFC 9421](https://www.rfc-editor.org/rfc/rfc9421.html) HTTP Message Signatures with an [RFC 9530](https://www.rfc-editor.org/rfc/rfc9530.html) `Content-Digest` over the raw body. The signer's public key is discovered from the profile URL in -the `UCP-Agent` header (its `signing_keys`). `ES256` (fixed-width raw `r||s`, not +the `UCP-Agent` header (its `keys[]`). `ES256` (fixed-width raw `r||s`, not ASN.1/DER) is the baseline; `Ed25519` is also supported. Behaviour is controlled by two flags: @@ -274,7 +274,7 @@ Response: } ] }, - "signing_keys": null + "keys": null } ``` diff --git a/rest/python/server/dependencies.py b/rest/python/server/dependencies.py index fe46b7e..fdc7913 100644 --- a/rest/python/server/dependencies.py +++ b/rest/python/server/dependencies.py @@ -68,7 +68,7 @@ async def verify_signature(request: Request) -> None: """Verify an inbound request's RFC 9421 signature per the UCP spec. The signer's public keys are discovered from the ``UCP-Agent`` header's - profile URL (its ``signing_keys``). Behaviour depends on + profile URL (its ``keys[]``). Behaviour depends on ``--require_signatures``: * When set, a missing or invalid signature is rejected with the spec's diff --git a/rest/python/server/signature_integration_test.py b/rest/python/server/signature_integration_test.py index 17e11d9..35f42e7 100644 --- a/rest/python/server/signature_integration_test.py +++ b/rest/python/server/signature_integration_test.py @@ -95,7 +95,7 @@ def setUp(self) -> None: version = config.get_server_version() good = json.dumps( - {"ucp": {"version": version, "signing_keys": [agent_jwk, rsa_jwk]}} + {"ucp": {"version": version, "keys": [agent_jwk, rsa_jwk]}} ).encode() _ProfileHandler.routes = { "/profile.json": (200, good), diff --git a/rest/python/server/ucp_signing.py b/rest/python/server/ucp_signing.py index cd4ff40..ac441e2 100644 --- a/rest/python/server/ucp_signing.py +++ b/rest/python/server/ucp_signing.py @@ -21,7 +21,7 @@ * RFC 9530 `Content-Digest` over the raw body bytes (`sha-256`). * ECDSA P-256 (`ES256`) verification with fixed-width raw `r||s` signatures (RFC 9421 Section 3.3.1) -- never ASN.1/DER -- and Ed25519 (RFC 8032). -* Signer public-key discovery from the `UCP-Agent` profile's `signing_keys`. +* Signer public-key discovery from the `UCP-Agent` profile's `keys[]`. Only the encoding and canonicalization are implemented here; every cryptographic primitive comes from the `cryptography` package. The module is @@ -466,6 +466,20 @@ def _authority(host: str) -> str: return authority +def _sig_capable(jwk: dict) -> bool: + """Whether a JWK is usable for signature verification. + + Per ucp#566 (and RFC 7517 Sections 4.2, 4.3), a profile's ``keys[]`` JWK Set + may carry non-signature keys. A key is skipped for verification when it is + marked ``use:"enc"``, or its ``key_ops`` is present but does not include + ``"verify"``. A key with no ``use``/``key_ops`` is eligible. + """ + if jwk.get("use") == "enc": + return False + key_ops = jwk.get("key_ops") + return key_ops is None or "verify" in key_ops + + def verify_request( method: str, authority: str, @@ -512,7 +526,10 @@ def verify_request( ) required = required_components(method, bool(query), headers, has_body) - keys_by_kid = {k.get("kid"): k for k in keys} + # Resolve keyid only among signature-capable keys (ucp#566): a profile's + # keys[] is a general JWK Set that may carry non-signature keys, so skip + # use:"enc" and any key_ops that omits "verify" before matching on kid. + keys_by_kid = {k.get("kid"): k for k in keys if _sig_capable(k)} def resolve(name: str) -> str | None: if name == "@method": @@ -729,10 +746,8 @@ async def fetch_signing_keys( ) -> list[dict]: """Fetch and cache a signer's published signing keys from its UCP profile. - Reads ``signing_keys`` (the canonical field on the current spec) and falls - back to a top-level ``keys`` array. Open spec PR #566 will promote ``keys`` - to canonical and drop ``signing_keys``; reading both keeps this resolver - correct across that change. + Reads the profile's ``keys[]`` (the canonical RFC 7517 JWK Set field as + of ucp#566, which removed ``signing_keys[]``). Args: profile_url: The signer's profile URL from the ``UCP-Agent`` header. @@ -781,12 +796,13 @@ async def fetch_signing_keys( def _extract_keys(document: dict) -> list[dict]: - """Pull the signing-key list out of a profile document.""" + """Pull the signing keys out of a profile document. + + ``keys[]`` is the canonical RFC 7517 JWK Set field per ucp#566, which removed + the earlier ``signing_keys[]``; this reference verifier reads only ``keys[]``. + """ if not isinstance(document, dict): return [] ucp = document.get("ucp", document) - for field in ("signing_keys", "keys"): - value = ucp.get(field) if isinstance(ucp, dict) else None - if isinstance(value, list) and value: - return value - return [] + value = ucp.get("keys") if isinstance(ucp, dict) else None + return value if isinstance(value, list) and value else [] diff --git a/rest/python/server/ucp_signing_test.py b/rest/python/server/ucp_signing_test.py index 64383e5..bba0f7a 100644 --- a/rest/python/server/ucp_signing_test.py +++ b/rest/python/server/ucp_signing_test.py @@ -368,22 +368,33 @@ def factory(*args, **kwargs): finally: signing.httpx.AsyncClient = real_client - def test_reads_signing_keys(self) -> None: - """signing_keys is read from the ucp envelope.""" + def test_reads_keys_from_ucp_envelope(self) -> None: + """keys[] (canonical per ucp#566) is read from the ucp envelope.""" keys = self._fetch( - lambda req: httpx.Response( - 200, json={"ucp": {"signing_keys": [{"kid": "a"}]}} - ) + lambda req: httpx.Response(200, json={"ucp": {"keys": [{"kid": "a"}]}}) ) self.assertEqual(keys[0]["kid"], "a") - def test_keys_fallback(self) -> None: - """A top-level keys array is read when signing_keys is absent.""" + def test_reads_top_level_keys(self) -> None: + """A top-level keys[] array (no ucp wrapper) is read.""" keys = self._fetch( lambda req: httpx.Response(200, json={"keys": [{"kid": "b"}]}) ) self.assertEqual(keys[0]["kid"], "b") + def test_legacy_signing_keys_is_not_read(self) -> None: + """A profile with only the removed signing_keys[] resolves to no keys. + + The reference verifier models the merged spec (keys[] only, ucp#566). + """ + with self.assertRaises(signing.SignatureError) as ctx: + self._fetch( + lambda req: httpx.Response( + 200, json={"ucp": {"signing_keys": [{"kid": "old"}]}} + ) + ) + self.assertEqual(ctx.exception.code, "profile_malformed") + def test_redirect_is_unreachable(self) -> None: """A 3xx response is treated as unreachable (no redirects allowed).""" with self.assertRaises(signing.SignatureError) as ctx: @@ -611,12 +622,81 @@ def resolve(name: str): class ExtractKeysTest(absltest.TestCase): - """_extract_keys tolerates a non-dict profile document.""" + """_extract_keys reads keys[] (canonical per ucp#566) and tolerates junk.""" def test_non_dict_document(self) -> None: """A non-object profile yields no keys, not an error.""" self.assertEqual(signing._extract_keys(["not", "a", "dict"]), []) + def test_reads_canonical_keys(self) -> None: + """keys[] under the ucp envelope is the canonical source.""" + doc = {"ucp": {"keys": [{"kid": "k"}]}} + self.assertEqual(signing._extract_keys(doc), [{"kid": "k"}]) + + def test_removed_signing_keys_is_ignored(self) -> None: + """The removed signing_keys[] field is not read (ucp#566).""" + doc = {"ucp": {"signing_keys": [{"kid": "old"}]}} + self.assertEqual(signing._extract_keys(doc), []) + + +class SigCapableTest(absltest.TestCase): + """Signature-capable key filtering (ucp#566). + + The verifier resolves keyid only among keys usable for verification, skipping + use:enc / key_ops-without-verify per RFC 7517 Sections 4.2 and 4.3. + """ + + def _signed(self, jwk_extra: dict): + """Sign a GET with a fresh key; return (published_jwk, headers).""" + key = ec.generate_private_key(ec.SECP256R1()) + jwk = signing.jwk_from_public_key(key.public_key(), "k1") + jwk = {**jwk, **jwk_extra} + add = signing.sign_request(key, "k1", "GET", "https://m.example/p", {}, b"") + headers = { + "signature-input": add["Signature-Input"], + "signature": add["Signature"], + } + return jwk, headers + + def test_use_sig_key_verifies(self) -> None: + """A key marked use:"sig" is signature-capable and verifies.""" + jwk, headers = self._signed({"use": "sig"}) + self.assertEqual( + signing.verify_request("GET", "m.example", "/p", "", headers, b"", [jwk]), + "k1", + ) + + def test_use_absent_key_verifies(self) -> None: + """A key with no `use` member is signature-capable (use is OPTIONAL).""" + jwk, headers = self._signed({}) + jwk.pop("use", None) + self.assertEqual( + signing.verify_request("GET", "m.example", "/p", "", headers, b"", [jwk]), + "k1", + ) + + def test_use_enc_key_is_skipped(self) -> None: + """A use:"enc" key with the matching kid is not used; key_not_found.""" + jwk, headers = self._signed({"use": "enc"}) + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_request("GET", "m.example", "/p", "", headers, b"", [jwk]) + self.assertEqual(ctx.exception.code, "key_not_found") + + def test_key_ops_without_verify_is_skipped(self) -> None: + """A key whose key_ops is present but omits "verify" is skipped.""" + jwk, headers = self._signed({"key_ops": ["encrypt", "decrypt"]}) + with self.assertRaises(signing.SignatureError) as ctx: + signing.verify_request("GET", "m.example", "/p", "", headers, b"", [jwk]) + self.assertEqual(ctx.exception.code, "key_not_found") + + def test_key_ops_with_verify_is_capable(self) -> None: + """A key whose key_ops includes "verify" is signature-capable.""" + jwk, headers = self._signed({"key_ops": ["verify"]}) + self.assertEqual( + signing.verify_request("GET", "m.example", "/p", "", headers, b"", [jwk]), + "k1", + ) + class KeyCacheTest(absltest.TestCase): """fetch_signing_keys serves a cached result on the second call.""" From 413978a608e6d416b99f9a104858e95e9b60722e Mon Sep 17 00:00:00 2001 From: Vishal Katyal Date: Fri, 17 Jul 2026 18:12:31 -0400 Subject: [PATCH 7/7] feat(python/server): verify signatures on the MCP transport route too Class sweep across every inbound route: /mcp was the one entry point without the verify_signature dependency, so unsigned POSTs succeeded even under --require_signatures. Discovery-only today (tools/call is rejected), but bridging tools/call later would have silently opened an unsigned transport into the checkout handlers. Same policy as every other route: verify-if-present by default, so unsigned MCP discovery keeps working in the default mode (covered by a new permissive-mode test); required under --require_signatures (covered by a new enforced-mode test mirroring the webhook one). --- rest/python/server/routes/mcp.py | 9 ++++++- .../server/signature_integration_test.py | 25 ++++++++++++++++++- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/rest/python/server/routes/mcp.py b/rest/python/server/routes/mcp.py index ed29a4e..9c9c985 100644 --- a/rest/python/server/routes/mcp.py +++ b/rest/python/server/routes/mcp.py @@ -27,7 +27,9 @@ from typing import Any import config +import dependencies from fastapi import APIRouter +from fastapi import Depends from fastapi import Request from fastapi import Response from fastapi.responses import JSONResponse @@ -110,12 +112,17 @@ def _error(request_id: Any, code: int, message: str) -> JSONResponse: ) -@router.post("/mcp") +@router.post("/mcp", dependencies=[Depends(dependencies.verify_signature)]) async def mcp_endpoint(request: Request) -> Response: """Minimal MCP (JSON-RPC 2.0) endpoint for capability discovery. Answers ``initialize`` and ``tools/list`` so MCP clients can complete discovery. ``tools/call`` is not yet bridged to the REST checkout handlers. + + Signature verification applies the same policy as every other inbound + route (verify-if-present by default, required under + ``--require_signatures``) so that bridging ``tools/call`` later cannot + silently open an unsigned transport into the checkout handlers. """ try: payload = await request.json() diff --git a/rest/python/server/signature_integration_test.py b/rest/python/server/signature_integration_test.py index 35f42e7..6873af0 100644 --- a/rest/python/server/signature_integration_test.py +++ b/rest/python/server/signature_integration_test.py @@ -72,7 +72,9 @@ class _SigTestBase(IntegrationTest): # Suppress every test inherited from IntegrationTest — not a hand-kept name # list, which would silently re-enable any test added to the base file later. - for _inherited in [_n for _n in dir(IntegrationTest) if _n.startswith("test_")]: + for _inherited in [ + _n for _n in dir(IntegrationTest) if _n.startswith("test_") + ]: locals()[_inherited] = None del _inherited @@ -162,6 +164,17 @@ class PermissiveModeTest(_SigTestBase): require_signatures = False + def test_mcp_unsigned_allowed(self) -> None: + """MCP discovery keeps working unsigned in the default mode.""" + with self.client: + response = self.client.post( + "/mcp", + headers={"UCP-Agent": f'profile="{self.profile_url}"'}, + json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"}, + ) + self.assertEqual(response.status_code, 200) + self.assertIn("result", response.json()) + def test_official_conformance_replay_no_fetch(self) -> None: """Legacy headers with no Signature-Input succeed without a key fetch.""" calls = [] @@ -441,6 +454,16 @@ def test_webhook_unsigned_rejected(self) -> None: ) self._assert_error(response, 401, "signature_missing") + def test_mcp_unsigned_rejected(self) -> None: + """The MCP transport endpoint also enforces signatures when required.""" + with self.client: + response = self.client.post( + "/mcp", + headers={"UCP-Agent": f'profile="{self.profile_url}"'}, + json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"}, + ) + self._assert_error(response, 401, "signature_missing") + def _resolver(body: bytes, headers: dict, port: int): """Build a signature-base component resolver for the DER-tamper test."""