diff --git a/packages/gnss-product-management/src/gnss_product_management/factories/remote_transport.py b/packages/gnss-product-management/src/gnss_product_management/factories/remote_transport.py index e46fd54..5932c20 100644 --- a/packages/gnss-product-management/src/gnss_product_management/factories/remote_transport.py +++ b/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 diff --git a/packages/gnss-product-management/test/test_download_integrity.py b/packages/gnss-product-management/test/test_download_integrity.py new file mode 100644 index 0000000..cb6470d --- /dev/null +++ b/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 == []