From 8eb696da6a5f8ea0de8781795207505834bb5b8e 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:17 -0500 Subject: [PATCH] fix(config,tdx): reject bool max_delegation_depth, non-mapping attestation config, and trailing/empty TDX quote containers (#273) - Reject boolean max_delegation_depth and non-mapping attestation sections in Ca2aConfig.from_dict, and validate max_depth and challenge_ttl_seconds in PeerNode.__init__. - Reject non-zero trailing bytes after TDX quotes, trailing slack inside the TDX signature section and QE certification data container, and empty PCK certificate chains in TdxQuote.parse. Closes #273 Signed-off-by: Raghuvar Karthik Durga <177662415+drk-ram@users.noreply.github.com> --- src/ca2a_runtime/config.py | 7 +++++-- src/ca2a_runtime/node.py | 8 ++++++++ src/ca2a_runtime/tee/tdx.py | 14 ++++++++++++++ tests/unit/test_config.py | 26 ++++++++++++++++++++++++++ tests/unit/test_tdx.py | 22 ++++++++++++++++++++++ 5 files changed, 75 insertions(+), 2 deletions(-) diff --git a/src/ca2a_runtime/config.py b/src/ca2a_runtime/config.py index ccbfe2ac..d5c494c2 100644 --- a/src/ca2a_runtime/config.py +++ b/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") diff --git a/src/ca2a_runtime/node.py b/src/ca2a_runtime/node.py index 2c5cfe92..0ea3a2a5 100644 --- a/src/ca2a_runtime/node.py +++ b/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 diff --git a/src/ca2a_runtime/tee/tdx.py b/src/ca2a_runtime/tee/tdx.py index 89047f09..36d23594 100644 --- a/src/ca2a_runtime/tee/tdx.py +++ b/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, diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index aebd8d70..c7068c89 100644 --- a/tests/unit/test_config.py +++ b/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}) diff --git a/tests/unit/test_tdx.py b/tests/unit/test_tdx.py index 51eefb1e..31193c9e 100644 --- a/tests/unit/test_tdx.py +++ b/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("