mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-21 17:01:13 +00:00
Oleksandr Bezdieniezhnykh
e2bebefdfc
AZ-507: codify cross-component import rule. Added _types/inference_errors.py shim re-exporting EngineBuildError + CalibrationCacheError from c7_inference; narrowed C10 EngineCompiler's except Exception to the two typed errors so unknown exceptions propagate (AC-3). Rewrote module-layout.md "Imports from" sections for 9 components + added Rule 9; appended an architecture.md ADR-009 note explaining why components must go through _types/*. AZ-323: ManifestBuilder + Ed25519ManifestSigner. Canonical JSON via orjson OPT_SORT_KEYS+OPT_INDENT_2, atomic-write Manifest.json + sha sidecar + .sig via AZ-280, operator-key fingerprint allowlist gate (C10-ST-01), ADR-010 takeoff_origin + flight_id baked into Manifest AND manifest_hash so re-planned routes change the cache identity (AC-15/AC-16). 20 unit tests cover all 16 ACs. AZ-324: ManifestVerifierImpl. Fail-closed Steps A-D: Manifest.json sidecar self-hash, Ed25519 trust-key set, schema parse with absolute/.. path rejection + takeoff_origin in-bbox check, stream SHA-256 per artifact with multi-failure accumulation. Operator mode re-derives tiles_coverage_sha256 from C6; airborne mode trusts the signed aggregate. 19 unit tests cover all 17 ACs. Composition root: c10_factory.build_manifest_builder + build_manifest_verifier + c6_tile_metadata_store_to_tiles_query adapter (the one place that legitimately imports both C6 and C10 without violating the AZ-270 lint). Dependency: pinned cryptography>=43.0,<46.0 in pyproject.toml. Tests: 1300 passed, 80 skipped (env-only), ruff clean for all AZ-323/324 files. AZ-306 (FAISS) intentionally deferred to batch 35 — needs C++ pybind11 toolchain not present in this environment. Co-authored-by: Cursor <cursoragent@cursor.com>
722 lines
24 KiB
Python
722 lines
24 KiB
Python
"""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
|