From f231b13c8afb5f98dacdbac71e1dc0a3c9046778 Mon Sep 17 00:00:00 2001 From: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> Date: Sat, 31 Oct 2026 10:07:13 -0500 Subject: [PATCH] fix(transport): enforce canonical base64url in b64url_decode and reject non-string fields in parse_channel_offer (#271) - Require len(value) % 4 != 1 and exact round-trip b64url_encode(raw) == value in b64url_decode. - Require non-empty str values for channel_public_key, platform, measurement, public_key, and nonce and a dict attestation object in parse_channel_offer. Closes #271 Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/ca2a_runtime/transport/_b64url.py | 12 +++++-- src/ca2a_runtime/transport/wire.py | 29 +++++++++++---- tests/unit/test_a2a_adapter.py | 52 +++++++++++++++++++++++++++ 3 files changed, 84 insertions(+), 9 deletions(-) diff --git a/src/ca2a_runtime/transport/_b64url.py b/src/ca2a_runtime/transport/_b64url.py index cf999ff1..b5987408 100644 --- a/src/ca2a_runtime/transport/_b64url.py +++ b/src/ca2a_runtime/transport/_b64url.py @@ -32,13 +32,19 @@ def b64url_decode(field: str, value: str) -> bytes: ``field`` names the value in the error message (e.g. ``"sealed_payload"`` or ``"raw_evidence"``), since the same helper serves several wire fields. """ - if not isinstance(value, str) or not _BASE64URL_RE.fullmatch(value): + if not isinstance(value, str) or not _BASE64URL_RE.fullmatch(value) or len(value) % 4 == 1: raise TransportError( f"{field} is not valid base64url", - detail=f"expected a non-empty base64url string, got {type(value).__name__}", + detail=f"expected a canonical non-empty base64url string, got {type(value).__name__}", ) try: padded = value + "=" * (-len(value) % 4) - return base64.urlsafe_b64decode(padded.encode("ascii")) + raw = base64.urlsafe_b64decode(padded.encode("ascii")) except (ValueError, UnicodeEncodeError) as exc: raise TransportError(f"{field} is not valid base64url", detail=str(exc)) from exc + if b64url_encode(raw) != value: + raise TransportError( + f"{field} is not valid base64url", + detail="non-canonical base64url encoding (non-zero trailing padding bits)", + ) + return raw diff --git a/src/ca2a_runtime/transport/wire.py b/src/ca2a_runtime/transport/wire.py index 72c899d3..dad57eb4 100644 --- a/src/ca2a_runtime/transport/wire.py +++ b/src/ca2a_runtime/transport/wire.py @@ -114,19 +114,36 @@ def parse_channel_offer(data: dict[str, Any]) -> ChannelOffer: "malformed channel offer", detail=f"expected an object, got {type(data).__name__}" ) try: - public_key = str(data["channel_public_key"]) + public_key = data["channel_public_key"] + if not isinstance(public_key, str) or not public_key: + raise TransportError( + "malformed channel offer", + detail="channel_public_key must be a non-empty string", + ) att = data["attestation"] + if not isinstance(att, dict): + raise TransportError( + "malformed channel offer", + detail=f"attestation must be an object, got {type(att).__name__}", + ) + for claim_field in _CLAIM_FIELDS: + val = att[claim_field] + if not isinstance(val, str) or not val: + raise TransportError( + "malformed channel offer", + detail=f"attestation.{claim_field} must be a non-empty string", + ) evidence = { field: b64url_decode(field, att[field]) for field in _EVIDENCE_FIELDS if field in att } role = att.get("role", ROLE_CALLEE) - if role not in ROLES: + if not isinstance(role, str) or role not in ROLES: raise TransportError("malformed channel offer", detail=f"unknown offer role {role!r}") report = AttestationReport( - platform=str(att["platform"]), - measurement=str(att["measurement"]), - public_key=str(att["public_key"]), - nonce=str(att["nonce"]), + platform=att["platform"], + measurement=att["measurement"], + public_key=att["public_key"], + nonce=att["nonce"], role=role, **evidence, ) diff --git a/tests/unit/test_a2a_adapter.py b/tests/unit/test_a2a_adapter.py index 7c4b4af4..0a19d39d 100644 --- a/tests/unit/test_a2a_adapter.py +++ b/tests/unit/test_a2a_adapter.py @@ -241,3 +241,55 @@ def test_sealed_payload_is_opaque_bytes_only() -> None: def test_extension_uri_is_stable() -> None: assert EXTENSION_URI == "https://agentrust-io.com/extensions/ca2a/v0.1" assert KEY_DELEGATION_CHAIN.startswith(EXTENSION_URI) + + +@pytest.mark.parametrize("bad_b64", ["A", "AB", "AC", "AD", "AAAAA", "AA="]) +def test_non_canonical_base64url_rejected_in_sealed_payload_and_offer(bad_b64: str) -> None: + from ca2a_runtime.transport import wire + from ca2a_runtime.transport._b64url import b64url_decode + + with pytest.raises(TransportError, match="not valid base64url"): + b64url_decode("sealed_payload", bad_b64) + with pytest.raises(TransportError, match="not valid base64url"): + parse_peer_request({"metadata": _ca2a_meta(sealed_payload=bad_b64)}) + offer_dict = { + "channel_public_key": "aa" * 32, + "attestation": { + "platform": "tpm", + "measurement": "sha256:abc", + "public_key": "aa" * 32, + "nonce": "n1", + "raw_evidence": bad_b64, + }, + } + with pytest.raises(TransportError, match="not valid base64url"): + wire.parse_channel_offer(offer_dict) + + +@pytest.mark.parametrize("bad_val", [None, 123, False, True, "", ["x"], {"k": "v"}]) +@pytest.mark.parametrize( + "field", + ["channel_public_key", "platform", "measurement", "public_key", "nonce"], +) +def test_parse_channel_offer_rejects_non_string_or_empty_claim_fields( + field: str, bad_val: object +) -> None: + from ca2a_runtime.transport import wire + + offer_dict: dict[str, object] = { + "channel_public_key": "aa" * 32, + "attestation": { + "platform": "software-only", + "measurement": "sha256:abc", + "public_key": "aa" * 32, + "nonce": "n1", + }, + } + if field == "channel_public_key": + offer_dict["channel_public_key"] = bad_val + else: + att = dict(offer_dict["attestation"]) # type: ignore[arg-type] + att[field] = bad_val + offer_dict["attestation"] = att + with pytest.raises(TransportError, match="malformed channel offer"): + wire.parse_channel_offer(offer_dict)