feat(uploads): ADR-0011 manual-upload safety gate (Now slice)

Standalone, non-agent-facing upload gate (quarantine -> validate -> admin register). Inert until wired to a UI: no MCP tool, no server.py change, no new dependency. See docs/adr/ADR-0011-upload-safety-gate.md. 24 new tests green. This PR does NOT deploy until merged into main.

.gitignore

@@ -10,3 +10,7 @@ tests/results/
 tmp/
 outputs/
 reports/
+# ADR-0011 upload safety gate — local quarantine / promoted-manifest dirs
+upload_staging/
+upload_registered/
+run_logs/

TODO.md

@@ -414,6 +414,16 @@ correct (single-dataset, Path B, ULM, 98.3% coverage); these are the gaps.
 
 ## Done (recent)
 
+- 2026-07-02 — **ADR-0011 upload safety gate ("Now" slice)** — branch
+  `feat/adr-0011-upload-gate` (off `origin/main`), **NOT pushed, no deploy**. New non-agent-facing
+  `src/uploads/` (`stage_upload` → `validate_upload` → `register_upload`) + shared
+  `src/core/integrity.py` (`compute_sha256`/`verify_sha256`, reused later by ADR-0010). Quarantine
+  + type/size allow-list + manifest-required + de-id attestation + validate-via-vetted-loader
+  (never-exec) + SHA-256 provenance into the ADR-0008 audit sink + admin-only registration
+  (`UPLOAD_ADMIN_IDS`, fail-closed). `tests/test_upload_gate.py` (24) green; reused suites (133)
+  green. ADR-0011 flipped Proposed→Accepted for this slice. **Follow-ups (open):** tabular
+  auto-validation (h5ad-only today); AWS staging bucket + pre-validation malware scan (ADR-0011
+  "At AWS"); wire `src/core/integrity.py` into `resolve_to_local_path` for ADR-0010 on-load verify.
 - 2026-06-22 — **TF semantic-annotation coverage + igraph network plot**
   (commit `ed98c5b` on `main`, **NOT pushed**). Fixed 70 valid HGNC TF symbols
   being mis-bucketed as `unresolved_label`: bundled `resources/semantic/hgnc_cache.tsv`

docs/adr/ADR-0010-dataset-integrity-verification.md

@@ -0,0 +1,67 @@
+# ADR-0010 — Dataset Integrity / Tamper Verification on Load
+
+**Status:** Proposed
+**Date:** 2026-07-01
+**Deciders:** Annie Voigt (project lead)
+**Driver:** OHSU security review Q7 — "detect unavailable/modified/compromised datasets." Today
+this is only *partial*: manifest validation catches schema/shape drift and load failures, but a
+silently altered file with the same shape would not be caught.
+**Related:** ADR-0007 Phase 4 (pre-stage-with-checksums), `biodata-registry` manifest schema
+(where the expected hash lives).
+
+---
+
+## Context
+
+External datasets are fetched dynamically at analysis time from their declared
+`expression_source` (GEO series matrix, GDC, hosted h5ad) and cached in-process (memory + disk)
+for the container lifetime; local resolution goes through `src/core/data_io.resolve_to_local_path`
+and the loaders (`src/tools/rna/loaders.py`, `src/workflows/geo.py`, `read_h5ad_cached`).
+
+`dataset_validate_manifest_against_data` verifies **semantics** (feature-ID type, sample counts,
+data level) — so a truncated download or a wrong-shape file is caught. What is *not* caught is
+**content tampering that preserves shape**: same dimensions, altered values. For restricted
+research data that is the gap the review is pointing at.
+
+## Decision
+
+Add **content-hash verification on load**, keyed off the manifest.
+
+- **Expected hash in the manifest.** Extend the `biodata_registry` manifest with an optional
+  `integrity:` block: `sha256` (and/or per-file hashes for multi-file sources) recorded when a
+  dataset is first validated/registered. Optional so existing manifests keep loading; verification
+  is enforced only where a hash is present.
+- **Verify at resolve time.** After `resolve_to_local_path` materializes a file (post-download,
+  pre-parse), compute its SHA-256 and compare to the manifest. On mismatch: **refuse to load**,
+  emit a clear integrity error naming the dataset, and never hand the file to the analysis tools
+  or the cache. This is a Layer-1-style *refusal* (cf. ADR-0002): it blocks, because a hash
+  mismatch on restricted data is not a "caution," it's a stop.
+- **Cache is keyed to the verified file.** `read_h5ad_cached` already keys on path+mtime+size;
+  integrity check runs before the cache admit so a bad file is never cached.
+- **Record-on-register.** A small helper computes and writes the hash when a dataset is validated
+  (external sources record on first successful validated fetch, acknowledging trust-on-first-use
+  for third-party sources).
+
+## Plan
+
+1. **Now (AWS-independent, ~2–3 days):** add the `integrity:` manifest field + schema test; add
+   SHA-256 verify in `resolve_to_local_path`; refusal path + test (good hash loads, altered file
+   refused, absent hash = load with a "no integrity baseline" note); backfill hashes for the
+   current registered datasets.
+   *(Available now: the shared hashing helper `src/core/integrity.py`
+   — `compute_sha256` / `verify_sha256`, streamed — landed with ADR-0011 and is the exact code
+   this step wires into `resolve_to_local_path`.)*
+2. **At AWS (ADR-0007 Phase 4):** the pre-stage sync job writes checksums into the read-only OHSU
+   source bucket, so integrity is anchored to an OHSU-controlled copy rather than trust-on-first-use
+   against the public source. This ADR's on-load check is the same code; only the hash's provenance
+   improves.
+
+## Consequences
+
+- **Positive:** closes the Q7 tamper gap with a deterministic control, independent of AWS; dovetails
+  with pre-staging (Phase 4) without rework.
+- **Cost / caveat:** trust-on-first-use for external sources until pre-staging exists — a mid-flight
+  change at the public source before first registration would be baselined as "correct." Pre-staging
+  (Phase 4) is what fully closes this; state that limitation to the review rather than implying the
+  on-load hash alone guarantees upstream authenticity. Hashing large h5ads adds seconds to first
+  load (negligible vs. parse; cached thereafter).

