Add spec-declared checksums for static tables; validate lockfile fast path

packages/gnss-product-management/src/gnss_product_management/factories/pipelines/resolve.py

@@ -5,8 +5,10 @@
 sidecar lockfiles, and persists an aggregate lockfile.
 
 Fast path: if an aggregate lockfile already exists for
-``(package, task, date, version)`` the pipeline returns immediately
-without searching or downloading.
+``(package, task, date, version)`` and every required entry still
+validates (file present, sidecar hash matching), the pipeline returns
+immediately without searching or downloading.  Otherwise it falls
+through to a full re-resolution.
 """
 
 from __future__ import annotations
@@ -23,15 +25,20 @@
 from gnss_product_management.factories.remote_transport import WormHole
 from gnss_product_management.factories.search_planner import SearchPlanner
 from gnss_product_management.lockfile.manager import LockfileManager
-from gnss_product_management.lockfile.operations import get_package_version
+from gnss_product_management.lockfile.operations import (
+    HashMismatchMode,
+    get_lock_product,
+    get_package_version,
+    validate_lock_product,
+)
 from gnss_product_management.specifications.dependencies.dependencies import (
     Dependency,
     DependencyResolution,
     DependencySpec,
     ResolvedDependency,
     SearchPreference,
 )
-from gnss_product_management.utilities.paths import AnyPath, as_path
+from gnss_product_management.utilities.paths import AnyPath
 
 logger = logging.getLogger(__name__)
 
@@ -44,8 +51,9 @@ class ResolvePipeline:
     in parallel via a :class:`~concurrent.futures.ThreadPoolExecutor`.
 
     Fast path: if an aggregate lockfile already exists for the
-    ``(package, task, date, version)`` identity, returns immediately
-    without searching or downloading.
+    ``(package, task, date, version)`` identity and every required
+    entry still validates, returns immediately without searching or
+    downloading; otherwise re-resolves.
 
     Args:
         env: The product registry with built catalogs.
@@ -119,15 +127,21 @@ def run(
             version=version,
         )
         if existing is not None:
-            logger.info(
-                "Lockfile already exists for %s on %s — skipping resolution: %s",
+            resolution = self._resolution_from_lockfile(existing, spec)
+            if resolution.all_required_fulfilled:
+                logger.info(
+                    "Lockfile already exists for %s on %s — skipping resolution: %s",
+                    spec.name,
+                    date.date(),
+                    lf_path,
+                )
+                logger.info(resolution.summary())
+                return resolution, lf_path
+            logger.warning(
+                "Lockfile for %s on %s has missing or invalid entries — re-resolving",
                 spec.name,
                 date.date(),
-                lf_path,
             )
-            resolution = self._resolution_from_lockfile(existing, spec)
-            logger.info(resolution.summary())
-            return resolution, lf_path
 
         # --- Full resolution -------------------------------------------------
         resolve_one = partial(
@@ -227,6 +241,29 @@ def _resolve_one(
             remote_url=found.uri,
         )
 
+    @staticmethod
+    def _lockfile_entry_is_valid(lp) -> bool:
+        """Check a lockfile entry's sink file: existence, then sidecar hash.
+
+        Aggregate lockfile entries carry no hash of their own, so after
+        the existence check the per-file sidecar ``_lock.json`` (written
+        at download time) provides the hash to validate against.  A sink
+        with no sidecar is only checked for existence.
+
+        Args:
+            lp: The :class:`LockProduct` entry from the aggregate lockfile.
+
+        Returns:
+            ``True`` if the sink file exists and matches its recorded hash.
+        """
+        # Also resolves .gz sinks to their decompressed file, mutating lp.sink.
+        if not validate_lock_product(lp, mode=HashMismatchMode.STRICT):
+            return False
+        sidecar = get_lock_product(lp.sink)
+        if sidecar is None:
+            return True
+        return validate_lock_product(sidecar, mode=HashMismatchMode.STRICT)
+
     def _resolution_from_lockfile(
         self,
         existing,
@@ -235,8 +272,8 @@ def _resolution_from_lockfile(
         """Reconstruct a :class:`DependencyResolution` from an existing lockfile.
 
         Iterates over every dependency in the spec (not just those in
-        the lockfile), marking any absent or file-missing entries as
-        ``'missing'``.
+        the lockfile), marking any absent, file-missing, or
+        hash-mismatched entries as ``'missing'``.
 
         Args:
             existing: The loaded :class:`DependencyLockFile`.
