diff --git a/backend/secuscan/finding_intelligence.py b/backend/secuscan/finding_intelligence.py index 8dcae71d..b96f4a38 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 @@ -543,6 +543,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 f4b3289c..97188e75 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -26,6 +26,19 @@ # 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 +# 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_CHUNK_SIZE = 1000 from .routes_report_helpers import ( _slugify_filename_part, build_report_filename, @@ -119,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, @@ -819,15 +832,57 @@ 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)): - """Get task execution result""" +async def get_task_result( + task_id: str, + owner: str = Depends(get_current_owner), + 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. + + `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 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: @@ -853,32 +908,53 @@ 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) + 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)] - - severity_counts: Dict[str, int] = {} - for finding in findings: - severity = str(finding.get("severity", "info")).lower() - severity_counts[severity] = severity_counts.get(severity, 0) + 1 + # 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) + findings = structured_findings[offset:offset + per_page] + 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(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(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): @@ -897,7 +973,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: @@ -947,7 +1023,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/frontend/src/api.ts b/frontend/src/api.ts index 58c6aae7..a11eaf56 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 { @@ -557,8 +561,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 new file mode 100644 index 00000000..0fa25cf9 --- /dev/null +++ b/testing/backend/integration/test_task_result_pagination.py @@ -0,0 +1,169 @@ +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_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 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) + + 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 + 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