From 32c48b8b5ff6502d0b3d0a2d0e3815c8f18d1704 Mon Sep 17 00:00:00 2001 From: Josh <162485031+wiyth00@users.noreply.github.com> Date: Sat, 25 Jul 2026 13:56:20 -0400 Subject: [PATCH] feat: verify ephemeris kernel integrity on load NAIF publishes no checksums for the generic kernels, so integrity is trust-on-first-use: the first load records a .sha256 sidecar and every later load must match it. A KERNEL_SHA256_PINS table supports pinning a known-good digest from a trusted local copy, which takes precedence over the sidecar. load_ephemeris exposes allow_unverified to bypass mismatches explicitly, and EphemerisContext now carries kernel_sha256/kernel_verified so manifests can record the verification status. --- chained_eclipse/ephemeris.py | 94 ++++++++++++++++++++++++++++++-- tests/test_kernel_integrity.py | 98 ++++++++++++++++++++++++++++++++++ 2 files changed, 189 insertions(+), 3 deletions(-) create mode 100644 tests/test_kernel_integrity.py diff --git a/chained_eclipse/ephemeris.py b/chained_eclipse/ephemeris.py index 15b8774..e11ee5c 100644 --- a/chained_eclipse/ephemeris.py +++ b/chained_eclipse/ephemeris.py @@ -7,6 +7,7 @@ from __future__ import annotations +import hashlib from dataclasses import dataclass from pathlib import Path from typing import Iterable @@ -26,6 +27,77 @@ ) from .models import RealEclipse +# Known-good kernel digests. NAIF publishes no checksums for the generic +# kernels, so pins must come from a trusted local copy: +# +# sha256sum data/ephemeris/de440s.bsp +# +# Until a kernel is pinned here, integrity falls back to trust-on-first-use: +# the first load records a `.sha256` sidecar next to the kernel and +# every later load must match it. +KERNEL_SHA256_PINS: dict[str, str] = {} + + +class EphemerisIntegrityError(RuntimeError): + """The on-disk ephemeris kernel does not match its expected SHA-256.""" + + +def _sha256_of_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def verify_kernel_integrity( + kernel_path: Path, *, allow_unverified: bool = False +) -> tuple[str, bool]: + """Check a kernel against its pin and/or recorded sidecar digest. + + Returns ``(sha256_hex, verified)``. ``verified`` is True only when the + digest matched a pre-existing expectation (a pin in + ``KERNEL_SHA256_PINS`` or a previously recorded sidecar). A fresh + recording is not a verification. + + Raises ``EphemerisIntegrityError`` on any mismatch unless + ``allow_unverified`` is True. + """ + + kernel = kernel_path.name + digest = _sha256_of_file(kernel_path) + verified = False + + pinned = KERNEL_SHA256_PINS.get(kernel) + if pinned is not None: + if digest == pinned.lower(): + verified = True + elif not allow_unverified: + raise EphemerisIntegrityError( + f"{kernel} has SHA-256 {digest}, but {pinned.lower()} is pinned in " + "KERNEL_SHA256_PINS. The kernel is corrupt or is a different " + "release. Delete it to re-download, or pass " + "allow_unverified=True to proceed anyway." + ) + + sidecar = kernel_path.with_name(kernel_path.name + ".sha256") + if sidecar.exists(): + recorded = sidecar.read_text(encoding="utf-8").split()[0].lower() + if digest == recorded: + if pinned is None: + verified = True + elif not allow_unverified: + raise EphemerisIntegrityError( + f"{kernel} has SHA-256 {digest}, but {recorded} was recorded in " + f"{sidecar.name} when it was first downloaded. The kernel " + "changed on disk. Delete the kernel and the sidecar to " + "re-download, or pass allow_unverified=True to proceed anyway." + ) + else: + sidecar.write_text(f"{digest} {kernel}\n", encoding="utf-8") + + return digest, verified + @dataclass(slots=True) class EphemerisContext: @@ -39,6 +111,8 @@ class EphemerisContext: sun: object cache_dir: Path kernel_path: Path + kernel_sha256: str = "" + kernel_verified: bool = False def time_utc(self, value: str) -> Time: return self.timescale.from_datetime(_parse_utc(value)) @@ -69,15 +143,27 @@ def _parse_utc(value: str): def load_ephemeris( - cache_dir: str | Path = "data/ephemeris", kernel: str = "de440s.bsp" + cache_dir: str | Path = "data/ephemeris", + kernel: str = "de440s.bsp", + *, + allow_unverified: bool = False, ) -> EphemerisContext: - """Load (and automatically download) the JPL kernel into a local cache.""" + """Load (and automatically download) the JPL kernel into a local cache. + + The kernel's SHA-256 is checked by ``verify_kernel_integrity``: against + ``KERNEL_SHA256_PINS`` when the kernel is pinned there, and against the + ``.sha256`` sidecar recorded on first download otherwise. + """ cache = Path(cache_dir).resolve() cache.mkdir(parents=True, exist_ok=True) loader = Loader(str(cache), verbose=True) timescale = loader.timescale() ephemeris = loader(kernel) + kernel_path = cache / kernel + kernel_sha256, kernel_verified = verify_kernel_integrity( + kernel_path, allow_unverified=allow_unverified + ) return EphemerisContext( loader=loader, timescale=timescale, @@ -86,7 +172,9 @@ def load_ephemeris( moon=ephemeris["moon"], sun=ephemeris["sun"], cache_dir=cache, - kernel_path=cache / kernel, + kernel_path=kernel_path, + kernel_sha256=kernel_sha256, + kernel_verified=kernel_verified, ) diff --git a/tests/test_kernel_integrity.py b/tests/test_kernel_integrity.py new file mode 100644 index 0000000..d016721 --- /dev/null +++ b/tests/test_kernel_integrity.py @@ -0,0 +1,98 @@ +"""Kernel integrity checks for the auto-downloaded JPL ephemeris. + +``verify_kernel_integrity`` is exercised directly on small temporary files so +no real kernel, network access, or Skyfield loader is needed. +""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +import pytest + +from chained_eclipse import ephemeris as ephemeris_module +from chained_eclipse.ephemeris import ( + EphemerisIntegrityError, + verify_kernel_integrity, +) + + +def _write_kernel(tmp_path: Path, payload: bytes = b"not a real kernel") -> Path: + kernel_path = tmp_path / "de440s.bsp" + kernel_path.write_bytes(payload) + return kernel_path + + +def _sidecar(kernel_path: Path) -> Path: + return kernel_path.with_name(kernel_path.name + ".sha256") + + +def test_first_load_records_sidecar_but_is_not_verified(tmp_path: Path) -> None: + kernel_path = _write_kernel(tmp_path) + digest, verified = verify_kernel_integrity(kernel_path) + assert digest == hashlib.sha256(kernel_path.read_bytes()).hexdigest() + assert verified is False, "a fresh recording must not count as verification" + assert _sidecar(kernel_path).read_text(encoding="utf-8").split()[0] == digest + + +def test_second_load_verifies_against_recorded_sidecar(tmp_path: Path) -> None: + kernel_path = _write_kernel(tmp_path) + verify_kernel_integrity(kernel_path) + digest, verified = verify_kernel_integrity(kernel_path) + assert verified is True + assert digest == hashlib.sha256(kernel_path.read_bytes()).hexdigest() + + +def test_changed_kernel_fails_sidecar_check(tmp_path: Path) -> None: + kernel_path = _write_kernel(tmp_path) + verify_kernel_integrity(kernel_path) + kernel_path.write_bytes(b"tampered or truncated kernel") + with pytest.raises(EphemerisIntegrityError, match="changed on disk"): + verify_kernel_integrity(kernel_path) + + +def test_allow_unverified_bypasses_sidecar_mismatch(tmp_path: Path) -> None: + kernel_path = _write_kernel(tmp_path) + verify_kernel_integrity(kernel_path) + kernel_path.write_bytes(b"tampered or truncated kernel") + digest, verified = verify_kernel_integrity(kernel_path, allow_unverified=True) + assert verified is False + assert digest == hashlib.sha256(kernel_path.read_bytes()).hexdigest() + + +def test_pinned_digest_verifies( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + kernel_path = _write_kernel(tmp_path) + good = hashlib.sha256(kernel_path.read_bytes()).hexdigest() + monkeypatch.setitem( + ephemeris_module.KERNEL_SHA256_PINS, "de440s.bsp", good.upper() + ) + digest, verified = verify_kernel_integrity(kernel_path) + assert digest == good + assert verified is True + + +def test_pinned_mismatch_raises_even_with_matching_sidecar( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + kernel_path = _write_kernel(tmp_path) + verify_kernel_integrity(kernel_path) # records a matching sidecar + monkeypatch.setitem( + ephemeris_module.KERNEL_SHA256_PINS, "de440s.bsp", "0" * 64 + ) + with pytest.raises(EphemerisIntegrityError, match="pinned"): + verify_kernel_integrity(kernel_path) + + +def test_pin_beats_sidecar_for_verified_status( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + kernel_path = _write_kernel(tmp_path) + good = hashlib.sha256(kernel_path.read_bytes()).hexdigest() + monkeypatch.setitem(ephemeris_module.KERNEL_SHA256_PINS, "de440s.bsp", good) + digest, verified = verify_kernel_integrity(kernel_path) + assert (digest, verified) == (good, True) + # Sidecar was still recorded for future unpinned loads. + assert _sidecar(kernel_path).exists()