@@ -254,10 +291,9 @@ def _resolution_from_lockfile(
                     ResolvedDependency(spec=dep.spec, required=dep.required, status="missing")
                 )
                 continue
-            sink_path = as_path(lp.sink) if lp.sink else None
-            if sink_path is None or not sink_path.exists():
+            if not lp.sink or not self._lockfile_entry_is_valid(lp):
                 logger.warning(
-                    "Lockfile entry for %s points to missing file %s — will re-resolve on next run",
+                    "Lockfile entry for %s points to a missing or corrupt file %s",
                     dep.spec,
                     lp.sink,
                 )

packages/gnss-product-management/src/gnss_product_management/factories/remote_transport.py

@@ -19,7 +19,7 @@
 )
 from gnss_product_management.specifications.products.product import PathTemplate
 from gnss_product_management.specifications.remote.resource import SearchTarget
-from gnss_product_management.utilities.helpers import decompress_gzip
+from gnss_product_management.utilities.helpers import decompress_gzip, hash_file
 from gnss_product_management.utilities.paths import AnyPath, as_path
 
 logger = logging.getLogger(__name__)
@@ -261,25 +261,34 @@ def _update_parameters(self, search_target: SearchTarget) -> SearchTarget:
         return updated
 
     @staticmethod
-    def _validate_cached_or_evict(path: AnyPath) -> bool:
-        """Validate a cached file against its sidecar lockfile hash.
+    def _validate_cached_or_evict(path: AnyPath, checksum: str | None = None) -> bool:
+        """Validate a cached file against a declared or sidecar hash.
 
-        Returns ``True`` when the file can be trusted: either its SHA-256
-        matches the sidecar ``_lock.json`` written at download time, or no
-        sidecar exists to validate against.  On a hash mismatch the file
-        and its stale sidecar are deleted so the caller re-downloads.
+        When *checksum* is given (declared in the resource spec) it is
+        authoritative: the file's SHA-256 must match it, regardless of
+        what the sidecar says — this catches caches poisoned before the
+        sidecar was written.  Otherwise the file is validated against
+        the sidecar ``_lock.json`` written at download time; a file with
+        no sidecar is trusted.  On mismatch the file and its sidecar are
+        deleted so the caller re-downloads.
 
         Args:
             path: Existing cached file to validate.
+            checksum: Expected ``sha256:<hex>`` from the resource spec.
 
         Returns:
             ``True`` if the cached file is usable, ``False`` if it was
             evicted and must be re-downloaded.
         """
-        lock = get_lock_product(path)
-        if lock is None or validate_lock_product(lock, mode=HashMismatchMode.STRICT):
-            return True
-        logger.warning("Cached file failed hash validation, evicting: %s", path)
+        if checksum is not None:
+            if hash_file(path) == checksum:
+                return True
+            logger.warning("Cached file does not match declared checksum, evicting: %s", path)
+        else:
+            lock = get_lock_product(path)
+            if lock is None or validate_lock_product(lock, mode=HashMismatchMode.STRICT):
+                return True
+            logger.warning("Cached file failed hash validation, evicting: %s", path)
         as_path(str(path) + "_lock.json").unlink(missing_ok=True)
         path.unlink(missing_ok=True)
         return False
@@ -294,11 +303,12 @@ def download_one(
         """Synchronously download matched files for one search target.
 
         Skips the download if the destination file already exists,
