Validate download integrity: evict corrupt cache hits, verify downloaded sizes

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

@@ -12,6 +12,11 @@
 
 from gnss_product_management.environments import WorkSpace
 from gnss_product_management.factories.connection_pool import ConnectionPoolFactory
+from gnss_product_management.lockfile.operations import (
+    HashMismatchMode,
+    get_lock_product,
+    validate_lock_product,
+)
 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
@@ -255,6 +260,30 @@ 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.
+
+        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.
+
+        Args:
+            path: Existing cached file to validate.
+
+        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)
+        as_path(str(path) + "_lock.json").unlink(missing_ok=True)
+        path.unlink(missing_ok=True)
+        return False
+
     def download_one(
         self,
         query: SearchTarget,
@@ -264,8 +293,12 @@ def download_one(
     ) -> AnyPath | None:
         """Synchronously download matched files for one search target.
 
-        Skips the download if the destination file already exists and
-        is non-empty.
+        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.
 
         Args:
             query: The resolved search target with filename value.
@@ -289,15 +322,24 @@ def download_one(
         # Prefer an already-decompressed version on disk
         if destination_path.suffix == ".gz":
             decompressed_path = destination_path.with_suffix("")
-            if decompressed_path.exists() and decompressed_path.stat().st_size > 0:
+            if (
+                decompressed_path.exists()
+                and decompressed_path.stat().st_size > 0
+                and self._validate_cached_or_evict(decompressed_path)
+            ):
                 logger.debug(
                     "Skipping download, decompressed file already exists: %s",
                     decompressed_path,
                 )
                 return decompressed_path
 
-        # Skip download if the file already exists and is non-empty
-        if destination_path.exists() and destination_path.stat().st_size > 0:
+        # Skip download if the file already exists, is non-empty, and
+        # passes sidecar hash validation.
+        if (
+            destination_path.exists()
+            and destination_path.stat().st_size > 0
+            and self._validate_cached_or_evict(destination_path)
+        ):
             logger.debug("Skipping download, file already exists: %s", destination_path)
             return destination_path
 
@@ -311,24 +353,47 @@ def download_one(
             logger.warning("Skipping zero-byte remote file: %s/%s", hostname, remote_file_path)
             return None
 
-        try:
-            result = self._connection_pool_factory.download_file(
-                hostname=hostname,
-                remote_path=remote_file_path,
-                target_dir=destination_dir,
-            )
-        except Exception as e:
+        result: Path | None = None
+        for attempt in range(2):
+            try:
+                result = self._connection_pool_factory.download_file(
+                    hostname=hostname,
+                    remote_path=remote_file_path,
+                    target_dir=destination_dir,
+                )
+            except Exception as e:
+                logger.warning(
+                    "Download failed for %s/%s/%s: %s",
+                    hostname,
+                    query.directory.value,
+                    query.product.filename.value,
+                    e,
+                )
+                return None
+            if result is None:
+                return None
+            if remote_size is None:
+                break
+            local_size = result.stat().st_size
+            if local_size == remote_size:
+                break
+            # Truncated/corrupt transfer: never leave it on disk where a
+            # later run would treat it as a satisfied dependency.
             logger.warning(
-                "Download failed for %s/%s/%s: %s",
-                hostname,
-                query.directory.value,
-                query.product.filename.value,
-                e,
+                "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
+
+        if result is None:
             return None
 
         # Decompress gzip files after download
-        if result is not None and result.suffix == ".gz":
+        if result.suffix == ".gz":
             decompressed = decompress_gzip(result)
             if decompressed is not None:
                 return decompressed

packages/gnss-product-management/test/test_download_integrity.py

@@ -0,0 +1,235 @@
+"""
+Tests: WormHole.download_one integrity validation.
+
+All tests are local-only (no network access) — the "remote" server is a
+directory on disk served through the ``file`` protocol, exercising the
+real ConnectionPoolFactory download path.
+
+Covers issue #26 symptom (c): truncated/corrupt downloads must never be
+cached and reused as satisfied dependencies.
+"""
+
+from __future__ import annotations
+
+import datetime
+import gzip
+from pathlib import Path
+
+import pytest
+from gnss_product_management.factories.connection_pool import ConnectionPoolFactory
+from gnss_product_management.factories.remote_transport import WormHole
+from gnss_product_management.lockfile.operations import (
+    build_lock_product,
+    get_lock_product_path,
+    write_lock_product,
+)
+from gnss_product_management.specifications.products.product import PathTemplate, Product
+from gnss_product_management.specifications.remote.resource import SearchTarget, Server
+
+# ── Constants ─────────────────────────────────────────────────────
+
+TEST_DATE = datetime.datetime(2025, 1, 15, tzinfo=datetime.timezone.utc)
+REMOTE_CONTENT = b"full remote product content, definitely not truncated"
+
+
+# ── Helpers ───────────────────────────────────────────────────────
+
+
+class _StubWorkspace:
+    """Minimal stand-in for WorkSpace.sink_product."""
+
+    def __init__(self, sink_root: Path):
+        self._sink_root = sink_root
+
+    def sink_product(self, product: Product, resource_id: str, date) -> SearchTarget:
+        return SearchTarget(
+            product=product,
+            server=Server(id="local_sink", hostname=str(self._sink_root)),
+            directory=PathTemplate(pattern="sink", value="sink"),
+        )
+
+
+def _make_query(remote_root: Path, filename: str) -> SearchTarget:
+    product = Product(
+        name="ORBIT",
+        parameters=[],
+        filename=PathTemplate(pattern=filename, value=filename),
+    )
+    return SearchTarget(
+        product=product,
+        server=Server(id="remote", hostname=str(remote_root)),
+        directory=PathTemplate(pattern="products", value="products"),
+    )
+
+
+@pytest.fixture
+def env(tmp_path: Path):
+    """A local 'remote' server dir, a sink dir, a WormHole, and a query."""
+    remote_root = tmp_path / "remote"
+    remote_dir = remote_root / "products"
+    remote_dir.mkdir(parents=True)
+    (remote_dir / "TEST.SP3").write_bytes(REMOTE_CONTENT)
+
+    sink_root = tmp_path / "workspace"
+    wh = WormHole()
+    wh._connection_pool_factory.add_connection(str(remote_root))
+
+    return {
+        "remote_root": remote_root,
+        "remote_dir": remote_dir,
+        "workspace": _StubWorkspace(sink_root),
+        "sink_dir": sink_root / "sink",
+        "wormhole": wh,
+    }
+
+
+def _download(env, filename: str = "TEST.SP3") -> Path | None:
+    query = _make_query(env["remote_root"], filename)
+    return env["wormhole"].download_one(
+        query=query,
+        local_resource_id="local_config",
+        local_factory=env["workspace"],
+        date=TEST_DATE,
+    )
+
+
+def _write_sidecar(path: Path) -> None:
+    write_lock_product(build_lock_product(sink=path, url="file://test", name="ORBIT"))
+
+
+# ── Fresh downloads ───────────────────────────────────────────────
+
+
+class TestFreshDownload:
+    def test_download_succeeds(self, env) -> None:
+        result = _download(env)
+        assert result is not None
+        assert result.read_bytes() == REMOTE_CONTENT
+
+    def test_zero_byte_remote_is_skipped(self, env) -> None:
+        (env["remote_dir"] / "TEST.SP3").write_bytes(b"")
+        assert _download(env) is None
+
+
+# ── Truncated transfers ───────────────────────────────────────────
+
+
+class TestTruncatedDownload:
+    def test_truncated_download_is_deleted_not_cached(self, env, monkeypatch) -> None:
+        """A transfer that keeps coming back short must fail — and leave
+        nothing on disk for a later run to treat as satisfied."""
+
+        def _truncated(self, hostname, remote_path, target_dir):
+            local = Path(target_dir) / Path(remote_path).name
+            local.write_bytes(REMOTE_CONTENT[: len(REMOTE_CONTENT) // 2])
+            return local
+
+        monkeypatch.setattr(ConnectionPoolFactory, "download_file", _truncated)
+        assert _download(env) is None
+        assert not (env["sink_dir"] / "TEST.SP3").exists()
+
+    def test_truncated_download_retries_and_succeeds(self, env, monkeypatch) -> None:
+        calls: list[int] = []
+
+        def _flaky(self, hostname, remote_path, target_dir):
+            calls.append(1)
+            local = Path(target_dir) / Path(remote_path).name
+            content = REMOTE_CONTENT if len(calls) > 1 else REMOTE_CONTENT[:10]
+            local.write_bytes(content)
+            return local
+
+        monkeypatch.setattr(ConnectionPoolFactory, "download_file", _flaky)
+        result = _download(env)
+        assert result is not None
+        assert result.read_bytes() == REMOTE_CONTENT
+        assert len(calls) == 2
+
+
+# ── Cached files ──────────────────────────────────────────────────
+
+
+class TestCachedFile:
+    def test_valid_cache_is_reused_without_download(self, env, monkeypatch) -> None:
+        cached = env["sink_dir"] / "TEST.SP3"
+        cached.parent.mkdir(parents=True)
+        cached.write_bytes(REMOTE_CONTENT)
+        _write_sidecar(cached)
+
+        calls: list[int] = []
+        monkeypatch.setattr(
+            ConnectionPoolFactory,
+            "download_file",
+            lambda self, **kw: calls.append(1),
+        )
+        assert _download(env) == cached
+        assert calls == []
+
+    def test_cache_without_sidecar_is_trusted(self, env, monkeypatch) -> None:
+        cached = env["sink_dir"] / "TEST.SP3"
+        cached.parent.mkdir(parents=True)
+        cached.write_bytes(b"pre-existing content, no sidecar")
+
+        calls: list[int] = []
+        monkeypatch.setattr(
+            ConnectionPoolFactory,
+            "download_file",
+            lambda self, **kw: calls.append(1),
+        )
+        assert _download(env) == cached
+        assert calls == []
+
+    def test_corrupt_cache_is_evicted_and_redownloaded(self, env) -> None:
+        """A cached file whose hash no longer matches its sidecar must be
+        evicted (with its stale sidecar) and fetched fresh."""
+        cached = env["sink_dir"] / "TEST.SP3"
+        cached.parent.mkdir(parents=True)
+        cached.write_bytes(REMOTE_CONTENT)
+        _write_sidecar(cached)
+        cached.write_bytes(b"corrupted after the sidecar was written")
+
+        result = _download(env)
+        assert result == cached
+        assert result.read_bytes() == REMOTE_CONTENT
+        assert not get_lock_product_path(cached).exists()
+
+
+# ── Gzip cache path ───────────────────────────────────────────────
+
+
+class TestGzipCache:
+    @pytest.fixture
+    def gz_env(self, env):
+        (env["remote_dir"] / "ATT.OBX.gz").write_bytes(gzip.compress(REMOTE_CONTENT))
+        return env
+
+    def test_gz_download_decompresses(self, gz_env) -> None:
+        result = _download(gz_env, filename="ATT.OBX.gz")
+        assert result is not None
+        assert result.name == "ATT.OBX"
+        assert result.read_bytes() == REMOTE_CONTENT
+
+    def test_corrupt_decompressed_cache_is_evicted_and_redownloaded(self, gz_env) -> None:
+        cached = gz_env["sink_dir"] / "ATT.OBX"
+        cached.parent.mkdir(parents=True)
+        cached.write_bytes(REMOTE_CONTENT)
+        _write_sidecar(cached)
+        cached.write_bytes(b"truncated OBX poisoning the cache")
+
+        result = _download(gz_env, filename="ATT.OBX.gz")
+        assert result == cached
+        assert result.read_bytes() == REMOTE_CONTENT
+
+    def test_valid_decompressed_cache_is_reused(self, gz_env, monkeypatch) -> None:
+        cached = gz_env["sink_dir"] / "ATT.OBX"
+        cached.parent.mkdir(parents=True)
+        cached.write_bytes(REMOTE_CONTENT)
+        _write_sidecar(cached)
+
+        calls: list[int] = []
+        monkeypatch.setattr(
+            ConnectionPoolFactory,
+            "download_file",
+            lambda self, **kw: calls.append(1),
+        )
+        assert _download(gz_env, filename="ATT.OBX.gz") == cached
+        assert calls == []