"""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. against-data checks (src.workflows.manifest_data_validation) Item 4 ("never execute uploaded content") is honoured structurally: the file is opened ONLY by a vetted reader — ``scanpy.read_h5ad`` for h5ad, ``pandas.read_csv`` for a flat matrix — never ``exec``/``eval``/``pickle``. Passing both gates flips the record to ``validated``; failing keeps it quarantined with the errors surfaced. Two file shapes are supported: - **h5ad** — carries expression AND sample metadata (obs), so the full five-check ``validate_manifest_against_data`` runs (data level, feature IDs, metadata columns, group columns, contrasts). - **flat matrix** (``.csv``/``.tsv``/``.txt`` + ``.gz``) — samples as rows, genes as columns (the convention ``decoupler_inspect_data`` uses). A bare matrix has no ``obs``, so only the two *file-content* checks that don't need metadata run (``data_level`` + ``feature_id_type``); the obs-dependent checks are recorded as skipped **caveats**, not silently passed. This verifies the file is what the manifest claims (correct units, correct feature IDs) — the security-relevant part — while stating honestly that grouping/metadata weren't validated. """ 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, ) # Flat-matrix suffixes (compound .gz handled in _kind_of / _read_tabular_matrix). _TABULAR_SUFFIXES = (".csv", ".tsv", ".txt", ".csv.gz", ".tsv.gz", ".txt.gz") # Cap the number of expression values fed to the data-level classifier, matching # the dataset_validate_manifest_against_data tool. _MAX_X = 50_000 def _report(status: str, errors: list[str], **extra: Any) -> dict: return {"status": status, "errors": errors, **extra} def _kind_of(path: str) -> str | None: """Return 'h5ad' | 'tabular' | None for a staged file name.""" name = Path(path).name.lower() if name.endswith(".h5ad"): return "h5ad" if name.endswith(_TABULAR_SUFFIXES): return "tabular" return None def _cap_flat(X_flat): import numpy as np if len(X_flat) > _MAX_X: rng = np.random.default_rng(0) return rng.choice(X_flat, size=_MAX_X, replace=False) return X_flat def _h5ad_report(staged_path: str, manifest_obj) -> dict: """Full five-check report from an h5ad, via the vetted scanpy loader.""" import scanpy as sc from src.workflows.manifest_data_validation import validate_manifest_against_data adata = sc.read_h5ad(staged_path) # vetted loader — never exec X = adata.X if hasattr(X, "toarray"): X = X.toarray() X_flat = _cap_flat(X.flatten()) return validate_manifest_against_data( X_flat, list(adata.var.index), adata.obs.copy(), manifest_obj ) def _tabular_report(staged_path: str, manifest_obj) -> dict: """Content-only report from a flat matrix, via the vetted pandas reader. Samples as rows, genes as columns (index_col=0), matching ``decoupler_inspect_data``. Only the two checks that a bare matrix can support are run; the obs-dependent checks are surfaced as caveats. """ import pandas as pd from src.workflows.manifest_data_validation import ( check_data_level, check_feature_id_type, ) name = Path(staged_path).name.lower() base = name[:-3] if name.endswith(".gz") else name sep = "\t" if base.endswith((".tsv", ".txt")) else "," # pandas infers .gz decompression from the filename. read_csv is a safe, # vetted reader — no code execution (unlike read_pickle). df = pd.read_csv(staged_path, index_col=0, sep=sep) X = df.to_numpy(dtype=float) # raises if the matrix isn't purely numeric X_flat = _cap_flat(X.flatten()) var_index = [str(c) for c in df.columns] checks = { "data_level": check_data_level(X_flat, manifest_obj.data_level), "feature_id_type": check_feature_id_type(var_index, manifest_obj.feature_id_type), } errors = [ f"[{name}] {c.get('message', '')}" for name, c in checks.items() if c.get("status") == "error" ] warnings = [ f"[{name}] {c.get('message', '')}" for name, c in checks.items() if c.get("status") == "warning" ] skipped = ["metadata_columns", "group_columns", "default_contrasts"] caveats = [ "Flat matrix carries no sample metadata (obs), so the obs-dependent checks " f"({', '.join(skipped)}) were NOT run — re-upload as an .h5ad with embedded " "obs to validate grouping and contrasts.", "Orientation assumed samples-as-rows, genes-as-columns (index_col=0); if the " "feature_id_type check flags a mismatch, the matrix may be transposed.", ] return { "dataset_id": manifest_obj.dataset_id, "n_samples": int(df.shape[0]), "n_features": int(df.shape[1]), "overall_valid": len(errors) == 0, "n_errors": len(errors), "n_warnings": len(warnings), "checks": checks, "errors": errors, "warnings": warnings, "skipped_checks": skipped, "caveats": caveats, } 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']``. A flat-matrix upload that validates carries ``record.caveats`` naming the checks that could not run. """ 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 0: content scan must run and clear before validation ──────────── # ADR-0011: "the optional malware scan runs on the staged object *before* # validation." The structural magic-byte check is unconditional; the AV pass # is best-effort (skipped-with-caveat unless UPLOAD_SCAN_REQUIRED). An # infected/disguised file never reaches the loaders below. from src.uploads.scanning import SCAN_INFECTED, scan_upload if record.scan_status is None: record, scan_report = scan_upload(record) if record.scan_status == SCAN_INFECTED: return record, _report("error", record.errors, scan=record.scan_detail) scan_caveats = list(record.caveats) # a skipped-AV caveat, if any # ── 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 reader only (item 4) ── kind = _kind_of(record.staged_path) if kind is None: record.errors = [f"Unsupported file type for validation: {Path(record.staged_path).name}."] persist_record(record) return record, _report("error", record.errors) try: if kind == "h5ad": data_report = _h5ad_report(record.staged_path, manifest_obj) else: data_report = _tabular_report(record.staged_path, 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) caveats = scan_caveats + [c for c in data_report.get("caveats", []) if c not in scan_caveats] 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." ] record.caveats = caveats 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 = [] record.caveats = caveats persist_record(record) return record, _report( "pass", [], against_data=data_report, schema_warnings=schema_result.warnings, caveats=caveats, )