@cryptotaxi247 / CoPilot / commits / 16aa7ad1

644 sca overview stream (#667)

* feat: add streaming endpoint for SCA results and implement client-side handling * Fix: Updates auth store property name Updates the property name used to retrieve the user token from the auth store in the streamScaOverview function. This ensures that the correct token is used for authorization when streaming SCA overview results. * Feat: Uses SSE client for SCA overview stream Replaces the direct fetch-event-source implementation for streaming SCA overview results with a reusable SSE client. This improves code maintainability and simplifies the handling of authentication and query parameters. The new SSE client automatically manages token injection and parameter serialization. * Feat: Enhances SSE client with lifecycle handlers Improves SSE client by adding support for `onOpen` and `onMessage` lifecycle handlers. This allows to execute code when the SSE connection is opened and when any message is received, improving flexibility and control over the SSE stream. It also ensures that the connection is properly handled based on the response status. Fixes errors thrown when the server does not send a data param with the event. Related to 644-sca-overview-stream * Refactor: Remove commented-out code for collecting SCA results for all agents * precommit-fixes --------- Co-authored-by: Davide Di Modica <webmaster.ddm@gmail.com>

taylor_socfortress committed Feb 2, 2026 at 13:08 UTC 16aa7ad14ac7487c77c24f5ef0e65d3ffef095a3
9 files changed +1022 -222
backend/app/agents/sca/routes/sca.py
+85
@@ -1,4 +1,6 @@
1 +import json
2 from typing import Any
3 +from typing import AsyncGenerator
4 from typing import Dict
5 from typing import Optional
6
@@ -9,6 +11,7 @@ from fastapi import HTTPException
11 from fastapi import Query
12 from fastapi import Response
13 from fastapi import Security
14 +from fastapi.responses import StreamingResponse
15 from loguru import logger
16 from sqlalchemy.ext.asyncio import AsyncSession
17
@@ -23,6 +26,7 @@ from app.agents.sca.services.sca import get_sca_report_download
26 from app.agents.sca.services.sca import get_sca_statistics
27 from app.agents.sca.services.sca import list_sca_reports
28 from app.agents.sca.services.sca import search_sca_overview
29 +from app.agents.sca.services.sca import stream_sca_for_all_agents
30 from app.auth.models.users import User
31 from app.auth.routes.auth import AuthHandler
32 from app.db.db_session import get_db
@@ -134,6 +138,87 @@ async def search_sca_results_overview(
138 raise HTTPException(status_code=500, detail=f"Failed to search SCA results: {e}")
139
140
141 +@sca_router.get(
142 + "/overview/stream",
143 + description="Stream SCA results across all agents as they are collected",
144 + dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))],
145 +)
146 +async def stream_sca_results_overview(
147 + customer_code: Optional[str] = Query(None, description="Filter by customer code"),
148 + agent_name: Optional[str] = Query(None, description="Filter by agent hostname"),
149 + policy_id: Optional[str] = Query(None, description="Filter by specific policy ID"),
150 + policy_name: Optional[str] = Query(None, description="Filter by policy name (partial matching)"),
151 + min_score: Optional[int] = Query(None, description="Filter by minimum score (0-100)", ge=0, le=100),
152 + max_score: Optional[int] = Query(None, description="Filter by maximum score (0-100)", ge=0, le=100),
153 + db: AsyncSession = Depends(get_db),
154 +) -> StreamingResponse:
155 + """
156 + Stream SCA results as Server-Sent Events (SSE).
157 +
158 + Results are sent as they are collected from each agent, allowing the frontend
159 + to display data progressively without waiting for all agents to complete.
160 +
161 + **Event Types:**
162 + - `start`: Initial event with total agent count
163 + - `agent_result`: SCA results for a single agent
164 + - `agent_error`: Error collecting data from an agent
165 + - `progress`: Progress update (agents processed so far)
166 + - `complete`: Final event with summary statistics
167 + - `error`: Fatal error that stops the stream
168 +
169 + **Example Events:**
170 + ```
171 + event: start
172 + data: {"total_agents": 50, "message": "Starting SCA collection..."}
173 +
174 + event: agent_result
175 + data: {"agent_id": "001", "agent_name": "server1", "policies": [...]}
176 +
177 + event: progress
178 + data: {"processed": 10, "total": 50, "successful": 8, "failed": 2}
179 +
180 + event: complete
181 + data: {"total_results": 150, "total_agents": 50, "average_score": 78.5, ...}
182 + ```
183 + """
184 + logger.info(
185 + f"Streaming SCA overview with filters: "
186 + f"customer_code={customer_code}, agent_name={agent_name}, "
187 + f"policy_id={policy_id}, policy_name={policy_name}, "
188 + f"min_score={min_score}, max_score={max_score}",
189 + )
190 +
191 + async def event_generator() -> AsyncGenerator[str, None]:
192 + try:
193 + async for event in stream_sca_for_all_agents(
194 + db_session=db,
195 + customer_code=customer_code,
196 + agent_name=agent_name,
197 + policy_id=policy_id,
198 + policy_name=policy_name,
199 + min_score=min_score,
200 + max_score=max_score,
201 + ):
202 + # Format as SSE
203 + event_type = event.get("event", "message")
204 + data = json.dumps(event.get("data", {}))
205 + yield f"event: {event_type}\ndata: {data}\n\n"
206 + except Exception as e:
207 + logger.error(f"Error in SSE stream: {e}")
208 + error_data = json.dumps({"error": str(e), "message": "Stream error occurred"})
209 + yield f"event: error\ndata: {error_data}\n\n"
210 +
211 + return StreamingResponse(
212 + event_generator(),
213 + media_type="text/event-stream",
214 + headers={
215 + "Cache-Control": "no-cache",
216 + "Connection": "keep-alive",
217 + "X-Accel-Buffering": "no", # Disable nginx buffering
218 + },
219 + )
220 +
221 +
222 @sca_router.get(
223 "/stats",
224 response_model=ScaStatsResponse,
backend/app/agents/sca/services/sca.py
+225 -160
@@ -6,6 +6,7 @@ import json
6 from asyncio import Semaphore
7 from datetime import datetime
8 from typing import Any
9 +from typing import AsyncGenerator
10 from typing import Dict
11 from typing import List
12 from typing import Optional
@@ -66,166 +67,6 @@ async def get_all_agents_from_db(
67 raise HTTPException(status_code=500, detail=f"Failed to fetch agents: {e}")
68
69
69 -# async def collect_sca_for_all_agents(
70 -# db_session: AsyncSession,
71 -# customer_code: Optional[str] = None,
72 -# agent_name: Optional[str] = None,
73 -# policy_id: Optional[str] = None,
74 -# policy_name: Optional[str] = None,
75 -# min_score: Optional[int] = None,
76 -# max_score: Optional[int] = None,
77 -# ) -> List[AgentScaOverviewItem]:
78 -# """
79 -# Collect SCA results for all agents from Wazuh Manager
80 -
81 -# Args:
82 -# db_session: Database session to use
83 -# customer_code: Optional customer code filter
84 -# agent_name: Optional agent name filter
85 -# policy_id: Optional policy ID filter
86 -# policy_name: Optional policy name filter (partial matching)
87 -# min_score: Optional minimum score filter
88 -# max_score: Optional maximum score filter
89 -
90 -# Returns:
91 -# List of AgentScaOverviewItem objects
92 -# """
93 -# try:
94 -# # Get agents from database
95 -# agents = await get_all_agents_from_db(db_session, customer_code)
96 -
97 -# if not agents:
98 -# logger.warning("No agents found" + (f" for customer {customer_code}" if customer_code else ""))
99 -# return []
100 -
101 -# all_sca_results = []
102 -
103 -# for agent in agents:
104 -# # Skip if agent name filter is specified and doesn't match
105 -# if agent_name and agent.hostname != agent_name:
106 -# continue
107 -
108 -# try:
109 -# logger.info(f"Collecting SCA results for agent: {agent.hostname}")
110 -
111 -# # Collect SCA data from Wazuh Manager for this agent
112 -# sca_response = await collect_agent_sca(agent.agent_id)
113 -
114 -# if not sca_response.success or not sca_response.sca:
115 -# logger.warning(f"No SCA data found for agent {agent.hostname}")
116 -# continue
117 -
118 -# # Process each SCA policy result for this agent
119 -# for sca_result in sca_response.sca:
120 -# # Apply filters
121 -# if policy_id and sca_result.policy_id != policy_id:
122 -# continue
123 -
124 -# if policy_name and policy_name.lower() not in sca_result.name.lower():
125 -# continue
126 -
127 -# if min_score is not None and sca_result.score < min_score:
128 -# continue
129 -
130 -# if max_score is not None and sca_result.score > max_score:
131 -# continue
132 -
133 -# # Create overview item
134 -# overview_item = AgentScaOverviewItem(
135 -# agent_id=agent.agent_id,
136 -# agent_name=agent.hostname,
137 -# customer_code=agent.customer_code,
138 -# policy_id=sca_result.policy_id,
139 -# policy_name=sca_result.name,
140 -# description=sca_result.description,
141 -# total_checks=sca_result.total_checks,
142 -# pass_count=sca_result.pass_count,
143 -# fail_count=sca_result.fail,
144 -# invalid_count=sca_result.invalid,
145 -# score=sca_result.score,
146 -# start_scan=sca_result.start_scan,
147 -# end_scan=sca_result.end_scan,
148 -# references=sca_result.references,
149 -# hash_file=sca_result.hash_file,
150 -# )
151 -
152 -# all_sca_results.append(overview_item)
153 -
154 -# except Exception as e:
155 -# logger.error(f"Error collecting SCA for agent {agent.hostname}: {e}")
156 -# # Continue with other agents even if one fails
157 -# continue
158 -
159 -# logger.info(f"Collected SCA results for {len(all_sca_results)} policy results across agents")
160 -# return all_sca_results
161 -
162 -# except Exception as e:
163 -# logger.error(f"Error collecting SCA for all agents: {e}")
164 -# raise HTTPException(status_code=500, detail=f"Failed to collect SCA results: {e}")
165 -
166 -
167 -# async def search_sca_overview(
168 -# db_session: AsyncSession,
169 -# customer_code: Optional[str] = None,
170 -# agent_name: Optional[str] = None,
171 -# policy_id: Optional[str] = None,
172 -# policy_name: Optional[str] = None,
173 -# min_score: Optional[int] = None,
174 -# max_score: Optional[int] = None,
175 -# page: int = 1,
176 -# page_size: int = 50,
177 -# ) -> ScaOverviewResponse:
178 -# """
179 -# Search SCA results across all agents with filtering and pagination
180 -
181 -# Args:
182 -# db_session: Database session for agent lookup
183 -# customer_code: Optional customer code filter
184 -# agent_name: Optional agent hostname filter
185 -# policy_id: Optional policy ID filter
186 -# policy_name: Optional policy name filter (partial matching)
187 -# min_score: Optional minimum score filter
188 -# max_score: Optional maximum score filter
189 -# page: Page number for pagination
190 -# page_size: Number of results per page
191 -
192 -# Returns:
193 -# ScaOverviewResponse with paginated results and statistics
194 -# """
195 -# logger.info(
196 -# f"Searching SCA overview with filters: customer_code={customer_code}, "
197 -# f"agent_name={agent_name}, policy_id={policy_id}, policy_name={policy_name}, "
198 -# f"min_score={min_score}, max_score={max_score}, page={page}, page_size={page_size}",
199 -# )
200 -
201 -# # Build filters applied dict for response
202 -# filters_applied = {}
203 -# if customer_code:
204 -# filters_applied["customer_code"] = customer_code
205 -# if agent_name:
206 -# filters_applied["agent_name"] = agent_name
207 -# if policy_id:
208 -# filters_applied["policy_id"] = policy_id
209 -# if policy_name:
210 -# filters_applied["policy_name"] = policy_name
211 -# if min_score is not None:
212 -# filters_applied["min_score"] = min_score
213 -# if max_score is not None:
214 -# filters_applied["max_score"] = max_score
215 -
216 -# try:
217 -# # Collect all SCA results with filtering
218 -# all_sca_results = await collect_sca_for_all_agents(
219 -# db_session=db_session,
220 -# customer_code=customer_code,
221 -# agent_name=agent_name,
222 -# policy_id=policy_id,
223 -# policy_name=policy_name,
224 -# min_score=min_score,
225 -# max_score=max_score,
226 -# )
227 -
228 -
70 async def collect_sca_for_single_agent(
71 agent: Agents,
72 semaphore: Semaphore,
@@ -1152,3 +993,227 @@ async def delete_sca_report(
993 except Exception as e:
994 logger.error(f"Error deleting SCA report: {e}")
995 raise HTTPException(status_code=500, detail=f"Failed to delete report: {e}")
996 +
997 +
998 +async def stream_sca_for_all_agents(
999 + db_session: AsyncSession,
1000 + customer_code: Optional[str] = None,
1001 + agent_name: Optional[str] = None,
1002 + policy_id: Optional[str] = None,
1003 + policy_name: Optional[str] = None,
1004 + min_score: Optional[int] = None,
1005 + max_score: Optional[int] = None,
1006 + max_concurrent_requests: int = DEFAULT_MAX_CONCURRENT_REQUESTS,
1007 +) -> AsyncGenerator[Dict[str, Any], None]:
1008 + """
1009 + Stream SCA results for all agents as they are collected.
1010 +
1011 + Yields SSE-formatted events as results come in from each agent.
1012 +
1013 + Args:
1014 + db_session: Database session to use
1015 + customer_code: Optional customer code filter
1016 + agent_name: Optional agent name filter
1017 + policy_id: Optional policy ID filter
1018 + policy_name: Optional policy name filter (partial matching)
1019 + min_score: Optional minimum score filter
1020 + max_score: Optional maximum score filter
1021 + max_concurrent_requests: Maximum concurrent API requests
1022 +
1023 + Yields:
1024 + Dict with 'event' type and 'data' payload
1025 + """
1026 + try:
1027 + # Get agents from database
1028 + agents = await get_all_agents_from_db(db_session, customer_code)
1029 +
1030 + if not agents:
1031 + yield {
1032 + "event": "complete",
1033 + "data": {
1034 + "total_results": 0,
1035 + "total_agents": 0,
1036 + "message": "No agents found",
1037 + },
1038 + }
1039 + return
1040 +
1041 + # Filter agents by name if specified
1042 + if agent_name:
1043 + agents = [a for a in agents if a.hostname == agent_name]
1044 + if not agents:
1045 + yield {
1046 + "event": "complete",
1047 + "data": {
1048 + "total_results": 0,
1049 + "total_agents": 0,
1050 + "message": f"No agents found matching hostname: {agent_name}",
1051 + },
1052 + }
1053 + return
1054 +
1055 + total_agents = len(agents)
1056 +
1057 + # Send start event
1058 + yield {
1059 + "event": "start",
1060 + "data": {
1061 + "total_agents": total_agents,
1062 + "message": f"Starting SCA collection for {total_agents} agents...",
1063 + },
1064 + }
1065 +
1066 + # Create semaphore for rate limiting
1067 + semaphore = Semaphore(max_concurrent_requests)
1068 +
1069 + # Track statistics
1070 + all_results: List[AgentScaOverviewItem] = []
1071 + processed_count = 0
1072 + successful_count = 0
1073 + failed_count = 0
1074 +
1075 + # Create a queue to receive results as they complete
1076 + result_queue: asyncio.Queue = asyncio.Queue()
1077 +
1078 + async def collect_and_queue(agent: Agents):
1079 + """Collect SCA for an agent and put result in queue"""
1080 + result = await collect_sca_for_single_agent(
1081 + agent=agent,
1082 + semaphore=semaphore,
1083 + policy_id=policy_id,
1084 + policy_name=policy_name,
1085 + min_score=min_score,
1086 + max_score=max_score,
1087 + )
1088 + await result_queue.put((agent, result))
1089 +
1090 + # Start all tasks
1091 + tasks = [asyncio.create_task(collect_and_queue(agent)) for agent in agents]
1092 +
1093 + # Process results as they come in
1094 + for _ in range(total_agents):
1095 + try:
1096 + # Wait for next result with timeout
1097 + agent, results = await asyncio.wait_for(result_queue.get(), timeout=60.0) # 60 second timeout per agent
1098 +
1099 + processed_count += 1
1100 +
1101 + if results:
1102 + successful_count += 1
1103 + all_results.extend(results)
1104 +
1105 + # Yield agent results
1106 + yield {
1107 + "event": "agent_result",
1108 + "data": {
1109 + "agent_id": agent.agent_id,
1110 + "agent_name": agent.hostname,
1111 + "customer_code": agent.customer_code,
1112 + "policy_count": len(results),
1113 + "policies": [
1114 + {
1115 + "policy_id": r.policy_id,
1116 + "policy_name": r.policy_name,
1117 + "description": r.description,
1118 + "total_checks": r.total_checks,
1119 + "pass_count": r.pass_count,
1120 + "fail_count": r.fail_count,
1121 + "invalid_count": r.invalid_count,
1122 + "score": r.score,
1123 + "start_scan": r.start_scan,
1124 + "end_scan": r.end_scan,
1125 + "references": r.references,
1126 + "hash_file": r.hash_file,
1127 + }
1128 + for r in results
1129 + ],
1130 + },
1131 + }
1132 + else:
1133 + # Agent had no SCA data (not necessarily an error)
1134 + yield {
1135 + "event": "agent_empty",
1136 + "data": {
1137 + "agent_id": agent.agent_id,
1138 + "agent_name": agent.hostname,
1139 + "message": "No SCA data available",
1140 + },
1141 + }
1142 +
1143 + # Yield progress update every 5 agents or on last agent
1144 + if processed_count % 5 == 0 or processed_count == total_agents:
1145 + yield {
1146 + "event": "progress",
1147 + "data": {
1148 + "processed": processed_count,
1149 + "total": total_agents,
1150 + "successful": successful_count,
1151 + "failed": failed_count,
1152 + "results_so_far": len(all_results),
1153 + "percent_complete": round((processed_count / total_agents) * 100, 1),
1154 + },
1155 + }
1156 +
1157 + except asyncio.TimeoutError:
1158 + failed_count += 1
1159 + processed_count += 1
1160 + yield {
1161 + "event": "agent_error",
1162 + "data": {
1163 + "agent_id": "unknown",
1164 + "message": "Timeout waiting for agent response",
1165 + },
1166 + }
1167 + except Exception as e:
1168 + failed_count += 1
1169 + processed_count += 1
1170 + logger.error(f"Error processing agent result: {e}")
1171 + yield {
1172 + "event": "agent_error",
1173 + "data": {
1174 + "message": str(e),
1175 + },
1176 + }
1177 +
1178 + # Wait for all tasks to complete (cleanup)
1179 + await asyncio.gather(*tasks, return_exceptions=True)
1180 +
1181 + # Calculate final statistics
1182 + unique_agents = set(item.agent_id for item in all_results)
1183 + unique_policies = set(item.policy_id for item in all_results)
1184 +
1185 + total_checks = sum(item.total_checks for item in all_results)
1186 + total_passes = sum(item.pass_count for item in all_results)
1187 + total_fails = sum(item.fail_count for item in all_results)
1188 + total_invalid = sum(item.invalid_count for item in all_results)
1189 +
1190 + average_score = sum(item.score for item in all_results) / len(all_results) if all_results else 0.0
1191 +
1192 + # Yield completion event
1193 + yield {
1194 + "event": "complete",
1195 + "data": {
1196 + "total_results": len(all_results),
1197 + "total_agents": len(unique_agents),
1198 + "total_policies": len(unique_policies),
1199 + "average_score": round(average_score, 2),
1200 + "total_checks": total_checks,
1201 + "total_passes": total_passes,
1202 + "total_fails": total_fails,
1203 + "total_invalid": total_invalid,
1204 + "agents_processed": processed_count,
1205 + "agents_successful": successful_count,
1206 + "agents_failed": failed_count,
1207 + "message": f"Completed SCA collection: {len(all_results)} results from {len(unique_agents)} agents",
1208 + },
1209 + }
1210 +
1211 + except Exception as e:
1212 + logger.error(f"Error in SCA streaming: {e}")
1213 + yield {
1214 + "event": "error",
1215 + "data": {
1216 + "error": str(e),
1217 + "message": "Fatal error during SCA collection",
1218 + },
1219 + }
frontend/package.json
+1
@@ -47,6 +47,7 @@
47 "@fontsource/jetbrains-mono": "^5.2.8",
48 "@fontsource/lexend": "^5.2.11",
49 "@fontsource/public-sans": "^5.2.7",
50 + "@microsoft/fetch-event-source": "^2.0.1",
51 "@shikijs/markdown-it": "^3.19.0",
52 "@singulio/app-auth-search": "^0.0.3",
53 "@types/codemirror": "^5.60.17",
frontend/pnpm-lock.yaml
+8
@@ -41,6 +41,9 @@ importers:
41 '@fontsource/public-sans':
42 specifier: ^5.2.7
43 version: 5.2.7
44 + '@microsoft/fetch-event-source':
45 + specifier: ^2.0.1
46 + version: 2.0.1
47 '@shikijs/markdown-it':
48 specifier: ^3.19.0
49 version: 3.19.0
@@ -990,6 +993,9 @@ packages:
993 '@marijn/find-cluster-break@1.0.2':
994 resolution: {integrity: sha512-l0h88YhZFyKdXIFNfSWpyjStDjGHwZ/U7iobcK1cQQD8sejsONdQtTVU+1wVN1PBw40PiiHB1vA5S7VTfQiP9g==}
995
996 + '@microsoft/fetch-event-source@2.0.1':
997 + resolution: {integrity: sha512-W6CLUJ2eBMw3Rec70qrsEW0jOm/3twwJv21mrmj2yORiaVmVYGS4sSS5yUwvQc1ZlDLYGPnClVWmUUMagKNsfA==}
998 +
999 '@nodelib/fs.scandir@2.1.5':
1000 resolution: {integrity: sha512-vq24Bq3ym5HEQm2NKCr3yXDwjc7vTsEThRDnkp2DK9p1uqLR+DHurm/NOTo0KG7HYHU7eppKZj3MyqYuMBf62g==}
1001 engines: {node: '>= 8'}
@@ -5717,6 +5723,8 @@ snapshots:
5723
5724 '@marijn/find-cluster-break@1.0.2': {}
5725
5726 + '@microsoft/fetch-event-source@2.0.1': {}
5727 +
5728 '@nodelib/fs.scandir@2.1.5':
5729 dependencies:
5730 '@nodelib/fs.stat': 2.0.5
frontend/src/api/endpoints/sca.ts
+103 -61
@@ -1,72 +1,114 @@
1 import type {
2 - ScaOverviewQuery,
3 - ScaOverviewResponse,
4 - SCAReportDeleteResponse,
5 - SCAReportGenerateRequest,
6 - SCAReportGenerateResponse,
7 - SCAReportListResponse,
8 - ScaStatsResponse
2 + ScaOverviewQuery,
3 + ScaOverviewResponse,
4 + SCAReportDeleteResponse,
5 + SCAReportGenerateRequest,
6 + SCAReportGenerateResponse,
7 + SCAReportListResponse,
8 + ScaStatsResponse
9 } from "@/types/sca.d"
10 import { HttpClient } from "../httpClient"
11 +import { createSSEStream } from "../sseClient"
12
13 export default {
13 - /**
14 - * Search SCA results across all agents with filtering and pagination
15 - */
16 - searchScaOverview(query?: ScaOverviewQuery, signal?: AbortSignal) {
17 - return HttpClient.get<ScaOverviewResponse>(`/sca/overview`, {
18 - params: {
19 - customer_code: query?.customer_code,
20 - agent_name: query?.agent_name,
21 - policy_id: query?.policy_id,
22 - policy_name: query?.policy_name,
23 - min_score: query?.min_score,
24 - max_score: query?.max_score,
25 - page: query?.page || 1,
26 - page_size: query?.page_size || 50
27 - },
28 - signal
29 - })
30 - },
14 + /**
15 + * Search SCA results across all agents with filtering and pagination
16 + */
17 + searchScaOverview(query?: ScaOverviewQuery, signal?: AbortSignal) {
18 + return HttpClient.get<ScaOverviewResponse>(`/sca/overview`, {
19 + params: {
20 + customer_code: query?.customer_code,
21 + agent_name: query?.agent_name,
22 + policy_id: query?.policy_id,
23 + policy_name: query?.policy_name,
24 + min_score: query?.min_score,
25 + max_score: query?.max_score,
26 + page: query?.page || 1,
27 + page_size: query?.page_size || 50
28 + },
29 + signal
30 + })
31 + },
32
32 - /**
33 - * Get SCA statistics
34 - */
35 - getScaStats(customer_code?: string) {
36 - return HttpClient.get<ScaStatsResponse>(`/sca/stats`, {
37 - params: customer_code ? { customer_code } : undefined
38 - })
39 - },
33 + /**
34 + * Stream SCA results using SSE client (token + params gestiti automaticamente)
35 + */
36 + async streamScaOverview(
37 + query: ScaOverviewQuery | undefined,
38 + handlers: {
39 + onStart?: (data: any) => void
40 + onAgentResult?: (data: any) => void
41 + onAgentEmpty?: (data: any) => void
42 + onProgress?: (data: any) => void
43 + onComplete?: (data: any) => void
44 + onError?: (error: any) => void
45 + },
46 + abortController?: AbortController
47 + ): Promise<void> {
48 + await createSSEStream({
49 + path: "/sca/overview/stream",
50 + params: query
51 + ? {
52 + customer_code: query.customer_code,
53 + agent_name: query.agent_name,
54 + policy_id: query.policy_id,
55 + policy_name: query.policy_name,
56 + min_score: query.min_score,
57 + max_score: query.max_score
58 + }
59 + : undefined,
60 + signal: abortController?.signal,
61 + handlers: {
62 + start: data => handlers.onStart?.(data),
63 + agent_result: data => handlers.onAgentResult?.(data),
64 + agent_empty: data => handlers.onAgentEmpty?.(data),
65 + progress: data => handlers.onProgress?.(data),
66 + complete: data => handlers.onComplete?.(data),
67 + error: data => handlers.onError?.(data),
68 + agent_error: data => handlers.onError?.(data),
69 + onError: err => handlers.onError?.(err)
70 + }
71 + })
72 + },
73
41 - /**
42 - * Generate an SCA report (synchronous)
43 - */
44 - generateReport(request: SCAReportGenerateRequest) {
45 - return HttpClient.post<SCAReportGenerateResponse>(`/sca/reports/generate`, request)
46 - },
74 + /**
75 + * Get SCA statistics
76 + */
77 + getScaStats(customer_code?: string) {
78 + return HttpClient.get<ScaStatsResponse>(`/sca/stats`, {
79 + params: customer_code ? { customer_code } : undefined
80 + })
81 + },
82
48 - /**
49 - * List all SCA reports
50 - */
51 - listReports(customer_code?: string) {
52 - return HttpClient.get<SCAReportListResponse>(`/sca/reports`, {
53 - params: customer_code ? { customer_code } : undefined
54 - })
55 - },
83 + /**
84 + * Generate an SCA report (synchronous)
85 + */
86 + generateReport(request: SCAReportGenerateRequest) {
87 + return HttpClient.post<SCAReportGenerateResponse>(`/sca/reports/generate`, request)
88 + },
89
57 - /**
58 - * Download an SCA report
59 - */
60 - downloadReport(reportId: number) {
61 - return HttpClient.get<Blob>(`/sca/reports/${reportId}/download`, {
62 - responseType: "blob"
63 - })
64 - },
90 + /**
91 + * List all SCA reports
92 + */
93 + listReports(customer_code?: string) {
94 + return HttpClient.get<SCAReportListResponse>(`/sca/reports`, {
95 + params: customer_code ? { customer_code } : undefined
96 + })
97 + },
98
66 - /**
67 - * Delete an SCA report
68 - */
69 - deleteReport(reportId: number) {
70 - return HttpClient.delete<SCAReportDeleteResponse>(`/sca/reports/${reportId}`)
71 - }
99 + /**
100 + * Download an SCA report
101 + */
102 + downloadReport(reportId: number) {
103 + return HttpClient.get<Blob>(`/sca/reports/${reportId}/download`, {
104 + responseType: "blob"
105 + })
106 + },
107 +
108 + /**
109 + * Delete an SCA report
110 + */
111 + deleteReport(reportId: number) {
112 + return HttpClient.delete<SCAReportDeleteResponse>(`/sca/reports/${reportId}`)
113 + }
114 }
frontend/src/api/sseClient.ts new
+92
@@ -0,0 +1,92 @@
1 +import { fetchEventSource } from "@microsoft/fetch-event-source"
2 +import { useAuthStore } from "@/stores/auth"
3 +import { HttpClient } from "./httpClient"
4 +
5 +/** Converte Record<string, string | number> in query string, escludendo undefined/null */
6 +function paramsToQueryString(params?: Record<string, string | number | undefined>): string {
7 + if (!params || Object.keys(params).length === 0) {
8 + return ""
9 + }
10 + const search = new URLSearchParams()
11 + for (const [key, value] of Object.entries(params)) {
12 + if (value !== undefined && value !== null) {
13 + search.append(key, String(value))
14 + }
15 + }
16 + const qs = search.toString()
17 + return qs ? `?${qs}` : ""
18 +}
19 +
20 +export interface SSEClientOptions {
21 + /** Path dell'endpoint (es. "/sca/overview/stream") */
22 + path: string
23 + /** HTTP method (default: "GET") */
24 + method?: "GET" | "POST" | "PUT" | "PATCH" | "DELETE"
25 + /** Base URL (default: HttpClient.baseURL ovvero "/api") */
26 + baseURL?: string
27 + /** Parametri convertiti automaticamente in query string */
28 + params?: Record<string, string | number | undefined>
29 + /** Handler per tipo di evento. Chiave = nome evento SSE (es. "start", "agent_result") + onOpen, onMessage, onError */
30 + handlers: Record<string, (data: unknown) => void> & {
31 + onOpen?: (response: Response) => void | Promise<void>
32 + onMessage?: (event: { event?: string; data: string }) => void
33 + onError?: (error: unknown) => void
34 + }
35 + /** AbortController per cancellare lo stream */
36 + signal?: AbortSignal
37 +}
38 +
39 +export async function createSSEStream(options: SSEClientOptions): Promise<void> {
40 + const { path, params, handlers, signal } = options
41 + const method = options.method ?? "GET"
42 + const baseURL = options.baseURL ?? HttpClient.defaults.baseURL ?? "/api"
43 + const authStore = useAuthStore()
44 +
45 + const base = (baseURL ?? "").replace(/\/$/, "")
46 + const cleanPath = path.startsWith("/") ? path : `/${path}`
47 + const queryString = paramsToQueryString(params)
48 + const url = `${base}${cleanPath}${queryString}`
49 +
50 + const headers: Record<string, string> = {}
51 + if (authStore.userToken) {
52 + headers.Authorization = `Bearer ${authStore.userToken}`
53 + }
54 +
55 + await fetchEventSource(url, {
56 + method,
57 + headers,
58 + signal,
59 + onopen(response) {
60 + if (!response.ok) {
61 + throw new Error(`Failed to connect: ${response.status} ${response.statusText}`)
62 + }
63 + handlers.onOpen?.(response)
64 + return Promise.resolve()
65 + },
66 + onmessage(event) {
67 + handlers.onMessage?.({ event: event.event, data: event.data ?? "" })
68 +
69 + if (!event.data) {
70 + return
71 + }
72 +
73 + try {
74 + const data = JSON.parse(event.data)
75 + const eventName = event.event || "message"
76 + const lifecycleKeys = ["onOpen", "onMessage", "onError"] // nomi handler lifecycle, non eventi SSE
77 + if (!lifecycleKeys.includes(eventName)) {
78 + const handler = handlers[eventName]
79 + if (handler) {
80 + handler(data)
81 + }
82 + }
83 + } catch (e) {
84 + console.error("Failed to parse SSE data:", e)
85 + }
86 + },
87 + onerror(err) {
88 + handlers.onError?.(err)
89 + throw err
90 + }
91 + })
92 +}
frontend/src/components/sca/StreamingList.vue new
+462
@@ -0,0 +1,462 @@
1 +<template>
2 + <div class="sca-streaming-list">
3 + <!-- Progress Header -->
4 + <n-card v-if="isStreaming || streamComplete" class="mb-4">
5 + <div class="flex flex-col gap-4">
6 + <!-- Progress Bar -->
7 + <div class="flex items-center gap-4">
8 + <n-progress
9 + type="line"
10 + :percentage="progress.percent_complete"
11 + :status="streamError ? 'error' : streamComplete ? 'success' : 'default'"
12 + :show-indicator="true"
13 + class="flex-grow"
14 + />
15 + <n-button
16 + v-if="!isStreaming && !streamComplete"
17 + type="primary"
18 + @click="startStream"
19 + :loading="isConnecting"
20 + >
21 + <template #icon>
22 + <Icon :name="RefreshIcon" />
23 + </template>
24 + Load SCA Data
25 + </n-button>
26 + <n-button
27 + v-if="isStreaming"
28 + type="error"
29 + @click="stopStream"
30 + >
31 + <template #icon>
32 + <Icon :name="StopIcon" />
33 + </template>
34 + Stop
35 + </n-button>
36 + <n-button
37 + v-if="streamComplete"
38 + @click="startStream"
39 + >
40 + <template #icon>
41 + <Icon :name="RefreshIcon" />
42 + </template>
43 + Refresh
44 + </n-button>
45 + </div>
46 +
47 + <!-- Status Text -->
48 + <div class="flex items-center justify-between text-sm">
49 + <span class="text-secondary">
50 + {{ statusMessage }}
51 + </span>
52 + <div class="flex gap-4">
53 + <span>
54 + Agents:
55 + <code class="text-success">{{ progress.successful }}</code>
56 + /
57 + <code>{{ progress.total }}</code>
58 + <code v-if="progress.failed > 0" class="text-error ml-1">
59 + ({{ progress.failed }} failed)
60 + </code>
61 + </span>
62 + <span>
63 + Results: <code>{{ results.length }}</code>
64 + </span>
65 + </div>
66 + </div>
67 + </div>
68 + </n-card>
69 +
70 + <!-- Statistics Summary (shown when complete) -->
71 + <n-card v-if="streamComplete && stats" class="mb-4">
72 + <div class="flex flex-wrap justify-between gap-4">
73 + <n-statistic label="Total Agents" :value="stats.total_agents" />
74 + <n-statistic label="Total Policies" :value="stats.total_policies" />
75 + <n-statistic label="Average Score">
76 + <template #default>
77 + <span :class="getScoreClass(stats.average_score)">
78 + {{ stats.average_score }}%
79 + </span>
80 + </template>
81 + </n-statistic>
82 + <n-statistic label="Checks" :value="stats.total_checks" />
83 + <n-statistic label="Passed" :value="stats.total_passes" class="text-success" />
84 + <n-statistic label="Failed" :value="stats.total_fails" class="text-error" />
85 + </div>
86 + </n-card>
87 +
88 + <!-- Filters -->
89 + <n-card class="mb-4">
90 + <div class="flex flex-wrap gap-4">
91 + <n-select
92 + v-model:value="filters.customer_code"
93 + placeholder="All Customers"
94 + :options="customerOptions"
95 + clearable
96 + class="w-48"
97 + @update:value="onFilterChange"
98 + />
99 + <n-input
100 + v-model:value="filters.agent_name"
101 + placeholder="Agent Name"
102 + clearable
103 + class="w-48"
104 + @update:value="onFilterChange"
105 + />
106 + <n-input
107 + v-model:value="filters.policy_name"
108 + placeholder="Policy Name"
109 + clearable
110 + class="w-48"
111 + @update:value="onFilterChange"
112 + />
113 + <n-input-number
114 + v-model:value="filters.min_score"
115 + placeholder="Min Score"
116 + :min="0"
117 + :max="100"
118 + clearable
119 + class="w-32"
120 + @update:value="onFilterChange"
121 + />
122 + <n-input-number
123 + v-model:value="filters.max_score"
124 + placeholder="Max Score"
125 + :min="0"
126 + :max="100"
127 + clearable
128 + class="w-32"
129 + @update:value="onFilterChange"
130 + />
131 + </div>
132 + </n-card>
133 +
134 + <!-- Results List -->
135 + <div class="results-container">
136 + <n-spin :show="isConnecting">
137 + <!-- Empty State -->
138 + <n-empty
139 + v-if="!isStreaming && !streamComplete && filteredResults.length === 0"
140 + description="Click 'Load SCA Data' to start collecting results"
141 + class="py-12"
142 + >
143 + <template #extra>
144 + <n-button type="primary" @click="startStream">
145 + Load SCA Data
146 + </n-button>
147 + </template>
148 + </n-empty>
149 +
150 + <!-- Results Table -->
151 + <n-data-table
152 + v-else
153 + :columns="columns"
154 + :data="paginatedResults"
155 + :pagination="pagination"
156 + :loading="isConnecting"
157 + :row-key="(row: AgentScaOverviewItem) => `${row.agent_id}-${row.policy_id}`"
158 + striped
159 + />
160 + </n-spin>
161 + </div>
162 +
163 + <!-- Error Display -->
164 + <n-alert v-if="streamError" type="error" class="mt-4" closable @close="streamError = null">
165 + <template #header>Stream Error</template>
166 + {{ streamError }}
167 + </n-alert>
168 + </div>
169 +</template>
170 +
171 +<script setup lang="ts">
172 +import type {
173 + AgentScaOverviewItem,
174 + ScaOverviewQuery,
175 + ScaStreamComplete,
176 + ScaStreamProgress
177 +} from "@/types/sca.d"
178 +import type { DataTableColumns } from "naive-ui"
179 +import {
180 + NAlert,
181 + NButton,
182 + NCard,
183 + NDataTable,
184 + NEmpty,
185 + NInput,
186 + NInputNumber,
187 + NProgress,
188 + NSelect,
189 + NSpin,
190 + NStatistic,
191 + useMessage
192 +} from "naive-ui"
193 +import { computed, h, onBeforeUnmount, reactive, ref } from "vue"
194 +import Api from "@/api"
195 +import Icon from "@/components/common/Icon.vue"
196 +import Badge from "@/components/common/Badge.vue"
197 +
198 +const RefreshIcon = "carbon:refresh"
199 +const StopIcon = "carbon:stop"
200 +
201 +const message = useMessage()
202 +
203 +// State
204 +const isConnecting = ref(false)
205 +const isStreaming = ref(false)
206 +const streamComplete = ref(false)
207 +const streamError = ref<string | null>(null)
208 +const results = ref<AgentScaOverviewItem[]>([])
209 +const stats = ref<ScaStreamComplete | null>(null)
210 +const abortController = ref<AbortController | null>(null)
211 +
212 +const progress = reactive<ScaStreamProgress>({
213 + processed: 0,
214 + total: 0,
215 + successful: 0,
216 + failed: 0,
217 + results_so_far: 0,
218 + percent_complete: 0
219 +})
220 +
221 +const filters = reactive<ScaOverviewQuery>({
222 + customer_code: undefined,
223 + agent_name: undefined,
224 + policy_name: undefined,
225 + min_score: undefined,
226 + max_score: undefined
227 +})
228 +
229 +// Customer options (you'd populate this from your API)
230 +const customerOptions = ref<{ label: string; value: string }[]>([])
231 +
232 +// Computed
233 +const statusMessage = computed(() => {
234 + if (isConnecting.value) return "Connecting..."
235 + if (isStreaming.value) return `Collecting SCA data... ${progress.processed}/${progress.total} agents`
236 + if (streamComplete.value) return stats.value?.message || "Collection complete"
237 + return "Ready to load SCA data"
238 +})
239 +
240 +const filteredResults = computed(() => {
241 + return results.value.filter(item => {
242 + if (filters.policy_name && !item.policy_name.toLowerCase().includes(filters.policy_name.toLowerCase())) {
243 + return false
244 + }
245 + return true
246 + })
247 +})
248 +
249 +const pagination = reactive({
250 + page: 1,
251 + pageSize: 25,
252 + showSizePicker: true,
253 + pageSizes: [10, 25, 50, 100],
254 + itemCount: computed(() => filteredResults.value.length),
255 + onChange: (page: number) => {
256 + pagination.page = page
257 + },
258 + onUpdatePageSize: (pageSize: number) => {
259 + pagination.pageSize = pageSize
260 + pagination.page = 1
261 + }
262 +})
263 +
264 +const paginatedResults = computed(() => {
265 + const start = (pagination.page - 1) * pagination.pageSize
266 + const end = start + pagination.pageSize
267 + return filteredResults.value.slice(start, end)
268 +})
269 +
270 +// Table columns
271 +const columns: DataTableColumns<AgentScaOverviewItem> = [
272 + {
273 + title: "Agent",
274 + key: "agent_name",
275 + width: 150,
276 + ellipsis: { tooltip: true }
277 + },
278 + {
279 + title: "Customer",
280 + key: "customer_code",
281 + width: 120
282 + },
283 + {
284 + title: "Policy",
285 + key: "policy_name",
286 + ellipsis: { tooltip: true }
287 + },
288 + {
289 + title: "Checks",
290 + key: "total_checks",
291 + width: 80,
292 + align: "center"
293 + },
294 + {
295 + title: "Passed",
296 + key: "pass_count",
297 + width: 80,
298 + align: "center",
299 + render: (row) => h("span", { class: "text-success" }, row.pass_count)
300 + },
301 + {
302 + title: "Failed",
303 + key: "fail_count",
304 + width: 80,
305 + align: "center",
306 + render: (row) => h("span", { class: "text-error" }, row.fail_count)
307 + },
308 + {
309 + title: "Score",
310 + key: "score",
311 + width: 100,
312 + align: "center",
313 + sorter: (a, b) => a.score - b.score,
314 + render: (row) => h(
315 + Badge,
316 + {
317 + type: "splitted",
318 + color: row.score >= 80 ? "success" : row.score >= 60 ? "warning" : "danger"
319 + },
320 + { label: () => `${row.score}%` }
321 + )
322 + },
323 + {
324 + title: "Last Scan",
325 + key: "end_scan",
326 + width: 160,
327 + render: (row) => new Date(row.end_scan).toLocaleString()
328 + }
329 +]
330 +
331 +// Methods
332 +function getScoreClass(score: number): string {
333 + if (score >= 80) return "text-success"
334 + if (score >= 60) return "text-warning"
335 + return "text-error"
336 +}
337 +
338 +async function startStream() {
339 + // Reset state
340 + results.value = []
341 + stats.value = null
342 + streamError.value = null
343 + streamComplete.value = false
344 + isConnecting.value = true
345 +
346 + Object.assign(progress, {
347 + processed: 0,
348 + total: 0,
349 + successful: 0,
350 + failed: 0,
351 + results_so_far: 0,
352 + percent_complete: 0
353 + })
354 +
355 + // Abort existing connection if any
356 + if (abortController.value) {
357 + abortController.value.abort()
358 + }
359 +
360 + // Create new abort controller
361 + abortController.value = new AbortController()
362 +
363 + // Build query params
364 + const query: ScaOverviewQuery = {}
365 + if (filters.customer_code) query.customer_code = filters.customer_code
366 + if (filters.agent_name) query.agent_name = filters.agent_name
367 + if (filters.policy_name) query.policy_name = filters.policy_name
368 + if (filters.min_score !== undefined) query.min_score = filters.min_score
369 + if (filters.max_score !== undefined) query.max_score = filters.max_score
370 +
371 + try {
372 + await Api.sca.streamScaOverview(
373 + query,
374 + {
375 + onStart(data) {
376 + isConnecting.value = false
377 + isStreaming.value = true
378 + progress.total = data.total_agents
379 + message.info(data.message)
380 + },
381 + onAgentResult(data) {
382 + // Add all policies from this agent
383 + for (const policy of data.policies) {
384 + results.value.push({
385 + agent_id: data.agent_id,
386 + agent_name: data.agent_name,
387 + customer_code: data.customer_code,
388 + ...policy
389 + })
390 + }
391 + },
392 + onAgentEmpty(data) {
393 + // Agent had no SCA data - could log or display if needed
394 + console.debug(`Agent ${data.agent_name} has no SCA data`)
395 + },
396 + onProgress(data) {
397 + Object.assign(progress, data)
398 + },
399 + onComplete(data) {
400 + stats.value = data
401 + isStreaming.value = false
402 + streamComplete.value = true
403 +
404 + // Sort results by score (lowest first)
405 + results.value.sort((a, b) => a.score - b.score)
406 +
407 + message.success(data.message)
408 + },
409 + onError(error) {
410 + const errorMessage = error?.message || error?.error || "Unknown error"
411 + console.warn("Stream error:", error)
412 +
413 + // Only set error if we haven't completed successfully
414 + if (!streamComplete.value) {
415 + streamError.value = errorMessage
416 + }
417 + progress.failed++
418 + }
419 + },
420 + abortController.value
421 + )
422 + } catch (error: any) {
423 + // Don't show error for intentional abort
424 + if (error.name !== "AbortError") {
425 + streamError.value = error.message || "Connection error"
426 + console.error("Stream connection error:", error)
427 + }
428 + } finally {
429 + isStreaming.value = false
430 + isConnecting.value = false
431 + }
432 +}
433 +
434 +function stopStream() {
435 + if (abortController.value) {
436 + abortController.value.abort()
437 + abortController.value = null
438 + }
439 + isStreaming.value = false
440 + isConnecting.value = false
441 + message.warning("Stream stopped by user")
442 +}
443 +
444 +function onFilterChange() {
445 + // Debounce and restart stream with new filters if already streaming
446 + // Or just filter client-side if data is already loaded
447 + pagination.page = 1
448 +}
449 +
450 +// Cleanup on unmount
451 +onBeforeUnmount(() => {
452 + if (abortController.value) {
453 + abortController.value.abort()
454 + }
455 +})
456 +</script>
457 +
458 +<style scoped>
459 +.results-container {
460 + min-height: 400px;
461 +}
462 +</style>
frontend/src/types/sca.d.ts
+45
@@ -125,3 +125,48 @@ export enum ScaComplianceLevel {
125 Poor = "Poor", // 60-69%
126 Critical = "Critical" // <60%
127 }
128 +
129 +// Streaming event types
130 +export interface ScaStreamStartEvent {
131 + total_agents: number
132 + message: string
133 +}
134 +
135 +export interface ScaStreamAgentResult {
136 + agent_id: string
137 + agent_name: string
138 + customer_code: string | null
139 + policy_count: number
140 + policies: AgentScaOverviewItem[]
141 +}
142 +
143 +export interface ScaStreamProgress {
144 + processed: number
145 + total: number
146 + successful: number
147 + failed: number
148 + results_so_far: number
149 + percent_complete: number
150 +}
151 +
152 +export interface ScaStreamComplete {
153 + total_results: number
154 + total_agents: number
155 + total_policies: number
156 + average_score: number
157 + total_checks: number
158 + total_passes: number
159 + total_fails: number
160 + total_invalid: number
161 + agents_processed: number
162 + agents_successful: number
163 + agents_failed: number
164 + message: string
165 +}
166 +
167 +export interface ScaStreamError {
168 + error?: string
169 + message: string
170 + agent_id?: string
171 + agent_name?: string
172 +}
frontend/src/views/agents/ScaOverview.vue
+1 -1
@@ -5,5 +5,5 @@
5 </template>
6
7 <script setup lang="ts">
8 -import List from "@/components/sca/List.vue"
8 +import List from "@/components/sca/StreamingList.vue"
9 </script>