"""Content-hash helpers for integrity / tamper verification. One dependency-free module backs two security controls so the *same* hashing code is shared rather than duplicated: - **ADR-0011** (manual-upload safety gate) records ``sha256`` in an upload's provenance at validation time — :func:`compute_sha256` / :func:`verify_sha256`. - **ADR-0010** (dataset integrity on load) refuses a tampered dataset at ``resolve_to_local_path`` time — :func:`verify_file`, which looks up the manifest's recorded baseline via the registry (:func:`expected_sha256_for_url`) and raises :class:`IntegrityError` on mismatch. The registry recorded an expected SHA-256 for each dataset's hosted source file (``integrity.sha256`` in the manifest). The load-time check verifies a materialized file against that baseline *before* it is parsed or cached, so a silently-altered file with the same shape — which ``dataset_validate_manifest_against_data`` cannot catch — is refused rather than analysed. That is a Layer-1-style *refusal* (cf. ADR-0002): a hash mismatch on restricted data is a stop, not a caution. Verification is enforced only where a baseline exists; datasets without one load unverified (trust-on-first-use until pre-staging anchors an OHSU-controlled copy — ADR-0007 Phase 4). All hashing is streamed so a multi-GB h5ad is never read into memory at once. """ from __future__ import annotations import hashlib from pathlib import Path # 4 MiB — matches the download chunk size used in src/core/data_io.py. _DEFAULT_CHUNK = 4 * 1024 * 1024 # --------------------------------------------------------------------------- # Low-level primitives (ADR-0011 upload gate + ADR-0010 on-load verify) # --------------------------------------------------------------------------- def compute_sha256(path: str | Path, chunk_size: int = _DEFAULT_CHUNK) -> str: """Return the hex SHA-256 of the file at ``path``, read in chunks. Raises ``FileNotFoundError`` if the path does not exist (callers that stage a file should have materialized it first). """ h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(chunk_size), b""): h.update(chunk) return h.hexdigest() def verify_sha256(path: str | Path, expected: str, chunk_size: int = _DEFAULT_CHUNK) -> bool: """Return True iff the file at ``path`` hashes to ``expected`` (case-insensitive). A missing/empty ``expected`` returns False — an absent baseline is not a pass. (ADR-0010's :func:`verify_file` handles "no baseline recorded" as a distinct, explicit case at its call site; this helper only answers "does it match".) """ if not expected: return False return compute_sha256(path, chunk_size).lower() == str(expected).strip().lower() # --------------------------------------------------------------------------- # On-load verification (ADR-0010): manifest-baselined refusal # --------------------------------------------------------------------------- class IntegrityError(RuntimeError): """Raised when a loaded file's SHA-256 does not match the manifest baseline. Refuses the load: the file is never handed to the analysis tools or the cache. """ def __init__( self, url: str, expected: str, actual: str, dataset_id: str | None = None, ) -> None: self.url = url self.expected = expected self.actual = actual self.dataset_id = dataset_id name = f" for dataset '{dataset_id}'" if dataset_id else "" super().__init__( f"Integrity check FAILED{name}: content of {url} does not match the " f"recorded baseline. Expected sha256={expected}, got sha256={actual}. " f"Refusing to load — the file may have been modified or replaced " f"(ADR-0010)." ) def _lookup_from_registry(url: str) -> tuple[str | None, str | None]: """Return ``(expected_sha256, dataset_id)`` for a source URL from the registry. Iterates the installed ``biodata_registry`` manifests and asks each for the hash matching ``url`` (via ``DatasetManifest.expected_sha256_for_url``). Any import/registry failure degrades to ``(None, None)`` so a missing or broken registry never blocks a load — it just means "no baseline". """ try: from biodata_registry import get_registry from biodata_registry.manifest_schema import DatasetManifest except Exception: return None, None try: registry = get_registry() for dataset_id in registry.list(): raw = registry.get(dataset_id) if not raw: continue try: manifest = DatasetManifest.from_dict(raw) except Exception: continue # expected_sha256_for_url is only present on biodata-registry >= 0.1.9; # getattr keeps this working against an older pinned wheel (→ no baseline). lookup = getattr(manifest, "expected_sha256_for_url", None) if lookup is None: continue expected = lookup(url) if expected: return expected, dataset_id except Exception: return None, None return None, None def expected_sha256_for_url(url: str) -> str | None: """Return the recorded SHA-256 baseline for a source URL, or None if unbaselined.""" expected, _ = _lookup_from_registry(url) return expected def verify_file(local_path: str | Path, url: str, *, dataset_id: str | None = None) -> None: """Verify a materialized file against its manifest SHA-256 baseline. No-op when the URL has no recorded baseline (returns silently). Raises :class:`IntegrityError` on mismatch. ``dataset_id`` is looked up from the registry when not supplied, purely to enrich the error message. """ expected, found_id = _lookup_from_registry(url) if not expected: return actual = compute_sha256(local_path) if actual.lower() != str(expected).strip().lower(): raise IntegrityError( url=url, expected=expected, actual=actual, dataset_id=dataset_id or found_id, ) def verify_bytes(data: bytes, url: str, *, dataset_id: str | None = None) -> None: """Verify an in-memory download against its manifest SHA-256 baseline. The bytes counterpart to :func:`verify_file`, for loaders that hold a source in memory rather than on disk (e.g. ``load_geo_series_matrix_lines``, which reads the raw — still-compressed — series-matrix response before parsing). ``data`` must be the source bytes exactly as served (pre-decompression), so it matches how the baseline was recorded from the same URL. No-op when the URL has no recorded baseline (returns silently). Raises :class:`IntegrityError` on mismatch. """ expected, found_id = _lookup_from_registry(url) if not expected: return actual = hashlib.sha256(data).hexdigest() if actual.lower() != str(expected).strip().lower(): raise IntegrityError( url=url, expected=expected, actual=actual, dataset_id=dataset_id or found_id, )