feat: verify ephemeris kernel integrity on load

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 `<kernel>.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
+    ``<kernel>.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,
     )
 
 

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()