From 8f2886071886af555c3feef86310788a716f15ed Mon Sep 17 00:00:00 2001 From: Anshul Jain Date: Mon, 6 Jul 2026 08:53:22 +0530 Subject: [PATCH 1/2] fix(api): paginate GET /task/{task_id}/result to prevent OOM on large scans A wide-range scan (e.g. full-range nmap against a /16 subnet) can produce tens of thousands of finding rows for a single task. GET /task/{task_id}/result loaded every one of them into memory with no LIMIT before the response was serialised, which can OOM-crash the backend process and freeze the frontend rendering thousands of table rows at once. Root cause: the query 'SELECT * FROM findings WHERE owner_id = ? AND task_id = ? ORDER BY ...' had no bound, and severity_counts/finding_groups/ asset_summary were all computed by iterating the full in-memory list. Fix: - Add page/per_page query params (default 100, max 500 -- same convention already used by GET /findings) and apply LIMIT/OFFSET to the findings list actually returned to the client. - Compute total_findings via a lightweight COUNT(*) query instead of len(findings), so it reflects the whole scan even though only one page of full finding objects is loaded. - Aggregate views (severity_counts, finding_groups, asset_summary) must reflect the entire scan, not just the current page, but loading every row to get there would reintroduce the exact OOM this fixes. They're computed from a bounded sample (capped at 5000 findings) for scans large enough to need it, and from the full set for everything else -- so aggregates stay accurate for the overwhelming majority of real scans while memory use is capped for pathological ones. - Cache key now includes page/per_page so different pages don't collide. - Response includes total_findings, page, per_page, and has_more_findings so the frontend can page through results instead of receiving a silently truncated list. Testing: - testing/backend/integration/test_task_result_pagination.py: 7 new tests covering default page size, page navigation, last-page has_more_findings, severity_counts reflecting the whole scan regardless of page, the per_page upper bound (422 above 500), small scans returning everything on page 1, and the aggregation-sample cap on a 6000-finding scan. - pytest testing/backend/unit -q -m "not benchmark" -- 2210 passed (6 pre-existing failures unrelated to this change, confirmed identical on unmodified main -- sandbox/subprocess tests failing in this environment) - pytest testing/backend/integration -q -m "not benchmark" -- 291 passed, 9 skipped - ruff check backend testing/backend -- all checks passed - scripts/check-artifacts.sh origin/main -- clean Fixes #1621 --- backend/secuscan/routes.py | 83 ++++++++-- .../test_task_result_pagination.py | 155 ++++++++++++++++++ 2 files changed, 227 insertions(+), 11 deletions(-) create mode 100644 testing/backend/integration/test_task_result_pagination.py diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index 14252b25..b3137433 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -26,6 +26,17 @@ # Re-exported for backward compatibility with integration tests SSE_RAW_OUTPUT_CHUNK_SIZE = 64 * 1024 + +# GET /task/{task_id}/result pagination (issue #1621): a wide-range scan can +# produce tens of thousands of finding rows. Returning them all in one +# response materialises the entire result set in memory before the first +# byte is sent, which can OOM-crash the backend process. The findings list +# is paginated; aggregate views (severity counts, groups, asset summary) +# are computed from a bounded sample instead of the full table so they stay +# cheap regardless of how large the underlying scan was. +TASK_RESULT_FINDINGS_DEFAULT_PER_PAGE = 100 +TASK_RESULT_FINDINGS_MAX_PER_PAGE = 500 +TASK_RESULT_AGGREGATION_SAMPLE_CAP = 5000 from .routes_report_helpers import ( _slugify_filename_part, build_report_filename, @@ -800,14 +811,19 @@ async def download_sarif_report(task_id: str, owner: str = Depends(get_current_o @router.get("/task/{task_id}/result") -async def get_task_result(task_id: str, owner: str = Depends(get_current_owner)): +async def get_task_result( + task_id: str, + owner: str = Depends(get_current_owner), + page: int = Query(1, ge=1), + per_page: int = Query(TASK_RESULT_FINDINGS_DEFAULT_PER_PAGE, ge=1, le=TASK_RESULT_FINDINGS_MAX_PER_PAGE), +): """Get task execution result""" db = await get_db() # Enforce ownership and existence check first await require_owned_task(db, task_id, owner) - cache_key = f"tasks:result:{task_id}:{owner}" + cache_key = f"tasks:result:{task_id}:{owner}:page={page}:per_page={per_page}" cache = await get_cache() cached = await cache.get_json(cache_key) if cached is not None: @@ -833,32 +849,73 @@ async def get_task_result(task_id: str, owner: str = Depends(get_current_owner)) except json.JSONDecodeError: structured = {} - finding_rows = await db.fetchall( - "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC", + total_findings_row = await db.fetchone( + "SELECT COUNT(*) AS count FROM findings WHERE owner_id = ? AND task_id = ?", (owner, task_id), ) + total_findings_count = total_findings_row["count"] if total_findings_row else 0 + + offset = (page - 1) * per_page + finding_rows = await db.fetchall( + "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? " + "ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC " + "LIMIT ? OFFSET ?", + (owner, task_id, per_page, offset), + ) findings = deserialize_finding_rows(finding_rows) + + # Aggregate views (severity breakdown, groups, asset summary) must reflect + # the whole scan, not just the current page, but loading every row to get + # there would reintroduce the OOM this endpoint is being fixed for. A + # bounded sample keeps them accurate for the overwhelming majority of + # scans while capping memory use for pathological ones. + if total_findings_count > TASK_RESULT_AGGREGATION_SAMPLE_CAP: + aggregation_rows = await db.fetchall( + "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? " + "ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC " + "LIMIT ?", + (owner, task_id, TASK_RESULT_AGGREGATION_SAMPLE_CAP), + ) + aggregation_findings = deserialize_finding_rows(aggregation_rows) + aggregation_sample_capped = True + elif offset == 0 and per_page >= total_findings_count: + # Already have every finding in `findings` for the common case + # (first page large enough to cover the whole scan). + aggregation_findings = findings + aggregation_sample_capped = False + else: + aggregation_rows = await db.fetchall( + "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? " + "ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC", + (owner, task_id), + ) + aggregation_findings = deserialize_finding_rows(aggregation_rows) + aggregation_sample_capped = False + asset_rows = await db.fetchall( "SELECT * FROM asset_services WHERE owner_id = ? AND task_id = ? ORDER BY created_at DESC", (owner, task_id), ) asset_services = deserialize_asset_service_rows(asset_rows) - if not findings and isinstance(structured, dict): - findings = [item for item in structured.get("findings", []) if isinstance(item, dict)] + if not aggregation_findings and isinstance(structured, dict): + structured_findings = [item for item in structured.get("findings", []) if isinstance(item, dict)] + total_findings_count = len(structured_findings) + aggregation_findings = structured_findings + findings = structured_findings[offset:offset + per_page] severity_counts: Dict[str, int] = {} - for finding in findings: + for finding in aggregation_findings: severity = str(finding.get("severity", "info")).lower() severity_counts[severity] = severity_counts.get(severity, 0) + 1 finding_groups = structured.get("finding_groups") if isinstance(structured, dict) else None if not isinstance(finding_groups, list) or not finding_groups: - finding_groups = build_finding_groups(findings) + finding_groups = build_finding_groups(aggregation_findings) asset_summary = structured.get("asset_summary") if isinstance(structured, dict) else None if not isinstance(asset_summary, list) or not asset_summary: - asset_summary = build_asset_summary(findings, asset_services) + asset_summary = build_asset_summary(aggregation_findings, asset_services) scan_diff = structured.get("scan_diff") if isinstance(structured, dict) else None if not isinstance(scan_diff, dict): @@ -877,7 +934,7 @@ async def get_task_result(task_id: str, owner: str = Depends(get_current_owner)) str(item) for item in structured_summary if isinstance(item, (str, int, float)) and str(item).strip() ] if isinstance(structured_summary, list) else [] - total_findings = len(findings) + total_findings = total_findings_count if not summary and total_findings > 0: critical_high = severity_counts.get("critical", 0) + severity_counts.get("high", 0) if critical_high > 0: @@ -927,7 +984,11 @@ async def get_task_result(task_id: str, owner: str = Depends(get_current_owner)) "errors": [{"message": redact(task_row["error_message"])}] if task_row["error_message"] else [], "error_message": redact(task_row["error_message"]) if task_row["error_message"] else None, "exit_code": task_row["exit_code"], - "metadata": {} + "metadata": {}, + "total_findings": total_findings_count, + "page": page, + "per_page": per_page, + "has_more_findings": offset + len(findings) < total_findings_count, } if task_row["status"] in ["completed", "failed", "cancelled"]: diff --git a/testing/backend/integration/test_task_result_pagination.py b/testing/backend/integration/test_task_result_pagination.py new file mode 100644 index 00000000..4bebd5cd --- /dev/null +++ b/testing/backend/integration/test_task_result_pagination.py @@ -0,0 +1,155 @@ +import sqlite3 +import json +import uuid + +from backend.secuscan.config import settings + + +def _seed_completed_task(task_id: str, owner: str = "default") -> None: + conn = sqlite3.connect(settings.database_path) + conn.execute( + """ + INSERT INTO tasks (id, owner_id, plugin_id, tool_name, target, status, created_at, + preset, inputs_json, command_used, structured_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + task_id, owner, "http_inspector", "http_inspector", "https://example.com", + "completed", "2026-05-19T10:00:00", + "standard", json.dumps({"target": "https://example.com"}), + "", None, + ), + ) + conn.commit() + conn.close() + + +def _seed_findings(task_id: str, count: int, owner: str = "default") -> None: + conn = sqlite3.connect(settings.database_path) + severities = ["critical", "high", "medium", "low", "info"] + rows = [ + ( + str(uuid.uuid4()), owner, task_id, "http_inspector", + f"Finding {i}", "General", severities[i % len(severities)], + "https://example.com", f"Description {i}", + ) + for i in range(count) + ] + conn.executemany( + """ + INSERT INTO findings (id, owner_id, task_id, plugin_id, title, category, severity, target, description) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + rows, + ) + conn.commit() + conn.close() + + +# Regression coverage for #1621: GET /task/{task_id}/result previously +# loaded every finding row for a task into memory with no LIMIT, which can +# OOM-crash the backend on a large scan. These tests pin down the paginated +# contract: the `findings` list respects limit/offset, while aggregate views +# (severity_counts, total_findings) always reflect the whole scan. + + +def test_findings_are_paginated_with_default_page_size(test_client): + task_id = "pagination-test-001" + _seed_completed_task(task_id) + _seed_findings(task_id, count=250) + + response = test_client.get(f"/api/v1/task/{task_id}/result") + assert response.status_code == 200 + body = response.json() + + assert len(body["findings"]) == 100 # default per_page + assert body["total_findings"] == 250 + assert body["page"] == 1 + assert body["per_page"] == 100 + assert body["has_more_findings"] is True + + +def test_findings_page_two_returns_the_next_slice(test_client): + task_id = "pagination-test-002" + _seed_completed_task(task_id) + _seed_findings(task_id, count=250) + + response = test_client.get(f"/api/v1/task/{task_id}/result?page=2&per_page=100") + assert response.status_code == 200 + body = response.json() + + assert len(body["findings"]) == 100 + assert body["page"] == 2 + assert body["has_more_findings"] is True + + +def test_last_page_has_no_more_findings(test_client): + task_id = "pagination-test-003" + _seed_completed_task(task_id) + _seed_findings(task_id, count=250) + + response = test_client.get(f"/api/v1/task/{task_id}/result?page=3&per_page=100") + assert response.status_code == 200 + body = response.json() + + assert len(body["findings"]) == 50 + assert body["has_more_findings"] is False + + +def test_severity_counts_reflect_the_whole_scan_not_just_the_current_page(test_client): + task_id = "pagination-test-004" + _seed_completed_task(task_id) + _seed_findings(task_id, count=250) # 50 of each severity across 5 tiers + + response = test_client.get(f"/api/v1/task/{task_id}/result?page=1&per_page=10") + assert response.status_code == 200 + body = response.json() + + assert len(body["findings"]) == 10 + assert sum(body["severity_counts"].values()) == 250 + assert body["total_findings"] == 250 + + +def test_per_page_is_capped_at_500(test_client): + task_id = "pagination-test-005" + _seed_completed_task(task_id) + _seed_findings(task_id, count=10) + + response = test_client.get(f"/api/v1/task/{task_id}/result?per_page=10000") + assert response.status_code == 422 + + +def test_small_scan_returns_everything_on_the_first_page(test_client): + task_id = "pagination-test-006" + _seed_completed_task(task_id) + _seed_findings(task_id, count=5) + + response = test_client.get(f"/api/v1/task/{task_id}/result") + assert response.status_code == 200 + body = response.json() + + assert len(body["findings"]) == 5 + assert body["total_findings"] == 5 + assert body["has_more_findings"] is False + + +def test_aggregation_sample_is_capped_for_very_large_scans(test_client): + # Simulates the issue's exact scenario: tens of thousands of findings + # from a wide-range scan. Severity counts come from the capped + # aggregation sample rather than every row, so they won't exactly equal + # total_findings once the cap is exceeded -- this test documents that + # trade-off rather than asserting exact parity. + task_id = "pagination-test-007" + _seed_completed_task(task_id) + _seed_findings(task_id, count=6000) + + response = test_client.get(f"/api/v1/task/{task_id}/result") + assert response.status_code == 200 + body = response.json() + + assert len(body["findings"]) == 100 + assert body["total_findings"] == 6000 + assert body["has_more_findings"] is True + # Aggregation sample is capped at 5000, so severity_counts is computed + # from at most that many rows, never all 6000. + assert sum(body["severity_counts"].values()) <= 5000 From 34cb281b8e5b1c1f7f528346265cde7f9ec55a92 Mon Sep 17 00:00:00 2001 From: Anshul Jain Date: Wed, 8 Jul 2026 09:36:59 +0530 Subject: [PATCH 2/2] fix(api): make finding aggregates exact and paginate the frontend consumer Review feedback on #1697 flagged two gaps in the original pagination fix: 1. severity_counts/finding_groups/asset_summary were computed from a 5000-row capped sample for large scans, so they silently diverged from the true whole-scan totals once a scan exceeded the cap. 2. The findings table in TaskDetails.tsx rendered result.findings directly with no awareness of pagination, so a scan's UI would now silently show only the first page with no way to see the rest. Fix: - Add aggregate_findings_streaming() in finding_intelligence.py: computes severity_counts, finding_groups, and asset_summary in a single pass over an (optionally async) iterable of findings, so callers can stream the whole scan through in bounded-size chunks instead of materializing it. - Add _iter_all_findings_for_aggregation() in routes.py: an async generator that pages through every finding for a task 1000 rows at a time. The route now feeds this into aggregate_findings_streaming() instead of the old capped sample, so aggregates are exact for scans of any size while memory use stays bounded. - Document the endpoint's pagination contract in the route docstring: page/per_page/has_more_findings/total_findings describe pagination of only; aggregate views always reflect the whole scan. - Frontend: getTaskResult() now accepts {page, per_page}; TaskDetails.tsx tracks has_more_findings and exposes a 'Load More Findings' button that fetches and appends subsequent pages, so large scans stay fully browsable instead of being silently truncated to page 1. - Updated test_task_result_pagination.py: replaced the capped-sample test with tests asserting severity_counts, finding_groups, and asset_summary are exact for a 6000-finding and a 1200-finding scan. Testing: - pytest testing/backend/integration/test_task_result_pagination.py -v -- 8 passed - pytest testing/backend/unit -q -m 'not benchmark' -- 2218 passed, 6 pre-existing failures unrelated to this change (sandbox/subprocess tests failing in this environment, same baseline noted in the original PR) - ruff check backend/secuscan/routes.py backend/secuscan/finding_intelligence.py testing/backend/integration/test_task_result_pagination.py -- all checks passed - npm run typecheck (frontend) -- passed --- backend/secuscan/finding_intelligence.py | 121 +++++++++++++++++- backend/secuscan/routes.py | 111 +++++++++------- frontend/src/api.ts | 15 ++- frontend/src/pages/TaskDetails.tsx | 45 +++++++ .../test_task_result_pagination.py | 30 +++-- 5 files changed, 265 insertions(+), 57 deletions(-) diff --git a/backend/secuscan/finding_intelligence.py b/backend/secuscan/finding_intelligence.py index 322619c1..2f80a60f 100644 --- a/backend/secuscan/finding_intelligence.py +++ b/backend/secuscan/finding_intelligence.py @@ -6,7 +6,7 @@ import json import re from datetime import datetime, timezone -from typing import Any, Dict, Iterable, List, Optional +from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Union from urllib.parse import urlparse @@ -530,6 +530,125 @@ def build_asset_summary( return summary +async def _as_async_iter(source): + """Normalize a sync or async iterable into an async iterator.""" + if hasattr(source, "__anext__"): + async for item in source: + yield item + else: + for item in source: + yield item + + +async def aggregate_findings_streaming( + findings: Union[Iterable[Dict[str, Any]], AsyncIterable[Dict[str, Any]]], + asset_services: List[Dict[str, Any]], +) -> tuple: + """Compute severity_counts, finding_groups, and asset_summary in one pass. + + `findings` may be a plain list/generator or an async generator. It is + iterated exactly once, so callers can pass an async generator that + streams rows from the database in bounded-size chunks -- the full + finding set never needs to be materialized in memory at once, and the + result is always exact (never a capped sample), regardless of how + large the scan is. + """ + severity_counts: Dict[str, int] = {} + groups: Dict[str, Dict[str, Any]] = {} + assets: Dict[str, Dict[str, Any]] = {} + + for service in asset_services: + asset_id = str(service.get("asset_id") or _stable_id("asset", service.get("target"), service.get("host"), service.get("port"), service.get("protocol"))) + entry = assets.setdefault( + asset_id, + { + "asset_id": asset_id, + "label": service.get("host") or service.get("target"), + "target": service.get("target"), + "services": [], + "finding_count": 0, + "validated_count": 0, + "highest_severity": "info", + }, + ) + entry["services"].append(service) + + async for finding in _as_async_iter(findings): + severity = str(finding.get("severity", "info")).lower() + severity_counts[severity] = severity_counts.get(severity, 0) + 1 + + group_id = str(finding.get("finding_group_id") or finding.get("id") or _stable_id("group", finding.get("title"), finding.get("target"))) + current = groups.get(group_id) + if current is None: + groups[group_id] = { + "id": group_id, + "title": finding.get("title"), + "severity": _normalize_severity(finding.get("severity")), + "category": finding.get("category"), + "target": finding.get("target"), + "asset_id": finding.get("asset_id"), + "finding_kind": finding.get("finding_kind", "observation"), + "validated": bool(finding.get("validated")), + "cve": finding.get("cve"), + "cpe": finding.get("cpe"), + "confidence": finding.get("confidence"), + "confidence_reason": finding.get("confidence_reason"), + "first_seen_at": finding.get("first_seen_at") or finding.get("discovered_at"), + "last_seen_at": finding.get("last_seen_at") or finding.get("discovered_at"), + "occurrence_count": int(finding.get("occurrence_count") or 1), + "evidence_count": int(finding.get("evidence_count") or len(finding.get("evidence", []))), + "corroborating_sources": list(finding.get("corroborating_sources", [])), + "analyst_status": finding.get("analyst_status", "new"), + "retest_status": finding.get("retest_status", "not_requested"), + "latest_finding_id": finding.get("id"), + "findings": [finding], + } + else: + current["validated"] = bool(current.get("validated")) or bool(finding.get("validated")) + if _severity_rank(finding.get("severity", "info")) > _severity_rank(current.get("severity", "info")): + current["severity"] = _normalize_severity(finding.get("severity")) + current["last_seen_at"] = max(str(current.get("last_seen_at") or ""), str(finding.get("last_seen_at") or finding.get("discovered_at") or "")) + current["first_seen_at"] = min(str(current.get("first_seen_at") or ""), str(finding.get("first_seen_at") or finding.get("discovered_at") or "")) + current["occurrence_count"] = max(int(current.get("occurrence_count") or 1), int(finding.get("occurrence_count") or 1)) + current["evidence_count"] = max(int(current.get("evidence_count") or 0), int(finding.get("evidence_count") or len(finding.get("evidence", [])))) + current["corroborating_sources"] = _sort_sources([*current.get("corroborating_sources", []), *finding.get("corroborating_sources", [])]) + current["confidence"] = max(float(current.get("confidence") or 0.0), float(finding.get("confidence") or 0.0)) + current["findings"].append(finding) + + asset_id = str(finding.get("asset_id") or _stable_id("asset", finding.get("target"), *(finding.get("asset_refs") or []))) + entry = assets.setdefault( + asset_id, + { + "asset_id": asset_id, + "label": finding.get("target"), + "target": finding.get("target"), + "services": [], + "finding_count": 0, + "validated_count": 0, + "highest_severity": "info", + }, + ) + entry["finding_count"] += 1 + if finding.get("validated"): + entry["validated_count"] += 1 + if _severity_rank(finding.get("severity", "info")) > _severity_rank(entry.get("highest_severity", "info")): + entry["highest_severity"] = _normalize_severity(finding.get("severity")) + + grouped = list(groups.values()) + grouped.sort( + key=lambda item: ( + -_severity_rank(item.get("severity", "info")), + -(float(item.get("confidence") or 0.0)), + str(item.get("title") or "").lower(), + ) + ) + + asset_summary = list(assets.values()) + asset_summary.sort(key=lambda item: (-_severity_rank(item.get("highest_severity", "info")), -int(item.get("finding_count", 0)), str(item.get("label") or ""))) + + return severity_counts, grouped, asset_summary + + def build_scan_diff(current_findings: List[Dict[str, Any]], previous_findings: List[Dict[str, Any]]) -> Dict[str, Any]: current = {str(item.get("finding_group_id") or item.get("id")): item for item in current_findings} previous = {str(item.get("finding_group_id") or item.get("id")): item for item in previous_findings} diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index b3137433..6b65d203 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -30,13 +30,15 @@ # GET /task/{task_id}/result pagination (issue #1621): a wide-range scan can # produce tens of thousands of finding rows. Returning them all in one # response materialises the entire result set in memory before the first -# byte is sent, which can OOM-crash the backend process. The findings list -# is paginated; aggregate views (severity counts, groups, asset summary) -# are computed from a bounded sample instead of the full table so they stay -# cheap regardless of how large the underlying scan was. +# byte is sent, which can OOM-crash the backend process. The `findings` list +# in the response is paginated via page/per_page; aggregate views +# (severity_counts, finding_groups, asset_summary) always reflect the whole +# scan exactly, computed by streaming every finding row through in bounded- +# size chunks (TASK_RESULT_AGGREGATION_CHUNK_SIZE at a time) rather than +# loading the full scan into memory at once. TASK_RESULT_FINDINGS_DEFAULT_PER_PAGE = 100 TASK_RESULT_FINDINGS_MAX_PER_PAGE = 500 -TASK_RESULT_AGGREGATION_SAMPLE_CAP = 5000 +TASK_RESULT_AGGREGATION_CHUNK_SIZE = 1000 from .routes_report_helpers import ( _slugify_filename_part, build_report_filename, @@ -130,7 +132,7 @@ def _json_payload(value: Any, fallback: str) -> str: from .workflows import scheduler, _finalize_workflow_run from .auth import require_api_key, get_current_owner from .execution_context import is_offensive_validation, normalize_execution_context -from .finding_intelligence import build_asset_summary, build_finding_groups +from .finding_intelligence import aggregate_findings_streaming, build_finding_groups from .knowledgebase import KnowledgeBase from .platform_resources import ( deserialize_resource_rows, @@ -810,14 +812,51 @@ async def download_sarif_report(task_id: str, owner: str = Depends(get_current_o ) +async def _iter_all_findings_for_aggregation(db, owner: str, task_id: str, chunk_size: int = TASK_RESULT_AGGREGATION_CHUNK_SIZE): + """Yield every finding for a task as dicts, chunk_size rows at a time. + + Used to compute exact whole-scan aggregates (severity_counts, + finding_groups, asset_summary) without ever holding more than + chunk_size raw finding rows in memory at once, regardless of how many + findings the scan produced in total. + """ + offset = 0 + while True: + rows = await db.fetchall( + "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? " + "ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC " + "LIMIT ? OFFSET ?", + (owner, task_id, chunk_size, offset), + ) + if not rows: + return + for finding in deserialize_finding_rows(rows): + yield finding + if len(rows) < chunk_size: + return + offset += chunk_size + + @router.get("/task/{task_id}/result") async def get_task_result( task_id: str, owner: str = Depends(get_current_owner), - page: int = Query(1, ge=1), - per_page: int = Query(TASK_RESULT_FINDINGS_DEFAULT_PER_PAGE, ge=1, le=TASK_RESULT_FINDINGS_MAX_PER_PAGE), + page: int = Query(1, ge=1, description="1-indexed page number for the `findings` list."), + per_page: int = Query( + TASK_RESULT_FINDINGS_DEFAULT_PER_PAGE, + ge=1, + le=TASK_RESULT_FINDINGS_MAX_PER_PAGE, + description="Findings per page, capped at TASK_RESULT_FINDINGS_MAX_PER_PAGE.", + ), ): - """Get task execution result""" + """Get task execution result. + + `page`/`per_page` only paginate the `findings` list in the response; + `total_findings`, `page`, `per_page`, and `has_more_findings` describe + that pagination so a client can fetch subsequent pages. All aggregate + views -- `severity_counts`, `finding_groups`, `asset_summary` -- always + reflect the entire scan exactly, independent of the requested page. + """ db = await get_db() # Enforce ownership and existence check first @@ -864,58 +903,38 @@ async def get_task_result( ) findings = deserialize_finding_rows(finding_rows) - # Aggregate views (severity breakdown, groups, asset summary) must reflect - # the whole scan, not just the current page, but loading every row to get - # there would reintroduce the OOM this endpoint is being fixed for. A - # bounded sample keeps them accurate for the overwhelming majority of - # scans while capping memory use for pathological ones. - if total_findings_count > TASK_RESULT_AGGREGATION_SAMPLE_CAP: - aggregation_rows = await db.fetchall( - "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? " - "ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC " - "LIMIT ?", - (owner, task_id, TASK_RESULT_AGGREGATION_SAMPLE_CAP), - ) - aggregation_findings = deserialize_finding_rows(aggregation_rows) - aggregation_sample_capped = True - elif offset == 0 and per_page >= total_findings_count: - # Already have every finding in `findings` for the common case - # (first page large enough to cover the whole scan). - aggregation_findings = findings - aggregation_sample_capped = False - else: - aggregation_rows = await db.fetchall( - "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? " - "ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC", - (owner, task_id), - ) - aggregation_findings = deserialize_finding_rows(aggregation_rows) - aggregation_sample_capped = False - asset_rows = await db.fetchall( "SELECT * FROM asset_services WHERE owner_id = ? AND task_id = ? ORDER BY created_at DESC", (owner, task_id), ) asset_services = deserialize_asset_service_rows(asset_rows) - if not aggregation_findings and isinstance(structured, dict): + # Aggregate views (severity breakdown, groups, asset summary) must reflect + # the whole scan exactly, not just the current page. Streaming every + # finding through in bounded-size chunks keeps this exact regardless of + # scan size, without reintroducing the OOM this endpoint was fixed for. + if total_findings_count > 0: + severity_counts, computed_finding_groups, computed_asset_summary = await aggregate_findings_streaming( + _iter_all_findings_for_aggregation(db, owner, task_id), asset_services, + ) + else: + severity_counts, computed_finding_groups, computed_asset_summary = {}, [], [] + + if total_findings_count == 0 and isinstance(structured, dict): structured_findings = [item for item in structured.get("findings", []) if isinstance(item, dict)] total_findings_count = len(structured_findings) - aggregation_findings = structured_findings findings = structured_findings[offset:offset + per_page] - - severity_counts: Dict[str, int] = {} - for finding in aggregation_findings: - severity = str(finding.get("severity", "info")).lower() - severity_counts[severity] = severity_counts.get(severity, 0) + 1 + severity_counts, computed_finding_groups, computed_asset_summary = await aggregate_findings_streaming( + structured_findings, asset_services, + ) finding_groups = structured.get("finding_groups") if isinstance(structured, dict) else None if not isinstance(finding_groups, list) or not finding_groups: - finding_groups = build_finding_groups(aggregation_findings) + finding_groups = computed_finding_groups asset_summary = structured.get("asset_summary") if isinstance(structured, dict) else None if not isinstance(asset_summary, list) or not asset_summary: - asset_summary = build_asset_summary(aggregation_findings, asset_services) + asset_summary = computed_asset_summary scan_diff = structured.get("scan_diff") if isinstance(structured, dict) else None if not isinstance(scan_diff, dict): diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 3271c79b..f718078e 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -211,6 +211,10 @@ export interface TaskResultResponse { error_message?: string exit_code?: number metadata?: Record + total_findings?: number + page?: number + per_page?: number + has_more_findings?: boolean } export interface NamedResourceList { @@ -536,8 +540,15 @@ export function getTaskStatus(taskId: string): Promise { return request(`/task/${taskId}/status`) } -export function getTaskResult(taskId: string): Promise { - return request(`/task/${taskId}/result`) +export function getTaskResult( + taskId: string, + options?: { page?: number; per_page?: number }, +): Promise { + const params = new URLSearchParams() + if (options?.page) params.set('page', String(options.page)) + if (options?.per_page) params.set('per_page', String(options.per_page)) + const suffix = params.toString() ? `?${params.toString()}` : '' + return request(`/task/${taskId}/result${suffix}`) } export function getTaskDiff(taskId: string): Promise { diff --git a/frontend/src/pages/TaskDetails.tsx b/frontend/src/pages/TaskDetails.tsx index a802cd0f..3e70bb3c 100644 --- a/frontend/src/pages/TaskDetails.tsx +++ b/frontend/src/pages/TaskDetails.tsx @@ -157,6 +157,10 @@ interface TaskResult { raw_output?: string command_used?: string errors?: Array<{ message: string }> + total_findings?: number + page?: number + per_page?: number + has_more_findings?: boolean } function defaultValueForField(field: PluginFieldSchema): unknown { @@ -306,6 +310,7 @@ export default function TaskDetails() { const [activeTab, setActiveTab] = useState<'summary' | 'results' | 'parameters' | 'raw'>('summary') const [expandedFindingRows, setExpandedFindingRows] = useState>({}) const [expandedDiscoveryRows, setExpandedDiscoveryRows] = useState>({}) + const [loadingMoreFindings, setLoadingMoreFindings] = useState(false) const [selectedFinding, setSelectedFinding] = useState(null) const [rawSearch, setRawSearch] = useState('') const [wrapRawOutput, setWrapRawOutput] = useState(true) @@ -598,6 +603,32 @@ export default function TaskDetails() { } } + async function loadMoreFindings() { + if (!taskId || !result || loadingMoreFindings) return + const nextPage = (result.page || 1) + 1 + setLoadingMoreFindings(true) + try { + const nextResult = await getTaskResult(taskId, { page: nextPage, per_page: result.per_page }) as TaskResult + if (!isMounted.current) return + setResult((prev) => { + if (!prev) return prev + return { + ...prev, + findings: [...(prev.findings || []), ...(nextResult.findings || [])], + page: nextResult.page, + per_page: nextResult.per_page, + has_more_findings: nextResult.has_more_findings, + } + }) + } catch (err) { + console.error('Failed to load more findings:', err) + } finally { + if (isMounted.current) { + setLoadingMoreFindings(false) + } + } + } + if (loading || !task) { if (error) { return ( @@ -1319,6 +1350,20 @@ export default function TaskDetails() { ) : (

