From 1afb332631bf2cad75b5c4a6105c52b6c18b8593 Mon Sep 17 00:00:00 2001 From: Susan Poudel Date: Tue, 8 Sep 2026 15:26:29 -0400 Subject: [PATCH] feat(cli): let ca2a start demand caller attestation from config PeerNode has carried require_caller_attestation, caller_verifier and challenge_ttl_seconds since the mutual-attestation work, but none of them were reachable from a config file, so a callee run through ca2a start could never appraise its callers. The attestation block now accepts all three. caller_verifier names a platform and a PEM roots file; only tpm can be built today (via ca2a_verify.tpm.tpm_verifier), and sev-snp or tdx fail at startup with the reason rather than appraising nothing. A hardware rung with no verifier is a CONFIG_ERROR naming the field. The startup line prints the rung, a node with no verifier is told hardware offers will be refused as unappraisable, and ConfigError.detail is now printed on the start path so the reason reaches the operator. Closes #160 Signed-off-by: Susan Poudel --- CHANGELOG.md | 15 ++++ docs/configuration.md | 15 +++- docs/spec/mutual-attestation.md | 5 ++ docs/spec/transport.md | 2 +- examples/minimal/ca2a-config.yaml | 4 + src/ca2a_runtime/bootstrap.py | 61 ++++++++++++++- src/ca2a_runtime/cli.py | 21 +++++- src/ca2a_runtime/config.py | 55 ++++++++++++++ tests/unit/test_bootstrap.py | 118 +++++++++++++++++++++++++++++- tests/unit/test_cli_start.py | 46 ++++++++++++ tests/unit/test_config.py | 59 +++++++++++++++ 11 files changed, 393 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 846c778..d067954 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`ca2a start` can demand caller attestation (#160).** The `attestation` block + gains `require_caller_attestation` (`none` | `any` | `hardware`), + `caller_verifier` (`platform` plus a `trusted_roots_path` PEM bundle), and + `challenge_ttl_seconds`, threaded through `bootstrap.build_peer_node` to the + `PeerNode` arguments that have existed since the mutual-attestation work. Until + now those arguments were reachable only from Python, so a config-driven callee + could never appraise its callers. All new paths fail closed: an unknown rung, a + `hardware` rung with no verifier, a missing or empty roots file, and a verifier + platform with no report-level verifier (`sev-snp`, `tdx`) are each a + `CONFIG_ERROR` at startup naming the field, not a callee that appears to appraise + and does not. Only `tpm` can be built today. The startup line now prints the rung + alongside the provider, and a node with no verifier is told that hardware offers + will be refused as unappraisable. `require_holder_proof` is deliberately not + exposed. + - **PLATFORM_INFO appraisal on the SEV-SNP path (LIMITATIONS: "Platform state is not appraised").** `SevSnpReport.platform_info` decodes the bitfield at offset `0x40`, and `verify_sev_snp_report` takes `require_platform`, `forbid_platform` and diff --git a/docs/configuration.md b/docs/configuration.md index 287581e..40ab9c3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -11,6 +11,11 @@ transport. attestation: provider: auto # auto | tpm | sev-snp | tdx | opaque | software-only enforcement_mode: enforcing # enforcing | advisory | silent + require_caller_attestation: none # none | any | hardware + # caller_verifier: # required for `hardware`, optional for `any` + # platform: tpm # tpm today; sev-snp and tdx are refused (see below) + # trusted_roots_path: vtpm-roots.pem + challenge_ttl_seconds: 60 max_delegation_depth: 8 # reject chains deeper than this listen_addr: "127.0.0.1:8443" @@ -33,6 +38,9 @@ local_policy: ["read", "write"] # allow-set for scope intersection (or use Ced |---|---|---| | `attestation.provider` | `auto` | TEE provider for peer attestation. `auto` selects a detected hardware provider and fails if there is none; it never falls back to `software-only`, which has to be named explicitly. `opaque` is not implemented. | | `attestation.enforcement_mode` | `enforcing` | Intended mode. The peer path always fails closed on cA2A denials today; advisory and silent are accepted in config but not applied on the wire. | +| `attestation.require_caller_attestation` | `none` | What the callee demands of a caller's own attestation, per [mutual-attestation.md](spec/mutual-attestation.md). `none` records the outcome and demands nothing; `any` requires an offer that appraises, software assurance included; `hardware` requires hardware assurance. At every rung an offer that is present and does not appraise is refused. | +| `attestation.caller_verifier` | none | How to appraise a hardware report a caller offers. `platform` plus `trusted_roots_path` (a PEM bundle, resolved relative to the config file). Required when the rung is `hardware`; without it, a hardware offer at `none` or `any` is refused as unappraisable rather than accepted. Only `tpm` can be built today, via `ca2a_verify.tpm.tpm_verifier`. `sev-snp` and `tdx` are accepted by the vocabulary and refused at startup with the reason: their verifiers take raw evidence and a certificate chain, and no report-level wrapper exists yet. | +| `attestation.challenge_ttl_seconds` | `60` | Lifetime of the challenge the callee issues for a caller to bind its offer into. The secret behind it is per-process and never persisted. | | `max_delegation_depth` | `8` | Chains deeper than this are rejected with `DELEGATION_DEPTH_EXCEEDED`. | | `listen_addr` | `127.0.0.1:8443` | Address `ca2a start` binds. The host is never defaulted, so serving on every interface has to be written out. | | `trusted_root_issuers` | none | Ed25519 public keys allowed to originate delegation chains. At least one is required by `ca2a start`; an internally valid chain from any other root is denied before policy evaluation. | @@ -55,9 +63,14 @@ ca2a validate-config --config examples/minimal/ca2a-config.yaml ca2a start --config examples/minimal/ca2a-config.yaml # note: software-only provider, callers appraise this channel key as # assurance="none" and the seal carries no hardware guarantee -# ca2a listening on 127.0.0.1:8443 (provider=software-only) +# note: require_caller_attestation=none with no caller_verifier; software offers +# appraise, hardware offers are refused as unappraisable +# ca2a listening on 127.0.0.1:8443 (provider=software-only, require_caller_attestation=none) ``` +`require_holder_proof` is not a config field. It is on for every node `ca2a start` +builds; a program with a reason to turn it off constructs the `PeerNode` itself. + `ca2a start` needs no extra install: the reference transport is standard library only. It is one way to run the peer path, not part of the profile. A program that already has a `Policy` and a provider can build a `PeerNode` and serve it from diff --git a/docs/spec/mutual-attestation.md b/docs/spec/mutual-attestation.md index 2209e45..522f581 100644 --- a/docs/spec/mutual-attestation.md +++ b/docs/spec/mutual-attestation.md @@ -166,6 +166,11 @@ client.send_task(base_url, chain, "read", "r0", payload=b"...", caller_provider=SoftwareProvider()) ``` +The same rung is reachable from a config file. `ca2a start` reads +`attestation.require_caller_attestation`, builds the `caller_verifier` from +`attestation.caller_verifier` (TPM roots today; see +[configuration](../configuration.md)), and passes both to the `PeerNode` it serves. + | Piece | Where | |---|---| | Stateless challenge (`v1...`) | `ca2a_runtime.challenge` | diff --git a/docs/spec/transport.md b/docs/spec/transport.md index 0d0f807..5f02a99 100644 --- a/docs/spec/transport.md +++ b/docs/spec/transport.md @@ -158,7 +158,7 @@ message = attach_ca2a_metadata(message, request) ## Running the reference transport -A library caller builds a `PeerNode` and hands it to `ca2a_runtime.transport.server.serve`. `ca2a start` is the same thing driven from a config file: it resolves the policy (`local_policy` or a Cedar `policy_bundle_path`) and the attestation provider, builds the node, and binds `listen_addr`. +A library caller builds a `PeerNode` and hands it to `ca2a_runtime.transport.server.serve`. `ca2a start` is the same thing driven from a config file: it resolves the policy (`local_policy` or a Cedar `policy_bundle_path`), the attestation provider, and the caller-appraisal posture (`require_caller_attestation` and, where one can be built, a `caller_verifier`), builds the node, and binds `listen_addr`. ```bash ca2a start --config examples/minimal/ca2a-config.yaml diff --git a/examples/minimal/ca2a-config.yaml b/examples/minimal/ca2a-config.yaml index 8760996..c5a0e54 100644 --- a/examples/minimal/ca2a-config.yaml +++ b/examples/minimal/ca2a-config.yaml @@ -1,6 +1,10 @@ attestation: provider: software-only # auto | tpm | sev-snp | tdx | opaque | software-only enforcement_mode: enforcing # the peer path always fails closed today + require_caller_attestation: none # none | any | hardware; see docs/configuration.md + # caller_verifier: # needed for `hardware`; TPM roots only today + # platform: tpm + # trusted_roots_path: vtpm-roots.pem max_delegation_depth: 8 listen_addr: "127.0.0.1:8443" diff --git a/src/ca2a_runtime/bootstrap.py b/src/ca2a_runtime/bootstrap.py index 0160e2e..5467682 100644 --- a/src/ca2a_runtime/bootstrap.py +++ b/src/ca2a_runtime/bootstrap.py @@ -10,6 +10,11 @@ Provider selection fails closed. ``software-only`` carries no hardware guarantee, so it is only ever used when the config names it: a hardware provider whose device node is absent is a startup error, not a silent downgrade. + +Caller appraisal is the same shape. ``attestation.caller_verifier`` names a +platform and a roots file; only platforms with a report-level verifier in +:mod:`ca2a_verify` can be built, and naming one that has none is a startup error +rather than a callee that looks like it appraises and does not. """ from __future__ import annotations @@ -21,6 +26,7 @@ load_agent_manifest_trust_anchor, verify_agent_manifest_binding, ) +from ca2a_runtime.attestation import Verifier from ca2a_runtime.cedar import CedarPolicy from ca2a_runtime.config import Ca2aConfig from ca2a_runtime.errors import ConfigError @@ -31,6 +37,7 @@ from ca2a_runtime.tee.software import SoftwareProvider from ca2a_runtime.tee.tdx import TdxProvider from ca2a_runtime.tee.tpm import TpmProvider +from ca2a_verify.tpm import tpm_verifier _HARDWARE_PROVIDERS: dict[str, type[BaseProvider]] = { "sev-snp": SevSnpProvider, @@ -38,6 +45,17 @@ "tpm": TpmProvider, } +# Platforms whose appraisal can be expressed as a Verifier over an +# AttestationReport. SEV-SNP and TDX have offline verifiers (ca2a_verify.sev_snp, +# ca2a_verify.tdx) but they take raw evidence plus a certificate chain, and the +# SNP collector deliberately drops the auxblob rather than guess at its format, so +# neither can yet be built from a report and a roots file alone. +_REPORT_VERIFIERS: dict[str, str] = { + "sev-snp": "verify_sev_snp_report needs the VCEK chain, which the collector does not " + "carry on the report yet", + "tdx": "verify_tdx_quote has no AttestationReport-level wrapper yet", +} + def load_policy(config: Ca2aConfig, *, config_dir: Path | None = None) -> Policy: """Resolve the callee policy from ``policy_bundle_path`` or ``local_policy``. @@ -97,8 +115,46 @@ def select_provider(config: Ca2aConfig) -> BaseProvider: raise ConfigError(f"attestation provider {name!r} is not implemented") +def build_caller_verifier(config: Ca2aConfig, *, config_dir: Path | None = None) -> Verifier | None: + """Resolve ``attestation.caller_verifier`` into a :data:`Verifier`, or ``None``. + + ``None`` means the config named no verifier. Under ``require_caller_attestation: + none`` or ``any`` that is a legitimate posture (a caller offering a hardware + report is then refused as present-and-unappraisable, per the mutual attestation + spec); under ``hardware`` the config layer has already rejected it. + """ + platform = config.caller_verifier_platform + if platform is None: + return None + + if platform in _REPORT_VERIFIERS: + raise ConfigError( + f"attestation.caller_verifier.platform {platform!r} cannot be appraised from a " + "config file yet", + detail=_REPORT_VERIFIERS[platform], + ) + if platform != "tpm": + raise ConfigError(f"attestation.caller_verifier.platform {platform!r} is not implemented") + + roots_path = Path(config.caller_verifier_trusted_roots_path or "") + if not roots_path.is_absolute() and config_dir is not None: + roots_path = config_dir / roots_path + if not roots_path.is_file(): + raise ConfigError(f"attestation.caller_verifier.trusted_roots_path not found: {roots_path}") + try: + roots_pem = roots_path.read_bytes() + except OSError as exc: + raise ConfigError( + f"cannot read attestation.caller_verifier.trusted_roots_path: {roots_path}", + detail=str(exc), + ) from exc + if not roots_pem.strip(): + raise ConfigError(f"attestation.caller_verifier.trusted_roots_path is empty: {roots_path}") + return tpm_verifier(roots_pem) + + def build_peer_node(config: Ca2aConfig, *, config_dir: Path | None = None) -> PeerNode: - """Build the node ``ca2a start`` serves: policy, provider, and depth limit.""" + """Build the node ``ca2a start`` serves: policy, provider, caller appraisal, depth.""" policy = load_policy(config, config_dir=config_dir) if not config.trusted_root_issuers: raise ConfigError( @@ -133,6 +189,9 @@ def build_peer_node(config: Ca2aConfig, *, config_dir: Path | None = None) -> Pe policy, provider=select_provider(config), max_depth=config.max_delegation_depth, + require_caller_attestation=config.require_caller_attestation, + caller_verifier=build_caller_verifier(config, config_dir=config_dir), + challenge_ttl_seconds=config.challenge_ttl_seconds, trusted_root_issuers=config.trusted_root_issuers, agent_manifest=manifest_binding, ) diff --git a/src/ca2a_runtime/cli.py b/src/ca2a_runtime/cli.py index 6eb2ae0..2a6edf3 100644 --- a/src/ca2a_runtime/cli.py +++ b/src/ca2a_runtime/cli.py @@ -161,7 +161,9 @@ def _cmd_start(args: argparse.Namespace) -> int: node = build_peer_node(cfg, config_dir=Path(args.config).resolve().parent) host, port = cfg.listen_host_port() except ConfigError as exc: - print(f"invalid config: {exc}", file=sys.stderr) + # The detail is the part that says what to change, so it goes with the message. + suffix = f" ({exc.detail})" if exc.detail else "" + print(f"invalid config: {exc}{suffix}", file=sys.stderr) return 1 if cfg.enforcement_mode != "enforcing": @@ -180,6 +182,17 @@ def _cmd_start(args: argparse.Namespace) -> int: file=sys.stderr, ) + if node.caller_verifier is None: + # "Demands nothing" still refuses a caller whose offer is present and does + # not appraise, and with no verifier every hardware offer is exactly that. + # Say so, or the first hardware-attested caller looks like a bug. + print( + f"note: require_caller_attestation={node.require_caller_attestation} with no " + "caller_verifier; software offers appraise, hardware offers are refused as " + "unappraisable", + file=sys.stderr, + ) + try: server = serve(node, host=host, port=port) except OSError as exc: @@ -188,7 +201,11 @@ def _cmd_start(args: argparse.Namespace) -> int: # Flushed, because an operator redirecting stdout to a log needs to see the # peer come up rather than wait on a full buffer. - print(f"ca2a listening on {host}:{port} (provider={node.provider.platform})", flush=True) + print( + f"ca2a listening on {host}:{port} (provider={node.provider.platform}, " + f"require_caller_attestation={node.require_caller_attestation})", + flush=True, + ) try: server.serve_forever() except KeyboardInterrupt: diff --git a/src/ca2a_runtime/config.py b/src/ca2a_runtime/config.py index c350b51..adb2365 100644 --- a/src/ca2a_runtime/config.py +++ b/src/ca2a_runtime/config.py @@ -12,11 +12,20 @@ import yaml +from ca2a_runtime.challenge import DEFAULT_TTL_SECONDS from ca2a_runtime.errors import ConfigError +from ca2a_runtime.peer import REQUIRE_HARDWARE, REQUIRE_NONE, REQUIREMENT_VALUES VALID_PROVIDERS = frozenset({"auto", "tpm", "sev-snp", "tdx", "opaque", "software-only"}) VALID_ENFORCEMENT = frozenset({"enforcing", "advisory", "silent"}) +# Platforms a config can name under ``attestation.caller_verifier.platform``. This +# is the set of platforms the config *vocabulary* knows, not the set that can be +# appraised today; bootstrap refuses the ones with no report-level verifier and +# says why, so a config that names ``sev-snp`` fails at startup rather than +# silently appraising nothing. +VALID_VERIFIER_PLATFORMS = frozenset({"tpm", "sev-snp", "tdx"}) + DEFAULT_LISTEN_ADDR = "127.0.0.1:8443" @@ -55,6 +64,10 @@ class Ca2aConfig: provider: str = "auto" enforcement_mode: str = "enforcing" + require_caller_attestation: str = REQUIRE_NONE + caller_verifier_platform: str | None = None + caller_verifier_trusted_roots_path: str | None = None + challenge_ttl_seconds: int = DEFAULT_TTL_SECONDS max_delegation_depth: int = 8 policy_bundle_path: str | None = None local_policy: frozenset[str] | None = None @@ -85,6 +98,44 @@ def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig: detail=f"expected one of {sorted(VALID_ENFORCEMENT)}", ) + requirement = attestation.get("require_caller_attestation", REQUIRE_NONE) + if requirement not in REQUIREMENT_VALUES: + raise ConfigError( + f"unknown attestation.require_caller_attestation: {requirement!r}", + detail=f"expected one of {sorted(REQUIREMENT_VALUES)}", + ) + + verifier = attestation.get("caller_verifier") + verifier_platform: str | None = None + verifier_roots: str | None = None + if verifier is not None: + if not isinstance(verifier, dict): + raise ConfigError("attestation.caller_verifier must be a mapping") + verifier_platform = verifier.get("platform") + verifier_roots = verifier.get("trusted_roots_path") + if verifier_platform not in VALID_VERIFIER_PLATFORMS: + raise ConfigError( + f"unknown attestation.caller_verifier.platform: {verifier_platform!r}", + detail=f"expected one of {sorted(VALID_VERIFIER_PLATFORMS)}", + ) + if not isinstance(verifier_roots, str) or not verifier_roots: + raise ConfigError( + "attestation.caller_verifier.trusted_roots_path must be a non-empty path", + detail="a PEM bundle of roots the caller's attestation key chain must reach", + ) + if requirement == REQUIRE_HARDWARE and verifier is None: + # PeerNode refuses this at construction too. Naming the config field + # here means the operator sees which line to add, not which argument. + raise ConfigError( + "attestation.require_caller_attestation 'hardware' needs attestation.caller_verifier", + detail="a hardware report cannot be appraised without one, so every call " + "would be refused", + ) + + ttl = attestation.get("challenge_ttl_seconds", DEFAULT_TTL_SECONDS) + if not isinstance(ttl, int) or isinstance(ttl, bool) or ttl < 1: + 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: raise ConfigError("max_delegation_depth must be a positive integer") @@ -140,6 +191,10 @@ def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig: return cls( provider=provider, enforcement_mode=enforcement, + require_caller_attestation=requirement, + caller_verifier_platform=verifier_platform, + caller_verifier_trusted_roots_path=verifier_roots, + challenge_ttl_seconds=ttl, max_delegation_depth=depth, policy_bundle_path=bundle, local_policy=local_policy, diff --git a/tests/unit/test_bootstrap.py b/tests/unit/test_bootstrap.py index 6ac5263..76d0411 100644 --- a/tests/unit/test_bootstrap.py +++ b/tests/unit/test_bootstrap.py @@ -14,16 +14,24 @@ import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey -from ca2a_runtime.bootstrap import build_peer_node, load_policy, select_provider +from ca2a_runtime.bootstrap import ( + build_caller_verifier, + build_peer_node, + load_policy, + select_provider, +) from ca2a_runtime.cedar import CedarPolicy from ca2a_runtime.config import Ca2aConfig from ca2a_runtime.delegation.credential import DelegationCredential, new_keypair -from ca2a_runtime.errors import CA2AError, ConfigError +from ca2a_runtime.errors import AttestationFailed, CA2AError, ConfigError from ca2a_runtime.policy import LocalPolicy +from ca2a_runtime.tee.base import AttestationReport from ca2a_runtime.tee.sev_snp import SevSnpProvider from ca2a_runtime.tee.software import SoftwareProvider from ca2a_runtime.transport import client, server +TPM_ROOT_PEM = Path(__file__).parent.parent / "fixtures/tpm/bare-rsa-prefix-0016/trusted-root.pem" + def test_load_local_policy() -> None: policy = load_policy(Ca2aConfig(local_policy=frozenset({"read"}))) @@ -85,6 +93,59 @@ def test_unimplemented_provider_rejected() -> None: select_provider(Ca2aConfig(provider="opaque")) +def test_no_caller_verifier_unless_configured() -> None: + assert build_caller_verifier(Ca2aConfig()) is None + + +def test_tpm_caller_verifier_is_built_from_roots_relative_to_config_dir(tmp_path: Path) -> None: + (tmp_path / "roots.pem").write_bytes(TPM_ROOT_PEM.read_bytes()) + cfg = Ca2aConfig.from_dict( + {"attestation": {"caller_verifier": {"platform": "tpm", "trusted_roots_path": "roots.pem"}}} + ) + verifier = build_caller_verifier(cfg, config_dir=tmp_path) + assert verifier is not None + # It is a real TPM verifier, not a stub: a report carrying no evidence fails closed. + bare = AttestationReport(platform="tpm", measurement="m", public_key="k", nonce="n") + with pytest.raises(AttestationFailed): + verifier(bare, "n") + + +def test_missing_roots_file_rejected(tmp_path: Path) -> None: + cfg = Ca2aConfig.from_dict( + {"attestation": {"caller_verifier": {"platform": "tpm", "trusted_roots_path": "nope.pem"}}} + ) + with pytest.raises(ConfigError, match="trusted_roots_path not found"): + build_caller_verifier(cfg, config_dir=tmp_path) + + +def test_empty_roots_file_rejected(tmp_path: Path) -> None: + (tmp_path / "roots.pem").write_text("\n", encoding="utf-8") + cfg = Ca2aConfig.from_dict( + {"attestation": {"caller_verifier": {"platform": "tpm", "trusted_roots_path": "roots.pem"}}} + ) + with pytest.raises(ConfigError, match="trusted_roots_path is empty"): + build_caller_verifier(cfg, config_dir=tmp_path) + + +@pytest.mark.parametrize("platform", ["sev-snp", "tdx"]) +def test_platforms_without_a_report_verifier_are_refused_with_a_reason( + tmp_path: Path, platform: str +) -> None: + # The config vocabulary knows these platforms; appraising them from a roots + # file does not exist yet. Naming one must fail at startup, not appraise nothing. + (tmp_path / "roots.pem").write_bytes(TPM_ROOT_PEM.read_bytes()) + cfg = Ca2aConfig.from_dict( + { + "attestation": { + "caller_verifier": {"platform": platform, "trusted_roots_path": "roots.pem"} + } + } + ) + with pytest.raises(ConfigError, match="cannot be appraised from a config file yet") as info: + build_caller_verifier(cfg, config_dir=tmp_path) + assert info.value.detail + + def _delegation_chain() -> tuple[list[DelegationCredential], Ed25519PrivateKey]: """A one-hop chain plus the leaf subject's key, which the caller must hold.""" root_priv, root_pub = new_keypair() @@ -99,12 +160,13 @@ def _delegation_chain() -> tuple[list[DelegationCredential], Ed25519PrivateKey]: return [cred], subject_priv -def _write_config(tmp_path: Path, root_issuer: str = "test-root") -> Path: +def _write_config(tmp_path: Path, root_issuer: str = "test-root", attestation: str = "") -> Path: path = tmp_path / "ca2a-config.yaml" path.write_text( "attestation:\n" " provider: software-only\n" " enforcement_mode: enforcing\n" + f"{attestation}" "max_delegation_depth: 3\n" "local_policy:\n" " - read\n" @@ -123,6 +185,29 @@ def test_build_peer_node_carries_config(tmp_path: Path) -> None: assert isinstance(node.provider, SoftwareProvider) assert node.max_depth == 3 assert node.trusted_root_issuers == frozenset({"test-root"}) + assert node.require_caller_attestation == "none" + assert node.caller_verifier is None + assert node.challenge_ttl_seconds == 60 + + +def test_build_peer_node_carries_caller_attestation_knobs(tmp_path: Path) -> None: + (tmp_path / "roots.pem").write_bytes(TPM_ROOT_PEM.read_bytes()) + cfg = Ca2aConfig.load( + _write_config( + tmp_path, + attestation=( + " require_caller_attestation: hardware\n" + " caller_verifier:\n" + " platform: tpm\n" + " trusted_roots_path: roots.pem\n" + " challenge_ttl_seconds: 15\n" + ), + ) + ) + node = build_peer_node(cfg, config_dir=tmp_path) + assert node.require_caller_attestation == "hardware" + assert node.caller_verifier is not None + assert node.challenge_ttl_seconds == 15 def test_build_peer_node_refuses_missing_trust_anchors() -> None: @@ -154,3 +239,30 @@ def test_config_built_node_serves_a_live_call(tmp_path: Path) -> None: finally: srv.shutdown() srv.server_close() + + +def test_config_demanding_caller_attestation_enforces_it_on_a_live_call(tmp_path: Path) -> None: + """The rung written in the config is the rung the served node applies.""" + chain, leaf_key = _delegation_chain() + cfg = Ca2aConfig.load( + _write_config(tmp_path, chain[0].issuer, attestation=" require_caller_attestation: any\n") + ) + host, _ = cfg.listen_host_port() + srv = server.serve(build_peer_node(cfg, config_dir=tmp_path), host=host, port=0) + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + try: + base = f"http://{host}:{srv.server_address[1]}" + + with pytest.raises(CA2AError) as exc_info: + client.send_task(base, chain, "read", "r0", holder_key=leaf_key) + assert exc_info.value.code == "ATTESTATION_FAILED" + + body = client.send_task( + base, chain, "read", "r1", holder_key=leaf_key, caller_provider=SoftwareProvider() + ) + assert body["accepted"] is True + assert body["caller_attestation"] == "software-only" + finally: + srv.shutdown() + srv.server_close() diff --git a/tests/unit/test_cli_start.py b/tests/unit/test_cli_start.py index 26fe11d..244203c 100644 --- a/tests/unit/test_cli_start.py +++ b/tests/unit/test_cli_start.py @@ -106,3 +106,49 @@ def test_start_warns_that_software_mode_has_no_guarantee( ) assert cli.main(["start", "--config", config]) == 0 assert 'assurance="none"' in capsys.readouterr().err + + +def test_start_reports_the_caller_attestation_rung( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setattr(server, "serve", lambda node, host, port: _FakeServer(node)) + config = _config( + tmp_path, + "attestation:\n provider: software-only\n require_caller_attestation: any\n" + "local_policy:\n - read\ntrusted_root_issuers:\n - test-root\n", + ) + assert cli.main(["start", "--config", config]) == 0 + captured = capsys.readouterr() + assert "require_caller_attestation=any" in captured.out + # No verifier was configured, so the operator is told what that does to + # hardware-attested callers before the first one is refused. + assert "hardware offers are refused as unappraisable" in captured.err + + +def test_start_refuses_hardware_rung_without_a_verifier( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + config = _config( + tmp_path, + "attestation:\n provider: software-only\n require_caller_attestation: hardware\n" + "local_policy:\n - read\ntrusted_root_issuers:\n - test-root\n", + ) + assert cli.main(["start", "--config", config]) == 1 + assert "needs attestation.caller_verifier" in capsys.readouterr().err + + +def test_start_refuses_a_verifier_platform_it_cannot_build( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + (tmp_path / "roots.pem").write_text("not empty\n", encoding="utf-8") + config = _config( + tmp_path, + "attestation:\n provider: software-only\n require_caller_attestation: hardware\n" + " caller_verifier:\n platform: sev-snp\n trusted_roots_path: roots.pem\n" + "local_policy:\n - read\ntrusted_root_issuers:\n - test-root\n", + ) + assert cli.main(["start", "--config", config]) == 1 + err = capsys.readouterr().err + assert "cannot be appraised from a config file yet" in err + # The reason travels as ConfigError.detail; the operator has to see it. + assert "VCEK chain" in err diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index af114d6..aebd8d7 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -18,6 +18,65 @@ def test_defaults_from_empty_dict() -> None: assert cfg.listen_host_port() == ("127.0.0.1", 8443) assert cfg.trusted_root_issuers == frozenset() assert cfg.agent_manifest_path is None + assert cfg.require_caller_attestation == "none" + assert cfg.caller_verifier_platform is None + assert cfg.caller_verifier_trusted_roots_path is None + assert cfg.challenge_ttl_seconds == 60 + + +def test_caller_attestation_block_is_loaded() -> None: + cfg = Ca2aConfig.from_dict( + { + "attestation": { + "provider": "software-only", + "require_caller_attestation": "hardware", + "caller_verifier": {"platform": "tpm", "trusted_roots_path": "roots.pem"}, + "challenge_ttl_seconds": 30, + } + } + ) + assert cfg.require_caller_attestation == "hardware" + assert cfg.caller_verifier_platform == "tpm" + assert cfg.caller_verifier_trusted_roots_path == "roots.pem" + assert cfg.challenge_ttl_seconds == 30 + + +def test_unknown_caller_attestation_rung_rejected() -> None: + with pytest.raises(ConfigError, match="require_caller_attestation"): + Ca2aConfig.from_dict({"attestation": {"require_caller_attestation": "always"}}) + + +def test_hardware_rung_without_a_verifier_is_a_config_error() -> None: + # PeerNode would refuse this too; the config layer names the field to add. + with pytest.raises(ConfigError, match="needs attestation.caller_verifier"): + Ca2aConfig.from_dict({"attestation": {"require_caller_attestation": "hardware"}}) + + +def test_any_rung_does_not_need_a_verifier() -> None: + cfg = Ca2aConfig.from_dict({"attestation": {"require_caller_attestation": "any"}}) + assert cfg.require_caller_attestation == "any" + assert cfg.caller_verifier_platform is None + + +@pytest.mark.parametrize( + "verifier", + [ + "tpm", + {"platform": "sgx", "trusted_roots_path": "roots.pem"}, + {"platform": "tpm"}, + {"platform": "tpm", "trusted_roots_path": ""}, + {"trusted_roots_path": "roots.pem"}, + ], +) +def test_malformed_caller_verifier_rejected(verifier: object) -> None: + with pytest.raises(ConfigError, match="caller_verifier"): + Ca2aConfig.from_dict({"attestation": {"caller_verifier": verifier}}) + + +@pytest.mark.parametrize("ttl", [0, -5, "60", 1.5, True]) +def test_bad_challenge_ttl_rejected(ttl: object) -> None: + with pytest.raises(ConfigError, match="challenge_ttl_seconds"): + Ca2aConfig.from_dict({"attestation": {"challenge_ttl_seconds": ttl}}) def test_trusted_root_issuers_are_loaded() -> None: