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/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..f9e2f44 --- /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, "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: diff --git a/rest/python/server/README.md b/rest/python/server/README.md index 1b6e7c9..3f0b3c2 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 `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 @@ -237,7 +274,7 @@ Response: } ] }, - "signing_keys": null + "keys": null } ``` 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..fdc7913 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 ``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/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/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 new file mode 100644 index 0000000..6873af0 --- /dev/null +++ b/rest/python/server/signature_integration_test.py @@ -0,0 +1,485 @@ +# 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, "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_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 = [] + 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_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: + 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_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: + 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 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.""" + 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() diff --git a/rest/python/server/ucp_signing.py b/rest/python/server/ucp_signing.py new file mode 100644 index 0000000..ac441e2 --- /dev/null +++ b/rest/python/server/ucp_signing.py @@ -0,0 +1,808 @@ +# 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 `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") + # 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 + + +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 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 _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, + 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) + # 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": + return method.upper() + if name == "@authority": + return _authority(authority) + if name == "@path": + return path or "/" # RFC 9421 Section 2.2.6: empty path is "/" + if name == "@query": + return "?" + query + 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" + ) + 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 + ): # 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}" + ) + 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 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. + 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 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) + 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 new file mode 100644 index 0000000..bba0f7a --- /dev/null +++ b/rest/python/server/ucp_signing_test.py @@ -0,0 +1,842 @@ +# 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 +import time + +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) + + 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).""" + + 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_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.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).""" + 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_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": {"keys": [{"kid": "a"}]}}) + ) + self.assertEqual(keys[0]["kid"], "a") + + 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: + 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)) + + +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 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.""" + + 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()