Reject alignment source dates that disagree with frozen inventories

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.

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:

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

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:

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,

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"

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"