docs/adr/ADR-0011-upload-safety-gate.md

@@ -0,0 +1,78 @@
+# ADR-0011 — Safety Gate for Manual Dataset Uploads
+
+**Status:** Accepted — "Now" (AWS-independent) slice implemented 2026-07-02; AWS pieces deferred (see Plan)
+**Date:** 2026-07-01
+**Deciders:** Annie Voigt (project lead)
+**Driver:** OHSU security review Q4 — researchers should be able to supply their own datasets.
+**This ADR is about the security controls wrapping any such capability, NOT about enabling
+free-form uploads.** An uploaded file is untrusted input; this defines the gate it must clear
+before it is ever loaded or made available.
+**Related:** ADR-0010 (upload records an integrity hash), ADR-0007 (uploaded data is only *read*
+by the vetted toolset / sandbox — never executed), `biodata-registry` (manifest is required),
+Data-Classification Q1–Q3 (uploads must stay de-identified / no PHI).
+
+---
+
+## Context
+
+Every dataset today enters through a **reviewed manifest** in `biodata-registry` +
+`dataset_validate_manifest_against_data`. That gate — semantics, refusal rules, prohibited
+inferences — is exactly what keeps the agent grounded. Any manual-upload feature must not bypass
+it; an uploaded file is untrusted input, and (per Data Classification) must remain de-identified
+with no PHI/PSI.
+
+The design principle: **an upload is not "available" until it has passed the same gate a
+registered dataset passes, plus an admin approval.** The value of this ADR is the gate, not the
+convenience of uploading.
+
+## Decision
+
+Treat uploaded files as quarantined-until-validated, never as executable content, routed through
+the existing manifest/validation machinery.
+
+1. **Quarantine on arrival.** Uploads land in an isolated staging area, not the active data path.
+   Enforce a **type allow-list** (expression matrices / h5ad / supported tabular only) and **size
+   limits**; reject anything else at the door.
+2. **Manifest required.** The uploader must supply (or the UI must capture) a manifest with the
+   same required fields as any `biodata_registry` dataset. No manifest ⇒ not ingestible.
+3. **Validate before availability.** Run `dataset_validate_manifest_against_data` on the staged
+   file. Failure ⇒ stays quarantined, surfaced back to the uploader; it never reaches the tools.
+4. **Never execute uploaded content.** Uploaded files are *data* opened by the vetted
+   decoupleR/scanpy loaders only — the same loaders as registered data, inside the sandbox
+   (ADR-0007) with read-only source mounts. Nothing in an upload is ever `exec`'d or interpreted
+   as code.
+5. **Record integrity + provenance.** On successful validation, compute and store the SHA-256
+   (ADR-0010) and uploader/time metadata, so the file is thereafter tamper-checked like any dataset.
+6. **De-identification attestation.** The upload step requires the uploader to confirm the data is
+   de-identified / contains no PHI/PSI (matching the Data-Classification answers); this is captured
+   in the audit trace (ADR-0008).
+7. **Admin approval to register.** An admin (the 1–2 admin accounts) promotes a validated upload
+   into the active registry. Researchers can stage + validate; only admins register. No shared
+   accounts (consistent with Auth Q6).
+
+## Plan
+
+- **Now (AWS-independent, ~1 week):** staging area + type/size gate + manifest-required flow +
+  validation hook + "never exec" guarantee (it's just a loader path) + integrity record (ADR-0010)
+  + admin-approval step. Testable locally end-to-end with a synthetic upload.
+  **✅ Implemented 2026-07-02** as `src/uploads/` (non-agent-facing Python API):
+  `stage_upload` (attestation → manifest → type allow-list → size → quarantine + SHA-256, via the
+  shared `src/core/integrity.py`), `validate_upload` (manifest-schema + `validate_manifest_against_data`
+  through the vetted `scanpy.read_h5ad` loader only — no `exec`/`eval`/`pickle`), and `register_upload`
+  (admin-only promotion, `UPLOAD_ADMIN_IDS`, into the live registry). Every state transition persists an
+  `UploadRecord` (with the de-id attestation + hash) through the always-on ADR-0008 audit sink.
+  End-to-end tests: `tests/test_upload_gate.py` (24). Auto-validation currently covers `.h5ad`;
+  tabular uploads stage but stay quarantined pending a tabular loader.
+- **At AWS:** staging bucket is a separate scoped prefix with its own encryption; the optional
+  malware scan runs on the staged object *before* validation. The basic type/size/quarantine gate
+  here does **not** require AWS; the malware scan does.
+
+## Consequences
+
+- **Positive:** delivers the requested capability without opening an ingress hole — uploads inherit
+  the same grounding gate as curated datasets, stay de-identified by attestation, are tamper-checked,
+  and are never executable. Clean story for the review.
+- **Cost / caveat:** a manifest requirement adds friction for uploaders (mitigate with a UI that
+  drafts the manifest from the file + a few prompts). Full anti-malware coverage is AWS-dependent;
+  until then the gate is type/size/quarantine/validation/attestation, which should be stated
+  honestly rather than described as malware-scanned.

memory.md

@@ -4,7 +4,45 @@ This file tracks current status, recent decisions, and next steps.
 Update it whenever meaningful work is completed or the direction changes.
 Stable architectural facts belong in `CLAUDE.md`, not here.
 
-Last updated: 2026-07-01
+Last updated: 2026-07-02
+
+---
+
+## 2026-07-02 — ADR-0011 upload safety gate "Now" slice IMPLEMENTED (branch `feat/adr-0011-upload-gate`, NOT pushed — no deploy)
+
+Built the AWS-independent slice of ADR-0011 (manual-dataset-upload safety gate) as a
+**standalone, non-agent-facing** Python package `src/uploads/` + a shared hash helper
+`src/core/integrity.py`. An upload is quarantined-until-validated, never executable, and only an
+admin can register it — the value is the gate, not the upload.
+
+- **Branch/worktree:** built off `origin/main` (`05fca82`, has the ADR-0008 audit sink) in an
+  isolated worktree so the unrelated `feat/sandbox-executor` work (incl. its uncommitted ADR-0009
+  Appendix A edit) was left untouched. The ADR-0010/0011 docs (which only existed on the sandbox
+  branch's `cc0ad55`) were pulled onto this branch so the ADR travels with its implementation.
+- **`src/core/integrity.py`** — `compute_sha256` / `verify_sha256` (streamed). Deliberately the
+  *shared* helper ADR-0010's on-load verification will reuse; ADR-0010 doc updated to point at it.
+- **`src/uploads/`** — `stage_upload` (gate order: de-id attestation → manifest required → type
+  allow-list `.h5ad/.csv/.tsv/.txt` + gz → size limit → isolated per-upload quarantine dir +
+  SHA-256), `validate_upload` (manifest-schema via `validate_manifest` **then**
+  `validate_manifest_against_data` through the vetted `scanpy.read_h5ad` loader only — the
+  "never exec" guarantee is structural: no `exec`/`eval`/`pickle` anywhere, asserted by a test),
+  `register_upload` (admin-only, `UPLOAD_ADMIN_IDS` fail-closed; requires `validated`; writes the
+  manifest into a registered-overlay dir + `get_registry().register`). Every transition persists an
+  `UploadRecord` (de-id attestation + hash + status) through the always-on ADR-0008 log sink, so the
+  attestation lands in the durable audit trail.
+- **Config resolved at call time** (`UPLOAD_STAGING_DIR`, `UPLOAD_REGISTERED_DIR`, `UPLOAD_MAX_BYTES`,
+  `UPLOAD_ADMIN_IDS`) — caught + fixed an import-time-constant bug that made env overrides inert and
+  leaked staging dirs into the repo root; tests are now hermetic (tmp dirs) and the default dirs are
+  gitignored.
+- **Tests:** `tests/test_upload_gate.py` (24) green — type/size/manifest/attestation gates,
+  integrity record + tamper detection, validate happy-path + data-level mismatch stays quarantined,
+  tabular-not-yet-supported, admin refusal (non-admin, and register-before-validate), promotion into
+  the live registry, never-exec source scan, attestation persisted to the audit sink. Reused
+  validator/registry suites still green (133). **Limitation:** auto-validation is h5ad-only this
+  phase (tabular stages but stays quarantined); AWS staging-bucket + malware scan deferred, stated
+  honestly in the ADR.
+- **No runtime/agent surface:** no MCP tool, no `server.py` change, no prompt change, **no Space
+  deploy/rebuild.** Local-only.
 
 ---
 

src/core/integrity.py

@@ -0,0 +1,43 @@
+"""Content-hash helpers for integrity / tamper verification.
+
+A tiny, dependency-free module so the *same* hashing code backs two controls:
+
+- **ADR-0011** (manual-upload safety gate) records ``sha256`` in an upload's
+  provenance at validation time.
+- **ADR-0010** (dataset integrity on load) will reuse :func:`verify_sha256` to
+  refuse a tampered file at ``resolve_to_local_path`` time.
+
+Streamed so a multi-GB h5ad never has to be 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
+
+
+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 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()

src/uploads/__init__.py

@@ -0,0 +1,37 @@
+"""Safety gate for manual dataset uploads (ADR-0011).
+
+Public, non-agent-facing API. An upload is quarantined-until-validated and never
+executable, routed through the same manifest/validation machinery a registered
+dataset passes, plus an admin approval:
+
+    stage_upload(...)      → quarantine + type/size/manifest/attestation gate
+    validate_upload(rec)   → manifest-schema + against-data validation
+    register_upload(rec)   → admin-only promotion into the live registry
+
+Only admins (UPLOAD_ADMIN_IDS) can register; researchers may stage + validate.
+"""
+from src.uploads.promotion import RegistrationRefused, register_upload
+from src.uploads.records import (
+    STATUS_QUARANTINED,
+    STATUS_REGISTERED,
+    STATUS_REJECTED,
+    STATUS_VALIDATED,
+    UploadRecord,
+    is_admin,
+)
+from src.uploads.staging import UploadRejected, stage_upload
+from src.uploads.validation import validate_upload
+
+__all__ = [
+    "stage_upload",
+    "validate_upload",
+    "register_upload",
+    "UploadRecord",
+    "UploadRejected",
+    "RegistrationRefused",
+    "is_admin",
+    "STATUS_REJECTED",
+    "STATUS_QUARANTINED",
+    "STATUS_VALIDATED",
+    "STATUS_REGISTERED",
+]

src/uploads/promotion.py

@@ -0,0 +1,80 @@
+"""Admin approval to register a validated upload (ADR-0011 item 7).
+
+Researchers can stage and validate; only a named admin can make an upload
+*available* to the pipeline. :func:`register_upload` is the promote step: it
+refuses unless the caller is an admin (``UPLOAD_ADMIN_IDS``) AND the upload has
+already reached ``validated``. On success it writes the manifest into a
+registered-overlay directory and registers it with the live
+:class:`DatasetRegistry`, so ``load_manifest(dataset_id)`` resolves thereafter.
+"""
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from pathlib import Path
+
+import yaml
+
+from src.uploads.records import (
+    STATUS_REGISTERED,
+    STATUS_VALIDATED,
+    UploadRecord,
+    is_admin,
+    persist_record,
+)
+
+# Where promoted manifests land. Kept separate from the read-only biodata_registry
+# package dir; the live registry is told about the file via .register(). Read at
+# call time (not import) so env overrides / per-test isolation take effect.
+def _registered_dir() -> str:
+    return os.environ.get("UPLOAD_REGISTERED_DIR", "./upload_registered")
+
+
+class RegistrationRefused(Exception):
+    """Raised when a promote request fails the admin / status precondition."""
+
+
+def register_upload(
+    record: UploadRecord,
+    *,
+    admin: str,
+    manifest_dir: str | Path | None = None,
+) -> UploadRecord:
+    """Promote a validated upload into the active registry. Admins only.
+
+    Raises
+    ------
+    RegistrationRefused
+        If ``admin`` is not in the admin allow-list, or the upload is not yet
+        ``validated``. The record is left unchanged.
+    """
+    # ── Admin gate (item 7) ──────────────────────────────────────────────────
+    if not is_admin(admin):
+        raise RegistrationRefused(
+            f"'{admin}' is not an authorized admin. Only accounts in "
+            "UPLOAD_ADMIN_IDS may register an upload."
+        )
+
+    # ── Must have cleared validation first ───────────────────────────────────
+    if record.status != STATUS_VALIDATED:
+        raise RegistrationRefused(
+            f"Upload '{record.upload_id}' is '{record.status}', not 'validated' — "
+            "it must pass validate_upload before it can be registered."
+        )
+
+    # ── Write the manifest into the registered overlay + register it ─────────
+    out_dir = Path(manifest_dir) if manifest_dir is not None else Path(_registered_dir())
+    out_dir.mkdir(parents=True, exist_ok=True)
+    manifest_path = out_dir / f"{record.dataset_id}.yaml"
+    with open(manifest_path, "w", encoding="utf-8") as fh:
+        yaml.safe_dump(dict(record.manifest), fh, sort_keys=False)
+
+    from src.datasets.registry import get_registry
+
+    get_registry().register(record.dataset_id, manifest_path)
+
+    record.status = STATUS_REGISTERED
+    record.registered_at = datetime.now().isoformat(timespec="seconds")
+    record.registered_manifest_path = str(manifest_path)
+    persist_record(record)
+    return record

src/uploads/records.py

@@ -0,0 +1,117 @@
+"""Provenance record + admin allow-list for the upload safety gate (ADR-0011).
+
+An :class:`UploadRecord` is the single, serialisable audit object that follows a
+manual upload through its whole lifecycle:
+
+    rejected  →  (never staged; failed the door gate)
+    quarantined  →  staged + hashed, not yet validated
+    validated    →  cleared manifest schema + against-data checks
+    registered   →  admin-approved into the active registry
+
+Every state transition calls :func:`persist_record`, which writes ``record.json``
+into the quarantine directory AND pushes the same dict through the always-on
+audit sink (ADR-0008) so the de-identification attestation and integrity hash
+land in the durable audit trail.
+"""
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Optional
+
+# Lifecycle states (kept as plain strings so the record stays JSON-trivial).
+STATUS_REJECTED = "rejected"
+STATUS_QUARANTINED = "quarantined"
+STATUS_VALIDATED = "validated"
+STATUS_REGISTERED = "registered"
+
+VALID_STATUSES = frozenset(
+    {STATUS_REJECTED, STATUS_QUARANTINED, STATUS_VALIDATED, STATUS_REGISTERED}
+)
+
+
+@dataclass
+class UploadRecord:
+    """Audit + provenance record for one manual dataset upload."""
+
+    upload_id: str
+    dataset_id: str
+    uploader: str
+    status: str
+    created_at: str
+
+    original_filename: str = ""
+    staged_path: Optional[str] = None
+    size_bytes: Optional[int] = None
+    sha256: Optional[str] = None
+    deidentified_attestation: bool = False
+
+    validated_at: Optional[str] = None
+    registered_at: Optional[str] = None
+    registered_manifest_path: Optional[str] = None
+
+    errors: list[str] = field(default_factory=list)
+    manifest: dict = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "upload_id": self.upload_id,
+            "dataset_id": self.dataset_id,
+            "uploader": self.uploader,
+            "status": self.status,
+            "created_at": self.created_at,
+            "original_filename": self.original_filename,
+            "staged_path": self.staged_path,
+            "size_bytes": self.size_bytes,
+            "sha256": self.sha256,
+            "deidentified_attestation": self.deidentified_attestation,
+            "validated_at": self.validated_at,
+            "registered_at": self.registered_at,
+            "registered_manifest_path": self.registered_manifest_path,
+            "errors": self.errors,
+            "manifest": self.manifest,
+        }
+
+
+def persist_record(record: UploadRecord) -> None:
+    """Write the record to disk (quarantine dir) and to the audit sink.
+
+    Best-effort: a persistence failure must never mask the gate decision. The
+    on-disk ``record.json`` is only written when the upload has a staged
+    directory (a door-rejected upload never gets one); the audit-sink push
+    always runs so even a rejection is traceable.
+    """
+    # 1) On-disk copy next to the staged file (when there is one).
+    if record.staged_path:
+        try:
+            out_dir = Path(record.staged_path).parent
+            out_dir.mkdir(parents=True, exist_ok=True)
+            with open(out_dir / "record.json", "w", encoding="utf-8") as fh:
+                json.dump(record.to_dict(), fh, indent=2, default=str)
+        except OSError:
+            pass  # disk copy is a convenience; the sink push below is the audit trail
+
+    # 2) Durable audit trail via the always-on log sink (ADR-0008/0009).
+    try:
+        from src.logging_sink import get_log_sink, persist_trace_safe
+
+        persist_trace_safe(get_log_sink(), f"upload_{record.upload_id}", record.to_dict())
+    except Exception:  # noqa: BLE001 — auditing must never crash the gate
+        pass
+
+
+def is_admin(identity: Optional[str]) -> bool:
+    """Whether ``identity`` may promote a validated upload into the registry.
+
+    The admin set is the ``UPLOAD_ADMIN_IDS`` env var — a comma-separated list of
+    the 1–2 named admin accounts (no shared accounts, Auth Q6). Empty/unset means
+    *no one* is an admin (fail closed): registration is impossible until an
+    operator names the admins, which is the correct default for a fresh deploy.
+    """
+    if not identity:
+        return False
+    raw = os.environ.get("UPLOAD_ADMIN_IDS", "")
+    admins = {a.strip() for a in raw.split(",") if a.strip()}
+    return identity.strip() in admins

src/uploads/staging.py

@@ -0,0 +1,199 @@
+"""Quarantine-on-arrival staging + the door gate (ADR-0011 items 1, 2, 5, 6).
+
+An uploaded file is untrusted input. :func:`stage_upload` is the *only* way a
+file enters the system, and it clears four gates before the file is written
+anywhere the pipeline can see it:
+
+    1. de-identification attestation present   (item 6)
+    2. manifest supplied                        (item 2)
+    3. type allow-list                          (item 1)
+    4. size limit                               (item 1)
+
+On success the file is copied into an isolated per-upload quarantine directory
+(NOT the active data path), its SHA-256 is recorded (item 5), and an
+``UploadRecord`` with ``status="quarantined"`` is returned. Nothing here loads,
+parses, or executes the file — that is the validation step's job, and only via
+the vetted loaders.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import uuid
+from datetime import datetime
+from pathlib import Path
+
+import yaml
+
+from src.core.integrity import compute_sha256
+from src.uploads.records import (
+    STATUS_QUARANTINED,
+    STATUS_REJECTED,
+    UploadRecord,
+    persist_record,
+)
+
+# --------------------------------------------------------------------------- #
+# Config (all env-overridable)
+# --------------------------------------------------------------------------- #
+
+# Expression matrices / h5ad / supported tabular only. A compound suffix like
+# ``.csv.gz`` is matched against the last two suffixes as well (see _suffix_of).
+ALLOWED_SUFFIXES = frozenset(
+    {".h5ad", ".csv", ".tsv", ".txt", ".csv.gz", ".tsv.gz", ".txt.gz"}
+)
+
+
+# Config is read at CALL time (not import time) so env overrides and per-test
+# isolation actually take effect. Isolated staging area — deliberately NOT under
+# any served data path, so a quarantined file is never resolvable by the loaders
+# until it is registered.
+def _staging_dir() -> str:
+    return os.environ.get("UPLOAD_STAGING_DIR", "./upload_staging")
+
+
+# Default 2 GiB ceiling; raise/lower via UPLOAD_MAX_BYTES for a given deployment.
+def _max_upload_bytes() -> int:
+    return int(os.environ.get("UPLOAD_MAX_BYTES", str(2 * 1024 * 1024 * 1024)))
+
+
+class UploadRejected(Exception):
+    """Raised when an upload fails the door gate.
+
+    Carries the ``UploadRecord`` (``status="rejected"``, with the reason in
+    ``.errors``) so a caller/UI has one structured failure path.
+    """
+
+    def __init__(self, record: UploadRecord):
+        self.record = record
+        super().__init__("; ".join(record.errors) or "upload rejected")
+
+
+def _suffix_of(filename: str) -> str:
+    """Return the allow-list suffix for ``filename`` (handles ``.csv.gz`` etc.).
+
+    Prefers the longest matching compound suffix so ``data.csv.gz`` maps to
+    ``.csv.gz`` rather than ``.gz``.
+    """
+    name = filename.lower()
+    parts = Path(name).suffixes  # e.g. ['.csv', '.gz']
+    if len(parts) >= 2:
+        compound = "".join(parts[-2:])
+        if compound in ALLOWED_SUFFIXES:
+            return compound
+    return Path(name).suffix  # single suffix, e.g. '.h5ad'
+
+
+def stage_upload(
+    src_path: str | Path,
+    *,
+    uploader: str,
+    dataset_id: str,
+    manifest: dict,
+    deidentified: bool,
+    staging_dir: str | Path | None = None,
+) -> UploadRecord:
+    """Gate an upload and quarantine it. Returns a ``quarantined`` record.
+
+    Parameters
+    ----------
+    src_path:
+        Path to the file the researcher wants to upload (already on local disk).
+    uploader:
+        Identity of the person uploading (recorded in provenance).
+    dataset_id:
+        The dataset_id the upload will register as if it passes.
+    manifest:
+        The manifest dict for the upload — required (ADR item 2). Schema
+        validity is checked later in :func:`validate_upload`; here it must
+        merely be present and non-empty.
+    deidentified:
+        The uploader's attestation that the data carries no PHI/PSI (ADR item 6).
+        Must be ``True`` or the upload is rejected at the door.
+
+    Raises
+    ------
+    UploadRejected
+        If any door gate fails. The attached record has ``status="rejected"``.
+    """
+    src_path = Path(src_path)
+    now = datetime.now().isoformat(timespec="seconds")
+    upload_id = uuid.uuid4().hex[:12]
+
+    def _reject(reason: str) -> "UploadRejected":
+        rec = UploadRecord(
+            upload_id=upload_id,
+            dataset_id=dataset_id,
+            uploader=uploader,
+            status=STATUS_REJECTED,
+            created_at=now,
+            original_filename=src_path.name,
+            deidentified_attestation=bool(deidentified),
+            manifest=dict(manifest) if isinstance(manifest, dict) else {},
+            errors=[reason],
+        )
+        persist_record(rec)
+        return UploadRejected(rec)
+
+    # ── Gate 1: de-identification attestation (item 6) ───────────────────────
+    if deidentified is not True:
+        raise _reject(
+            "De-identification attestation required: the uploader must confirm "
+            "the data contains no PHI/PSI (Data-Classification Q1–Q3)."
+        )
+
+    # ── Gate 2: manifest required (item 2) ───────────────────────────────────
+    if not isinstance(manifest, dict) or not manifest:
+        raise _reject(
+            "A manifest is required — an upload with no manifest is not "
+            "ingestible (ADR-0011 item 2)."
+        )
+
+    # ── Gate 3: type allow-list (item 1) ─────────────────────────────────────
+    suffix = _suffix_of(src_path.name)
+    if suffix not in ALLOWED_SUFFIXES:
+        raise _reject(
+            f"File type '{src_path.suffix or src_path.name}' is not allowed. "
+            f"Permitted: {sorted(ALLOWED_SUFFIXES)}."
+        )
+
+    # ── Gate 4: existence + size limit (item 1) ──────────────────────────────
+    if not src_path.is_file():
+        raise _reject(f"Source file not found: {src_path}")
+    size = src_path.stat().st_size
+    limit = _max_upload_bytes()
+    if size > limit:
+        raise _reject(
+            f"File is {size} bytes, exceeding the {limit}-byte upload limit "
+            f"(UPLOAD_MAX_BYTES)."
+        )
+
+    # ── Quarantine: copy into an isolated per-upload dir (item 1) ────────────
+    base = Path(staging_dir) if staging_dir is not None else Path(_staging_dir())
+    quarantine = base / upload_id
+    quarantine.mkdir(parents=True, exist_ok=True)
+    staged_path = quarantine / f"data{suffix}"
+    shutil.copy2(src_path, staged_path)
+
+    # Manifest travels with the staged file.
+    with open(quarantine / "manifest.yaml", "w", encoding="utf-8") as fh:
+        yaml.safe_dump(dict(manifest), fh, sort_keys=False)
+
+    # ── Integrity record (item 5) ────────────────────────────────────────────
+    sha256 = compute_sha256(staged_path)
+
+    record = UploadRecord(
+        upload_id=upload_id,
+        dataset_id=dataset_id,
+        uploader=uploader,
+        status=STATUS_QUARANTINED,
+        created_at=now,
+        original_filename=src_path.name,
+        staged_path=str(staged_path),
+        size_bytes=size,
+        sha256=sha256,
+        deidentified_attestation=True,
+        manifest=dict(manifest),
+    )
+    persist_record(record)
+    return record

src/uploads/validation.py

@@ -0,0 +1,116 @@
+"""Validate-before-availability for a quarantined upload (ADR-0011 items 3, 4, 5).
+
+A staged file stays quarantined until it clears the SAME grounding gate a
+registered dataset clears:
+
+    1. manifest schema validation  (src.datasets.manifest_schema.validate_manifest)
+    2. semantic against-data checks (src.workflows.manifest_data_validation)
+
+Item 4 ("never execute uploaded content") is honoured structurally: the file is
+opened ONLY by the vetted scanpy loader — the same loader registered data uses —
+never ``exec``/``eval``/``pickle``. Passing both checks flips the record to
+``validated``; failing keeps it quarantined with the errors surfaced.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from src.uploads.records import (
+    STATUS_QUARANTINED,
+    STATUS_VALIDATED,
+    UploadRecord,
+    persist_record,
+)
+
+
+def _report(status: str, errors: list[str], **extra: Any) -> dict:
+    return {"status": status, "errors": errors, **extra}
+
+
+def validate_upload(record: UploadRecord) -> tuple[UploadRecord, dict]:
+    """Run the manifest-schema + against-data gate on a quarantined upload.
+
+    Returns ``(record, report)``. The record's ``status`` becomes ``validated``
+    only when both gates pass; otherwise it stays ``quarantined`` and the reasons
+    are in ``record.errors`` / ``report['errors']``.
+    """
+    if record.status != STATUS_QUARANTINED:
+        return record, _report(
+            "error",
+            [f"Upload is '{record.status}', not 'quarantined' — nothing to validate."],
+        )
+    if not record.staged_path or not Path(record.staged_path).is_file():
+        record.errors = [f"Staged file missing: {record.staged_path}"]
+        persist_record(record)
+        return record, _report("error", record.errors)
+
+    # ── Gate 1: manifest schema validity ─────────────────────────────────────
+    from src.datasets.manifest_schema import DatasetManifest, validate_manifest
+
+    schema_result = validate_manifest(record.manifest)
+    if not schema_result.valid:
+        record.errors = [f"manifest schema: {e}" for e in schema_result.errors]
+        persist_record(record)
+        return record, _report(
+            "error", record.errors, schema_warnings=schema_result.warnings
+        )
+
+    manifest_obj = DatasetManifest.from_dict(record.manifest)
+
+    # ── Gate 2: against-data validation via the VETTED loader only (item 4) ──
+    suffix = Path(record.staged_path).suffix.lower()
+    if suffix != ".h5ad":
+        # The gate accepts tabular for staging, but auto against-data validation
+        # is h5ad-only this phase (every served dataset is an h5ad). Stay
+        # quarantined and say so honestly rather than pretend it validated.
+        record.errors = [
+            f"Auto-validation for '{suffix}' uploads is not implemented yet — "
+            "convert to .h5ad or validate manually. File stays quarantined."
+        ]
+        persist_record(record)
+        return record, _report("error", record.errors)
+
+    try:
+        import numpy as np
+        import scanpy as sc
+
+        from src.workflows.manifest_data_validation import validate_manifest_against_data
+
+        adata = sc.read_h5ad(record.staged_path)  # vetted loader — never exec
+
+        X = adata.X
+        if hasattr(X, "toarray"):
+            X = X.toarray()
+        X_flat = X.flatten()
+        if len(X_flat) > 50_000:
+            rng = np.random.default_rng(0)
+            X_flat = rng.choice(X_flat, size=50_000, replace=False)
+
+        data_report = validate_manifest_against_data(
+            X_flat, list(adata.var.index), adata.obs.copy(), manifest_obj
+        )
+    except Exception as exc:  # noqa: BLE001 — a load/validate failure is a rejection
+        record.errors = [f"against-data validation failed to run: {exc}"]
+        persist_record(record)
+        return record, _report("error", record.errors)
+
+    if not data_report.get("overall_valid"):
+        record.errors = list(data_report.get("errors", [])) or [
+            "against-data validation reported the manifest inconsistent with the file."
+        ]
+        persist_record(record)
+        return record, _report(
+            "error", record.errors, against_data=data_report,
+            schema_warnings=schema_result.warnings,
+        )
+
+    # ── Passed both gates ────────────────────────────────────────────────────
+    record.status = STATUS_VALIDATED
+    record.validated_at = datetime.now().isoformat(timespec="seconds")
+    record.errors = []
+    persist_record(record)
+    return record, _report(
+        "pass", [], against_data=data_report, schema_warnings=schema_result.warnings
+    )

tests/test_upload_gate.py

@@ -0,0 +1,268 @@
+"""End-to-end tests for the manual-upload safety gate (ADR-0011).
+
+Synthetic fixtures only — a small AnnData written to a tmp .h5ad, a matching
+manifest dict, no network. Exercises the whole lifecycle:
+
+    stage_upload  →  validate_upload  →  register_upload
+
+plus every gate that can reject at each step.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import anndata as ad
+import numpy as np
+import pandas as pd
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.core.integrity import compute_sha256, verify_sha256
+from src.datasets.registry import get_registry, load_manifest
+from src.uploads import (
+    RegistrationRefused,
+    UploadRejected,
+    register_upload,
+    stage_upload,
+    validate_upload,
+)
+from src.uploads.records import (
+    STATUS_QUARANTINED,
+    STATUS_REGISTERED,
+    STATUS_VALIDATED,
+)
+
+
+# --------------------------------------------------------------------------- #
+# Fixtures / helpers
+# --------------------------------------------------------------------------- #
+
+def _write_h5ad(path: Path, *, data_level: str = "log", n_obs: int = 30, n_var: int = 60) -> Path:
+    """Write a synthetic h5ad. data_level='log' → log2-ish; 'counts' → integers."""
+    rng = np.random.default_rng(0)
+    if data_level == "counts":
+        X = rng.poisson(200, size=(n_obs, n_var)).astype(np.float32)
+    else:
+        X = rng.uniform(2.0, 14.0, size=(n_obs, n_var)).astype(np.float32)
+    var = pd.DataFrame(index=[f"GENE{i:04d}" for i in range(n_var)])
+    cond = (["A"] * (n_obs // 2)) + (["B"] * (n_obs - n_obs // 2))
+    obs = pd.DataFrame({"condition": cond}, index=[f"S{i}" for i in range(n_obs)])
+    adata = ad.AnnData(X=X, obs=obs, var=var)
+    adata.write_h5ad(path)
+    return path
+
+
+def _manifest(dataset_id: str, *, data_level: str = "log_expression") -> dict:
+    return {
+        "dataset_id": dataset_id,
+        "title": "Synthetic upload test dataset",
+        "accession": "UPLOAD-TEST",
+        "organism": "human",
+        "modality": "bulk_rnaseq",
+        "platform": "synthetic",
+        "data_level": data_level,
+        "feature_id_type": "gene_symbol",
+        "expression_source": {"type": "local"},
+        "metadata_source": {"type": "local", "embedded": True},
+        "group_columns": ["condition"],
+        "valid_workflows": ["activity_scoring"],
+        "limitations": ["synthetic dataset for tests"],
+    }
+
+
+@pytest.fixture
+def good_h5ad(tmp_path: Path) -> Path:
+    return _write_h5ad(tmp_path / "source.h5ad")
+
+
+@pytest.fixture(autouse=True)
+def _isolated_env(tmp_path, monkeypatch):
+    """Point staging/registered/audit dirs at tmp and default to no admins."""
+    monkeypatch.setenv("UPLOAD_STAGING_DIR", str(tmp_path / "staging"))
+    monkeypatch.setenv("UPLOAD_REGISTERED_DIR", str(tmp_path / "registered"))
+    monkeypatch.setenv("LOG_SINK", "local")
+    monkeypatch.setenv("LOG_SINK_LOCAL_DIR", str(tmp_path / "audit"))
+    monkeypatch.delenv("UPLOAD_ADMIN_IDS", raising=False)
+    monkeypatch.delenv("UPLOAD_MAX_BYTES", raising=False)
+
+
+def _stage(good_h5ad, dataset_id="up_test", **over):
+    kwargs = dict(
+        uploader="researcher@ohsu.edu",
+        dataset_id=dataset_id,
+        manifest=_manifest(dataset_id),
+        deidentified=True,
+    )
+    kwargs.update(over)
+    return stage_upload(good_h5ad, **kwargs)
+
+
+# --------------------------------------------------------------------------- #
+# Door gate
+# --------------------------------------------------------------------------- #
+
+@pytest.mark.parametrize("suffix", [".py", ".sh", ".pkl", ".bin", ".exe"])
+def test_type_gate_rejects_disallowed(tmp_path, suffix):
+    bad = tmp_path / f"payload{suffix}"
+    bad.write_bytes(b"not data")
+    with pytest.raises(UploadRejected) as exc:
+        _stage(bad)
+    assert exc.value.record.status == "rejected"
+    assert "not allowed" in exc.value.record.errors[0]
+
+
+@pytest.mark.parametrize("suffix", [".h5ad", ".csv", ".tsv", ".csv.gz"])
+def test_type_gate_accepts_allowed_suffixes(tmp_path, suffix):
+    f = tmp_path / f"data{suffix}"
+    f.write_bytes(b"col1,col2\n1,2\n")  # content irrelevant at the door
+    rec = _stage(f)  # tabular is accepted for STAGING (validation is h5ad-only)
+    assert rec.status == STATUS_QUARANTINED
+
+
+def test_size_gate_rejects_oversize(good_h5ad, monkeypatch):
+    monkeypatch.setenv("UPLOAD_MAX_BYTES", "10")  # 10 bytes — the h5ad is bigger
+    with pytest.raises(UploadRejected) as exc:
+        _stage(good_h5ad)
+    assert "upload limit" in exc.value.record.errors[0]
+
+
+def test_manifest_required(good_h5ad):
+    with pytest.raises(UploadRejected) as exc:
+        _stage(good_h5ad, manifest={})
+    assert "manifest is required" in exc.value.record.errors[0].lower()
+
+
+def test_attestation_required(good_h5ad):
+    with pytest.raises(UploadRejected) as exc:
+        _stage(good_h5ad, deidentified=False)
+    assert "attestation" in exc.value.record.errors[0].lower()
+
+
+def test_staged_record_carries_attestation_and_hash(good_h5ad):
+    rec = _stage(good_h5ad)
+    assert rec.status == STATUS_QUARANTINED
+    assert rec.deidentified_attestation is True
+    assert rec.sha256 and len(rec.sha256) == 64
+    assert Path(rec.staged_path).is_file()
+
+
+# --------------------------------------------------------------------------- #
+# Integrity helper
+# --------------------------------------------------------------------------- #
+
+def test_integrity_hash_recorded_and_verifiable(good_h5ad):
+    rec = _stage(good_h5ad)
+    assert rec.sha256 == compute_sha256(rec.staged_path)
+    assert verify_sha256(rec.staged_path, rec.sha256) is True
+
+
+def test_integrity_detects_tamper(good_h5ad):
+    rec = _stage(good_h5ad)
+    with open(rec.staged_path, "ab") as fh:
+        fh.write(b"tampered")
+    assert verify_sha256(rec.staged_path, rec.sha256) is False
+
+
+def test_verify_sha256_empty_baseline_is_false(good_h5ad):
+    rec = _stage(good_h5ad)
+    assert verify_sha256(rec.staged_path, "") is False
+
+
+# --------------------------------------------------------------------------- #
+# Validation
+# --------------------------------------------------------------------------- #
+
+def test_validate_happy_path(good_h5ad):
+    rec = _stage(good_h5ad, dataset_id="up_valid")
+    rec, report = validate_upload(rec)
+    assert rec.status == STATUS_VALIDATED
+    assert report["status"] == "pass"
+    assert rec.validated_at
+
+
+def test_validate_mismatch_stays_quarantined(tmp_path):
+    # Manifest declares raw_counts, but the data is log-scaled → against-data error.
+    src = _write_h5ad(tmp_path / "log.h5ad", data_level="log")
+    m = _manifest("up_mismatch", data_level="raw_counts")
+    rec = stage_upload(
+        src, uploader="r@ohsu.edu", dataset_id="up_mismatch",
+        manifest=m, deidentified=True,
+    )
+    rec, report = validate_upload(rec)
+    assert rec.status == STATUS_QUARANTINED
+    assert report["status"] == "error"
+    assert rec.errors
+
+
+def test_validate_tabular_not_supported(tmp_path):
+    f = tmp_path / "matrix.csv"
+    f.write_bytes(b"gene,S1,S2\nGENE1,1,2\n")
+    rec = _stage(f, dataset_id="up_csv")
+    rec, report = validate_upload(rec)
+    assert rec.status == STATUS_QUARANTINED
+    assert "not implemented" in report["errors"][0].lower()
+
+
+# --------------------------------------------------------------------------- #
+# Admin approval / promotion
+# --------------------------------------------------------------------------- #
+
+def test_register_refused_for_non_admin(good_h5ad, monkeypatch):
+    monkeypatch.setenv("UPLOAD_ADMIN_IDS", "alice@ohsu.edu")
+    rec = _stage(good_h5ad, dataset_id="up_nonadmin")
+    rec, _ = validate_upload(rec)
+    assert rec.status == STATUS_VALIDATED
+    with pytest.raises(RegistrationRefused, match="not an authorized admin"):
+        register_upload(rec, admin="bob@ohsu.edu")
+
+
+def test_register_refused_before_validation(good_h5ad, monkeypatch):
+    monkeypatch.setenv("UPLOAD_ADMIN_IDS", "alice@ohsu.edu")
+    rec = _stage(good_h5ad, dataset_id="up_early")  # still quarantined
+    with pytest.raises(RegistrationRefused, match="not 'validated'"):
+        register_upload(rec, admin="alice@ohsu.edu")
+
+
+def test_admin_register_promotes_into_registry(good_h5ad, monkeypatch):
+    monkeypatch.setenv("UPLOAD_ADMIN_IDS", "alice@ohsu.edu")
+    dsid = "up_registered_ds"
+    rec = _stage(good_h5ad, dataset_id=dsid)
+    rec, _ = validate_upload(rec)
+    rec = register_upload(rec, admin="alice@ohsu.edu")
+    assert rec.status == STATUS_REGISTERED
+    assert rec.registered_at
+    # The live registry now resolves the freshly-registered dataset.
+    assert get_registry().get(dsid) is not None
+    assert load_manifest(dsid).dataset_id == dsid
+
+
+# --------------------------------------------------------------------------- #
+# Never-execute guarantee (ADR item 4)
+# --------------------------------------------------------------------------- #
+
+def test_uploads_package_has_no_exec_paths():
+    pkg = Path(__file__).parent.parent / "src" / "uploads"
+    forbidden = ("exec(", "eval(", "pickle.load", "__import__(")
+    for py in pkg.glob("*.py"):
+        text = py.read_text()
+        for token in forbidden:
+            assert token not in text, f"{py.name} contains forbidden call: {token}"
+
+
+# --------------------------------------------------------------------------- #
+# Provenance / audit trail (ADR item 6)
+# --------------------------------------------------------------------------- #
+
+def test_attestation_persisted_to_audit_sink(good_h5ad, tmp_path):
+    rec = _stage(good_h5ad, dataset_id="up_audit")
+    rec, _ = validate_upload(rec)
+    audit = tmp_path / "audit" / f"upload_{rec.upload_id}_trace.json"
+    assert audit.is_file()
+    import json
+
+    blob = json.loads(audit.read_text())
+    assert blob["deidentified_attestation"] is True
+    assert blob["sha256"] == rec.sha256
+    assert blob["status"] == STATUS_VALIDATED