diff --git a/Dockerfile b/Dockerfile index c9e6852..085ab41 100644 --- a/Dockerfile +++ b/Dockerfile @@ -110,5 +110,9 @@ RUN --mount=type=secret,id=attestation_secret,required=true \ # Build-time only (BuildKit secret mount → file, not ENV): # attestation_secret → /run/prism/attestation_hmac_key (mode 0400) +# Sidecar listen mode (optional). Publish this port in the Lium template's +# internal_ports so BASE can dial POST /v1/sidecar/attest on the instance. +EXPOSE 8787 + ENTRYPOINT ["prism-recipe"] CMD ["preflight"] diff --git a/Dockerfile.cuda b/Dockerfile.cuda index 6caa846..f93d158 100644 --- a/Dockerfile.cuda +++ b/Dockerfile.cuda @@ -88,5 +88,9 @@ RUN --mount=type=secret,id=attestation_secret,required=true \ && test -s /run/prism/attestation_hmac_key \ && touch -h -d "@${SOURCE_DATE_EPOCH}" /run/prism /run/prism/attestation_hmac_key +# Sidecar listen mode (optional). Publish this port in the Lium template's +# internal_ports so BASE can dial POST /v1/sidecar/attest on the instance. +EXPOSE 8787 + ENTRYPOINT ["prism-recipe"] CMD ["preflight"] diff --git a/src/prism_recipe/sidecar/__main__.py b/src/prism_recipe/sidecar/__main__.py index 10f535b..9a9b977 100644 --- a/src/prism_recipe/sidecar/__main__.py +++ b/src/prism_recipe/sidecar/__main__.py @@ -14,6 +14,7 @@ from prism_recipe.sidecar.config import SidecarConfig from prism_recipe.sidecar.errors import SidecarError, SidecarReachabilityError +from prism_recipe.sidecar.listen import serve_forever from prism_recipe.sidecar.service import AttestationSidecar from prism_recipe.sidecar.transport import FakeChallengeTransport, HttpxChallengeTransport from prism_recipe.sidecar.types import Challenge, ChallengePhase @@ -50,21 +51,42 @@ def main(argv: Sequence[str] | None = None) -> int: p_once = sub.add_parser("answer-once", help="Answer a single challenge JSON") p_once.add_argument("--nonce", required=True) - p_once.add_argument( - "--phase", choices=("start", "interval", "end"), default="start" - ) + p_once.add_argument("--phase", choices=("start", "interval", "end"), default="start") p_once.add_argument("--pod-id", required=True) p_once.add_argument("--digest", required=True) p_once.add_argument("--variant", choices=("cpu", "cuda"), default="cpu") p_once.add_argument("--secret-path", type=Path, required=True) p_once.add_argument("--root", type=Path, default=None) + p_serve = sub.add_parser( + "serve", + help="Listen for BASE attest requests (stdlib HTTP; no pull loop)", + ) + p_serve.add_argument( + "--host", + default="0.0.0.0", + help="Bind address (explicit; default 0.0.0.0 for container publish)", + ) + p_serve.add_argument( + "--port", + type=int, + default=8787, + help="Bind port (default 8787; publish via Lium internal_ports)", + ) + p_serve.add_argument("--pod-id", default=None) + p_serve.add_argument("--digest", default=None) + p_serve.add_argument("--variant", choices=("cpu", "cuda"), default=None) + p_serve.add_argument("--secret-path", type=Path, default=None) + p_serve.add_argument("--root", type=Path, default=None) + args = parser.parse_args(list(sys.argv[1:] if argv is None else argv)) if args.command == "answer-once": return _cmd_answer_once(args) if args.command == "run": return _cmd_run(args) + if args.command == "serve": + return _cmd_serve(args) return 2 @@ -100,20 +122,16 @@ def _cmd_run(args: argparse.Namespace) -> int: try: if args.fake: cfg = _config_from_args_or_env(args, require_identity=True) - transport: FakeChallengeTransport | HttpxChallengeTransport = ( - FakeChallengeTransport( - challenges_by_phase={ - ChallengePhase.START: Challenge( - nonce="fake-nonce-start", phase=ChallengePhase.START - ), - ChallengePhase.INTERVAL: Challenge( - nonce="fake-nonce-interval", phase=ChallengePhase.INTERVAL - ), - ChallengePhase.END: Challenge( - nonce="fake-nonce-end", phase=ChallengePhase.END - ), - } - ) + transport: FakeChallengeTransport | HttpxChallengeTransport = FakeChallengeTransport( + challenges_by_phase={ + ChallengePhase.START: Challenge( + nonce="fake-nonce-start", phase=ChallengePhase.START + ), + ChallengePhase.INTERVAL: Challenge( + nonce="fake-nonce-interval", phase=ChallengePhase.INTERVAL + ), + ChallengePhase.END: Challenge(nonce="fake-nonce-end", phase=ChallengePhase.END), + } ) else: cfg = _config_from_args_or_env(args, require_identity=True) @@ -121,18 +139,31 @@ def _cmd_run(args: argparse.Namespace) -> int: if not base_url: raise SidecarError("base_url required (env PRISM_SIDECAR_BASE_URL)") transport = HttpxChallengeTransport(base_url=base_url) - code = AttestationSidecar(cfg).run( - transport=transport, rng=random.Random(args.seed) - ) + code = AttestationSidecar(cfg).run(transport=transport, rng=random.Random(args.seed)) return code except SidecarError as exc: sys.stderr.write(f"sidecar error: {exc}\n") return 1 -def _config_from_args_or_env( - args: argparse.Namespace, *, require_identity: bool -) -> SidecarConfig: +def _cmd_serve(args: argparse.Namespace) -> int: + try: + cfg = _config_from_args_or_env(args, require_identity=True) + # Attach optional CLI-only knobs that _config_from_args_or_env may omit + # when falling back to env (base_url unused in listen mode). + serve_forever(cfg, host=str(args.host), port=int(args.port)) + return 0 + except (SidecarError, SidecarReachabilityError, ValueError, OSError) as exc: + sys.stderr.write(f"sidecar error: {exc}\n") + return 1 + + +def _config_from_args_or_env(args: argparse.Namespace, *, require_identity: bool) -> SidecarConfig: + retry_budget = getattr(args, "retry_budget", None) + interval_count = getattr(args, "interval_count", None) + interval_min_s = getattr(args, "interval_min_s", None) + interval_max_s = getattr(args, "interval_max_s", None) + base_url = getattr(args, "base_url", None) if args.pod_id and args.digest: variant = args.variant or "cpu" secret = args.secret_path or SidecarConfig.default_secret_path() @@ -142,13 +173,11 @@ def _config_from_args_or_env( variant=variant, secret_path=secret, root=args.root, - retry_budget=args.retry_budget or 3, - interval_count=( - args.interval_count if args.interval_count is not None else 0 - ), - interval_min_s=args.interval_min_s if args.interval_min_s is not None else 0.0, - interval_max_s=args.interval_max_s if args.interval_max_s is not None else 0.0, - base_url=args.base_url, + retry_budget=retry_budget or 3, + interval_count=(interval_count if interval_count is not None else 0), + interval_min_s=interval_min_s if interval_min_s is not None else 0.0, + interval_max_s=interval_max_s if interval_max_s is not None else 0.0, + base_url=base_url, ) if require_identity and not (args.pod_id and args.digest): # Fall back to env for container runs. @@ -159,23 +188,17 @@ def _config_from_args_or_env( variant=args.variant or env_cfg.variant, secret_path=args.secret_path or env_cfg.secret_path, root=args.root, - retry_budget=args.retry_budget or env_cfg.retry_budget, + retry_budget=retry_budget or env_cfg.retry_budget, interval_count=( - args.interval_count - if args.interval_count is not None - else env_cfg.interval_count + interval_count if interval_count is not None else env_cfg.interval_count ), interval_min_s=( - args.interval_min_s - if args.interval_min_s is not None - else env_cfg.interval_min_s + interval_min_s if interval_min_s is not None else env_cfg.interval_min_s ), interval_max_s=( - args.interval_max_s - if args.interval_max_s is not None - else env_cfg.interval_max_s + interval_max_s if interval_max_s is not None else env_cfg.interval_max_s ), - base_url=args.base_url or env_cfg.base_url, + base_url=base_url or env_cfg.base_url, ) raise SidecarError("pod_id and digest required") diff --git a/src/prism_recipe/sidecar/listen.py b/src/prism_recipe/sidecar/listen.py new file mode 100644 index 0000000..9cfa93d --- /dev/null +++ b/src/prism_recipe/sidecar/listen.py @@ -0,0 +1,208 @@ +"""Stdlib HTTP listen mode for the in-image attestation sidecar. + +BASE dials the running instance and POSTs a fresh nonce. No third-party +server stack — ``http.server.ThreadingHTTPServer`` only (hermetic image). +""" + +from __future__ import annotations + +import json +import logging +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Final +from urllib.parse import urlparse + +from prism_recipe.sidecar.config import SidecarConfig +from prism_recipe.sidecar.errors import SidecarError +from prism_recipe.sidecar.service import AttestationSidecar, ChallengeAnswer +from prism_recipe.sidecar.types import Challenge, ChallengePhase +from prism_recipe.sidecar.wire import signed_attestation_to_wire + +logger = logging.getLogger(__name__) + +ATTEST_PATH: Final[str] = "/v1/sidecar/attest" +HEALTHZ_PATH: Final[str] = "/healthz" +MAX_BODY_BYTES: Final[int] = 64 * 1024 +_VALID_PHASES: Final[frozenset[str]] = frozenset({"start", "interval", "end"}) + + +def build_server( + config: SidecarConfig, + *, + host: str, + port: int, +) -> ThreadingHTTPServer: + """Bind an explicit host/port and return a ready ``ThreadingHTTPServer``. + + ``host`` must be caller-provided (no implicit ``0.0.0.0`` default here). + Pass ``port=0`` in tests for an ephemeral port. + """ + if not host.strip(): + msg = "host must be non-empty" + raise ValueError(msg) + sidecar = AttestationSidecar(config) + handler = _make_handler(sidecar) + return ThreadingHTTPServer((host, port), handler) + + +def serve_forever(config: SidecarConfig, *, host: str, port: int) -> None: + """Build the server and block serving until interrupted.""" + server = build_server(config, host=host, port=port) + bound_host, bound_port = server.server_address[:2] + logger.info("sidecar listen mode on http://%s:%s", bound_host, bound_port) + try: + server.serve_forever() + finally: + server.server_close() + + +def answer_to_wire(answer: ChallengeAnswer) -> dict[str, Any]: + """Serialize a ``ChallengeAnswer`` to the answer-once wire shape.""" + wire = signed_attestation_to_wire(answer.signed) + wire["phase"] = answer.phase.value + wire["baked_manifest_match"] = answer.baked_manifest_match + wire["mismatched_paths"] = list(answer.mismatched_paths) + return wire + + +def _make_handler(sidecar: AttestationSidecar) -> type[BaseHTTPRequestHandler]: + class SidecarHTTPRequestHandler(BaseHTTPRequestHandler): + server_version = "PrismSidecar/1.0" + protocol_version = "HTTP/1.1" + + def log_message(self, fmt: str, *args: object) -> None: + # Access log only — never include body/secret material. + logger.info("%s - %s", self.address_string(), fmt % args) + + def do_GET(self) -> None: # noqa: N802 — stdlib handler API + path = urlparse(self.path).path + if path == HEALTHZ_PATH: + self._json_response(200, {"status": "ok"}) + return + if path == ATTEST_PATH: + self._json_response(405, {"error": "method not allowed"}) + return + self._json_response(404, {"error": "not found"}) + + def do_POST(self) -> None: # noqa: N802 — stdlib handler API + path = urlparse(self.path).path + if path != ATTEST_PATH: + if path == HEALTHZ_PATH: + self._json_response(405, {"error": "method not allowed"}) + return + self._json_response(404, {"error": "not found"}) + return + self._handle_attest() + + def do_PUT(self) -> None: # noqa: N802 + self._method_not_allowed_or_404() + + def do_DELETE(self) -> None: # noqa: N802 + self._method_not_allowed_or_404() + + def do_PATCH(self) -> None: # noqa: N802 + self._method_not_allowed_or_404() + + def _method_not_allowed_or_404(self) -> None: + path = urlparse(self.path).path + if path in {ATTEST_PATH, HEALTHZ_PATH}: + self._json_response(405, {"error": "method not allowed"}) + return + self._json_response(404, {"error": "not found"}) + + def _handle_attest(self) -> None: + try: + raw = self._read_body() + except _ClientBodyError as exc: + self._json_response(exc.status, {"error": exc.message}) + return + try: + challenge = _parse_attest_body(raw) + except ValueError as exc: + self._json_response(400, {"error": str(exc)}) + return + try: + answer = sidecar.answer_challenge(challenge) + wire = answer_to_wire(answer) + except SidecarError: + logger.exception("sidecar attest failed") + self._json_response(500, {"error": "internal error"}) + return + except Exception: + logger.exception("unexpected attest failure") + self._json_response(500, {"error": "internal error"}) + return + self._json_response(200, wire) + + def _read_body(self) -> bytes: + length_hdr = self.headers.get("Content-Length") + if length_hdr is None: + # No body / chunked not supported — treat as empty. + return b"" + try: + length = int(length_hdr) + except ValueError as exc: + raise _ClientBodyError(400, "invalid Content-Length") from exc + if length < 0: + raise _ClientBodyError(400, "invalid Content-Length") + if length > MAX_BODY_BYTES: + raise _ClientBodyError(413, "request body too large") + return self.rfile.read(length) + + def _json_response(self, status: int, body: dict[str, Any]) -> None: + payload = json.dumps(body, separators=(",", ":"), sort_keys=True).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.send_header("Connection", "close") + self.end_headers() + self.wfile.write(payload) + + return SidecarHTTPRequestHandler + + +def _parse_attest_body(raw: bytes) -> Challenge: + """Parse POST /v1/sidecar/attest JSON into a ``Challenge`` (boundary).""" + if not raw.strip(): + msg = "request body required" + raise ValueError(msg) + try: + data = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + msg = "malformed JSON" + raise ValueError(msg) from exc + if not isinstance(data, dict): + msg = "JSON body must be an object" + raise ValueError(msg) + nonce = data.get("nonce") + if not isinstance(nonce, str) or not nonce.strip(): + msg = "nonce must be a non-empty string" + raise ValueError(msg) + phase_raw = data.get("phase", "start") + if not isinstance(phase_raw, str): + msg = "phase must be a string" + raise ValueError(msg) + phase_key = phase_raw.strip().lower() + if phase_key not in _VALID_PHASES: + msg = f"invalid phase: {phase_raw!r}" + raise ValueError(msg) + match phase_key: + case "start": + phase = ChallengePhase.START + case "interval": + phase = ChallengePhase.INTERVAL + case "end": + phase = ChallengePhase.END + case unreachable: + msg = f"invalid phase: {unreachable!r}" + raise ValueError(msg) + return Challenge(nonce=nonce.strip(), phase=phase) + + +class _ClientBodyError(Exception): + """Client-caused body read failure (maps to 4xx).""" + + def __init__(self, status: int, message: str) -> None: + super().__init__(message) + self.status = status + self.message = message diff --git a/tests/test_sidecar_listen.py b/tests/test_sidecar_listen.py new file mode 100644 index 0000000..c91bff9 --- /dev/null +++ b/tests/test_sidecar_listen.py @@ -0,0 +1,305 @@ +"""TDD: sidecar listen/serve mode (stdlib HTTP server). + +BASE dials the running instance and POSTs a fresh nonce; the sidecar answers +with the same wire shape as ``answer-once``. +""" + +from __future__ import annotations + +import threading +from collections.abc import Iterator +from http.server import ThreadingHTTPServer +from pathlib import Path +from typing import Any + +import httpx +import pytest + +from prism_recipe.attestation.payload import ( + derive_attestation_key, + verify_attestation_payload, +) +from prism_recipe.sealed_surface import bake_manifest +from prism_recipe.sidecar.config import SidecarConfig +from prism_recipe.sidecar.service import AttestationSidecar +from prism_recipe.sidecar.types import Challenge, ChallengePhase +from prism_recipe.sidecar.wire import signed_attestation_to_wire, signed_from_wire + +BUILD_SECRET = b"sidecar-listen-unit-test-build-secret-v1" +DIGEST = "sha256:" + ("ab" * 32) +POD_ID = "pod_listen_001" +NONCE = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +MAX_BODY = 64 * 1024 + + +@pytest.fixture +def recipe_root() -> Path: + root = Path(__file__).resolve().parents[1] + assert (root / "src" / "prism_recipe").is_dir() + return root + + +@pytest.fixture +def baked_manifest_path(recipe_root: Path, tmp_path: Path) -> Path: + out = tmp_path / "sealed_surface_manifest.json" + bake_manifest(recipe_root, out) + return out + + +@pytest.fixture +def secret_path(tmp_path: Path) -> Path: + path = tmp_path / "attestation_hmac_key" + path.write_bytes(BUILD_SECRET) + path.chmod(0o400) + return path + + +@pytest.fixture +def sidecar_config( + recipe_root: Path, + secret_path: Path, + baked_manifest_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> SidecarConfig: + monkeypatch.setenv("PRISM_RECIPE_SEALED_MANIFEST", str(baked_manifest_path)) + monkeypatch.setenv("PRISM_RECIPE_HOME", str(recipe_root)) + return SidecarConfig( + pod_id=POD_ID, + digest=DIGEST, + variant="cpu", + secret_path=secret_path, + root=recipe_root, + retry_budget=1, + interval_count=0, + ) + + +@pytest.fixture +def listen_server(sidecar_config: SidecarConfig) -> Iterator[tuple[ThreadingHTTPServer, str]]: + from prism_recipe.sidecar.listen import build_server + + server = build_server(sidecar_config, host="127.0.0.1", port=0) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + host, port = server.server_address[:2] + base = f"http://{host}:{port}" + try: + yield server, base + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5.0) + + +def test_healthz_ok(listen_server: tuple[ThreadingHTTPServer, str]) -> None: + """Given running server; When GET /healthz; Then 200 JSON liveness.""" + _server, base = listen_server + with httpx.Client(timeout=5.0) as client: + resp = client.get(f"{base}/healthz") + assert resp.status_code == 200 + assert resp.headers.get("content-type", "").startswith("application/json") + body = resp.json() + assert body.get("status") == "ok" + + +def test_listen_attest_returns_verifiable_signature( + listen_server: tuple[ThreadingHTTPServer, str], +) -> None: + """Given POST attest with nonce; When BASE verifies HMAC; Then ok + digest.""" + _server, base = listen_server + with httpx.Client(timeout=30.0) as client: + resp = client.post( + f"{base}/v1/sidecar/attest", + json={"nonce": NONCE, "phase": "start"}, + ) + assert resp.status_code == 200 + wire = resp.json() + signed = signed_from_wire(wire) + verify_key = derive_attestation_key(BUILD_SECRET) + result = verify_attestation_payload(signed, verify_key=verify_key) + assert result.ok is True + assert signed.payload.digest == DIGEST + assert signed.payload.pod_id == POD_ID + + +def test_listen_attest_reflects_requested_nonce( + listen_server: tuple[ThreadingHTTPServer, str], +) -> None: + """Given a specific nonce; When attest; Then signed payload binds that nonce.""" + _server, base = listen_server + nonce = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + with httpx.Client(timeout=30.0) as client: + resp = client.post( + f"{base}/v1/sidecar/attest", + json={"nonce": nonce, "phase": "interval"}, + ) + assert resp.status_code == 200 + wire = resp.json() + assert wire["payload"]["nonce"] == nonce + assert wire["phase"] == "interval" + + +def test_unknown_path_404(listen_server: tuple[ThreadingHTTPServer, str]) -> None: + """Given unknown path; When GET; Then 404 JSON.""" + _server, base = listen_server + with httpx.Client(timeout=5.0) as client: + resp = client.get(f"{base}/nope") + assert resp.status_code == 404 + assert resp.headers.get("content-type", "").startswith("application/json") + assert "error" in resp.json() + + +def test_wrong_method_405(listen_server: tuple[ThreadingHTTPServer, str]) -> None: + """Given known path wrong method; When GET attest; Then 405 JSON.""" + _server, base = listen_server + with httpx.Client(timeout=5.0) as client: + resp = client.get(f"{base}/v1/sidecar/attest") + assert resp.status_code == 405 + assert resp.headers.get("content-type", "").startswith("application/json") + assert "error" in resp.json() + + +def test_bad_json_returns_400_not_500( + listen_server: tuple[ThreadingHTTPServer, str], +) -> None: + """Given malformed JSON body; When POST attest; Then 400 not 500.""" + _server, base = listen_server + with httpx.Client(timeout=5.0) as client: + resp = client.post( + f"{base}/v1/sidecar/attest", + content=b"not-json{", + headers={"content-type": "application/json"}, + ) + assert resp.status_code == 400 + assert "error" in resp.json() + + +def test_missing_nonce_returns_400( + listen_server: tuple[ThreadingHTTPServer, str], +) -> None: + """Given body without nonce; When POST attest; Then 400.""" + _server, base = listen_server + with httpx.Client(timeout=5.0) as client: + resp = client.post( + f"{base}/v1/sidecar/attest", + json={"phase": "start"}, + ) + assert resp.status_code == 400 + assert "error" in resp.json() + + +def test_blank_nonce_returns_400( + listen_server: tuple[ThreadingHTTPServer, str], +) -> None: + """Given blank nonce; When POST attest; Then 400.""" + _server, base = listen_server + with httpx.Client(timeout=5.0) as client: + resp = client.post( + f"{base}/v1/sidecar/attest", + json={"nonce": " ", "phase": "start"}, + ) + assert resp.status_code == 400 + assert "error" in resp.json() + + +def test_invalid_phase_returns_400( + listen_server: tuple[ThreadingHTTPServer, str], +) -> None: + """Given invalid phase; When POST attest; Then 400.""" + _server, base = listen_server + with httpx.Client(timeout=5.0) as client: + resp = client.post( + f"{base}/v1/sidecar/attest", + json={"nonce": NONCE, "phase": "bogus"}, + ) + assert resp.status_code == 400 + assert "error" in resp.json() + + +def test_error_response_does_not_leak_secret( + listen_server: tuple[ThreadingHTTPServer, str], +) -> None: + """Given any error path; When inspecting bodies; Then secret never appears.""" + _server, base = listen_server + secret_hex = BUILD_SECRET.hex() + probes: list[httpx.Response] = [] + with httpx.Client(timeout=5.0) as client: + probes.append(client.post(f"{base}/v1/sidecar/attest", content=b"{")) + probes.append(client.post(f"{base}/v1/sidecar/attest", json={"phase": "start"})) + probes.append( + client.post( + f"{base}/v1/sidecar/attest", + json={"nonce": NONCE, "phase": "nope"}, + ) + ) + probes.append(client.get(f"{base}/missing")) + probes.append(client.get(f"{base}/v1/sidecar/attest")) + # Happy path body must not echo the raw secret either. + probes.append( + client.post( + f"{base}/v1/sidecar/attest", + json={"nonce": NONCE, "phase": "start"}, + ) + ) + for resp in probes: + text = resp.text + assert BUILD_SECRET.decode("ascii") not in text + assert secret_hex not in text + assert BUILD_SECRET not in resp.content + + +def test_wire_shape_matches_answer_once( + listen_server: tuple[ThreadingHTTPServer, str], + sidecar_config: SidecarConfig, +) -> None: + """Given same challenge; When listen vs answer-once; Then JSON keys match.""" + _server, base = listen_server + with httpx.Client(timeout=30.0) as client: + resp = client.post( + f"{base}/v1/sidecar/attest", + json={"nonce": NONCE, "phase": "end"}, + ) + assert resp.status_code == 200 + listen_wire: dict[str, Any] = resp.json() + + sidecar = AttestationSidecar(sidecar_config) + answer = sidecar.answer_challenge(Challenge(nonce=NONCE, phase=ChallengePhase.END)) + once_wire = signed_attestation_to_wire(answer.signed) + once_wire["phase"] = answer.phase.value + once_wire["baked_manifest_match"] = answer.baked_manifest_match + once_wire["mismatched_paths"] = list(answer.mismatched_paths) + + assert set(listen_wire.keys()) == set(once_wire.keys()) + assert set(listen_wire["payload"].keys()) == set(once_wire["payload"].keys()) + for key in ( + "schema_version", + "algorithm", + "signature", + "phase", + "baked_manifest_match", + "mismatched_paths", + "hardware_root_of_trust", + "sufficient_alone_for_tier_elevation", + "proves", + ): + assert key in listen_wire + assert listen_wire["payload"]["nonce"] == once_wire["payload"]["nonce"] + assert listen_wire["payload"]["digest"] == once_wire["payload"]["digest"] + assert listen_wire["signature"] == once_wire["signature"] + + +def test_oversized_body_rejected( + listen_server: tuple[ThreadingHTTPServer, str], +) -> None: + """Given body larger than max; When POST attest; Then 400/413 not 500.""" + _server, base = listen_server + huge = b'{"nonce":"' + (b"x" * (MAX_BODY + 8)) + b'","phase":"start"}' + with httpx.Client(timeout=5.0) as client: + resp = client.post( + f"{base}/v1/sidecar/attest", + content=huge, + headers={"content-type": "application/json"}, + ) + assert resp.status_code in {400, 413} + assert resp.status_code != 500 + assert "error" in resp.json()