diff --git a/agentscore_commerce/identity/__init__.py b/agentscore_commerce/identity/__init__.py index 9a4f7b6..e2e4c85 100644 --- a/agentscore_commerce/identity/__init__.py +++ b/agentscore_commerce/identity/__init__.py @@ -64,9 +64,11 @@ ) from agentscore_commerce.identity.ucp_jwks import ( GeneratedUCPKey, + LoadUCPSigningKeyOptions, UCPVerificationError, build_jwks_response, generate_ucp_signing_key, + load_ucp_signing_key_from_env, sign_ucp_profile, verify_ucp_profile, ) @@ -116,6 +118,7 @@ def _load_asgi_middleware() -> tuple[Any, Any]: "GateResult", "GeneratedUCPKey", "IdentityStatus", + "LoadUCPSigningKeyOptions", "OperatorVerification", "PolicyBlock", "SignerSanctions", @@ -141,6 +144,7 @@ def _load_asgi_middleware() -> tuple[Any, Any]: "generate_ucp_signing_key", "hash_operator_token", "is_fixable_denial", + "load_ucp_signing_key_from_env", "mpp_payment_handler", "run_gate_with_enforcement", "shipping_country_allowed", diff --git a/agentscore_commerce/identity/ucp_jwks.py b/agentscore_commerce/identity/ucp_jwks.py index 42d6a1a..dbac555 100644 --- a/agentscore_commerce/identity/ucp_jwks.py +++ b/agentscore_commerce/identity/ucp_jwks.py @@ -27,6 +27,9 @@ import contextlib import hmac import json +import logging +import os +import threading import warnings from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Literal, cast @@ -532,11 +535,202 @@ async def jwks(): return {"keys": keys} +# ── env-driven loader (extracted from store + martin + signed_ucp_merchant) ── + +_logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class LoadUCPSigningKeyOptions: + """Configuration for :func:`load_ucp_signing_key_from_env`. + + Env-var names are overridable so a merchant can run multiple distinct signing + keys from different env namespaces (e.g. ``PROD_UCP_JWK`` vs ``STAGING_UCP_JWK``). + ``default_kid`` and ``default_alg`` are used when the env JWK is absent or + doesn't carry its own ``kid`` / can't dictate alg via kty+crv. + """ + + env_jwk_var: str = "UCP_SIGNING_KEY_JWK_PRIVATE" + env_kid_var: str = "UCP_SIGNING_KEY_KID" + env_alg_var: str = "UCP_SIGNING_KEY_ALG" + default_kid: str = "merchant-default" + default_alg: Literal["EdDSA", "ES256"] = "EdDSA" + + +_env_loader_cache: dict[tuple[str, str, str, str, str], GeneratedUCPKey] = {} +_env_loader_lock = threading.Lock() + + +def _read_env_trimmed(name: str) -> str | None: + r"""Read ``name`` from env, strip whitespace, treat whitespace-only as unset. + + A secret-manager export piped through ``xargs`` appends ``\n``, which would otherwise + make ``UCP_SIGNING_KEY_JWK_PRIVATE`` fail ``json.loads`` with a misleading error. + """ + raw = os.environ.get(name) + if raw is None: + return None + trimmed = raw.strip() + return trimmed or None + + +def _detect_alg_from_jwk(jwk: dict[str, Any]) -> Literal["EdDSA", "ES256"] | None: + """Detect signing alg from JWK shape; returns ``None`` for unsupported kty/crv.""" + kty = jwk.get("kty") + crv = jwk.get("crv") + if kty == "OKP" and crv == "Ed25519": + return "EdDSA" + if kty == "EC" and crv == "P-256": + return "ES256" + return None + + +def _build_env_signing_key(opts: LoadUCPSigningKeyOptions) -> GeneratedUCPKey: + """Load (or generate) one signing key per env state. No locking (caller wraps).""" + kid_default = _read_env_trimmed(opts.env_kid_var) or opts.default_kid + raw_alg = (_read_env_trimmed(opts.env_alg_var) or "").upper() + # Case-insensitive env-alg comparison: secret configs commonly carry casing + # drift (``"es256"``, ``" ES256 "``, ``"eS256"``). Strict exact-match would + # silently downgrade to the default and operators would publish a JWKS + # containing the wrong key family. + alg_fallback: Literal["EdDSA", "ES256"] = "ES256" if raw_alg == "ES256" else opts.default_alg + + env_jwk = _read_env_trimmed(opts.env_jwk_var) + if env_jwk: + from joserfc.jwk import ECKey, OKPKey # type: ignore[import-not-found] + + try: + jwk_dict = json.loads(env_jwk) + except json.JSONDecodeError as exc: + msg = f"{opts.env_jwk_var} is not valid JSON: {exc}" + raise ValueError(msg) from exc + + if not isinstance(jwk_dict, dict) or not jwk_dict: + msg = f"{opts.env_jwk_var} must be a non-empty JWK object; got {type(jwk_dict).__name__}." + raise ValueError(msg) + + detected_alg = _detect_alg_from_jwk(jwk_dict) + if not detected_alg: + msg = ( + f"{opts.env_jwk_var} has unsupported kty/crv " + f"(got kty={jwk_dict.get('kty')!r} crv={jwk_dict.get('crv')!r}); " + "expected OKP+Ed25519 or EC+P-256." + ) + raise ValueError(msg) + + try: + priv = OKPKey.import_key(jwk_dict) if detected_alg == "EdDSA" else ECKey.import_key(jwk_dict) + except Exception as exc: + # Do NOT interpolate the underlying exception message; some import paths echo + # back fields of the input JWK including private key material. Surface only + # the exception class so logs never carry key bytes through stderr / CloudWatch. + msg = ( + f"{opts.env_jwk_var} has malformed key material ({type(exc).__name__}). " + "Verify the JWK is well-formed and matches the declared kty/crv. " + "Underlying details suppressed to avoid leaking key bytes." + ) + raise ValueError(msg) from exc + + # Project to canonical public fields per kty so unknown env JWK fields + # (key_ops, x5c, x5t, x5u, etc.) don't leak into the published JWKS. + raw = priv.as_dict(private=False) + if detected_alg == "EdDSA": + public_jwk: dict[str, Any] = { + "kty": raw["kty"], + "crv": raw["crv"], + "x": raw["x"], + } + else: + public_jwk = { + "kty": raw["kty"], + "crv": raw["crv"], + "x": raw["x"], + "y": raw["y"], + } + # Empty-string kid in env JWK falls through to the configured default — + # publishing `"kid": ""` would break every kid-pinning verifier. + public_jwk["kid"] = jwk_dict.get("kid") or kid_default + public_jwk["alg"] = detected_alg + public_jwk["use"] = "sig" + _logger.info( + "Loaded persistent UCP signing key kid=%s alg=%s from %s", + public_jwk["kid"], + detected_alg, + opts.env_jwk_var, + ) + return GeneratedUCPKey(private_key=priv, public_jwk=public_jwk) + + _logger.error( + "%s not set; generating ephemeral signing key. Verifier caches will break across restarts. " + "NOT SAFE FOR PRODUCTION.", + opts.env_jwk_var, + ) + return generate_ucp_signing_key(kid=kid_default, alg=alg_fallback) + + +def load_ucp_signing_key_from_env(opts: LoadUCPSigningKeyOptions | None = None) -> GeneratedUCPKey: + """Load the merchant's UCP signing key from env, with concurrent-safe caching. + + On first call (per ``opts``): reads ``opts.env_jwk_var``, parses it as a JWK, + validates kty/crv (OKP+Ed25519 or EC+P-256), and projects to a canonical + public JWK. Falls back to an ephemeral keypair when the env var is missing + or whitespace-only (dev-friendly; logs a loud warning). + + Subsequent calls with the same ``opts`` return the cached key without + re-reading env. Concurrent first-callers serialize on a lock so only one + key generation runs; the rest receive the cached result. + + Different ``opts`` values get separate cache entries: a merchant running + one signing key per env namespace (e.g. prod vs staging) does not collide. + + Env-driven precedence: + + * Embedded ``kid`` in the JWK wins over ``opts.env_kid_var`` env value; + empty-string ``kid`` in the env JWK falls through to ``opts.default_kid``. + * Structural ``kty``+``crv`` in the JWK wins over ``opts.env_alg_var`` env + value (which is only consulted in the ephemeral fallback path). + + Raises ``ValueError`` with a sanitized message for malformed env JWKs; + raw exception detail is intentionally suppressed so key bytes can never + reach logs. + """ + resolved = opts if opts is not None else LoadUCPSigningKeyOptions() + cache_key = ( + resolved.env_jwk_var, + resolved.env_kid_var, + resolved.env_alg_var, + resolved.default_kid, + resolved.default_alg, + ) + cached = _env_loader_cache.get(cache_key) + if cached is not None: + return cached + with _env_loader_lock: + cached = _env_loader_cache.get(cache_key) + if cached is not None: + return cached + result = _build_env_signing_key(resolved) + _env_loader_cache[cache_key] = result + return result + + +def _reset_ucp_signing_key_cache() -> None: + """Test-only: clear the env-loader cache. + + Use after ``monkeypatch.setenv(...)`` / ``monkeypatch.delenv(...)`` to force + the next ``load_ucp_signing_key_from_env`` call to re-read the env state. + """ + with _env_loader_lock: + _env_loader_cache.clear() + + __all__ = [ "GeneratedUCPKey", + "LoadUCPSigningKeyOptions", "UCPVerificationError", "build_jwks_response", "generate_ucp_signing_key", + "load_ucp_signing_key_from_env", "sign_ucp_profile", "verify_ucp_profile", ] diff --git a/tests/test_load_ucp_signing_key_from_env.py b/tests/test_load_ucp_signing_key_from_env.py new file mode 100644 index 0000000..2585fcf --- /dev/null +++ b/tests/test_load_ucp_signing_key_from_env.py @@ -0,0 +1,288 @@ +"""Tests for ``load_ucp_signing_key_from_env`` — env-driven UCP signing-key loader. + +Locked behavior contract (shared with the Node sibling at +``node-commerce/tests/identity/load-ucp-signing-key-from-env.test.ts``): + +* env JWK present → load + validate kty/crv (OKP+Ed25519 or EC+P-256), project to canonical public JWK +* env JWK absent → generate ephemeral key (logs loud warning) +* malformed JSON → ValueError naming the env var +* unsupported kty/crv → ValueError naming the actual kty/crv +* malformed key material → sanitized ValueError (no key bytes in the message) +* whitespace-only env value → treated as absent +* embedded kid in JWK wins over env kid; empty-string kid falls through to default +* concurrent first-callers see the same cached key (lock-protected) +* different opts get separate cache entries +""" + +from __future__ import annotations + +import json + +import pytest +from joserfc.jwk import ECKey, OKPKey + +from agentscore_commerce.identity.ucp_jwks import ( + LoadUCPSigningKeyOptions, + _reset_ucp_signing_key_cache, + load_ucp_signing_key_from_env, +) + + +@pytest.fixture(autouse=True) +def _reset_cache() -> None: + _reset_ucp_signing_key_cache() + + +def _build_ed25519_jwk() -> dict: + """Generate an Ed25519 JWK (with the private ``d`` field) for env-loading tests.""" + return OKPKey.generate_key(crv="Ed25519").as_dict(private=True) + + +def _build_p256_jwk() -> dict: + """Generate a P-256 JWK (with the private ``d`` field) for env-loading tests.""" + return ECKey.generate_key(crv="P-256").as_dict(private=True) + + +# ─── env JWK present: happy paths ──────────────────────────────────────────── + + +def test_loads_ed25519_jwk_from_env(monkeypatch) -> None: + private_jwk = _build_ed25519_jwk() + private_jwk["kid"] = "test-ed25519-key" + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", json.dumps(private_jwk)) + + result = load_ucp_signing_key_from_env() + + assert result.public_jwk["kty"] == "OKP" + assert result.public_jwk["crv"] == "Ed25519" + assert result.public_jwk["alg"] == "EdDSA" + assert result.public_jwk["use"] == "sig" + assert result.public_jwk["kid"] == "test-ed25519-key" + assert "d" not in result.public_jwk # private field stripped + + +def test_loads_es256_jwk_from_env(monkeypatch) -> None: + private_jwk = _build_p256_jwk() + private_jwk["kid"] = "test-p256-key" + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", json.dumps(private_jwk)) + + result = load_ucp_signing_key_from_env() + + assert result.public_jwk["kty"] == "EC" + assert result.public_jwk["crv"] == "P-256" + assert result.public_jwk["alg"] == "ES256" + assert result.public_jwk["kid"] == "test-p256-key" + assert "d" not in result.public_jwk + + +# ─── kid precedence ────────────────────────────────────────────────────────── + + +def test_embedded_kid_wins_over_env_kid_default(monkeypatch) -> None: + private_jwk = _build_ed25519_jwk() + private_jwk["kid"] = "embedded-kid" + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", json.dumps(private_jwk)) + monkeypatch.setenv("UCP_SIGNING_KEY_KID", "env-kid") + + result = load_ucp_signing_key_from_env() + assert result.public_jwk["kid"] == "embedded-kid" + + +def test_empty_string_embedded_kid_falls_through_to_env(monkeypatch) -> None: + """An env JWK with ``kid: ""`` would publish empty kid; helper falls back to env kid.""" + private_jwk = _build_ed25519_jwk() + private_jwk["kid"] = "" + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", json.dumps(private_jwk)) + monkeypatch.setenv("UCP_SIGNING_KEY_KID", "fallback-kid") + + result = load_ucp_signing_key_from_env() + assert result.public_jwk["kid"] == "fallback-kid" + + +def test_missing_embedded_kid_falls_through_to_default(monkeypatch) -> None: + private_jwk = _build_ed25519_jwk() + private_jwk.pop("kid", None) + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", json.dumps(private_jwk)) + monkeypatch.delenv("UCP_SIGNING_KEY_KID", raising=False) + + result = load_ucp_signing_key_from_env(LoadUCPSigningKeyOptions(default_kid="opts-default")) + assert result.public_jwk["kid"] == "opts-default" + + +# ─── canonical public JWK projection ───────────────────────────────────────── + + +def test_unknown_env_jwk_fields_dropped_from_public_jwk(monkeypatch) -> None: + """``key_ops``, ``x5c``, ``x5t``, ``x5u`` etc. on the env JWK don't leak into JWKS.""" + private_jwk = _build_ed25519_jwk() + private_jwk["kid"] = "test-kid" + private_jwk["key_ops"] = ["sign", "verify"] + private_jwk["x5c"] = ["fake-cert"] + private_jwk["x5t"] = "fake-thumbprint" + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", json.dumps(private_jwk)) + + result = load_ucp_signing_key_from_env() + + assert "key_ops" not in result.public_jwk + assert "x5c" not in result.public_jwk + assert "x5t" not in result.public_jwk + + +# ─── env JWK absent: ephemeral fallback ────────────────────────────────────── + + +def test_generates_ephemeral_key_when_env_jwk_missing(monkeypatch) -> None: + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + monkeypatch.delenv("UCP_SIGNING_KEY_KID", raising=False) + + result = load_ucp_signing_key_from_env() + + assert result.public_jwk["kty"] == "OKP" # default alg is EdDSA + assert result.public_jwk["alg"] == "EdDSA" + assert result.public_jwk["kid"] == "merchant-default" # default from LoadUCPSigningKeyOptions + + +def test_ephemeral_respects_default_alg_options(monkeypatch) -> None: + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + result = load_ucp_signing_key_from_env(LoadUCPSigningKeyOptions(default_alg="ES256")) + assert result.public_jwk["alg"] == "ES256" + assert result.public_jwk["kty"] == "EC" + + +def test_env_alg_overrides_default_in_ephemeral_path(monkeypatch) -> None: + """When env JWK is absent, env alg (case-insensitive) is honored.""" + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + monkeypatch.setenv("UCP_SIGNING_KEY_ALG", "es256") # lowercase + result = load_ucp_signing_key_from_env() + assert result.public_jwk["alg"] == "ES256" + + +def test_env_alg_case_insensitive_strict_match(monkeypatch) -> None: + """Unrecognized env alg falls back to default.""" + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + monkeypatch.setenv("UCP_SIGNING_KEY_ALG", "rs256") # not supported + result = load_ucp_signing_key_from_env() + assert result.public_jwk["alg"] == "EdDSA" # falls to default + + +# ─── whitespace handling ───────────────────────────────────────────────────── + + +def test_whitespace_only_env_treated_as_absent(monkeypatch) -> None: + """``aws secretsmanager get-secret-value | xargs`` appends ``\\n``; helper trims it.""" + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", " \n\t ") + result = load_ucp_signing_key_from_env() + # Falls through to ephemeral + assert result.public_jwk["alg"] == "EdDSA" + + +def test_env_kid_whitespace_trimmed(monkeypatch) -> None: + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + monkeypatch.setenv("UCP_SIGNING_KEY_KID", " trimmed-kid ") + result = load_ucp_signing_key_from_env() + assert result.public_jwk["kid"] == "trimmed-kid" + + +# ─── error paths ───────────────────────────────────────────────────────────── + + +def test_malformed_json_raises_naming_env_var(monkeypatch) -> None: + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", "{not valid json") + with pytest.raises(ValueError, match="UCP_SIGNING_KEY_JWK_PRIVATE is not valid JSON"): + load_ucp_signing_key_from_env() + + +def test_unsupported_kty_crv_raises_naming_actual(monkeypatch) -> None: + monkeypatch.setenv( + "UCP_SIGNING_KEY_JWK_PRIVATE", + json.dumps({"kty": "RSA", "n": "abc", "e": "AQAB"}), + ) + with pytest.raises(ValueError, match=r"unsupported kty/crv.*kty='RSA'"): + load_ucp_signing_key_from_env() + + +def test_jwk_not_an_object_raises(monkeypatch) -> None: + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", "[1, 2, 3]") + with pytest.raises(ValueError, match="must be a non-empty JWK object"): + load_ucp_signing_key_from_env() + + +def test_empty_jwk_object_raises(monkeypatch) -> None: + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", "{}") + with pytest.raises(ValueError, match="must be a non-empty JWK object"): + load_ucp_signing_key_from_env() + + +def test_malformed_key_material_sanitizes_underlying_error(monkeypatch) -> None: + """Underlying joserfc exception is replaced with a class-only message; key bytes never leak.""" + bad_jwk = { + "kty": "OKP", + "crv": "Ed25519", + "x": "this-is-not-base64-key-material", + "d": "leaked-secret-should-not-appear", + } + monkeypatch.setenv("UCP_SIGNING_KEY_JWK_PRIVATE", json.dumps(bad_jwk)) + with pytest.raises(ValueError) as exc_info: + load_ucp_signing_key_from_env() + # The secret in `d` must NOT appear in the surfaced error message + assert "leaked-secret-should-not-appear" not in str(exc_info.value) + assert "Underlying details suppressed to avoid leaking key bytes" in str(exc_info.value) + + +# ─── caching + concurrency ─────────────────────────────────────────────────── + + +def test_repeated_calls_return_cached_key(monkeypatch) -> None: + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + first = load_ucp_signing_key_from_env() + second = load_ucp_signing_key_from_env() + assert first is second + + +def test_different_opts_get_separate_cache_entries(monkeypatch) -> None: + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + first = load_ucp_signing_key_from_env(LoadUCPSigningKeyOptions(default_kid="kid-a")) + second = load_ucp_signing_key_from_env(LoadUCPSigningKeyOptions(default_kid="kid-b")) + assert first is not second + assert first.public_jwk["kid"] == "kid-a" + assert second.public_jwk["kid"] == "kid-b" + + +def test_concurrent_first_callers_share_same_key(monkeypatch) -> None: + """Lock prevents two concurrent ephemeral generations from racing.""" + import threading + + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + results: list = [] + + def call() -> None: + results.append(load_ucp_signing_key_from_env()) + + threads = [threading.Thread(target=call) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + # All 8 callers received the same key object (lock-protected single generation) + assert all(r is results[0] for r in results) + + +def test_reset_cache_clears_entries(monkeypatch) -> None: + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + first = load_ucp_signing_key_from_env() + _reset_ucp_signing_key_cache() + second = load_ucp_signing_key_from_env() + # New ephemeral key generated after reset (not the same object) + assert first is not second + + +def test_env_var_overridable_via_opts(monkeypatch) -> None: + """A merchant running multiple keys from different env namespaces sees separate state.""" + private_jwk = _build_ed25519_jwk() + private_jwk["kid"] = "prod-key" + monkeypatch.setenv("PROD_UCP_JWK", json.dumps(private_jwk)) + monkeypatch.delenv("UCP_SIGNING_KEY_JWK_PRIVATE", raising=False) + + result = load_ucp_signing_key_from_env(LoadUCPSigningKeyOptions(env_jwk_var="PROD_UCP_JWK")) + assert result.public_jwk["kid"] == "prod-key"