mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-22 19:31:15 +00:00
[AZ-507] [AZ-323] [AZ-324] C10 Manifest build + verify + AZ-270 hygiene
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>
This commit is contained in:
Oleksandr Bezdieniezhnykh
@@ -0,0 +1,721 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user