"""Interim content scan for a quarantined upload (ADR-0011 — AWS-independent). The ADR's malware scan is deferred to AWS ("the optional malware scan runs on the staged object *before* validation"). Until then, a fully local scan stands in so the "content scan" box is checked honestly rather than left empty. It runs between :func:`stage_upload` and :func:`validate_upload` and has two layers: 1. **Structural magic-byte check (always on, zero dependency).** The door gate only trusts the *filename suffix*; this opens the first bytes and confirms the content matches. An ``.h5ad`` must be an HDF5 container; a ``.gz`` must be gzip; a flat-text matrix must contain no executable/archive/pickle magic, no shebang, and no NUL bytes. This catches the obvious attack the suffix gate misses — a renamed ELF/Mach-O/PE/ZIP/pickle wearing a ``.csv`` extension. 2. **Optional anti-virus pass (ClamAV, if present).** If ``clamdscan``/``clamscan`` is on PATH (or ``UPLOAD_SCAN_CMD`` names a scanner), the staged object is run through it. This is an external AV *reading* the file — the file is still never executed or interpreted here (no ``exec``/``eval``/``pickle``). Honesty rule (ADR Consequences): when no AV is available the scan is recorded as ``skipped`` with a caveat, never silently reported as "malware-scanned". Set ``UPLOAD_SCAN_REQUIRED=1`` to fail closed instead (reject when no AV is present). A structural-magic failure is *always* a hard stop regardless of that flag. Which of those two the *deployment* actually gets is no longer implicit in whatever is on the host's PATH: it is declared in ``deploy/scan_posture.yaml`` and read via :mod:`src.uploads.posture`. Every record is stamped with the posture it was scanned under, so a ``clean`` produced by a dev laptop's ClamAV is distinguishable from coverage the deployment guarantees. """ from __future__ import annotations import os import shlex import shutil import subprocess from datetime import datetime from pathlib import Path from typing import Any from src.uploads.posture import ( POSTURE_STRUCTURAL_ONLY, av_required, declared_posture, ) from src.uploads.records import ( STATUS_QUARANTINED, UploadRecord, persist_record, ) # Scan outcomes recorded on the record's ``scan_status``. SCAN_CLEAN = "clean" SCAN_INFECTED = "infected" SCAN_SKIPPED = "skipped" # --------------------------------------------------------------------------- # # Magic-byte signatures # --------------------------------------------------------------------------- # # The signatures a well-formed upload MUST start with, keyed by staged suffix. _HDF5_MAGIC = b"\x89HDF\r\n\x1a\n" _GZIP_MAGIC = b"\x1f\x8b" # Executable / archive / serialized-object leaders that must NEVER appear at the # head of a file we treat as a plain-text expression matrix. A match means the # "csv" is really a disguised binary. _FORBIDDEN_TEXT_LEADERS: tuple[bytes, ...] = ( b"\x7fELF", # ELF (Linux executable / .so) b"MZ", # DOS/PE (.exe / .dll) b"\xfe\xed\xfa\xce", # Mach-O 32 big-endian b"\xfe\xed\xfa\xcf", # Mach-O 64 big-endian b"\xce\xfa\xed\xfe", # Mach-O 32 little-endian b"\xcf\xfa\xed\xfe", # Mach-O 64 little-endian b"\xca\xfe\xba\xbe", # Mach-O universal / Java .class b"PK\x03\x04", # ZIP (also xlsx/jar/docx containers) b"PK\x05\x06", # empty ZIP b"\x80\x04", # pickle protocol 4 b"\x80\x05", # pickle protocol 5 b"#!", # shebang script ) # How many bytes to sniff for the NUL / leader checks. _SNIFF_BYTES = 65536 class ScanError(Exception): """Raised only when the record is not in a scannable state.""" # --------------------------------------------------------------------------- # # Config (read at call time so env overrides / tests take effect) # --------------------------------------------------------------------------- # def _scan_required() -> bool: """Fail closed when no clean AV result? Env knob OR the declared posture.""" return av_required() def _resolve_av_commands(staged_path: str) -> list[list[str]]: """Return AV scanner argvs to try, in preference order (may be empty). ``UPLOAD_SCAN_CMD`` (a shell-style template, ``{path}`` substituted) takes precedence and is the *only* candidate when set — an operator's explicit choice is honoured verbatim, no fallback. Otherwise auto-detect ClamAV and return both the daemon client (``clamdscan``, fast when ``clamd`` is warm) AND the standalone binary (``clamscan``, works with no daemon). ``_run_av`` tries them in order, so a missing/misconfigured daemon degrades to a standalone scan instead of being reported as unscanned. """ override = os.environ.get("UPLOAD_SCAN_CMD", "").strip() if override: return [[tok.replace("{path}", staged_path) for tok in shlex.split(override)]] candidates: list[list[str]] = [] if shutil.which("clamdscan"): candidates.append(["clamdscan", "--no-summary", "--fdpass", staged_path]) if shutil.which("clamscan"): candidates.append(["clamscan", "--no-summary", staged_path]) return candidates # --------------------------------------------------------------------------- # # Layer 1: structural magic-byte check # --------------------------------------------------------------------------- # def _suffix_kind(name: str) -> str: """Return 'h5ad' | 'gzip' | 'xlsx' | 'text' for a staged filename.""" low = name.lower() if low.endswith(".h5ad"): return "h5ad" if low.endswith(".xlsx"): return "xlsx" if low.endswith(".gz"): return "gzip" return "text" def _xlsx_structural_check(path: Path, head: bytes) -> dict[str, Any]: """An ``.xlsx`` must be a real OOXML workbook, not just any zip. A bare ``PK`` check would pass a jar, a docx, or an ODS — anything zipped. So after the magic bytes we open the container (read-only, no extraction) and require the two entries every xlsx has: ``[Content_Types].xml`` and a ``xl/`` part. Nothing is decompressed to disk and no macro part is ever evaluated; a workbook carrying ``xl/vbaProject.bin`` is rejected outright, since a metadata sheet has no business shipping macros. """ import zipfile if not head.startswith((b"PK\x03\x04", b"PK\x05\x06")): return { "ok": False, "reason": "declared .xlsx but the file is not a zip container " "(missing PK magic bytes).", } try: with zipfile.ZipFile(path) as zf: names = zf.namelist() except zipfile.BadZipFile as exc: return {"ok": False, "reason": f"declared .xlsx but the zip container is corrupt: {exc}"} if "[Content_Types].xml" not in names or not any(n.startswith("xl/") for n in names): return { "ok": False, "reason": "declared .xlsx but the zip is not an OOXML workbook " "(no [Content_Types].xml / xl/ parts) — a renamed archive.", } macros = [n for n in names if n.lower().endswith("vbaproject.bin")] if macros: return { "ok": False, "reason": "the workbook contains a VBA macro project " f"({macros[0]}); macro-bearing workbooks are refused.", } return {"ok": True, "reason": "OOXML workbook verified (no macro project)."} def structural_check(staged_path: str) -> dict[str, Any]: """Confirm the file's leading bytes match the type its suffix claims. Returns ``{"ok": bool, "reason": str}``. ``ok=False`` is a hard rejection — the content contradicts the declared type (e.g. an executable renamed ``.csv``), which no legitimate expression matrix does. """ path = Path(staged_path) with open(path, "rb") as fh: head = fh.read(_SNIFF_BYTES) kind = _suffix_kind(path.name) if kind == "h5ad": if not head.startswith(_HDF5_MAGIC): return { "ok": False, "reason": "declared .h5ad but the file is not an HDF5 container " "(missing HDF5 magic bytes).", } return {"ok": True, "reason": "HDF5 container header verified."} if kind == "xlsx": return _xlsx_structural_check(path, head) if kind == "gzip": if not head.startswith(_GZIP_MAGIC): return { "ok": False, "reason": "declared .gz but the file is not gzip-compressed " "(missing gzip magic bytes).", } return {"ok": True, "reason": "gzip header verified."} # Plain-text matrix: reject disguised binaries and non-text content. for leader in _FORBIDDEN_TEXT_LEADERS: if head.startswith(leader): return { "ok": False, "reason": "declared a text matrix but the file begins with a " f"binary/executable/archive signature ({leader!r}).", } if b"\x00" in head: return { "ok": False, "reason": "declared a text matrix but contains NUL bytes — it is " "binary content, not a delimited text file.", } return {"ok": True, "reason": "text matrix — no binary/executable signature."} # --------------------------------------------------------------------------- # # Layer 2: optional AV pass # --------------------------------------------------------------------------- # def _run_one_av(argv: list[str]) -> dict[str, Any]: """Run a single AV scanner argv. ``result`` ∈ {clean, infected, error}. ClamAV exit codes: 0 = clean, 1 = virus found, 2 = error. The subprocess only *reads* the staged file; nothing in it is executed. """ try: proc = subprocess.run( # noqa: S603 — fixed argv, no shell, read-only scan argv, capture_output=True, text=True, timeout=int(os.environ.get("UPLOAD_SCAN_TIMEOUT", "600")), check=False, ) except (OSError, subprocess.SubprocessError) as exc: return {"result": "error", "detail": f"scanner failed to run: {exc}"} tool = Path(argv[0]).name if proc.returncode == 0: return {"result": "clean", "detail": f"{tool}: clean."} if proc.returncode == 1: signature = (proc.stdout or proc.stderr or "").strip().splitlines() hit = signature[0] if signature else "malware signature match" return {"result": "infected", "detail": f"{tool} flagged the file: {hit}"} return { "result": "error", "detail": f"{tool} exited {proc.returncode}: " f"{(proc.stderr or proc.stdout or '').strip()[:200]}", } def _run_av(staged_path: str) -> dict[str, Any]: """Run the available AV scanner(s) in preference order. A definitive result (clean or infected) from any candidate is returned immediately. If a candidate errors (e.g. ``clamdscan`` with no running daemon), the next candidate is tried, so a standalone ``clamscan`` still covers the file. Returns ``unavailable`` when nothing is installed, or the last error if every candidate errored. """ candidates = _resolve_av_commands(staged_path) if not candidates: return {"result": "unavailable", "detail": "no anti-virus scanner on PATH."} last = {"result": "unavailable", "detail": "no anti-virus scanner on PATH."} for argv in candidates: last = _run_one_av(argv) if last["result"] in {"clean", "infected"}: return last return last # every candidate errored — surface the last error # --------------------------------------------------------------------------- # # Public entry point # --------------------------------------------------------------------------- # def scan_upload(record: UploadRecord) -> tuple[UploadRecord, dict[str, Any]]: """Content-scan a quarantined upload before it may be validated. Sets ``record.scan_status`` to one of ``clean`` / ``infected`` / ``skipped``, stamps ``record.scanned_at``, and persists. The record's lifecycle ``status`` is left ``quarantined`` in every case — a scan does not promote an upload; it only decides whether validation may proceed: - **clean** — magic-byte check passed and (if an AV was present) it reported clean. Validation may proceed. - **skipped** — magic-byte check passed but no AV was available. A caveat is recorded; validation may proceed unless ``UPLOAD_SCAN_REQUIRED`` is set, in which case this is treated as ``infected`` (fail closed). - **infected** — the structural magic-byte check failed (always a hard stop), or the AV flagged the file. ``record.errors`` explains; validation refuses. """ if record.status != STATUS_QUARANTINED: raise ScanError(f"Upload is '{record.status}', not 'quarantined' — nothing to scan.") if not record.staged_path or not Path(record.staged_path).is_file(): record.scan_status = SCAN_INFECTED record.scanned_at = datetime.now().isoformat(timespec="seconds") record.scan_detail = f"staged file missing: {record.staged_path}" record.errors = [record.scan_detail] persist_record(record) return record, {"status": "error", "errors": record.errors} record.scanned_at = datetime.now().isoformat(timespec="seconds") posture = declared_posture() record.scan_posture = posture # ── Layer 1: structural magic-byte check (always a hard stop on failure) ── struct = structural_check(record.staged_path) if not struct["ok"]: record.scan_status = SCAN_INFECTED record.scan_detail = f"structural check: {struct['reason']}" record.errors = ["Content scan rejected the upload — " + struct["reason"]] persist_record(record) return record, {"status": "error", "errors": record.errors, "scan": struct} # ── Layer 2: optional AV pass ──────────────────────────────────────────── av = _run_av(record.staged_path) if av["result"] == "infected": record.scan_status = SCAN_INFECTED record.scan_detail = f"antivirus: {av['detail']}" record.errors = ["Content scan rejected the upload — " + av["detail"]] persist_record(record) return record, {"status": "error", "errors": record.errors, "scan": av} if av["result"] == "clean": record.scan_status = SCAN_CLEAN record.scan_detail = av["detail"] record.errors = [] if posture == POSTURE_STRUCTURAL_ONLY: # An AV was on THIS host, but the deployment does not ship one (see # deploy/scan_posture.yaml). Record that, so a dev-laptop `clean` # is never mistaken for coverage the Space actually provides. host_caveat = ( "Anti-virus ran on this host and reported clean, but the declared " "deployment posture is 'structural_only' (deploy/scan_posture.yaml) " "— the deployed Space has no AV binary. Do not generalise this " "clean result to uploads made in production." ) if host_caveat not in record.caveats: record.caveats.append(host_caveat) persist_record(record) return record, {"status": "pass", "scan": av, "posture": posture} # AV unavailable or errored → skipped, unless required (fail closed). if _scan_required(): record.scan_status = SCAN_INFECTED record.scan_detail = f"scan required but no clean AV result: {av['detail']}" record.errors = [ "Content scan could not complete and UPLOAD_SCAN_REQUIRED is set: " + av["detail"] ] persist_record(record) return record, {"status": "error", "errors": record.errors, "scan": av} record.scan_status = SCAN_SKIPPED record.scan_detail = av["detail"] caveat = ( "Malware scan SKIPPED — " + av["detail"] + " Structural magic-byte checks " "passed, but no anti-virus scan ran. This upload is " "structurally-verified, NOT malware-scanned — do not describe it as " "malware-scanned. This is the declared posture for this deployment " "(posture='" + posture + "', deploy/scan_posture.yaml); managed AV is " "deferred to the ADR-0011 AWS staged-object pass." ) if caveat not in record.caveats: record.caveats.append(caveat) persist_record(record) return record, {"status": "skipped", "scan": av, "caveat": caveat, "posture": posture}