diff --git a/src/adcp/governance.py b/src/adcp/governance.py index 22cc05fb01..aae5bad3f7 100644 --- a/src/adcp/governance.py +++ b/src/adcp/governance.py @@ -36,6 +36,7 @@ from adcp.signing._strict_json import parse_strict_json from adcp.signing.brand_jwks import BrandJsonResolverError, _pick_agent from adcp.signing.canonical import canonicalize_target_uri +from adcp.signing.constants import ADCP_USE_GOVERNANCE from adcp.signing.crypto import b64url_decode, b64url_encode, import_jwk from adcp.signing.etld import host_from from adcp.signing.jwks import AsyncCachingJwksResolver, AsyncJwksResolver, JwksResolver @@ -1210,7 +1211,7 @@ def reject( if ( not isinstance(jwk, Mapping) or jwk.get("kid") != kid - or jwk.get("adcp_use") != "governance-signing" + or jwk.get("adcp_use") != ADCP_USE_GOVERNANCE or jwk.get("use") != "sig" or not isinstance(jwk.get("key_ops"), list) or "verify" not in jwk["key_ops"] diff --git a/src/adcp/signing/__init__.py b/src/adcp/signing/__init__.py index 7e0f37d0fe..5d01a6dc3c 100644 --- a/src/adcp/signing/__init__.py +++ b/src/adcp/signing/__init__.py @@ -172,6 +172,10 @@ signing_operation, ) from adcp.signing.constants import ( + ADCP_USE_GOVERNANCE, + ADCP_USE_REQUEST, + ADCP_USE_RESPONSE, + ADCP_USE_WEBHOOK, DEFAULT_EXPIRES_IN_SECONDS, DEFAULT_SKEW_SECONDS, DEFAULT_TAG, @@ -444,6 +448,10 @@ def __init__(self, *args: object, **kwargs: object) -> None: "SignerKeyResolver", "WbaSignerKeyResolver", "StaticSignerKeys", + "ADCP_USE_GOVERNANCE", + "ADCP_USE_REQUEST", + "ADCP_USE_RESPONSE", + "ADCP_USE_WEBHOOK", "ALG_ED25519", "ALG_ES256", "ALLOWED_ALGS", diff --git a/src/adcp/signing/constants.py b/src/adcp/signing/constants.py index 6feddf4223..92aaf310ec 100644 --- a/src/adcp/signing/constants.py +++ b/src/adcp/signing/constants.py @@ -5,7 +5,12 @@ DEFAULT_TAG = "adcp/request-signing/v1" WEBHOOK_TAG = "adcp/webhook-signing/v1" ADCP_USE_REQUEST = "request-signing" +#: Deprecated upstream (adcp#5555): new webhook signers publish ``request-signing``. ADCP_USE_WEBHOOK = "webhook-signing" +#: Designated-task response JWS (``core/response-payload-jws-envelope.json``). +ADCP_USE_RESPONSE = "response-signing" +#: Governance attestation JWS (``check_governance`` ``governance_context``). +ADCP_USE_GOVERNANCE = "governance-signing" #: Current request keys and deprecated webhook keys are both valid for webhooks. WEBHOOK_ACCEPTED_ADCP_USES: frozenset[str] = frozenset({ADCP_USE_REQUEST, ADCP_USE_WEBHOOK}) MAX_WINDOW_SECONDS = 300 @@ -15,7 +20,9 @@ SIG_LABEL_DEFAULT = "sig1" __all__ = [ + "ADCP_USE_GOVERNANCE", "ADCP_USE_REQUEST", + "ADCP_USE_RESPONSE", "ADCP_USE_WEBHOOK", "DEFAULT_EXPIRES_IN_SECONDS", "DEFAULT_SKEW_SECONDS", diff --git a/src/adcp/signing/keygen.py b/src/adcp/signing/keygen.py index 76fb49b47d..e9b3348a1b 100644 --- a/src/adcp/signing/keygen.py +++ b/src/adcp/signing/keygen.py @@ -10,9 +10,10 @@ ergonomic. Both paths produce the same PEM and JWK. Publish the JWK (public half) -at your agent's ``jwks_uri`` — the ``adcp_use`` claim (``request-signing`` -or ``webhook-signing``) pins the key to one signing profile; verifiers -reject keys used in the wrong surface. +at your agent's ``jwks_uri`` — the ``adcp_use`` claim pins the key to one +signing purpose (``request-signing``, ``response-signing``, +``governance-signing``, the deprecated ``webhook-signing``, or any other +lowercase hyphenated token); verifiers reject keys used in the wrong surface. Usage (CLI): adcp-keygen --alg ed25519 --out private-key.pem @@ -43,6 +44,7 @@ import getpass import json import os +import re import secrets import sys import unicodedata @@ -54,6 +56,12 @@ from cryptography.hazmat.primitives.asymmetric import ec, ed25519 from joserfc.jwk import ECKey, OKPKey +from adcp.signing.constants import ( + ADCP_USE_GOVERNANCE, + ADCP_USE_REQUEST, + ADCP_USE_RESPONSE, + ADCP_USE_WEBHOOK, +) from adcp.signing.crypto import ALG_ED25519, ALG_ES256, load_private_key_pem # Key generation, PKCS#8 PEM export and RFC 7517 / 8037 public JWK export are @@ -61,7 +69,11 @@ # the JWS ``alg`` each curve is published under, and the AdCP-specific JWK # members (``use``, ``key_ops``, ``adcp_use``, ``kid``). -_ADCP_USE_VALUES = ("request-signing", "webhook-signing") +#: Purposes the spec names today, for help text. The accepted vocabulary is +#: open: any token matching ``_ADCP_USE_PATTERN`` is a valid ``adcp_use``. +_KNOWN_ADCP_USES = (ADCP_USE_REQUEST, ADCP_USE_RESPONSE, ADCP_USE_GOVERNANCE, ADCP_USE_WEBHOOK) +_ADCP_USE_PATTERN = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*") +_ADCP_USE_MAX_LENGTH = 64 # JWS ``alg`` (RFC 7518 / RFC 8037) published for each AdCP-allowed curve. Not # the RFC 9421 ``alg`` used in ``Signature-Input``. @@ -88,6 +100,25 @@ def _adcp_public_jwk(key: OKPKey | ECKey, *, kid: str, adcp_use: str) -> dict[st } +def _check_purpose(purpose: object) -> str: + """``purpose`` if it is a lowercase hyphenated token, else ``ValueError``. + + Validation is syntactic, not by membership: an AdCP verifier accepts a + key only when ``adcp_use`` equals the purpose of the surface it verifies + (fail-closed), so a purpose no verifier implements authorizes nothing. + """ + if ( + not isinstance(purpose, str) + or len(purpose) > _ADCP_USE_MAX_LENGTH + or _ADCP_USE_PATTERN.fullmatch(purpose) is None + ): + raise ValueError( + "purpose must be a lowercase hyphenated token of at most " + f"{_ADCP_USE_MAX_LENGTH} characters (e.g. {ADCP_USE_REQUEST!r}), got {purpose!r}" + ) + return purpose + + def _private_pem(key: OKPKey | ECKey, passphrase: bytes | None) -> bytes: # joserfc serializes as PKCS#8, using BestAvailableEncryption when a # password is given. Its ``password`` is annotated ``str``, but the value @@ -114,7 +145,7 @@ def pem_to_adcp_jwk( pem: bytes, *, kid: str, - purpose: Literal["request-signing", "webhook-signing"], + purpose: str, password: bytes | None = None, ) -> dict[str, Any]: """Derive the public JWK for an existing AdCP signing PEM. @@ -134,8 +165,8 @@ def pem_to_adcp_jwk( * ``alg`` — MUST be ``"EdDSA"`` for Ed25519, ``"ES256"`` for P-256 (NOT the RFC 9421 ``alg`` casing used in ``Signature-Input``). * ``adcp_use`` — required by AdCP #2423; verifiers reject keys - lacking it. MUST match the signing surface (``"request-signing"`` - vs. ``"webhook-signing"``). + lacking it. MUST match the signing surface (``"request-signing"``, + ``"response-signing"``, ``"governance-signing"``, ...). * ``key_ops`` — MUST be ``["verify"]`` (the public half cannot sign). @@ -146,22 +177,20 @@ def pem_to_adcp_jwk( the loader handles both forms. :param kid: JWK ``kid`` to embed. MUST match the value the signer will advertise via :meth:`SigningProvider.key_id`. - :param purpose: Which AdCP signing profile this key is for. Sets - ``adcp_use``. Generate distinct keys per purpose — sharing - material across request-signing and webhook-signing is a spec - violation, not just a convention. + :param purpose: Which AdCP signing purpose this key is for. Sets + ``adcp_use``. Any lowercase hyphenated token is accepted; see + :func:`generate_signing_keypair` for the vocabulary and why an + open one is safe. Generate distinct keys per purpose. :param password: Passphrase if ``pem`` is an encrypted private key. :returns: Public JWK ready to publish in the agent's ``jwks_uri``. The private scalar (``d``) is NEVER included in the output. - :raises ValueError: ``purpose`` is not in - ``("request-signing", "webhook-signing")``; the PEM is not - Ed25519 or ECDSA-P-256; the EC curve is not P-256. + :raises ValueError: ``purpose`` is not a lowercase hyphenated token; + the PEM is not Ed25519 or ECDSA-P-256; the EC curve is not P-256. """ - if purpose not in _ADCP_USE_VALUES: - raise ValueError(f"purpose must be one of {_ADCP_USE_VALUES}, got {purpose!r}") + purpose = _check_purpose(purpose) if not kid: raise ValueError("kid must be a non-empty string") @@ -215,7 +244,7 @@ def generate_signing_keypair( *, alg: Literal["ed25519", "es256"] = "ed25519", kid: str | None = None, - purpose: Literal["request-signing", "webhook-signing"] = "request-signing", + purpose: str = ADCP_USE_REQUEST, passphrase: bytes | None = None, ) -> tuple[bytes, dict[str, Any]]: """Generate an AdCP signing keypair. @@ -234,11 +263,20 @@ def generate_signing_keypair( collision-resistant within a single process but does not guarantee uniqueness across processes; rotation tooling needs its own identifier scheme to track retirement / revocation. - :param purpose: Which AdCP signing profile this key is for. Sets the - JWK ``adcp_use`` claim. **Request-signing and webhook-signing - keys MUST be distinct** — a signature from one surface cannot - replay as the other, and every conformant verifier enforces the - claim. Generate separate keys per purpose. + :param purpose: Which AdCP signing purpose this key is for. Sets the + JWK ``adcp_use`` claim. The spec names ``"request-signing"`` (RFC + 9421 requests, and webhooks from new signers), + ``"response-signing"`` (designated-task response JWS), + ``"governance-signing"`` and the deprecated ``"webhook-signing"``; + the constants are in :mod:`adcp.signing.constants` + (``ADCP_USE_REQUEST``, ...). Any other lowercase hyphenated token + is accepted for a surface the vocabulary does not name yet. + An open vocabulary is safe because purpose matching is + fail-closed: a verifier accepts a key only when ``adcp_use`` + equals the purpose of the surface it verifies (webhook verifiers + accept ``request-signing`` or ``webhook-signing``), so a key + published under any other purpose verifies nothing there. + Generate separate keys per purpose. :param passphrase: When provided, the PEM is encrypted with ``BestAvailableEncryption``. Typical only for dev-laptop keys; automated deployments usually leave the PEM unencrypted and @@ -256,7 +294,8 @@ def generate_signing_keypair( publish at your ``jwks_uri``. The private scalar is NOT in the JWK — only in the PEM. - :raises ValueError: ``alg`` or ``purpose`` is unsupported. + :raises ValueError: ``alg`` is unsupported, or ``purpose`` is not a + lowercase hyphenated token. Example: >>> pem, jwk = generate_signing_keypair( @@ -267,8 +306,7 @@ def generate_signing_keypair( """ if alg not in ("ed25519", "es256"): raise ValueError(f"alg must be 'ed25519' or 'es256', got {alg!r}") - if purpose not in _ADCP_USE_VALUES: - raise ValueError(f"purpose must be one of {_ADCP_USE_VALUES}, got {purpose!r}") + purpose = _check_purpose(purpose) resolved_kid = kid or _default_kid(alg) if alg == "ed25519": return generate_ed25519(resolved_kid, passphrase=passphrase, adcp_use=purpose) @@ -291,10 +329,17 @@ def _prompt_passphrase_bytes() -> bytes: return unicodedata.normalize("NFC", first).encode("utf-8") +def _purpose_arg(value: str) -> str: + try: + return _check_purpose(value) + except ValueError as exc: + raise argparse.ArgumentTypeError(str(exc)) from None + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="adcp-keygen", - description="Generate a signing keypair for the AdCP request-signing profile.", + description="Generate an AdCP signing keypair (PEM private key + public JWK).", ) parser.add_argument( "--alg", @@ -328,13 +373,17 @@ def main(argv: list[str] | None = None) -> int: ) parser.add_argument( "--purpose", - choices=list(_ADCP_USE_VALUES), - default="request-signing", + type=_purpose_arg, + default=ADCP_USE_REQUEST, + metavar="PURPOSE", help=( - "Which AdCP signing profile this key is for (sets the JWK `adcp_use` " - "claim). `request-signing` for outbound tool calls; `webhook-signing` " - "for keys a sender uses to sign outbound webhooks to buyers. Verifiers " - "enforce the claim, so mixing the two silently fails at first delivery." + "Which AdCP signing purpose this key is for (sets the JWK `adcp_use` " + f"claim): {', '.join(_KNOWN_ADCP_USES)}, or another lowercase " + "hyphenated token. `request-signing` (default) signs outbound tool " + "calls and, for new signers, webhooks; `response-signing` signs " + "designated-task response envelopes; `webhook-signing` is deprecated. " + "Verifiers enforce the claim, so a key under the wrong purpose fails " + "at first use." ), ) args = parser.parse_args(argv) diff --git a/tests/conformance/signing/test_keygen_programmatic.py b/tests/conformance/signing/test_keygen_programmatic.py index 77d6344458..3e4bbfda4b 100644 --- a/tests/conformance/signing/test_keygen_programmatic.py +++ b/tests/conformance/signing/test_keygen_programmatic.py @@ -18,7 +18,15 @@ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ec, ed25519 -from adcp.signing import generate_signing_keypair, public_key_from_jwk +from adcp.signing import ( + ADCP_USE_GOVERNANCE, + ADCP_USE_REQUEST, + ADCP_USE_RESPONSE, + ADCP_USE_WEBHOOK, + generate_signing_keypair, + public_key_from_jwk, +) +from adcp.signing.constants import WEBHOOK_ACCEPTED_ADCP_USES # --------------------------------------------------------------------------- # Core contract — returns (pem_bytes, public_jwk) @@ -88,15 +96,81 @@ def test_webhook_signing_purpose_sets_adcp_use() -> None: assert jwk["adcp_use"] == "webhook-signing" -def test_invalid_purpose_raises_value_error() -> None: - """Typos in ``purpose`` should fail loud — a silent fallback would - silently mint a key with an unexpected ``adcp_use``.""" - with pytest.raises(ValueError, match="purpose must be one of"): - generate_signing_keypair( - alg="ed25519", - kid="test-kid", - purpose="requets-signing", # type: ignore[arg-type] # typo - ) +@pytest.mark.parametrize( + "purpose", + [ADCP_USE_RESPONSE, ADCP_USE_GOVERNANCE, "oauth-access-token-signing", "x1"], +) +def test_purpose_vocabulary_is_open(purpose: str) -> None: + """``response-signing`` and purposes the spec does not name yet are + expressible (#1677); only the token's syntax is checked.""" + _, jwk = generate_signing_keypair(alg="es256", kid="test-kid", purpose=purpose) + assert jwk["adcp_use"] == purpose + + +@pytest.mark.parametrize( + "purpose", + [ + "", + "Request-Signing", + "request_signing", + "request signing", + "-request", + "request-", + "request--signing", + "request-signing\n", + "a" * 65, + None, + b"request-signing", + ], +) +def test_malformed_purpose_raises_value_error(purpose: object) -> None: + """The vocabulary is open but the input is still validated: a value + that is not a lowercase hyphenated token fails loud rather than mint + a key with an unexpected ``adcp_use``.""" + with pytest.raises(ValueError, match="purpose must be a lowercase hyphenated token"): + generate_signing_keypair(alg="ed25519", kid="test-kid", purpose=purpose) # type: ignore[arg-type] + + +def test_unregistered_purpose_fails_closed_at_the_request_verifier() -> None: + """An open vocabulary is safe because purpose matching is fail-closed: + a key minted under any purpose other than the verified surface's is + refused with ``request_signature_key_purpose_invalid``.""" + from adcp.signing.errors import ( + REQUEST_SIGNATURE_KEY_PURPOSE_INVALID, + SignatureVerificationError, + ) + from adcp.signing.verifier import _check_key_purpose + + for purpose in (ADCP_USE_RESPONSE, ADCP_USE_GOVERNANCE, "oauth-access-token-signing"): + _, jwk = generate_signing_keypair(alg="ed25519", kid="k", purpose=purpose) + with pytest.raises(SignatureVerificationError) as excinfo: + _check_key_purpose(jwk, "ed25519", expected_adcp_use=ADCP_USE_REQUEST) + assert excinfo.value.code == REQUEST_SIGNATURE_KEY_PURPOSE_INVALID + with pytest.raises(SignatureVerificationError): + _check_key_purpose( + jwk, + "ed25519", + expected_adcp_use=ADCP_USE_WEBHOOK, + accepted_adcp_uses=WEBHOOK_ACCEPTED_ADCP_USES, + ) + + +def test_cli_accepts_open_purpose_and_rejects_malformed(tmp_path, capsys) -> None: + import json + + from adcp.signing import keygen + + out = tmp_path / "response.pem" + rc = keygen.main(["--out", str(out), "--kid", "resp", "--purpose", "response-signing"]) + assert rc == 0 + printed = json.loads(capsys.readouterr().out) + assert printed["keys"][0]["adcp_use"] == "response-signing" + + with pytest.raises(SystemExit) as excinfo: + keygen.main(["--out", str(tmp_path / "bad.pem"), "--purpose", "Response_Signing"]) + assert excinfo.value.code == 2 + assert "lowercase hyphenated token" in capsys.readouterr().err + assert not (tmp_path / "bad.pem").exists() def test_invalid_alg_raises_value_error() -> None: diff --git a/tests/conformance/signing/test_signing_provider.py b/tests/conformance/signing/test_signing_provider.py index ff709a855e..c89b8559a3 100644 --- a/tests/conformance/signing/test_signing_provider.py +++ b/tests/conformance/signing/test_signing_provider.py @@ -303,8 +303,14 @@ def test_pem_to_adcp_jwk_accepts_spki_public_pem() -> None: def test_pem_to_adcp_jwk_rejects_invalid_purpose() -> None: pem, _ = generate_signing_keypair(alg="ed25519", kid="x", purpose="request-signing") - with pytest.raises(ValueError, match="purpose must be one of"): - pem_to_adcp_jwk(pem, kid="x", purpose="some-other-purpose") # type: ignore[arg-type] + with pytest.raises(ValueError, match="purpose must be a lowercase hyphenated token"): + pem_to_adcp_jwk(pem, kid="x", purpose="Request Signing") + + +def test_pem_to_adcp_jwk_accepts_response_signing() -> None: + pem, minted = generate_signing_keypair(alg="es256", kid="x", purpose="response-signing") + derived = pem_to_adcp_jwk(pem, kid="x", purpose="response-signing") + assert derived == minted def test_pem_to_adcp_jwk_rejects_empty_kid() -> None: diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 8ac65a79a0..1dabac09d4 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -938,6 +938,10 @@ }, "adcp.signing": { "resolves": { + "ADCP_USE_GOVERNANCE": "value:'governance-signing'", + "ADCP_USE_REQUEST": "value:'request-signing'", + "ADCP_USE_RESPONSE": "value:'response-signing'", + "ADCP_USE_WEBHOOK": "value:'webhook-signing'", "ALG_ED25519": "value:'ed25519'", "ALG_ES256": "value:'ecdsa-p256-sha256'", "ALLOWED_ALGS": "instance:builtins:frozenset",