Annie Voigt
style: apply ruff lint --fix + ruff format across the tree
c3b49d6
Raw
History Blame Contribute Delete
9.69 kB
"""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,
)