diff --git a/CHANGELOG.md b/CHANGELOG.md index caaf5e60..7b8e6ae6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Reject RFC 7517 / RFC 7518 / RFC 8037 private or symmetric JWK members (`d`, + `p`, `q`, `dp`, `dq`, `qi`, `oth`, `k`) and enforce canonical unpadded 32-byte + base64url public keys in `ca2a_verify.dag` and `ca2a_runtime.agent_manifest` + (#267). +- Reject lone UTF-16 surrogates (`U+D800..U+DFFF`) in `ca2a_runtime.canonical` + string values and object keys with a deterministic `ValueError` instead of + leaking `UnicodeEncodeError` (#267). +- Reject duplicate JSON object member names when loading chain and revocation + files in `ca2a_verify.verify` (#267). - Sealing to a low-order X25519 peer key raises `SealedChannelError`, consistent with other invalid channel keys, instead of leaking a backend `ValueError`. These keys were already refused; no ciphertext format or acceptance rule changes. diff --git a/src/ca2a_runtime/agent_manifest.py b/src/ca2a_runtime/agent_manifest.py index 594cfb85..9ec48935 100644 --- a/src/ca2a_runtime/agent_manifest.py +++ b/src/ca2a_runtime/agent_manifest.py @@ -14,7 +14,8 @@ from ca2a_runtime.errors import ConfigError -_B64URL_RE = re.compile(r"^[A-Za-z0-9\-_]*$") +_B64URL_RE = re.compile(r"^[A-Za-z0-9_-]+\Z") +_PRIVATE_JWK_PARAMS = frozenset({"d", "p", "q", "dp", "dq", "qi", "oth", "k"}) @dataclass(frozen=True) @@ -40,13 +41,16 @@ class AgentManifestBinding: def _b64url_decode(value: str) -> bytes: - if not _B64URL_RE.fullmatch(value): + if not isinstance(value, str) or not _B64URL_RE.fullmatch(value) or len(value) % 4 == 1: raise ConfigError("Agent Manifest signature/key must use base64url encoding") padding = (-len(value)) % 4 try: - return base64.urlsafe_b64decode(value + "=" * padding) + raw = base64.urlsafe_b64decode((value + "=" * padding).encode("ascii")) except ValueError as exc: raise ConfigError("Agent Manifest signature/key is not valid base64url") from exc + if base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") != value: + raise ConfigError("Agent Manifest signature/key must use canonical base64url encoding") + return raw def _b64url_encode(value: bytes) -> str: @@ -97,6 +101,11 @@ def load_agent_manifest_trust_anchor(path: str | Path) -> dict[str, bytes]: except (OSError, json.JSONDecodeError) as exc: raise ConfigError(f"cannot read Agent Manifest trust anchor: {exc}") from exc + if isinstance(raw, dict) and _PRIVATE_JWK_PARAMS.intersection(raw): + raise ConfigError( + "Agent Manifest trust anchor must not contain private or symmetric key parameters" + ) + items: list[Any] if isinstance(raw, dict) and "public_key_base64url" in raw: items = [raw] @@ -109,7 +118,14 @@ def load_agent_manifest_trust_anchor(path: str | Path) -> dict[str, bytes]: for item in items: if not isinstance(item, dict): raise ConfigError("Agent Manifest trust anchor keys must be objects") - public_key = _b64url_decode(str(item.get("public_key_base64url", ""))) + if _PRIVATE_JWK_PARAMS.intersection(item): + raise ConfigError( + "Agent Manifest trust anchor must not contain private or symmetric key parameters" + ) + raw_pub = item.get("public_key_base64url") + if not isinstance(raw_pub, str): + raise ConfigError("Agent Manifest trust anchor public_key_base64url must be a string") + public_key = _b64url_decode(raw_pub) if len(public_key) != 32: raise ConfigError("Agent Manifest trust anchor contains an invalid Ed25519 key") key_id = str(item.get("key_id") or _key_id(public_key)) diff --git a/src/ca2a_runtime/canonical.py b/src/ca2a_runtime/canonical.py index 956b5dd5..5659bbad 100644 --- a/src/ca2a_runtime/canonical.py +++ b/src/ca2a_runtime/canonical.py @@ -41,6 +41,10 @@ def _escape_string(s: str) -> str: out: list[str] = ['"'] for ch in s: code = ord(ch) + if 0xD800 <= code <= 0xDFFF: + raise ValueError( + f"lone UTF-16 surrogate U+{code:04X} is not valid in RFC 8785 canonical JSON" + ) if code in _SHORT_ESCAPES: out.append(_SHORT_ESCAPES[code]) elif code < 0x20: @@ -82,6 +86,12 @@ def _serialize(value: Any) -> str: for key in value: if not isinstance(key, str): raise TypeError(f"object keys must be strings, got {type(key).__name__}") + for ch in key: + code = ord(ch) + if 0xD800 <= code <= 0xDFFF: + raise ValueError( + f"lone UTF-16 surrogate U+{code:04X} is not valid in RFC 8785 canonical JSON" + ) items = sorted(value.items(), key=lambda kv: kv[0].encode("utf-16-be")) return "{" + ",".join(f"{_escape_string(k)}:{_serialize(v)}" for k, v in items) + "}" raise TypeError(f"unsupported type for canonicalization: {type(value).__name__}") diff --git a/src/ca2a_verify/dag.py b/src/ca2a_verify/dag.py index fa6a323b..16e9219c 100644 --- a/src/ca2a_verify/dag.py +++ b/src/ca2a_verify/dag.py @@ -60,6 +60,8 @@ class TraceDagResult: # The TRACE digest form (sha256:/sha384: + lowercase hex), same as the schema. _DIGEST_RE = re.compile(r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$") +_B64URL_RE = re.compile(r"^[A-Za-z0-9_-]+\Z") +_PRIVATE_JWK_PARAMS = frozenset({"d", "p", "q", "dp", "dq", "qi", "oth", "k"}) def _validate_delegation_block(block: Any, index: int) -> None: @@ -88,12 +90,25 @@ def _validate_delegation_block(block: Any, index: int) -> None: def _b64url_decode(value: str) -> bytes: + if not isinstance(value, str) or not _B64URL_RE.match(value) or len(value) % 4 == 1: + raise ValueError("value must be canonical unpadded base64url") padded = value + "=" * (-len(value) % 4) - return base64.urlsafe_b64decode(padded) + raw = base64.urlsafe_b64decode(padded.encode("ascii")) + if base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") != value: + raise ValueError("base64url value has non-canonical trailing bits") + return raw def _jwk_raw(jwk: dict[str, Any]) -> bytes: """Raw Ed25519 public-key bytes from an OKP JWK, or raise TraceRecordInvalid.""" + if not isinstance(jwk, dict): + raise TraceRecordInvalid("cnf.jwk must be a JSON object") + private_params = sorted(_PRIVATE_JWK_PARAMS.intersection(jwk)) + if private_params: + raise TraceRecordInvalid( + "cnf.jwk must not contain private or symmetric key parameters", + detail=f"forbidden parameters: {private_params}", + ) if jwk.get("kty") != "OKP" or jwk.get("crv") != "Ed25519": raise TraceRecordInvalid( "unsupported cnf.jwk key type", detail="expected an OKP/Ed25519 key" @@ -102,9 +117,15 @@ def _jwk_raw(jwk: dict[str, Any]) -> bytes: if not isinstance(x, str): raise TraceRecordInvalid("cnf.jwk is missing its 'x' parameter") try: - return _b64url_decode(x) + raw = _b64url_decode(x) except (binascii.Error, ValueError) as exc: raise TraceRecordInvalid("cnf.jwk 'x' is not valid base64url", detail=str(exc)) from exc + if len(raw) != 32: + raise TraceRecordInvalid( + "cnf.jwk 'x' must decode to 32 Ed25519 public-key bytes", + detail=f"got {len(raw)} bytes", + ) + return raw def _trusted_raw_keys(trusted_keys: Iterable[Any]) -> set[bytes]: diff --git a/src/ca2a_verify/verify.py b/src/ca2a_verify/verify.py index 853e3dff..a7b923fe 100644 --- a/src/ca2a_verify/verify.py +++ b/src/ca2a_verify/verify.py @@ -79,6 +79,19 @@ def verify_delegation_chain( ) +def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + obj: dict[str, Any] = {} + for key, value in pairs: + if key in obj: + raise ValueError(f"duplicate JSON object key: {key!r}") + obj[key] = value + return obj + + +def _loads_no_duplicate_keys(raw: str) -> Any: + return json.loads(raw, object_pairs_hook=_reject_duplicate_keys) + + def _parse_chain(data: Any) -> list[DelegationCredential]: if isinstance(data, dict) and "chain" in data: data = data["chain"] @@ -101,8 +114,8 @@ def verify_chain_file( if not p.is_file(): raise InvalidCredential(f"chain file not found: {p}") try: - data = json.loads(p.read_text(encoding="utf-8")) - except json.JSONDecodeError as exc: + data = _loads_no_duplicate_keys(p.read_text(encoding="utf-8")) + except (json.JSONDecodeError, ValueError) as exc: raise InvalidCredential(f"invalid JSON in {p}", detail=str(exc)) from exc return verify_delegation_chain( _parse_chain(data), @@ -124,7 +137,7 @@ def load_revocation_snapshot(path: str | Path) -> RevocationSnapshot: if not p.is_file(): raise InvalidRevocation(f"revocation file not found: {p}") try: - data = json.loads(p.read_text(encoding="utf-8")) - except json.JSONDecodeError as exc: + data = _loads_no_duplicate_keys(p.read_text(encoding="utf-8")) + except (json.JSONDecodeError, ValueError) as exc: raise InvalidRevocation(f"invalid JSON in {p}", detail=str(exc)) from exc return RevocationSnapshot.from_dict(data) diff --git a/tests/unit/test_agent_manifest_cose.py b/tests/unit/test_agent_manifest_cose.py index 3377f89e..4a2f5c9b 100644 --- a/tests/unit/test_agent_manifest_cose.py +++ b/tests/unit/test_agent_manifest_cose.py @@ -200,3 +200,49 @@ def test_trust_anchor_key_id_must_match_key(tmp_path: Path) -> None: path = _trust_file(tmp_path / "manifest-key.json", "0" * 64, keypair.public_bytes) with pytest.raises(ConfigError, match="key_id does not match"): load_agent_manifest_trust_anchor(path) + + +@pytest.mark.parametrize("param", ["d", "p", "q", "dp", "dq", "qi", "oth", "k"]) +def test_trust_anchor_rejects_private_or_symmetric_jwk_parameters( + tmp_path: Path, param: str +) -> None: + keypair = sdk.generate_ed25519() + pub_b64 = base64.urlsafe_b64encode(keypair.public_bytes).rstrip(b"=").decode("ascii") + single = tmp_path / f"single-{param}.json" + single.write_text( + json.dumps({"key_id": keypair.key_id, "public_key_base64url": pub_b64, param: "secret"}), + encoding="utf-8", + ) + with pytest.raises(ConfigError, match="private or symmetric"): + load_agent_manifest_trust_anchor(single) + + nested = tmp_path / f"nested-{param}.json" + nested.write_text( + json.dumps( + {"keys": [{"key_id": keypair.key_id, "public_key_base64url": pub_b64, param: "secret"}]} + ), + encoding="utf-8", + ) + with pytest.raises(ConfigError, match="private or symmetric"): + load_agent_manifest_trust_anchor(nested) + + +@pytest.mark.parametrize( + "bad_pub", + [ + "", + "A", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA+A", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA/A", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAB", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + None, + 123, + ], +) +def test_trust_anchor_rejects_non_canonical_base64url(tmp_path: Path, bad_pub: object) -> None: + path = tmp_path / "bad-anchor.json" + path.write_text(json.dumps({"public_key_base64url": bad_pub}), encoding="utf-8") + with pytest.raises(ConfigError): + load_agent_manifest_trust_anchor(path) diff --git a/tests/unit/test_canonical.py b/tests/unit/test_canonical.py index 9bef6503..a9ca7e38 100644 --- a/tests/unit/test_canonical.py +++ b/tests/unit/test_canonical.py @@ -77,3 +77,24 @@ def test_non_string_object_keys_rejected() -> None: for value in ({1: "a", "1": "b"}, {True: 1}, {None: 1}, {"outer": {2: "x"}}): with pytest.raises(TypeError, match="keys must be strings"): canonicalize(value) + + +@pytest.mark.parametrize( + "value", + [ + "\ud800", + "\udbff", + "\udc00", + "\udfff", + "prefix-\ud800-suffix", + ["\ud83d"], + {"k": "\ude00"}, + {"\ud800": 1}, + {"a": 1, "\udfff": 2}, + ], +) +def test_lone_utf16_surrogates_rejected_with_deterministic_value_error(value: object) -> None: + """RFC 8785 (I-JSON) forbids unpaired UTF-16 surrogates in strings and keys.""" + with pytest.raises(ValueError, match="lone UTF-16 surrogate") as exc_info: + canonicalize(value) + assert type(exc_info.value) is ValueError diff --git a/tests/unit/test_trace_binding.py b/tests/unit/test_trace_binding.py index 873e1aa0..4d470fb1 100644 --- a/tests/unit/test_trace_binding.py +++ b/tests/unit/test_trace_binding.py @@ -385,3 +385,52 @@ def test_non_object_record_is_invalid_not_a_crash(record: object) -> None: """ with pytest.raises(TraceRecordInvalid, match="not a JSON object"): verify_trace_dag([record], trusted_keys=[]) # type: ignore[list-item] + + +@pytest.mark.parametrize("param", ["d", "p", "q", "dp", "dq", "qi", "oth", "k"]) +def test_jwk_with_private_or_symmetric_parameter_rejected(param: str) -> None: + """Public verification JWKs must never carry private or symmetric key material.""" + chain, keys = build_chain_with_keys([frozenset({"cap:a"})]) + records = emit_dag(_hops(chain, keys)) + root = records[0] + valid_jwk = dict(root["cnf"]["jwk"]) + tainted_jwk = {**valid_jwk, param: "AAAA"} + + with pytest.raises(TraceRecordInvalid, match="private or symmetric"): + verify_trace_dag([root], trusted_keys=[tainted_jwk]) + + tainted_record = {**root, "cnf": {"jwk": tainted_jwk}} + with pytest.raises(TraceRecordInvalid, match="private or symmetric"): + cross_check_trace_dag([tainted_record], chain) + with pytest.raises(TraceRecordInvalid): + verify_trace_dag([tainted_record], trusted_keys=[valid_jwk]) + + +@pytest.mark.parametrize( + "bad_x", + [ + "", + "A", + "AAAA=", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA+A", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA/A", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA ", + # 43 chars where the final 4-bit group has non-zero trailing bits ('B' = 1). + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAB", + # Valid canonical base64url of 31 bytes (42 chars) instead of 32 bytes. + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + ], +) +def test_non_canonical_or_wrong_length_jwk_x_rejected(bad_x: str) -> None: + """cnf.jwk 'x' and trusted JWK dicts must be canonical unpadded 32-byte base64url.""" + key = generate_key() + root = sign_trace_record( + build_trace_record( + subject="spiffe://ca2a.example/root", iat=_NOW, context=_software_context() + ), + key, + ) + bad_jwk = {"kty": "OKP", "crv": "Ed25519", "x": bad_x} + with pytest.raises(TraceRecordInvalid): + verify_trace_dag([root], trusted_keys=[bad_jwk]) diff --git a/tests/unit/test_verify.py b/tests/unit/test_verify.py index bdd95a0c..7d06880d 100644 --- a/tests/unit/test_verify.py +++ b/tests/unit/test_verify.py @@ -9,8 +9,13 @@ from ca2a_runtime.cli import main as cli_main from ca2a_runtime.delegation import DelegationCredential -from ca2a_runtime.errors import CA2AError, InvalidCredential, ScopeEscalation -from ca2a_verify import ChainResult, verify_chain_file, verify_delegation_chain +from ca2a_runtime.errors import CA2AError, InvalidCredential, InvalidRevocation, ScopeEscalation +from ca2a_verify import ( + ChainResult, + load_revocation_snapshot, + verify_chain_file, + verify_delegation_chain, +) from tests.unit.conftest import build_chain @@ -137,3 +142,34 @@ def test_verify_dag_cli_requires_a_trusted_root_when_cross_checking_chain( ) assert exc.value.code == 2 + + +def test_verify_chain_file_rejects_duplicate_json_keys( + tmp_path: Path, valid_chain: list[DelegationCredential] +) -> None: + dumped = _dump(valid_chain) + hop_json = json.dumps(dumped[0]) + # Inject a duplicate credential_id key before the legitimate one. + dup_hop = '{"credential_id": "shadowed", ' + hop_json[1:] + p = tmp_path / "dup-hop.json" + p.write_text(f"[{dup_hop}]", encoding="utf-8") + with pytest.raises(InvalidCredential, match="invalid JSON") as exc_info: + verify_chain_file(p, trusted_root_issuers={valid_chain[0].issuer}) + assert "duplicate JSON object key" in (exc_info.value.detail or "") + + p_top = tmp_path / "dup-top.json" + p_top.write_text( + f'{{"chain": [], "chain": {json.dumps(dumped)}}}', + encoding="utf-8", + ) + with pytest.raises(InvalidCredential, match="invalid JSON") as exc_top: + verify_chain_file(p_top, trusted_root_issuers={valid_chain[0].issuer}) + assert "duplicate JSON object key" in (exc_top.value.detail or "") + + +def test_load_revocation_snapshot_rejects_duplicate_json_keys(tmp_path: Path) -> None: + p = tmp_path / "dup-revocations.json" + p.write_text('{"as_of": 100, "as_of": 200, "revocations": []}', encoding="utf-8") + with pytest.raises(InvalidRevocation, match="invalid JSON") as exc_info: + load_revocation_snapshot(p) + assert "duplicate JSON object key" in (exc_info.value.detail or "")