fix(transport): enforce canonical base64url in b64url_decode and reject non-string fields in parse_channel_offer (#271)

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

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,
         )

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)