From 4d988012ecb4329783b67adfab26130e0002824c Mon Sep 17 00:00:00 2001 From: Alex Rubinsteyn Date: Tue, 6 Oct 2026 17:26:43 -0400 Subject: [PATCH] Enforce frozen inventory modification dates during alignment access --- docs/alignment-inventory-time-spec.md | 28 +++++ docs/reads.md | 15 +++ osteosarc/cache.py | 29 +++++ osteosarc/dataset.py | 11 +- osteosarc/reads.py | 13 +- pyproject.toml | 2 +- tests/test_alignment_inventory_time.py | 159 +++++++++++++++++++++++++ 7 files changed, 246 insertions(+), 11 deletions(-) create mode 100644 docs/alignment-inventory-time-spec.md create mode 100644 tests/test_alignment_inventory_time.py diff --git a/docs/alignment-inventory-time-spec.md b/docs/alignment-inventory-time-spec.md new file mode 100644 index 0000000..a91cc3f --- /dev/null +++ b/docs/alignment-inventory-time-spec.md @@ -0,0 +1,28 @@ +# Enforce snapshot modification times during alignment access + +Issue: https://github.com/iskandr/osteosarc/issues/112 + +1. Share the whole-file downloader's numeric inventory timestamp comparison + with remote alignment header inspection and indexed extraction. + Compare UTC epochs using the existing one-second tolerance. HTTP + Last-Modified has second precision; see RFC 9110 sections + [5.6.7](https://www.rfc-editor.org/rfc/rfc9110.html#section-5.6.7) and + [8.8.2](https://www.rfc-editor.org/rfc/rfc9110.html#section-8.8.2). +2. Reject older/newer remote dates before reading a header or downloading an + index. Validate recorded identities when reopening cached headers and read + derivatives, including old receipts, without network access or rewriting + preserved provenance. Existing request identities and byte pins remain valid. +3. Retain the existing available-evidence policy: a nonnumeric/missing inventory + time or missing server date cannot establish a date match. New alignment + receipts explicitly record `inventory_modification` as `matched`, + `inventory_timestamp_unavailable`, or `last_modified_unavailable`. + Invalid supplied dates or nonfinite numeric inventory times raise + IntegrityError. Local files continue to use their local identity checks. +4. Cover actual indexed BAM access via a local HTTP server, equal/fractional + times, mismatch in both directions, missing evidence, malformed dates, + extraction/recovery entry points and offline legacy cache validation. + Confirm mismatches publish no completed acquisition. Run lint, full tests, + package build and Python/consumer CI. Bump to 0.15.5, merge and publish. + +No biological consequence or read-filter semantics change. This is the source +identity prerequisite for resuming the remaining selected Sid SV acquisitions. diff --git a/docs/reads.md b/docs/reads.md index 90e70c6..2646615 100644 --- a/docs/reads.md +++ b/docs/reads.md @@ -141,6 +141,21 @@ checksums, header, tool versions and read count. Osteosarc checks a remote BAM's HTTP headers before and after reading, to catch a file that changed mid-read. `osteosarc downloads` lists every extract with its local path. +When the source is a `File` from a snapshot, header inspection and regional +extraction also compare its inventory size and numeric modification timestamp +with the server's headers. A date mismatch raises `IntegrityError` before +reading the alignment or fetching its index. Cached headers and subsets are +checked against their recorded source dates too, including older receipts, +without network access or rewriting the saved provenance. + +New receipts record `inventory_modification`: `matched` means both dates were +available and agree within the existing one-second tolerance. HTTP dates have +second precision ([RFC 9110](https://www.rfc-editor.org/rfc/rfc9110.html#section-8.8.2)). +`inventory_timestamp_unavailable` or `last_modified_unavailable` means that date +check could not be performed; other source identity and byte checks still run. +Malformed server dates fail when a numeric inventory timestamp is present. +An arbitrary URL without a snapshot timestamp has no inventory date to verify. + ## Local BAMs The same extraction works on any local indexed BAM: diff --git a/osteosarc/cache.py b/osteosarc/cache.py index 3b62702..355a005 100644 --- a/osteosarc/cache.py +++ b/osteosarc/cache.py @@ -15,6 +15,7 @@ import fcntl import hashlib import json +import math import os import random import secrets @@ -25,6 +26,7 @@ from contextlib import contextmanager from dataclasses import asdict, dataclass from datetime import datetime, timezone +from email.utils import parsedate_to_datetime from pathlib import Path from urllib.parse import unquote, urlsplit @@ -37,6 +39,33 @@ ATTEMPTS = 5 +def _check_inventory_time(modified, last_modified, *, source): + """Compare available inventory/HTTP dates and explicitly identify missing evidence.""" + if type(modified) not in (int, float): + return "inventory_timestamp_unavailable" + try: + finite = math.isfinite(modified) + except OverflowError: + finite = False + if not finite: + raise IntegrityError(f"Invalid inventory modification timestamp for {source}: {modified!r}") + if not last_modified: + return "last_modified_unavailable" + try: + served = parsedate_to_datetime(last_modified) + # HTTP's obsolete asctime format has no timezone, but still means UTC. + if served.tzinfo is None: + served = served.replace(tzinfo=timezone.utc) + served = served.timestamp() + except (TypeError, ValueError, OverflowError, AttributeError) as error: + raise IntegrityError(f"Invalid Last-Modified date for {source}: {last_modified!r}") from error + if abs(served - modified) > 1: + raise IntegrityError( + f"{source} was modified at {last_modified}, not at the time this snapshot's " + f"inventory lists ({modified}); create a new snapshot to use the current object") + return "matched" + + def _transient(error): """Whether a request may succeed if tried again: a server error or too many requests (500, 502, 503, 504, 429), or a dropped connection. Not a refusal diff --git a/osteosarc/dataset.py b/osteosarc/dataset.py index 6f43988..4dd62d6 100644 --- a/osteosarc/dataset.py +++ b/osteosarc/dataset.py @@ -9,12 +9,12 @@ import warnings from collections import Counter, defaultdict from datetime import datetime, timezone -from email.utils import parsedate_to_datetime from functools import cached_property from pathlib import Path, PurePosixPath from urllib.parse import urlsplit from .cache import Cache, Receipt, file_lock, place, stable_id, write_json +from .cache import _check_inventory_time as _inventory_time_status from .catalog import ( BUCKET, SNAPSHOT_SOURCES, @@ -49,14 +49,7 @@ def _check_inventory_time(file, receipt): """Refuse bytes newer (or older) than the object this snapshot's inventory listed.""" - if not isinstance(file.modified, (int, float)) or not receipt.last_modified: - return - served = parsedate_to_datetime(receipt.last_modified).timestamp() - if abs(served - file.modified) > 1: - raise IntegrityError( - f"{file.key} was modified at {receipt.last_modified}, not at the time this snapshot's " - f"inventory lists ({datetime.fromtimestamp(file.modified, timezone.utc).isoformat()}); " - "create a new snapshot to use the current object") + return _inventory_time_status(file.modified, receipt.last_modified, source=file.key) GUIDE = """What's here: diff --git a/osteosarc/reads.py b/osteosarc/reads.py index 0f6e71b..0508392 100644 --- a/osteosarc/reads.py +++ b/osteosarc/reads.py @@ -19,6 +19,7 @@ from .cache import ( ATTEMPTS, Cache, + _check_inventory_time, digest, file_identity, file_lock, @@ -361,6 +362,10 @@ def _verified_receipt(directory, request): receipt = read_read_receipt(path, directory.parent.parent) if receipt["request"] != request: raise IntegrityError("Request differs from its cached receipt") + remote = receipt.get("remote_identity") + if remote is not None: + _check_inventory_time(request.get("source_modified"), remote.get("last-modified"), + source=request["source"]) for filename, expected in receipt["files"].items(): if (Path(filename).name != filename or not (directory / filename).is_file() or digest(directory / filename) != expected): @@ -390,7 +395,7 @@ def inspect_alignment(source, *, cache=None, snapshot_id=None, timeout=600): """Inspect and cache a BAM/CRAM header without requiring an index. Accepts the same sources as extract_reads. Remote headers are pinned on - first inspection for a snapshot, checked against inventory size, and reused + first inspection for a snapshot, checked against inventory size and time, and reused offline. This establishes assembly from contig lengths, not viewer labels. Extraction rejects a remote object that changed since header inspection. """ @@ -413,6 +418,8 @@ def inspect_alignment(source, *, cache=None, snapshot_id=None, timeout=600): if before and file and file.size is not None and before["content-length"] is not None: if int(before["content-length"]) != file.size: raise IntegrityError("Remote alignment size differs from the pinned inventory") + inventory_modification = _check_inventory_time( + request["source_modified"], before.get("last-modified"), source=location) if remote else None identity = None if remote else file_identity(location) command = ["samtools", "view", "--no-PG", "-H", location] header_text = _run(command, timeout).stdout.decode() @@ -425,6 +432,7 @@ def inspect_alignment(source, *, cache=None, snapshot_id=None, timeout=600): work = Path(temporary) (work / "header.sam").write_text(header_text) receipt = dict(request=request, files={"header.sam": digest(work / "header.sam")}, + inventory_modification=inventory_modification, remote_identity=before, command=_recorded_command(command, work, cache), samtools_version=_samtools_version()) write_json(work / "receipt.json", receipt) @@ -542,6 +550,8 @@ def extract_reads(source, regions, *, cache=None, index=None, filters=None, refe if before and file and file.size is not None and before["content-length"] is not None: if int(before["content-length"]) != file.size: raise IntegrityError("Remote alignment size differs from the pinned inventory") + inventory_modification = _check_inventory_time( + request["source_modified"], before.get("last-modified"), source=location) if remote else None if remote and before != info.receipt["remote_identity"]: raise IntegrityError("Remote alignment changed since header inspection; use a new snapshot") index_receipt = None @@ -660,6 +670,7 @@ def acquire(): files = {name: digest(work / name) for name in ("reads.bam", "reads.bam.bai", "header.sam", "regions.bed", "mates.bed", "templates.txt") if (work / name).exists()} receipt = dict(request=request, files=files, records=count, + inventory_modification=inventory_modification, resolved_regions=[asdict(r) for r in resolved], remote_identity=before, header_receipt=info.receipt, diff --git a/pyproject.toml b/pyproject.toml index 031616d..2ea950c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "osteosarc" -version = "0.15.4" +version = "0.15.5" description = "Reproducible access to the public osteosarc.com dataset" readme = "README.md" requires-python = ">=3.9" diff --git a/tests/test_alignment_inventory_time.py b/tests/test_alignment_inventory_time.py new file mode 100644 index 0000000..05f4b55 --- /dev/null +++ b/tests/test_alignment_inventory_time.py @@ -0,0 +1,159 @@ +from collections import Counter +from dataclasses import replace +from email.utils import formatdate +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from threading import Thread + +import pytest + +from osteosarc import Cache, File, IntegrityError, RecoveryPolicy, Region, extract_reads +from osteosarc.cache import _check_inventory_time, stable_id, write_json +from osteosarc.read_receipts import write_read_receipt +from osteosarc.reads import inspect_alignment + +EPOCH = 1784586023 +REGIONS = [Region("chr1", 100, 160, "GRCh38")] + + +@pytest.fixture +def remote_bam(bam): + """Serve actual BAM bytes with a controllable HTTP modification date.""" + state = dict(last_modified=formatdate(EPOCH, usegmt=True), requests=[]) + + class Handler(SimpleHTTPRequestHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, directory=str(bam.parent), **kwargs) + + def send_header(self, keyword, value): + if keyword.lower() == "last-modified": + value = state["last_modified"] + if value is None: + return + super().send_header(keyword, value) + + def do_HEAD(self): + state["requests"].append("HEAD") + super().do_HEAD() + + def do_GET(self): + state["requests"].append("GET") + super().do_GET() + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + source = File("remote", "input.bam", f"http://127.0.0.1:{server.server_port}/input.bam", + "alignment", "bam", size=bam.stat().st_size, modified=EPOCH) + try: + yield source, bam, state + finally: + server.shutdown() + server.server_close() + thread.join() + + +def acquire(operation, source, bam, cache): + if operation == "inspect": + return inspect_alignment(source, cache=cache, snapshot_id="frozen") + kwargs = dict(cache=cache, index=str(bam) + ".bai", snapshot_id="frozen", max_records=6) + if operation == "recover": + kwargs.pop("max_records") + kwargs["recovery"] = RecoveryPolicy(max_rounds=1, max_records=100) + return extract_reads(source, REGIONS, **kwargs) + + +@pytest.mark.parametrize("operation", ["inspect", "extract", "recover"]) +@pytest.mark.parametrize("offset", [-86400, 86400]) +def test_remote_inventory_date_mismatch_fails_before_reading(remote_bam, tmp_path, operation, offset): + source, bam, state = remote_bam + cache = Cache(tmp_path / "cache") + with pytest.raises(IntegrityError, match="inventory lists.*new snapshot"): + acquire(operation, replace(source, modified=EPOCH + offset), bam, cache) + assert "HEAD" in state["requests"] + assert "GET" not in state["requests"] + assert not list(cache.workspace.rglob("receipt.json")) + + +@pytest.mark.parametrize("offset", [0, 0.5]) +def test_matching_inventory_date_preserves_exact_regional_reads(remote_bam, tmp_path, offset): + import pysam + + source, bam, state = remote_bam + subset = acquire("extract", replace(source, modified=EPOCH + offset), bam, Cache(tmp_path / "cache")) + assert subset.receipt["inventory_modification"] == "matched" + assert subset.receipt["header_receipt"]["inventory_modification"] == "matched" + assert subset.receipt["request"]["source_modified"] == EPOCH + offset + with pysam.AlignmentFile(bam) as original, subset.open() as selected: + expected = Counter(r.to_string() for r in original if r.reference_start < 160) + assert Counter(r.to_string() for r in selected) == expected + assert subset.receipt["records"] == 6 + assert "GET" in state["requests"] + + +@pytest.mark.parametrize("modified,last_modified,status", [ + (None, formatdate(EPOCH, usegmt=True), "inventory_timestamp_unavailable"), + ("unknown", formatdate(EPOCH, usegmt=True), "inventory_timestamp_unavailable"), + (EPOCH, None, "last_modified_unavailable"), +]) +def test_missing_timestamp_evidence_is_explicit(remote_bam, tmp_path, modified, last_modified, status): + source, bam, state = remote_bam + state["last_modified"] = last_modified + subset = acquire("extract", replace(source, modified=modified), bam, Cache(tmp_path / "cache")) + assert subset.receipt["inventory_modification"] == status + assert subset.receipt["header_receipt"]["inventory_modification"] == status + assert subset.receipt["remote_identity"]["last-modified"] == last_modified + + +def test_invalid_server_date_is_an_integrity_error(remote_bam, tmp_path): + source, bam, state = remote_bam + state["last_modified"] = "not a date" + cache = Cache(tmp_path / "cache") + with pytest.raises(IntegrityError, match="Invalid Last-Modified"): + acquire("extract", source, bam, cache) + assert "GET" not in state["requests"] + assert not list(cache.workspace.rglob("receipt.json")) + + +@pytest.mark.parametrize("operation", ["inspect", "extract"]) +@pytest.mark.parametrize("legacy", [False, True]) +def test_offline_cached_date_mismatch_is_rejected(remote_bam, tmp_path, monkeypatch, operation, legacy): + import osteosarc.reads as reads + + source, bam, state = remote_bam + cache = Cache(tmp_path / "cache") + result = acquire(operation, source, bam, cache) + receipt = result.receipt + directory = result.path.parent + # Model a result accepted by the old implementation under a mismatched inventory. + receipt["request"]["source_modified"] += 86400 + if legacy: + receipt.pop("inventory_modification") + if operation == "inspect": + write_json(directory / "receipt.json", receipt) + else: + write_read_receipt(directory / "receipt.json", receipt, cache.workspace) + directory.rename(directory.parent / stable_id(receipt["request"])) + monkeypatch.setattr(reads, "_remote_identity", lambda *a, **k: pytest.fail("offline HTTP access")) + monkeypatch.setattr(reads, "_run", lambda *a, **k: pytest.fail("offline SAMtools execution")) + with pytest.raises(IntegrityError, match="inventory lists.*new snapshot"): + acquire(operation, replace(source, modified=EPOCH + 86400), bam, Cache(cache.root, offline=True)) + + +def test_size_check_still_precedes_reading(remote_bam, tmp_path): + source, bam, state = remote_bam + with pytest.raises(IntegrityError, match="size differs"): + acquire("inspect", replace(source, size=source.size + 1), bam, Cache(tmp_path / "cache")) + assert "GET" not in state["requests"] + + +@pytest.mark.parametrize("modified", [float("nan"), float("inf"), 10 ** 1000]) +def test_invalid_numeric_inventory_times_are_rejected(modified): + with pytest.raises(IntegrityError, match="Invalid inventory modification timestamp"): + _check_inventory_time(modified, formatdate(EPOCH, usegmt=True), source="source.bam") + + +def test_obsolete_http_date_is_interpreted_as_utc(): + assert _check_inventory_time(EPOCH, "Mon Jul 20 22:20:23 2026", source="source.bam") == "matched"