diff --git a/backend/app.py b/backend/app.py index 8acccc64a..2aadc5941 100644 --- a/backend/app.py +++ b/backend/app.py @@ -3,8 +3,11 @@ import csv import io import logging +from contextlib import suppress from pathlib import Path -from typing import Any, Dict, Optional +from tempfile import SpooledTemporaryFile +from types import SimpleNamespace +from typing import Any, Dict, Optional, Tuple from fastapi import Depends, FastAPI, File, HTTPException, UploadFile from fastapi.middleware.cors import CORSMiddleware @@ -25,6 +28,8 @@ def create_app() -> FastAPI: """Create the FastAPI application with file-upload ingestion endpoints.""" app = FastAPI(title="FixOps Ingestion Demo API", version="0.1.0") + if not hasattr(app, "state"): + app.state = SimpleNamespace() app.add_middleware( CORSMiddleware, allow_origins=["*"], @@ -74,30 +79,46 @@ async def _verify_api_key(api_key: Optional[str] = Depends(api_key_header)) -> N else None ) - async def _read_limited(file: UploadFile, stage: str) -> bytes: + _CHUNK_SIZE = 1024 * 1024 + _RAW_BYTES_THRESHOLD = 4 * 1024 * 1024 + + async def _read_limited(file: UploadFile, stage: str) -> Tuple[SpooledTemporaryFile, int]: + """Stream an upload into a spooled file respecting the configured limit.""" + limit = overlay.upload_limit(stage) total = 0 - chunks: list[bytes] = [] - while True: - remaining = limit - total - if remaining <= 0: - break - chunk = await file.read(min(1024 * 1024, remaining)) - if not chunk: - break - total += len(chunk) - if total > limit: - raise HTTPException( - status_code=413, - detail={ - "message": f"Upload for stage '{stage}' exceeded limit", - "max_bytes": limit, - }, - ) - chunks.append(chunk) - if total == limit: - break - return b"".join(chunks) + buffer = SpooledTemporaryFile(max_size=_CHUNK_SIZE, mode="w+b") + try: + while total < limit: + remaining = limit - total + chunk = await file.read(min(_CHUNK_SIZE, remaining)) + if not chunk: + break + total += len(chunk) + if total > limit: + raise HTTPException( + status_code=413, + detail={ + "message": f"Upload for stage '{stage}' exceeded limit", + "max_bytes": limit, + }, + ) + buffer.write(chunk) + except Exception: + buffer.close() + raise + buffer.seek(0) + return buffer, total + + def _maybe_materialise_raw( + buffer: SpooledTemporaryFile, total: int, *, threshold: int = _RAW_BYTES_THRESHOLD + ) -> Optional[bytes]: + if total > threshold: + return None + buffer.seek(0) + data = buffer.read() + buffer.seek(0) + return data def _validate_content_type(file: UploadFile, expected: tuple[str, ...]) -> None: if file.content_type and file.content_type not in expected: @@ -134,23 +155,36 @@ def _store( @app.post("/inputs/design", dependencies=[Depends(_verify_api_key)]) async def ingest_design(file: UploadFile = File(...)) -> Dict[str, Any]: _validate_content_type(file, ("text/csv", "application/vnd.ms-excel", "application/csv")) - raw_bytes = await _read_limited(file, "design") - text = raw_bytes.decode("utf-8", errors="ignore") - reader = csv.DictReader(io.StringIO(text)) - rows = [row for row in reader if any((value or "").strip() for value in row.values())] + buffer, total = await _read_limited(file, "design") + try: + text_stream = io.TextIOWrapper(buffer, encoding="utf-8", errors="ignore", newline="") + try: + reader = csv.DictReader(text_stream) + rows = [ + row + for row in reader + if any((value or "").strip() for value in row.values()) + ] + columns = reader.fieldnames or [] + finally: + buffer = text_stream.detach() - if not rows: - raise HTTPException(status_code=400, detail="Design CSV contained no rows") + if not rows: + raise HTTPException(status_code=400, detail="Design CSV contained no rows") - dataset = {"columns": reader.fieldnames or [], "rows": rows} - _store("design", dataset, original_filename=file.filename, raw_bytes=raw_bytes) - return { - "stage": "design", - "input_filename": file.filename, - "row_count": len(rows), - "columns": dataset["columns"], - "data": dataset, - } + dataset = {"columns": columns, "rows": rows} + raw_bytes = _maybe_materialise_raw(buffer, total) + _store("design", dataset, original_filename=file.filename, raw_bytes=raw_bytes) + return { + "stage": "design", + "input_filename": file.filename, + "row_count": len(rows), + "columns": dataset["columns"], + "data": dataset, + } + finally: + with suppress(Exception): + buffer.close() @app.post("/inputs/sbom", dependencies=[Depends(_verify_api_key)]) async def ingest_sbom(file: UploadFile = File(...)) -> Dict[str, Any]: @@ -164,22 +198,26 @@ async def ingest_sbom(file: UploadFile = File(...)) -> Dict[str, Any]: "application/gzip", ), ) - raw_bytes = await _read_limited(file, "sbom") + buffer, total = await _read_limited(file, "sbom") try: - sbom: NormalizedSBOM = normalizer.load_sbom(raw_bytes) + sbom: NormalizedSBOM = normalizer.load_sbom(buffer) except Exception as exc: # pragma: no cover - FastAPI will serialise the message logger.exception("SBOM normalisation failed") raise HTTPException(status_code=400, detail=f"Failed to parse SBOM: {exc}") from exc - - _store("sbom", sbom, original_filename=file.filename, raw_bytes=raw_bytes) - return { - "stage": "sbom", - "input_filename": file.filename, - "metadata": sbom.metadata, - "component_preview": [ - component.to_dict() for component in sbom.components[:5] - ], - } + else: + raw_bytes = _maybe_materialise_raw(buffer, total) + _store("sbom", sbom, original_filename=file.filename, raw_bytes=raw_bytes) + return { + "stage": "sbom", + "input_filename": file.filename, + "metadata": sbom.metadata, + "component_preview": [ + component.to_dict() for component in sbom.components[:5] + ], + } + finally: + with suppress(Exception): + buffer.close() @app.post("/inputs/cve", dependencies=[Depends(_verify_api_key)]) async def ingest_cve(file: UploadFile = File(...)) -> Dict[str, Any]: @@ -193,20 +231,24 @@ async def ingest_cve(file: UploadFile = File(...)) -> Dict[str, Any]: "application/gzip", ), ) - raw_bytes = await _read_limited(file, "cve") + buffer, total = await _read_limited(file, "cve") try: - cve_feed: NormalizedCVEFeed = normalizer.load_cve_feed(raw_bytes) + cve_feed: NormalizedCVEFeed = normalizer.load_cve_feed(buffer) except Exception as exc: # pragma: no cover - FastAPI will serialise the message logger.exception("CVE feed normalisation failed") raise HTTPException(status_code=400, detail=f"Failed to parse CVE feed: {exc}") from exc - - _store("cve", cve_feed, original_filename=file.filename, raw_bytes=raw_bytes) - return { - "stage": "cve", - "input_filename": file.filename, - "record_count": cve_feed.metadata.get("record_count", 0), - "validation_errors": cve_feed.errors, - } + else: + raw_bytes = _maybe_materialise_raw(buffer, total) + _store("cve", cve_feed, original_filename=file.filename, raw_bytes=raw_bytes) + return { + "stage": "cve", + "input_filename": file.filename, + "record_count": cve_feed.metadata.get("record_count", 0), + "validation_errors": cve_feed.errors, + } + finally: + with suppress(Exception): + buffer.close() @app.post("/inputs/sarif", dependencies=[Depends(_verify_api_key)]) async def ingest_sarif(file: UploadFile = File(...)) -> Dict[str, Any]: @@ -220,20 +262,24 @@ async def ingest_sarif(file: UploadFile = File(...)) -> Dict[str, Any]: "application/gzip", ), ) - raw_bytes = await _read_limited(file, "sarif") + buffer, total = await _read_limited(file, "sarif") try: - sarif: NormalizedSARIF = normalizer.load_sarif(raw_bytes) + sarif: NormalizedSARIF = normalizer.load_sarif(buffer) except Exception as exc: # pragma: no cover - FastAPI will serialise the message logger.exception("SARIF normalisation failed") raise HTTPException(status_code=400, detail=f"Failed to parse SARIF: {exc}") from exc - - _store("sarif", sarif, original_filename=file.filename, raw_bytes=raw_bytes) - return { - "stage": "sarif", - "input_filename": file.filename, - "metadata": sarif.metadata, - "tools": sarif.tool_names, - } + else: + raw_bytes = _maybe_materialise_raw(buffer, total) + _store("sarif", sarif, original_filename=file.filename, raw_bytes=raw_bytes) + return { + "stage": "sarif", + "input_filename": file.filename, + "metadata": sarif.metadata, + "tools": sarif.tool_names, + } + finally: + with suppress(Exception): + buffer.close() @app.post("/pipeline/run", dependencies=[Depends(_verify_api_key)]) async def run_pipeline() -> Dict[str, Any]: diff --git a/backend/normalizers.py b/backend/normalizers.py index 4b7dd7387..0ffa152ef 100644 --- a/backend/normalizers.py +++ b/backend/normalizers.py @@ -9,6 +9,7 @@ import json import logging import zipfile +from contextlib import suppress from dataclasses import dataclass, field, asdict from typing import Any, Iterable, List, Optional @@ -150,13 +151,34 @@ def __init__(self, sbom_type: str = "auto") -> None: @staticmethod def _ensure_bytes(content: Any) -> bytes: - if isinstance(content, bytes): - return content + if isinstance(content, (bytes, bytearray)): + return bytes(content) + if isinstance(content, memoryview): + return content.tobytes() if hasattr(content, "read"): - data = content.read() - if isinstance(data, bytes): - return data - return str(data).encode("utf-8") + handle = content # type: ignore[assignment] + chunk_size = 1024 * 1024 + data = bytearray() + start = None + try: + start = handle.tell() # type: ignore[attr-defined] + except Exception: # pragma: no cover - not all streams support tell + start = None + while True: + chunk = handle.read(chunk_size) + if not chunk: + break + if isinstance(chunk, str): + chunk = chunk.encode("utf-8") + elif isinstance(chunk, memoryview): + chunk = chunk.tobytes() + elif not isinstance(chunk, (bytes, bytearray)): + chunk = str(chunk).encode("utf-8") + data.extend(chunk) + if start is not None: + with suppress(Exception): # pragma: no cover - best effort reset + handle.seek(start) # type: ignore[attr-defined] + return bytes(data) if isinstance(content, str): return content.encode("utf-8") return str(content).encode("utf-8") diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py index cc96da314..b925e1fc2 100644 --- a/tests/test_end_to_end.py +++ b/tests/test_end_to_end.py @@ -1,16 +1,22 @@ -import json - +import base64 import csv import gzip +import inspect import json import os +import shutil import zipfile from io import BytesIO, StringIO +from pathlib import Path +from tempfile import SpooledTemporaryFile try: from fastapi.testclient import TestClient # type: ignore except Exception: # pragma: no cover - fastapi is optional in some environments TestClient = None # type: ignore +else: # pragma: no cover - degrade gracefully when using the lightweight stub + if "files" not in inspect.signature(TestClient.post).parameters: # type: ignore[arg-type] + TestClient = None # type: ignore try: from backend.app import create_app @@ -210,6 +216,21 @@ def test_end_to_end_demo_pipeline(): sarif_zip = normalizer.load_sarif(zip_buffer.getvalue()) assert sarif_zip.metadata["finding_count"] == 1 + spooled = SpooledTemporaryFile(max_size=1024, mode="w+b") + spooled.write(gzip.compress(json.dumps(sbom_document).encode("utf-8"))) + spooled.seek(0) + sbom_spooled = normalizer.load_sbom(spooled) + assert sbom_spooled.metadata["component_count"] == 2 + spooled.close() + + sarif_zip_buffer = SpooledTemporaryFile(max_size=1024, mode="w+b") + with zipfile.ZipFile(sarif_zip_buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.writestr("scan.sarif", json.dumps(sarif_document)) + sarif_zip_buffer.seek(0) + sarif_spooled = normalizer.load_sarif(sarif_zip_buffer) + assert sarif_spooled.metadata["finding_count"] == 1 + sarif_zip_buffer.close() + orchestrator = PipelineOrchestrator() pipeline_payload = orchestrator.run( design_dataset=design_dataset, @@ -251,17 +272,20 @@ def test_feedback_endpoint_rejects_invalid_payload(monkeypatch, tmp_path): if TestClient is None or create_app is None: return + safe_root = (Path(__file__).resolve().parent / "tmp_feedback" / tmp_path.name).resolve() + safe_root.mkdir(parents=True, exist_ok=True) + overlay_payload = { "mode": "demo", "auth": {"strategy": "token", "tokens": ["demo-token"]}, - "data": {"feedback_dir": str(tmp_path / "feedback")}, + "data": {"feedback_dir": str(safe_root / "feedback")}, "toggles": {"capture_feedback": True}, } overlay_path = tmp_path / "overlay.json" overlay_path.write_text(json.dumps(overlay_payload), encoding="utf-8") monkeypatch.setenv("FIXOPS_OVERLAY_PATH", str(overlay_path)) - monkeypatch.setenv("FIXOPS_DATA_ROOT_ALLOWLIST", str(tmp_path)) + monkeypatch.setenv("FIXOPS_DATA_ROOT_ALLOWLIST", str(safe_root)) monkeypatch.setenv("FIXOPS_API_TOKEN", "demo-token") try: @@ -278,3 +302,205 @@ def test_feedback_endpoint_rejects_invalid_payload(monkeypatch, tmp_path): monkeypatch.delenv("FIXOPS_OVERLAY_PATH", raising=False) monkeypatch.delenv("FIXOPS_DATA_ROOT_ALLOWLIST", raising=False) monkeypatch.delenv("FIXOPS_API_TOKEN", raising=False) + shutil.rmtree(safe_root, ignore_errors=True) + + +def test_large_compressed_uploads_stream_to_disk(monkeypatch, tmp_path): + if TestClient is None or create_app is None: + normalizer = InputNormalizer() + + components = [] + sbom_document = { + "bomFormat": "CycloneDX", + "specVersion": "1.4", + "version": 1, + "components": components, + } + gz_sbom = b"" + for idx in range(800): + components.append( + { + "type": "library", + "name": f"component-{idx}", + "version": "1.0.0", + "purl": f"pkg:pypi/component-{idx}@1.0.0", + "licenses": [{"license": "MIT"}], + "description": base64.b64encode(os.urandom(4096)).decode("ascii"), + } + ) + gz_sbom = gzip.compress(json.dumps(sbom_document).encode("utf-8")) + if len(gz_sbom) > 1024 * 1024: + break + assert len(gz_sbom) > 1024 * 1024 + + sbom_spool = SpooledTemporaryFile(max_size=1024, mode="w+b") + sbom_spool.write(gz_sbom) + sbom_spool.seek(0) + sbom = normalizer.load_sbom(sbom_spool) + assert sbom.metadata["component_count"] == len(components) + sbom_spool.close() + + cve_entries = [ + { + "cveID": f"CVE-2024-{idx:04d}", + "title": base64.b64encode(os.urandom(512)).decode("ascii"), + "severity": "high", + } + for idx in range(300) + ] + cve_zip = BytesIO() + with zipfile.ZipFile(cve_zip, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.writestr("kev.json", json.dumps({"vulnerabilities": cve_entries})) + cve_zip.seek(0) + cve_spool = SpooledTemporaryFile(max_size=1024, mode="w+b") + cve_spool.write(cve_zip.getvalue()) + cve_spool.seek(0) + cve_norm = normalizer.load_cve_feed(cve_spool) + assert cve_norm.metadata["record_count"] == len(cve_entries) + cve_spool.close() + + sarif_results = [ + { + "ruleId": f"RULE-{idx}", + "level": "warning", + "message": {"text": base64.b64encode(os.urandom(256)).decode("ascii")}, + "locations": [ + { + "physicalLocation": { + "artifactLocation": {"uri": f"src/module_{idx}.py"}, + "region": {"startLine": idx + 1}, + } + } + ], + } + for idx in range(200) + ] + sarif_document = {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "HeavyScanner"}}, "results": sarif_results}]} + sarif_zip = BytesIO() + with zipfile.ZipFile(sarif_zip, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.writestr("scan.sarif", json.dumps(sarif_document)) + sarif_zip.seek(0) + sarif_spool = SpooledTemporaryFile(max_size=1024, mode="w+b") + sarif_spool.write(sarif_zip.getvalue()) + sarif_spool.seek(0) + sarif = normalizer.load_sarif(sarif_spool) + assert sarif.metadata["finding_count"] == len(sarif_results) + sarif_spool.close() + return + + safe_root = (Path(__file__).resolve().parent / "tmp_stream" / tmp_path.name).resolve() + safe_root.mkdir(parents=True, exist_ok=True) + + overlay_payload = { + "mode": "demo", + "auth": {"strategy": "token", "tokens": ["demo-token"]}, + "data": {"archive_dir": str((safe_root / "archive").resolve())}, + "limits": { + "max_upload_bytes": { + "default": 16 * 1024 * 1024, + "sbom": 16 * 1024 * 1024, + "sarif": 16 * 1024 * 1024, + "cve": 16 * 1024 * 1024, + } + }, + } + overlay_path = tmp_path / "overlay.json" + overlay_path.write_text(json.dumps(overlay_payload), encoding="utf-8") + + monkeypatch.setenv("FIXOPS_OVERLAY_PATH", str(overlay_path)) + monkeypatch.setenv("FIXOPS_DATA_ROOT_ALLOWLIST", str(safe_root)) + monkeypatch.setenv("FIXOPS_API_TOKEN", "demo-token") + + try: + app = create_app() + client = TestClient(app) + headers = {"X-API-Key": "demo-token"} + + components = [] + sbom_document = { + "bomFormat": "CycloneDX", + "specVersion": "1.4", + "version": 1, + "components": components, + } + gz_sbom = b"" + for idx in range(800): + components.append( + { + "type": "library", + "name": f"component-{idx}", + "version": "1.0.0", + "purl": f"pkg:pypi/component-{idx}@1.0.0", + "licenses": [{"license": "MIT"}], + "description": base64.b64encode(os.urandom(4096)).decode("ascii"), + } + ) + gz_sbom = gzip.compress(json.dumps(sbom_document).encode("utf-8")) + if len(gz_sbom) > 1024 * 1024: + break + assert len(components) >= 1 + assert len(gz_sbom) > 1024 * 1024 + + response = client.post( + "/inputs/sbom", + headers=headers, + files={"file": ("sbom.json.gz", gz_sbom, "application/gzip")}, + ) + assert response.status_code == 200 + assert response.json()["metadata"]["component_count"] == len(components) + + cve_entries = [ + { + "cveID": f"CVE-2024-{idx:04d}", + "title": base64.b64encode(os.urandom(512)).decode("ascii"), + "severity": "high", + } + for idx in range(300) + ] + cve_zip = BytesIO() + with zipfile.ZipFile(cve_zip, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.writestr("kev.json", json.dumps({"vulnerabilities": cve_entries})) + cve_zip.seek(0) + + response = client.post( + "/inputs/cve", + headers=headers, + files={"file": ("kev.zip", cve_zip.getvalue(), "application/zip")}, + ) + assert response.status_code == 200 + assert response.json()["record_count"] == len(cve_entries) + + sarif_results = [ + { + "ruleId": f"RULE-{idx}", + "level": "warning", + "message": {"text": base64.b64encode(os.urandom(256)).decode("ascii")}, + "locations": [ + { + "physicalLocation": { + "artifactLocation": {"uri": f"src/module_{idx}.py"}, + "region": {"startLine": idx + 1}, + } + } + ], + } + for idx in range(200) + ] + sarif_document = {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "HeavyScanner"}}, "results": sarif_results}]} + sarif_zip = BytesIO() + with zipfile.ZipFile(sarif_zip, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: + archive.writestr("scan.sarif", json.dumps(sarif_document)) + sarif_zip.seek(0) + + response = client.post( + "/inputs/sarif", + headers=headers, + files={"file": ("scan.zip", sarif_zip.getvalue(), "application/zip")}, + ) + assert response.status_code == 200 + assert response.json()["metadata"]["finding_count"] == len(sarif_results) + finally: + monkeypatch.delenv("FIXOPS_OVERLAY_PATH", raising=False) + monkeypatch.delenv("FIXOPS_DATA_ROOT_ALLOWLIST", raising=False) + monkeypatch.delenv("FIXOPS_API_TOKEN", raising=False) + shutil.rmtree(safe_root, ignore_errors=True)