fix(config,tdx): reject bool max_delegation_depth, non-mapping attestation config, and trailing/empty TDX quote containers (#273)

src/ca2a_runtime/config.py

@@ -83,7 +83,10 @@ def listen_host_port(self) -> tuple[str, int]:
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig:
-        attestation = data.get("attestation", {}) or {}
+        raw_attestation = data.get("attestation")
+        if raw_attestation is not None and not isinstance(raw_attestation, dict):
+            raise ConfigError("attestation must be a mapping")
+        attestation = raw_attestation or {}
         provider = attestation.get("provider", "auto")
         enforcement = attestation.get("enforcement_mode", "enforcing")
 
@@ -137,7 +140,7 @@ def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig:
             raise ConfigError("attestation.challenge_ttl_seconds must be a positive integer")
 
         depth = data.get("max_delegation_depth", 8)
-        if not isinstance(depth, int) or depth < 1:
+        if not isinstance(depth, int) or isinstance(depth, bool) or depth < 1:
             raise ConfigError("max_delegation_depth must be a positive integer")
 
         raw_local = data.get("local_policy")

src/ca2a_runtime/node.py

@@ -75,6 +75,14 @@ def __init__(
                 detail="a hardware report cannot be appraised without one, so every "
                 "call would be refused",
             )
+        if not isinstance(max_depth, int) or isinstance(max_depth, bool) or max_depth < 1:
+            raise ConfigError("max_depth must be a positive integer")
+        if (
+            not isinstance(challenge_ttl_seconds, int)
+            or isinstance(challenge_ttl_seconds, bool)
+            or challenge_ttl_seconds < 1
+        ):
+            raise ConfigError("challenge_ttl_seconds must be a positive integer")
         self.policy = policy
         self.provider: BaseProvider = provider if provider is not None else SoftwareProvider()
         self.max_depth = max_depth

src/ca2a_runtime/tee/tdx.py

@@ -145,6 +145,10 @@ def parse(cls, blob: bytes) -> TdxQuote:
             )
         if pos + outer_len > len(sig):
             raise AttestationFailed("QE certification data is truncated")
+        if pos + outer_len != len(sig):
+            raise AttestationFailed(
+                "TDX signature section carries trailing bytes after QE certification data"
+            )
         cert_data = sig[pos : pos + outer_len]
         if len(cert_data) < QE_REPORT_LEN + QUOTE_SIG_LEN + 2 + CERT_DATA_HEADER_LEN:
             raise AttestationFailed("QE certification data is truncated")
@@ -164,6 +168,10 @@ def parse(cls, blob: bytes) -> TdxQuote:
         inner += CERT_DATA_HEADER_LEN
         if inner + cert_len > len(cert_data):
             raise AttestationFailed("PCK certificate chain is truncated")
+        if inner + cert_len != len(cert_data):
+            raise AttestationFailed(
+                "QE certification data carries trailing bytes after PCK certificate chain"
+            )
         cert_bytes = cert_data[inner : inner + cert_len]
         if cert_type != CERT_TYPE_PCK_CHAIN:
             raise AttestationFailed(
@@ -176,6 +184,12 @@ def parse(cls, blob: bytes) -> TdxQuote:
             raise AttestationFailed(
                 "could not parse PCK certificate chain", detail=str(exc)
             ) from exc
+        if not chain:
+            raise AttestationFailed("PCK certificate chain is empty")
+        if any(blob[end:]):
+            raise AttestationFailed(
+                "TDX quote carries non-zero trailing bytes after signature section"
+            )
 
         return cls(
             version=version,

tests/unit/test_config.py

@@ -190,3 +190,29 @@ def test_agent_manifest_config_is_loaded_as_a_complete_set() -> None:
 def test_incomplete_or_unbound_agent_manifest_config_is_rejected(manifest: dict) -> None:
     with pytest.raises(ConfigError, match="agent_manifest"):
         Ca2aConfig.from_dict({"agent_manifest": manifest})
+
+
+@pytest.mark.parametrize("bad_depth", [True, False, 0, -1, "8"])
+def test_bool_or_non_positive_max_delegation_depth_rejected(bad_depth: object) -> None:
+    from ca2a_runtime.node import PeerNode
+    from ca2a_runtime.policy import LocalPolicy
+
+    with pytest.raises(ConfigError, match="max_delegation_depth"):
+        Ca2aConfig.from_dict({"max_delegation_depth": bad_depth})
+    with pytest.raises(ConfigError, match="max_depth"):
+        PeerNode(LocalPolicy.of(["read"]), max_depth=bad_depth)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize("bad_ttl", [True, False, 0, -1, "60"])
+def test_peer_node_rejects_bool_or_non_positive_challenge_ttl(bad_ttl: object) -> None:
+    from ca2a_runtime.node import PeerNode
+    from ca2a_runtime.policy import LocalPolicy
+
+    with pytest.raises(ConfigError, match="challenge_ttl_seconds"):
+        PeerNode(LocalPolicy.of(["read"]), challenge_ttl_seconds=bad_ttl)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize("bad_att", ["software-only", [1], 42, True])
+def test_non_mapping_attestation_config_rejected(bad_att: object) -> None:
+    with pytest.raises(ConfigError, match="attestation must be a mapping"):
+        Ca2aConfig.from_dict({"attestation": bad_att})

tests/unit/test_tdx.py

@@ -260,3 +260,25 @@ def test_trailing_padding_after_the_signature_section_is_ignored(quote_and_root)
         expected_mrtd=quote_and_root["mrtd"],
         expected_report_data=quote_and_root["rd"],
     )
+
+
+def test_nonzero_trailing_bytes_or_nested_container_slack_rejected(quote_and_root) -> None:
+    quote: bytes = quote_and_root["quote"]
+    with pytest.raises(AttestationFailed, match="non-zero trailing bytes"):
+        TdxQuote.parse(quote + b"\x01")
+
+    signed = 48 + 584
+    sig = quote[signed + 4 :]
+    head = quote[:signed]
+
+    # Trailing byte inside sig after outer QE certification data
+    sig_with_slack = sig + b"\x00"
+    with pytest.raises(AttestationFailed, match="trailing bytes after QE certification data"):
+        TdxQuote.parse(head + struct.pack("<I", len(sig_with_slack)) + sig_with_slack)
+
+    # Trailing byte inside outer QE certification data after PCK chain
+    qe = 128 + 6
+    cert_data = sig[qe:] + b"\x00"
+    new_sig = sig[:128] + struct.pack("<HI", 6, len(cert_data)) + cert_data
+    with pytest.raises(AttestationFailed, match="trailing bytes after PCK certificate chain"):
+        TdxQuote.parse(head + struct.pack("<I", len(new_sig)) + new_sig)