"""Unit tests for AZ-324 :class:`ManifestVerifierImpl`. Covers all 17 ACs in the AZ-324 task spec plus a Protocol-conformance check. Uses the real AZ-323 :class:`ManifestBuilder` to materialise fixtures so the sign/verify round trip exercises production code on both sides. """ from __future__ import annotations import hashlib import logging from pathlib import Path from uuid import UUID, uuid4 import orjson from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) from gps_denied_onboard._types.geo import BoundingBox, LatLonAlt from gps_denied_onboard._types.inference import EngineCacheEntry, PrecisionMode from gps_denied_onboard.clock.wall_clock import WallClock from gps_denied_onboard.components.c10_provisioning import ( C10ManifestConfig, Ed25519ManifestSigner, ManifestBuilder, ManifestBuildInput, ManifestVerifier, ManifestVerifierImpl, TileHashRecord, TilesByBboxQuery, VerifyFailReason, VerifyOutcome, ) from gps_denied_onboard.helpers.sha256_sidecar import Sha256Sidecar _BBOX = BoundingBox(50.0, 36.0, 50.5, 36.5) _ZOOM_LEVELS = (16, 17, 18) def _write_pkcs8_key(tmp_path: Path, name: str = "operator.key") -> Path: priv = Ed25519PrivateKey.generate() pem = priv.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ) path = tmp_path / name path.write_bytes(pem) return path def _public_key_from_pem(path: Path) -> Ed25519PublicKey: priv = serialization.load_pem_private_key(path.read_bytes(), password=None) assert isinstance(priv, Ed25519PrivateKey) return priv.public_key() def _make_engines(cache_root: Path) -> tuple[EngineCacheEntry, ...]: engines = cache_root / "engines" engines.mkdir(parents=True, exist_ok=True) entries: list[EngineCacheEntry] = [] for name in ("dinov2_vpr", "lightglue", "aliked"): path = engines / f"{name}_sm87_jp62_trt103_fp16.engine" payload = f"engine-{name}".encode() path.write_bytes(payload) entries.append( EngineCacheEntry( engine_path=path, sha256_hex=hashlib.sha256(payload).hexdigest(), sm=87, jp="6.2", trt="10.3", precision=PrecisionMode.FP16, extras={}, ) ) return tuple(entries) def _make_descriptor_index(cache_root: Path) -> Path: desc_dir = cache_root / "descriptors" desc_dir.mkdir(parents=True, exist_ok=True) path = desc_dir / "corpus.index" Sha256Sidecar.write_atomic_and_sidecar(path, b"faiss-binary-payload") return path def _make_calibration(cache_root: Path) -> Path: cal_dir = cache_root / "calibration" cal_dir.mkdir(parents=True, exist_ok=True) path = cal_dir / "int8_calibration.json" path.write_bytes(b'{"calibration": "data"}') return path def _make_tiles(count: int = 10) -> tuple[TileHashRecord, ...]: return tuple( TileHashRecord( zoom=16 + (i % 3), lat=50.0 + 0.001 * i, lon=36.0 + 0.001 * i, source="googlemaps", sha256_hex=hashlib.sha256(f"tile-{i}".encode()).hexdigest(), ) for i in range(count) ) class _StaticTiles: def __init__(self, records: tuple[TileHashRecord, ...]) -> None: self._records = records def query_by_bbox(self, *, bbox, zoom_levels, sector_class): # type: ignore[no-untyped-def] return self._records def _build_signed_manifest( tmp_path: Path, *, tiles: tuple[TileHashRecord, ...] | None = None, takeoff_origin: LatLonAlt | None = None, flight_id: UUID | None = None, ) -> tuple[Path, Ed25519PublicKey, tuple[TileHashRecord, ...]]: """Materialise a complete signed Manifest set on disk. The builder writes absolute paths verbatim; the verifier expects cache-root-relative paths (AC-7 bans absolute paths). We post-process the Manifest body to relative paths and re-sign with the same key, so each fixture is a realistic v1.1 Manifest. """ cache_root = tmp_path / "cache_root" cache_root.mkdir(parents=True, exist_ok=True) engine_entries = _make_engines(cache_root) descriptor_index = _make_descriptor_index(cache_root) calibration = _make_calibration(cache_root) key_path = _write_pkcs8_key(tmp_path) tiles_used = tiles if tiles is not None else _make_tiles(10) builder = ManifestBuilder( sidecar=Sha256Sidecar(), signer=Ed25519ManifestSigner(), tile_metadata_store=_StaticTiles(tiles_used), logger=logging.getLogger(f"build-{id(tmp_path)}"), clock=WallClock(), config=C10ManifestConfig(), ) request = ManifestBuildInput( cache_root=cache_root, bbox=_BBOX, zoom_levels=_ZOOM_LEVELS, sector_class="stable_rear", engine_entries=engine_entries, descriptor_index_path=descriptor_index, calibration_path=calibration, key_path=key_path, takeoff_origin=takeoff_origin, flight_id=flight_id, ) artifact = builder.build_manifest(request) # Rewrite to relative paths + re-sign. body = orjson.loads(artifact.manifest_path.read_bytes()) for entry in body["artifacts"]["engines"]: entry["path"] = str(Path(entry["path"]).relative_to(cache_root)) body["artifacts"]["descriptor_index"]["path"] = str( Path(body["artifacts"]["descriptor_index"]["path"]).relative_to(cache_root) ) body["artifacts"]["calibration"]["path"] = str( Path(body["artifacts"]["calibration"]["path"]).relative_to(cache_root) ) body_bytes = orjson.dumps( body, option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2 ) if not body_bytes.endswith(b"\n"): body_bytes += b"\n" Sha256Sidecar.write_atomic_and_sidecar(artifact.manifest_path, body_bytes) priv = serialization.load_pem_private_key(key_path.read_bytes(), password=None) assert isinstance(priv, Ed25519PrivateKey) Sha256Sidecar.write_atomic(artifact.signature_path, priv.sign(body_bytes)) pub = _public_key_from_pem(key_path) return artifact.manifest_path, pub, tiles_used def _build_verifier( *, tiles: tuple[TileHashRecord, ...] | None = None ) -> tuple[ManifestVerifierImpl, list[logging.LogRecord]]: records: list[logging.LogRecord] = [] class _ListHandler(logging.Handler): def emit(self, record: logging.LogRecord) -> None: records.append(record) logger = logging.getLogger(f"verify-{id(records)}") logger.setLevel(logging.DEBUG) logger.handlers.clear() logger.addHandler(_ListHandler()) logger.propagate = False tile_store: TilesByBboxQuery | None = ( _StaticTiles(tiles) if tiles is not None else None ) verifier = ManifestVerifierImpl( sidecar=Sha256Sidecar(), logger=logger, clock=WallClock(), tile_metadata_store=tile_store, ) return verifier, records # ---------------------------------------------------------------------- # AC-1 # ---------------------------------------------------------------------- def test_ac1_pass_on_valid_manifest(tmp_path: Path) -> None: # Arrange manifest_path, pub, _ = _build_signed_manifest(tmp_path) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) # Assert assert result.outcome is VerifyOutcome.PASS assert result.fail_reasons == () assert all(c.matched for c in result.per_artifact_checks) assert result.signing_public_key_fingerprint is not None assert result.elapsed_ms >= 0 # ---------------------------------------------------------------------- # AC-2 # ---------------------------------------------------------------------- def test_ac2_fail_on_missing_manifest(tmp_path: Path) -> None: # Arrange verifier, _ = _build_verifier() missing = tmp_path / "nope" / "Manifest.json" # Act result = verifier.verify_manifest( manifest_path=missing, trusted_public_keys=(Ed25519PrivateKey.generate().public_key(),), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert result.fail_reasons == (VerifyFailReason.MANIFEST_NOT_FOUND,) assert result.per_artifact_checks == () assert result.signing_public_key_fingerprint is None # ---------------------------------------------------------------------- # AC-3 # ---------------------------------------------------------------------- def test_ac3_fail_on_missing_signature(tmp_path: Path) -> None: # Arrange manifest_path, pub, _ = _build_signed_manifest(tmp_path) (manifest_path.parent / "Manifest.json.sig").unlink() verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert result.fail_reasons == (VerifyFailReason.SIGNATURE_NOT_FOUND,) assert result.per_artifact_checks == () # ---------------------------------------------------------------------- # AC-4 # ---------------------------------------------------------------------- def test_ac4_fail_on_tampered_manifest_body(tmp_path: Path) -> None: # Arrange: flip one byte in Manifest.json (sidecar untouched) manifest_path, pub, _ = _build_signed_manifest(tmp_path) body = bytearray(manifest_path.read_bytes()) body[10] ^= 0x01 manifest_path.write_bytes(bytes(body)) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.MANIFEST_SELF_HASH_MISMATCH in result.fail_reasons assert result.per_artifact_checks == () # ---------------------------------------------------------------------- # AC-5 # ---------------------------------------------------------------------- def test_ac5_fail_on_untrusted_public_key(tmp_path: Path) -> None: # Arrange: verify with a different keypair manifest_path, _signed_with, _ = _build_signed_manifest(tmp_path) other_key = Ed25519PrivateKey.generate().public_key() verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(other_key,), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.UNTRUSTED_PUBLIC_KEY in result.fail_reasons assert result.signing_public_key_fingerprint is not None assert result.per_artifact_checks == () # ---------------------------------------------------------------------- # AC-6 # ---------------------------------------------------------------------- def test_ac6_schema_violation_names_offending_field(tmp_path: Path) -> None: # Arrange manifest_path, _pub, _ = _build_signed_manifest(tmp_path) body = orjson.loads(manifest_path.read_bytes()) body.pop("signing_public_key_fingerprint") new_bytes = orjson.dumps(body, option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2) + b"\n" Sha256Sidecar.write_atomic_and_sidecar(manifest_path, new_bytes) # Re-sign so we get past Step B; we want Step C to be the failure. priv = Ed25519PrivateKey.generate() Sha256Sidecar.write_atomic( manifest_path.parent / "Manifest.json.sig", priv.sign(new_bytes), ) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(priv.public_key(),), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.SCHEMA_VIOLATION in result.fail_reasons assert any("signing_public_key_fingerprint" in d for d in result.fail_details) # ---------------------------------------------------------------------- # AC-7 # ---------------------------------------------------------------------- def test_ac7_absolute_path_in_artifact_is_schema_violation(tmp_path: Path) -> None: # Arrange manifest_path, _pub, _ = _build_signed_manifest(tmp_path) body = orjson.loads(manifest_path.read_bytes()) body["artifacts"]["engines"][0]["path"] = "/etc/passwd" new_bytes = orjson.dumps(body, option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2) + b"\n" Sha256Sidecar.write_atomic_and_sidecar(manifest_path, new_bytes) priv = Ed25519PrivateKey.generate() Sha256Sidecar.write_atomic( manifest_path.parent / "Manifest.json.sig", priv.sign(new_bytes), ) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(priv.public_key(),), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.SCHEMA_VIOLATION in result.fail_reasons assert any("/etc/passwd" in d for d in result.fail_details) # No per-artifact disk reads happened — the walk is Step D only. assert result.per_artifact_checks == () def test_ac7_dot_dot_segment_in_artifact_is_schema_violation(tmp_path: Path) -> None: # Arrange manifest_path, _pub, _ = _build_signed_manifest(tmp_path) body = orjson.loads(manifest_path.read_bytes()) body["artifacts"]["calibration"]["path"] = "../calibration/int8.json" new_bytes = orjson.dumps(body, option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2) + b"\n" Sha256Sidecar.write_atomic_and_sidecar(manifest_path, new_bytes) priv = Ed25519PrivateKey.generate() Sha256Sidecar.write_atomic( manifest_path.parent / "Manifest.json.sig", priv.sign(new_bytes), ) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(priv.public_key(),), ) # Assert assert VerifyFailReason.SCHEMA_VIOLATION in result.fail_reasons # ---------------------------------------------------------------------- # AC-8 # ---------------------------------------------------------------------- def test_ac8_multiple_fail_reasons_accumulate(tmp_path: Path) -> None: # Arrange: 1 engine missing, 1 engine drifted, 1 engine OK manifest_path, pub, _ = _build_signed_manifest(tmp_path) cache_root = manifest_path.parent body = orjson.loads(manifest_path.read_bytes()) # Delete first engine first_engine_rel = body["artifacts"]["engines"][0]["path"] (cache_root / first_engine_rel).unlink() # Mutate second engine second_engine_rel = body["artifacts"]["engines"][1]["path"] (cache_root / second_engine_rel).write_bytes(b"drifted-bytes") verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.ARTIFACT_MISSING in result.fail_reasons assert VerifyFailReason.ARTIFACT_HASH_MISMATCH in result.fail_reasons engine_checks = [c for c in result.per_artifact_checks if "engines" in c.relative_path] assert len(engine_checks) == 3 matched_flags = [c.matched for c in engine_checks] assert matched_flags.count(True) == 1 assert matched_flags.count(False) == 2 # ---------------------------------------------------------------------- # AC-9 # ---------------------------------------------------------------------- def test_ac9_operator_mode_re_derives_tiles_coverage(tmp_path: Path) -> None: # Arrange: build with tiles X; verify with tiles Y → mismatch tiles_built = _make_tiles(10) tiles_drifted = ( *tiles_built[:-1], TileHashRecord( zoom=18, lat=50.99, lon=36.99, source="googlemaps", sha256_hex=hashlib.sha256(b"drifted-tile").hexdigest(), ), ) manifest_path, pub, _ = _build_signed_manifest(tmp_path, tiles=tiles_built) verifier, _ = _build_verifier(tiles=tiles_drifted) # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.TILES_COVERAGE_MISMATCH in result.fail_reasons assert any("tiles_coverage" in d for d in result.fail_details) def test_ac9_operator_mode_pass_when_tiles_match(tmp_path: Path) -> None: # Arrange tiles = _make_tiles(10) manifest_path, pub, _ = _build_signed_manifest(tmp_path, tiles=tiles) verifier, _ = _build_verifier(tiles=tiles) # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) # Assert assert result.outcome is VerifyOutcome.PASS tiles_check = next( c for c in result.per_artifact_checks if c.relative_path == "tiles_coverage" ) assert tiles_check.matched is True assert tiles_check.actual_sha256 == tiles_check.expected_sha256 # ---------------------------------------------------------------------- # AC-10 # ---------------------------------------------------------------------- def test_ac10_airborne_mode_trusts_tiles_coverage(tmp_path: Path) -> None: # Arrange manifest_path, pub, _ = _build_signed_manifest(tmp_path) verifier, _ = _build_verifier() # tile_metadata_store=None # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) # Assert assert result.outcome is VerifyOutcome.PASS tiles_check = next( c for c in result.per_artifact_checks if c.relative_path == "tiles_coverage" ) assert tiles_check.matched is True # ---------------------------------------------------------------------- # AC-11 # ---------------------------------------------------------------------- def test_ac11_protocol_conformance() -> None: # Assert verifier, _ = _build_verifier() assert isinstance(verifier, ManifestVerifier) # ---------------------------------------------------------------------- # AC-12 # ---------------------------------------------------------------------- def test_ac12_elapsed_ms_recorded_on_every_outcome(tmp_path: Path) -> None: # Arrange verifier, _ = _build_verifier() manifest_path, pub, _ = _build_signed_manifest(tmp_path) # Act pass_result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) fail_result = verifier.verify_manifest( manifest_path=tmp_path / "missing.json", trusted_public_keys=(pub,), ) # Assert assert pass_result.elapsed_ms >= 0 assert fail_result.elapsed_ms >= 0 # ---------------------------------------------------------------------- # AC-13 # ---------------------------------------------------------------------- def test_ac13_empty_trusted_public_keys_fails_closed(tmp_path: Path) -> None: # Arrange verifier, records = _build_verifier() manifest_path, _pub, _ = _build_signed_manifest(tmp_path) # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.UNTRUSTED_PUBLIC_KEY in result.fail_reasons assert result.per_artifact_checks == () errors = [r for r in records if r.levelno == logging.ERROR] assert any(r.__dict__.get("kind") == "c10.manifest.verify.untrusted" for r in errors) # ---------------------------------------------------------------------- # AC-14 # ---------------------------------------------------------------------- def test_ac14_v10_manifest_without_flight_block_parses(tmp_path: Path) -> None: # Arrange: take a built manifest and strip the entire `flight` block. manifest_path, _pub, _ = _build_signed_manifest(tmp_path) body = orjson.loads(manifest_path.read_bytes()) body.pop("flight", None) body["schema_version"] = "1.0" new_bytes = orjson.dumps(body, option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2) + b"\n" Sha256Sidecar.write_atomic_and_sidecar(manifest_path, new_bytes) priv = Ed25519PrivateKey.generate() Sha256Sidecar.write_atomic( manifest_path.parent / "Manifest.json.sig", priv.sign(new_bytes), ) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(priv.public_key(),), ) # Assert assert result.outcome is VerifyOutcome.PASS assert result.takeoff_origin is None assert result.flight_id is None # ---------------------------------------------------------------------- # AC-15 # ---------------------------------------------------------------------- def test_ac15_well_formed_in_bbox_takeoff_origin_passes(tmp_path: Path) -> None: # Arrange flight = uuid4() origin = LatLonAlt(50.0, 36.2, 200.0) manifest_path, pub, _ = _build_signed_manifest( tmp_path, takeoff_origin=origin, flight_id=flight ) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(pub,), ) # Assert assert result.outcome is VerifyOutcome.PASS assert result.takeoff_origin == origin assert result.flight_id == flight # ---------------------------------------------------------------------- # AC-16 # ---------------------------------------------------------------------- def test_ac16_malformed_takeoff_origin_fails_closed(tmp_path: Path) -> None: # Arrange manifest_path, _pub, _ = _build_signed_manifest(tmp_path) body = orjson.loads(manifest_path.read_bytes()) body["flight"] = { "flight_id": str(uuid4()), "takeoff_origin": {"lat_deg": 200.0, "lon_deg": 36.2, "alt_m": 100.0}, } new_bytes = orjson.dumps(body, option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2) + b"\n" Sha256Sidecar.write_atomic_and_sidecar(manifest_path, new_bytes) priv = Ed25519PrivateKey.generate() Sha256Sidecar.write_atomic( manifest_path.parent / "Manifest.json.sig", priv.sign(new_bytes), ) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(priv.public_key(),), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.TAKEOFF_ORIGIN_INVALID in result.fail_reasons assert any("lat_deg" in d for d in result.fail_details) # Diagnostics: takeoff_origin still populated even on FAIL assert result.takeoff_origin is not None # ---------------------------------------------------------------------- # AC-17 # ---------------------------------------------------------------------- def test_ac17_out_of_bbox_takeoff_origin_fails_closed(tmp_path: Path) -> None: # Arrange manifest_path, _pub, _ = _build_signed_manifest(tmp_path) body = orjson.loads(manifest_path.read_bytes()) body["flight"] = { "flight_id": str(uuid4()), "takeoff_origin": {"lat_deg": 10.0, "lon_deg": 10.0, "alt_m": 0.0}, } new_bytes = orjson.dumps(body, option=orjson.OPT_SORT_KEYS | orjson.OPT_INDENT_2) + b"\n" Sha256Sidecar.write_atomic_and_sidecar(manifest_path, new_bytes) priv = Ed25519PrivateKey.generate() Sha256Sidecar.write_atomic( manifest_path.parent / "Manifest.json.sig", priv.sign(new_bytes), ) verifier, _ = _build_verifier() # Act result = verifier.verify_manifest( manifest_path=manifest_path, trusted_public_keys=(priv.public_key(),), ) # Assert assert result.outcome is VerifyOutcome.FAIL assert VerifyFailReason.TAKEOFF_ORIGIN_OUT_OF_BBOX in result.fail_reasons