-        is non-empty, and matches its sidecar lockfile hash (when one
-        exists) — a corrupt cached file is evicted and re-downloaded.
-        Downloaded files are verified against the remote size (when the
-        server reports one) and retried once on mismatch, so truncated
-        transfers are never cached.
+        is non-empty, and matches the spec-declared checksum (when the
+        resource spec pins one) or its sidecar lockfile hash — a corrupt
+        cached file is evicted and re-downloaded.  Downloaded files are
+        verified against the remote size (when the server reports one)
+        and the declared checksum, retried once on mismatch, so
+        truncated or corrupt transfers are never cached.
 
         Args:
             query: The resolved search target with filename value.
@@ -334,11 +344,13 @@ def download_one(
                 return decompressed_path
 
         # Skip download if the file already exists, is non-empty, and
-        # passes sidecar hash validation.
+        # passes checksum/sidecar hash validation.  A declared checksum
+        # describes the file as served, so it only applies to the
+        # non-decompressed destination path.
         if (
             destination_path.exists()
             and destination_path.stat().st_size > 0
-            and self._validate_cached_or_evict(destination_path)
+            and self._validate_cached_or_evict(destination_path, query.checksum)
         ):
             logger.debug("Skipping download, file already exists: %s", destination_path)
             return destination_path
@@ -372,20 +384,26 @@ def download_one(
                 return None
             if result is None:
                 return None
-            if remote_size is None:
-                break
+            # Truncated/corrupt transfers must never be left on disk where
+            # a later run would treat them as satisfied dependencies.
             local_size = result.stat().st_size
-            if local_size == remote_size:
+            if remote_size is not None and local_size != remote_size:
+                logger.warning(
+                    "Size mismatch for %s: %d bytes local vs %d remote — deleting%s",
+                    result,
+                    local_size,
+                    remote_size,
+                    ", retrying" if attempt == 0 else "",
+                )
+            elif query.checksum is not None and hash_file(result) != query.checksum:
+                logger.warning(
+                    "Checksum mismatch for %s: expected %s — deleting%s",
+                    result,
+                    query.checksum,
+                    ", retrying" if attempt == 0 else "",
+                )
+            else:
                 break
-            # Truncated/corrupt transfer: never leave it on disk where a
-            # later run would treat it as a satisfied dependency.
-            logger.warning(
-                "Size mismatch for %s: %d bytes local vs %d remote — deleting%s",
-                result,
-                local_size,
-                remote_size,
-                ", retrying" if attempt == 0 else "",
-            )
             result.unlink(missing_ok=True)
             result = None
 

packages/gnss-product-management/src/gnss_product_management/specifications/remote/resource.py

@@ -45,6 +45,11 @@ class ResourceProductSpec(BaseModel):
         description: Human-readable description.
         parameters: Parameter overrides (values pinned by this center).
         directory: Directory template for this product.
+        checksum: Expected ``sha256:<hex>`` of the served file, as
+            downloaded (i.e. before any decompression).  Only meaningful
+            for entries that resolve to exactly one file — static tables
+            pinned to an immutable source.  When set, downloads and cache
+            hits are validated against it.
     """
 
     id: str
@@ -55,6 +60,7 @@ class ResourceProductSpec(BaseModel):
     description: str | None = None
     parameters: list[Parameter]
     directory: PathTemplate
+    checksum: str | None = None
 
 
 class ResourceSpec(BaseModel):
@@ -98,11 +104,15 @@ class SearchTarget(BaseModel):
         product: The product being queried.
         server: The server endpoint to query.
         directory: Directory path template for the product.
+        checksum: Expected ``sha256:<hex>`` of the served file, carried
+            from the resource spec.  ``None`` when the source declares no
+            checksum.
     """
 
     product: Product
     server: Server
     directory: PathTemplate
+    checksum: str | None = None
 
     def narrow(self) -> "SearchTarget":
         """Substitute already-known parameter values into directory/filename patterns.

packages/gnss-product-management/src/gnss_product_management/specifications/remote/resource_catalog.py

@@ -123,6 +123,7 @@ def build(cls, resource_spec: ResourceSpec, product_catalog) -> "ResourceCatalog
                                 product=pinned_product,
                                 server=server,
                                 directory=rp_spec.directory,
+                                checksum=rp_spec.checksum,
                             ).narrow()
                         )