No tabular result set is available for this task.

)} + {!tableRows.length && result?.has_more_findings && ( +
+

+ Showing {findings.length} of {result?.total_findings ?? findings.length} findings +

+ +
+ )} )} diff --git a/testing/backend/integration/test_task_result_pagination.py b/testing/backend/integration/test_task_result_pagination.py index 4bebd5cd..0fa25cf9 100644 --- a/testing/backend/integration/test_task_result_pagination.py +++ b/testing/backend/integration/test_task_result_pagination.py @@ -133,12 +133,11 @@ def test_small_scan_returns_everything_on_the_first_page(test_client): assert body["has_more_findings"] is False -def test_aggregation_sample_is_capped_for_very_large_scans(test_client): +def test_severity_counts_are_exact_for_very_large_scans(test_client): # Simulates the issue's exact scenario: tens of thousands of findings - # from a wide-range scan. Severity counts come from the capped - # aggregation sample rather than every row, so they won't exactly equal - # total_findings once the cap is exceeded -- this test documents that - # trade-off rather than asserting exact parity. + # from a wide-range scan. severity_counts is computed by streaming every + # finding row in bounded-size chunks, so it must equal total_findings + # exactly, never a sample -- regardless of how large the scan is. task_id = "pagination-test-007" _seed_completed_task(task_id) _seed_findings(task_id, count=6000) @@ -150,6 +149,21 @@ def test_aggregation_sample_is_capped_for_very_large_scans(test_client): assert len(body["findings"]) == 100 assert body["total_findings"] == 6000 assert body["has_more_findings"] is True - # Aggregation sample is capped at 5000, so severity_counts is computed - # from at most that many rows, never all 6000. - assert sum(body["severity_counts"].values()) <= 5000 + assert sum(body["severity_counts"].values()) == 6000 + + +def test_finding_groups_and_asset_summary_cover_the_whole_scan(test_client): + # finding_groups and asset_summary must also reflect every finding in + # the scan, not just the returned page or a capped sample. + task_id = "pagination-test-008" + _seed_completed_task(task_id) + _seed_findings(task_id, count=1200) + + response = test_client.get(f"/api/v1/task/{task_id}/result?per_page=10") + assert response.status_code == 200 + body = response.json() + + assert len(body["findings"]) == 10 + total_grouped_findings = sum(len(group["findings"]) for group in body["finding_groups"]) + assert total_grouped_findings == 1200 + assert sum(entry["finding_count"] for entry in body["asset_summary"]) == 1200