src/uploads/scanning.py

"""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}