| 1 | import asyncio |
| 2 | import csv |
| 3 | import hashlib |
| 4 | import io |
| 5 | 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 |
| 13 | |
| 14 | import httpx |
| 15 | from fastapi import HTTPException |
| 16 | from loguru import logger |
| 17 | from sqlalchemy import desc |
| 18 | from sqlalchemy import select |
| 19 | from sqlalchemy.ext.asyncio import AsyncSession |
| 20 | |
| 21 | from app.agents.sca.schema.sca import AgentPackageMatch |
| 22 | from app.agents.sca.schema.sca import AgentScaOverviewItem |
| 23 | from app.agents.sca.schema.sca import ScaOverviewResponse |
| 24 | from app.agents.sca.schema.sca import ScaPackageAgentsResponse |
| 25 | from app.agents.sca.schema.sca import ScaPackageRegistryItem |
| 26 | from app.agents.sca.schema.sca import ScaPackageRegistryResponse |
| 27 | from app.agents.sca.schema.sca import ScaPoliciesIndexResponse |
| 28 | from app.agents.sca.schema.sca import ScaPolicyContentResponse |
| 29 | from app.agents.sca.schema.sca import ScaPolicyItem |
| 30 | from app.agents.sca.schema.sca import SCAReportGenerateRequest |
| 31 | from app.agents.sca.schema.sca import SCAReportGenerateResponse |
| 32 | from app.agents.sca.schema.sca import SCAReportListResponse |
| 33 | from app.agents.sca.schema.sca import SCAReportResponse |
| 34 | from app.agents.sca.schema.sca import ScaStatsResponse |
| 35 | from app.agents.wazuh.services.sca import collect_agent_sca |
| 36 | from app.auth.models.users import User |
| 37 | from app.data_store.data_store_operations import delete_file_from_minio |
| 38 | from app.data_store.data_store_operations import retrieve_file_from_minio |
| 39 | from app.data_store.data_store_operations import store_file_in_minio |
| 40 | from app.db.universal_models import Agents |
| 41 | from app.db.universal_models import SCAReport |
| 42 | from app.middleware.customer_access import customer_access_handler |
| 43 | |
| 44 | # Default concurrency limit for parallel API requests |
| 45 | DEFAULT_MAX_CONCURRENT_REQUESTS = 100 |
| 46 | |
| 47 | # CoPilot-SCA public repository base URL |
| 48 | COPILOT_SCA_RAW_BASE = "https://raw.githubusercontent.com/socfortress/CoPilot-SCA/refs/heads/main" |
| 49 | COPILOT_SCA_INDEX_URL = f"{COPILOT_SCA_RAW_BASE}/index.json" |
| 50 | |
| 51 | |
| 52 | async def get_all_agents_from_db( |
| 53 | db_session: AsyncSession, |
| 54 | customer_code: Optional[str] = None, |
| 55 | ) -> List[Agents]: |
| 56 | """ |
| 57 | Get all agents from database, optionally filtered by customer code |
| 58 | |
| 59 | Args: |
| 60 | db_session: Database session to use |
| 61 | customer_code: Optional customer code to filter agents by |
| 62 | |
| 63 | Returns: |
| 64 | List of Agent objects |
| 65 | """ |
| 66 | try: |
| 67 | query = select(Agents) |
| 68 | if customer_code: |
| 69 | query = query.filter(Agents.customer_code == customer_code) |
| 70 | |
| 71 | result = await db_session.execute(query) |
| 72 | agents = result.scalars().all() |
| 73 | |
| 74 | logger.info(f"Found {len(agents)} agents" + (f" for customer {customer_code}" if customer_code else "")) |
| 75 | return agents |
| 76 | |
| 77 | except Exception as e: |
| 78 | logger.error(f"Error fetching agents from database: {e}") |
| 79 | raise HTTPException(status_code=500, detail=f"Failed to fetch agents: {e}") |
| 80 | |
| 81 | |
| 82 | async def collect_sca_for_single_agent( |
| 83 | agent: Agents, |
| 84 | semaphore: Semaphore, |
| 85 | policy_id: Optional[str] = None, |
| 86 | policy_name: Optional[str] = None, |
| 87 | min_score: Optional[int] = None, |
| 88 | max_score: Optional[int] = None, |
| 89 | ) -> List[AgentScaOverviewItem]: |
| 90 | """ |
| 91 | Collect SCA results for a single agent with semaphore-based rate limiting. |
| 92 | |
| 93 | Args: |
| 94 | agent: Agent to collect SCA data for |
| 95 | semaphore: Semaphore to limit concurrent requests |
| 96 | policy_id: Optional policy ID filter |
| 97 | policy_name: Optional policy name filter (partial matching) |
| 98 | min_score: Optional minimum score filter |
| 99 | max_score: Optional maximum score filter |
| 100 | |
| 101 | Returns: |
| 102 | List of AgentScaOverviewItem objects for this agent |
| 103 | """ |
| 104 | async with semaphore: |
| 105 | try: |
| 106 | logger.debug(f"Collecting SCA results for agent: {agent.hostname} (ID: {agent.agent_id})") |
| 107 | |
| 108 | # Collect SCA data from Wazuh Manager for this agent |
| 109 | sca_response = await collect_agent_sca(agent.agent_id) |
| 110 | |
| 111 | if not sca_response.success or not sca_response.sca: |
| 112 | logger.debug(f"No SCA data for agent {agent.hostname}") |
| 113 | return [] |
| 114 | |
| 115 | results = [] |
| 116 | |
| 117 | # Process each SCA policy result for this agent |
| 118 | for sca_result in sca_response.sca: |
| 119 | # Apply filters |
| 120 | if policy_id and sca_result.policy_id != policy_id: |
| 121 | continue |
| 122 | if policy_name and policy_name.lower() not in sca_result.name.lower(): |
| 123 | continue |
| 124 | if min_score is not None and sca_result.score < min_score: |
| 125 | continue |
| 126 | if max_score is not None and sca_result.score > max_score: |
| 127 | continue |
| 128 | |
| 129 | # Create overview item |
| 130 | overview_item = AgentScaOverviewItem( |
| 131 | agent_id=agent.agent_id, |
| 132 | agent_name=agent.hostname, |
| 133 | customer_code=agent.customer_code, |
| 134 | policy_id=sca_result.policy_id, |
| 135 | policy_name=sca_result.name, |
| 136 | description=sca_result.description, |
| 137 | total_checks=sca_result.total_checks, |
| 138 | pass_count=sca_result.pass_count, |
| 139 | fail_count=sca_result.fail, |
| 140 | invalid_count=sca_result.invalid, |
| 141 | score=sca_result.score, |
| 142 | start_scan=sca_result.start_scan, |
| 143 | end_scan=sca_result.end_scan, |
| 144 | references=sca_result.references, |
| 145 | hash_file=sca_result.hash_file, |
| 146 | ) |
| 147 | results.append(overview_item) |
| 148 | |
| 149 | logger.debug(f"Collected {len(results)} SCA results for agent {agent.hostname}") |
| 150 | return results |
| 151 | |
| 152 | except Exception as e: |
| 153 | logger.error(f"Error collecting SCA for agent {agent.hostname}: {e}") |
| 154 | # Return empty list instead of raising - allows other agents to continue |
| 155 | return [] |
| 156 | |
| 157 | |
| 158 | async def collect_sca_for_all_agents( |
| 159 | db_session: AsyncSession, |
| 160 | customer_code: Optional[str] = None, |
| 161 | agent_name: Optional[str] = None, |
| 162 | policy_id: Optional[str] = None, |
| 163 | policy_name: Optional[str] = None, |
| 164 | min_score: Optional[int] = None, |
| 165 | max_score: Optional[int] = None, |
| 166 | max_concurrent_requests: int = DEFAULT_MAX_CONCURRENT_REQUESTS, |
| 167 | ) -> List[AgentScaOverviewItem]: |
| 168 | """ |
| 169 | Collect SCA results for all agents from Wazuh Manager using parallel requests. |
| 170 | |
| 171 | Args: |
| 172 | db_session: Database session to use |
| 173 | customer_code: Optional customer code filter |
| 174 | agent_name: Optional agent name filter |
| 175 | policy_id: Optional policy ID filter |
| 176 | policy_name: Optional policy name filter (partial matching) |
| 177 | min_score: Optional minimum score filter |
| 178 | max_score: Optional maximum score filter |
| 179 | max_concurrent_requests: Maximum number of concurrent API requests (default: 10) |
| 180 | |
| 181 | Returns: |
| 182 | List of AgentScaOverviewItem objects |
| 183 | """ |
| 184 | try: |
| 185 | # Get agents from database |
| 186 | agents = await get_all_agents_from_db(db_session, customer_code) |
| 187 | |
| 188 | if not agents: |
| 189 | logger.warning("No agents found" + (f" for customer {customer_code}" if customer_code else "")) |
| 190 | return [] |
| 191 | |
| 192 | # Filter agents by name if specified (do this before parallel processing) |
| 193 | if agent_name: |
| 194 | agents = [a for a in agents if a.hostname == agent_name] |
| 195 | if not agents: |
| 196 | logger.info(f"No agents found matching hostname: {agent_name}") |
| 197 | return [] |
| 198 | |
| 199 | logger.info(f"Collecting SCA data for {len(agents)} agents " f"(max concurrent: {max_concurrent_requests})") |
| 200 | |
| 201 | # Create semaphore to limit concurrent requests to Wazuh Manager |
| 202 | semaphore = Semaphore(max_concurrent_requests) |
| 203 | |
| 204 | # Create tasks for parallel execution |
| 205 | tasks = [ |
| 206 | collect_sca_for_single_agent( |
| 207 | agent=agent, |
| 208 | semaphore=semaphore, |
| 209 | policy_id=policy_id, |
| 210 | policy_name=policy_name, |
| 211 | min_score=min_score, |
| 212 | max_score=max_score, |
| 213 | ) |
| 214 | for agent in agents |
| 215 | ] |
| 216 | |
| 217 | # Execute all tasks concurrently with gather |
| 218 | # return_exceptions=True prevents one failure from canceling all tasks |
| 219 | results = await asyncio.gather(*tasks, return_exceptions=True) |
| 220 | |
| 221 | # Flatten results and filter out exceptions |
| 222 | all_sca_results = [] |
| 223 | successful_agents = 0 |
| 224 | failed_agents = 0 |
| 225 | |
| 226 | for i, result in enumerate(results): |
| 227 | if isinstance(result, Exception): |
| 228 | logger.error(f"Task for agent {agents[i].hostname} failed: {result}") |
| 229 | failed_agents += 1 |
| 230 | continue |
| 231 | if isinstance(result, list): |
| 232 | if result: # Only count as successful if we got data |
| 233 | successful_agents += 1 |
| 234 | all_sca_results.extend(result) |
| 235 | |
| 236 | logger.info(f"Collected {len(all_sca_results)} SCA policy results from " f"{successful_agents} agents ({failed_agents} failed)") |
| 237 | |
| 238 | return all_sca_results |
| 239 | |
| 240 | except Exception as e: |
| 241 | logger.error(f"Error collecting SCA for all agents: {e}") |
| 242 | raise HTTPException(status_code=500, detail=f"Failed to collect SCA results: {e}") |
| 243 | |
| 244 | |
| 245 | async def search_sca_overview( |
| 246 | db_session: AsyncSession, |
| 247 | customer_code: Optional[str] = None, |
| 248 | agent_name: Optional[str] = None, |
| 249 | policy_id: Optional[str] = None, |
| 250 | policy_name: Optional[str] = None, |
| 251 | min_score: Optional[int] = None, |
| 252 | max_score: Optional[int] = None, |
| 253 | page: int = 1, |
| 254 | page_size: int = 50, |
| 255 | max_concurrent_requests: int = DEFAULT_MAX_CONCURRENT_REQUESTS, |
| 256 | ) -> ScaOverviewResponse: |
| 257 | """ |
| 258 | Search SCA results across all agents with filtering and pagination |
| 259 | |
| 260 | Args: |
| 261 | db_session: Database session for agent lookup |
| 262 | customer_code: Optional customer code filter |
| 263 | agent_name: Optional agent hostname filter |
| 264 | policy_id: Optional policy ID filter |
| 265 | policy_name: Optional policy name filter (partial matching) |
| 266 | min_score: Optional minimum score filter |
| 267 | max_score: Optional maximum score filter |
| 268 | page: Page number for pagination |
| 269 | page_size: Number of results per page |
| 270 | max_concurrent_requests: Maximum concurrent API requests (default: 10) |
| 271 | |
| 272 | Returns: |
| 273 | ScaOverviewResponse with paginated results and statistics |
| 274 | """ |
| 275 | logger.info( |
| 276 | f"Searching SCA overview with filters: customer_code={customer_code}, " |
| 277 | f"agent_name={agent_name}, policy_id={policy_id}, policy_name={policy_name}, " |
| 278 | f"min_score={min_score}, max_score={max_score}, page={page}, page_size={page_size}", |
| 279 | ) |
| 280 | |
| 281 | # Build filters applied dict for response |
| 282 | filters_applied = {} |
| 283 | if customer_code: |
| 284 | filters_applied["customer_code"] = customer_code |
| 285 | if agent_name: |
| 286 | filters_applied["agent_name"] = agent_name |
| 287 | if policy_id: |
| 288 | filters_applied["policy_id"] = policy_id |
| 289 | if policy_name: |
| 290 | filters_applied["policy_name"] = policy_name |
| 291 | if min_score is not None: |
| 292 | filters_applied["min_score"] = min_score |
| 293 | if max_score is not None: |
| 294 | filters_applied["max_score"] = max_score |
| 295 | |
| 296 | try: |
| 297 | # Collect all SCA results with filtering (now uses parallel requests) |
| 298 | all_sca_results = await collect_sca_for_all_agents( |
| 299 | db_session=db_session, |
| 300 | customer_code=customer_code, |
| 301 | agent_name=agent_name, |
| 302 | policy_id=policy_id, |
| 303 | policy_name=policy_name, |
| 304 | min_score=min_score, |
| 305 | max_score=max_score, |
| 306 | max_concurrent_requests=max_concurrent_requests, |
| 307 | ) |
| 308 | |
| 309 | # Sort results by agent's minimum score (lowest first) |
| 310 | if all_sca_results: |
| 311 | # Group results by agent to find minimum score per agent |
| 312 | agent_min_scores = {} |
| 313 | agent_results = {} |
| 314 | |
| 315 | for result in all_sca_results: |
| 316 | agent_id = result.agent_id |
| 317 | if agent_id not in agent_min_scores: |
| 318 | agent_min_scores[agent_id] = result.score |
| 319 | agent_results[agent_id] = [] |
| 320 | else: |
| 321 | agent_min_scores[agent_id] = min(agent_min_scores[agent_id], result.score) |
| 322 | agent_results[agent_id].append(result) |
| 323 | |
| 324 | # Sort agents by their minimum score (lowest first) |
| 325 | sorted_agent_ids = sorted(agent_min_scores.keys(), key=lambda x: agent_min_scores[x]) |
| 326 | |
| 327 | # Rebuild the results list with agents sorted by their minimum score |
| 328 | all_sca_results = [] |
| 329 | for agent_id in sorted_agent_ids: |
| 330 | # Sort policies within each agent by score (lowest first) |
| 331 | agent_policies = sorted(agent_results[agent_id], key=lambda x: x.score) |
| 332 | all_sca_results.extend(agent_policies) |
| 333 | |
| 334 | logger.info("Sorted SCA results by agent minimum scores (lowest first)") |
| 335 | |
| 336 | total_count = len(all_sca_results) |
| 337 | |
| 338 | # Calculate pagination |
| 339 | total_pages = (total_count + page_size - 1) // page_size |
| 340 | start_idx = (page - 1) * page_size |
| 341 | end_idx = start_idx + page_size |
| 342 | |
| 343 | # Get paginated results |
| 344 | paginated_results = all_sca_results[start_idx:end_idx] |
| 345 | |
| 346 | # Calculate statistics |
| 347 | unique_agents = set(item.agent_id for item in all_sca_results) |
| 348 | unique_policies = set(item.policy_id for item in all_sca_results) |
| 349 | |
| 350 | total_checks_all = sum(item.total_checks for item in all_sca_results) |
| 351 | total_passes_all = sum(item.pass_count for item in all_sca_results) |
| 352 | total_fails_all = sum(item.fail_count for item in all_sca_results) |
| 353 | total_invalid_all = sum(item.invalid_count for item in all_sca_results) |
| 354 | |
| 355 | # Calculate average score |
| 356 | average_score = sum(item.score for item in all_sca_results) / len(all_sca_results) if all_sca_results else 0.0 |
| 357 | |
| 358 | return ScaOverviewResponse( |
| 359 | sca_results=paginated_results, |
| 360 | total_count=total_count, |
| 361 | total_agents=len(unique_agents), |
| 362 | total_policies=len(unique_policies), |
| 363 | average_score=round(average_score, 2), |
| 364 | total_checks_all_agents=total_checks_all, |
| 365 | total_passes_all_agents=total_passes_all, |
| 366 | total_fails_all_agents=total_fails_all, |
| 367 | total_invalid_all_agents=total_invalid_all, |
| 368 | page=page, |
| 369 | page_size=page_size, |
| 370 | total_pages=total_pages, |
| 371 | has_next=page < total_pages, |
| 372 | has_previous=page > 1, |
| 373 | success=True, |
| 374 | message=f"Found {total_count} SCA results across {len(unique_agents)} agents (sorted by agent minimum score, lowest first)", |
| 375 | filters_applied=filters_applied, |
| 376 | ) |
| 377 | |
| 378 | except Exception as e: |
| 379 | logger.error(f"Error in SCA overview search: {e}") |
| 380 | raise HTTPException(status_code=500, detail=f"Failed to search SCA results: {e}") |
| 381 | |
| 382 | |
| 383 | async def get_sca_statistics( |
| 384 | db_session: AsyncSession, |
| 385 | customer_code: Optional[str] = None, |
| 386 | ) -> ScaStatsResponse: |
| 387 | """ |
| 388 | Get SCA statistics across all agents or for a specific customer |
| 389 | |
| 390 | Args: |
| 391 | db_session: Database session to use |
| 392 | customer_code: Optional customer code to filter by |
| 393 | |
| 394 | Returns: |
| 395 | ScaStatsResponse with SCA statistics |
| 396 | """ |
| 397 | try: |
| 398 | logger.info("Getting SCA statistics" + (f" for customer {customer_code}" if customer_code else " for all customers")) |
| 399 | |
| 400 | # Collect all SCA results |
| 401 | all_sca_results = await collect_sca_for_all_agents( |
| 402 | db_session=db_session, |
| 403 | customer_code=customer_code, |
| 404 | ) |
| 405 | |
| 406 | if not all_sca_results: |
| 407 | return ScaStatsResponse( |
| 408 | total_agents_with_sca=0, |
| 409 | total_policies=0, |
| 410 | average_score_across_all=0.0, |
| 411 | total_checks_all_agents=0, |
| 412 | total_passes_all_agents=0, |
| 413 | total_fails_all_agents=0, |
| 414 | total_invalid_all_agents=0, |
| 415 | by_customer={}, |
| 416 | success=True, |
| 417 | message="No SCA results found", |
| 418 | ) |
| 419 | |
| 420 | # Calculate overall statistics |
| 421 | unique_agents = set(item.agent_id for item in all_sca_results) |
| 422 | unique_policies = set(item.policy_id for item in all_sca_results) |
| 423 | |
| 424 | total_checks_all = sum(item.total_checks for item in all_sca_results) |
| 425 | total_passes_all = sum(item.pass_count for item in all_sca_results) |
| 426 | total_fails_all = sum(item.fail_count for item in all_sca_results) |
| 427 | total_invalid_all = sum(item.invalid_count for item in all_sca_results) |
| 428 | |
| 429 | average_score = sum(item.score for item in all_sca_results) / len(all_sca_results) |
| 430 | |
| 431 | # Group by customer if no specific customer requested |
| 432 | by_customer = {} |
| 433 | if not customer_code: |
| 434 | customer_groups = {} |
| 435 | for item in all_sca_results: |
| 436 | cust_code = item.customer_code or "unknown" |
| 437 | if cust_code not in customer_groups: |
| 438 | customer_groups[cust_code] = [] |
| 439 | customer_groups[cust_code].append(item) |
| 440 | |
| 441 | for cust_code, items in customer_groups.items(): |
| 442 | unique_agents_cust = set(item.agent_id for item in items) |
| 443 | unique_policies_cust = set(item.policy_id for item in items) |
| 444 | avg_score_cust = sum(item.score for item in items) / len(items) |
| 445 | |
| 446 | by_customer[cust_code] = { |
| 447 | "total_agents": len(unique_agents_cust), |
| 448 | "total_policies": len(unique_policies_cust), |
| 449 | "average_score": round(avg_score_cust, 2), |
| 450 | "total_checks": sum(item.total_checks for item in items), |
| 451 | "total_passes": sum(item.pass_count for item in items), |
| 452 | "total_fails": sum(item.fail_count for item in items), |
| 453 | "total_invalid": sum(item.invalid_count for item in items), |
| 454 | } |
| 455 | |
| 456 | return ScaStatsResponse( |
| 457 | total_agents_with_sca=len(unique_agents), |
| 458 | total_policies=len(unique_policies), |
| 459 | average_score_across_all=round(average_score, 2), |
| 460 | total_checks_all_agents=total_checks_all, |
| 461 | total_passes_all_agents=total_passes_all, |
| 462 | total_fails_all_agents=total_fails_all, |
| 463 | total_invalid_all_agents=total_invalid_all, |
| 464 | by_customer=by_customer, |
| 465 | success=True, |
| 466 | message=f"SCA statistics calculated for {len(unique_agents)} agents", |
| 467 | ) |
| 468 | |
| 469 | except Exception as e: |
| 470 | logger.error(f"Error getting SCA statistics: {e}") |
| 471 | raise HTTPException(status_code=500, detail=f"Failed to get SCA statistics: {e}") |
| 472 | |
| 473 | |
| 474 | async def collect_sca_for_report( |
| 475 | db_session: AsyncSession, |
| 476 | current_user: User, |
| 477 | customer_code: str, |
| 478 | agent_name: Optional[str] = None, |
| 479 | policy_id: Optional[str] = None, |
| 480 | min_score: Optional[int] = None, |
| 481 | max_score: Optional[int] = None, |
| 482 | ) -> List[AgentScaOverviewItem]: |
| 483 | """ |
| 484 | Collect ALL SCA results from Wazuh Manager API for CSV report generation. |
| 485 | |
| 486 | This uses the same data collection method as the overview search, |
| 487 | but without pagination to get all results for comprehensive reports. |
| 488 | |
| 489 | Args: |
| 490 | db_session: Database session for agent lookup |
| 491 | current_user: Current authenticated user for customer access filtering |
| 492 | customer_code: Customer code to filter by |
| 493 | agent_name: Optional agent hostname filter |
| 494 | policy_id: Optional policy ID filter |
| 495 | min_score: Optional minimum score filter |
| 496 | max_score: Optional maximum score filter |
| 497 | |
| 498 | Returns: |
| 499 | List of all matching SCA policy results (no pagination) |
| 500 | """ |
| 501 | logger.info( |
| 502 | f"Collecting ALL SCA results from Wazuh Manager for report: customer_code={customer_code}, " |
| 503 | f"agent_name={agent_name}, policy_id={policy_id}, " |
| 504 | f"min_score={min_score}, max_score={max_score}", |
| 505 | ) |
| 506 | |
| 507 | # Apply customer access filtering |
| 508 | accessible_customers = await customer_access_handler.get_user_accessible_customers(current_user, db_session) |
| 509 | |
| 510 | if "*" not in accessible_customers and customer_code not in accessible_customers: |
| 511 | logger.warning(f"User {current_user.username} denied access to customer {customer_code}") |
| 512 | return [] |
| 513 | |
| 514 | try: |
| 515 | # Use existing collect function to get all SCA results from Wazuh Manager |
| 516 | all_sca_results = await collect_sca_for_all_agents( |
| 517 | db_session=db_session, |
| 518 | customer_code=customer_code, |
| 519 | agent_name=agent_name, |
| 520 | policy_id=policy_id, |
| 521 | policy_name=None, # Not used for reports |
| 522 | min_score=min_score, |
| 523 | max_score=max_score, |
| 524 | ) |
| 525 | |
| 526 | logger.info(f"Successfully collected {len(all_sca_results)} SCA policy results from Wazuh Manager for report") |
| 527 | return all_sca_results |
| 528 | |
| 529 | except Exception as e: |
| 530 | logger.error(f"Error collecting SCA results for report: {e}") |
| 531 | raise |
| 532 | |
| 533 | |
| 534 | async def generate_sca_csv_report( |
| 535 | db_session: AsyncSession, |
| 536 | current_user: User, |
| 537 | request: SCAReportGenerateRequest, |
| 538 | report_id: Optional[int] = None, |
| 539 | ) -> SCAReportGenerateResponse: |
| 540 | """ |
| 541 | Generate a CSV SCA report for a specific customer and store it in MinIO. |
| 542 | |
| 543 | Uses Wazuh Manager API to fetch all SCA policy-level data for comprehensive reports. |
| 544 | |
| 545 | Args: |
| 546 | db_session: Database session |
| 547 | current_user: Current authenticated user |
| 548 | request: Report generation request with filters |
| 549 | report_id: Optional existing report ID (for background task updates) |
| 550 | |
| 551 | Returns: |
| 552 | SCAReportGenerateResponse with report details |
| 553 | """ |
| 554 | try: |
| 555 | # Verify customer access |
| 556 | accessible_customers = await customer_access_handler.get_user_accessible_customers(current_user, db_session) |
| 557 | |
| 558 | if "*" not in accessible_customers and request.customer_code not in accessible_customers: |
| 559 | # If we have a report_id, update it to failed status |
| 560 | if report_id: |
| 561 | stmt = select(SCAReport).filter(SCAReport.id == report_id) |
| 562 | result = await db_session.execute(stmt) |
| 563 | report = result.scalars().first() |
| 564 | if report: |
| 565 | report.status = "failed" |
| 566 | report.error_message = "Insufficient permissions" |
| 567 | await db_session.commit() |
| 568 | |
| 569 | return SCAReportGenerateResponse( |
| 570 | success=False, |
| 571 | message=f"Access denied to customer {request.customer_code}", |
| 572 | error="Insufficient permissions", |
| 573 | ) |
| 574 | |
| 575 | # If report_id exists, get the existing report name and object_key |
| 576 | # Otherwise generate new ones |
| 577 | if report_id: |
| 578 | stmt = select(SCAReport).filter(SCAReport.id == report_id) |
| 579 | result = await db_session.execute(stmt) |
| 580 | existing_report = result.scalars().first() |
| 581 | |
| 582 | if not existing_report: |
| 583 | return SCAReportGenerateResponse( |
| 584 | success=False, |
| 585 | message=f"Report ID {report_id} not found", |
| 586 | error="Report not found", |
| 587 | ) |
| 588 | |
| 589 | # Use existing report name and paths |
| 590 | report_name = existing_report.report_name |
| 591 | file_name = existing_report.file_name |
| 592 | object_key = existing_report.object_key |
| 593 | bucket_name = existing_report.bucket_name |
| 594 | |
| 595 | logger.info(f"Generating SCA report for existing record: {report_name} (ID: {report_id})") |
| 596 | else: |
| 597 | # Generate new report name for synchronous generation |
| 598 | timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") |
| 599 | report_name = request.report_name or f"sca_report_{timestamp}" |
| 600 | file_name = f"{report_name}.csv" |
| 601 | object_key = f"{request.customer_code}/{file_name}" |
| 602 | bucket_name = "sca-reports" |
| 603 | |
| 604 | logger.info(f"Generating new SCA report: {report_name}") |
| 605 | |
| 606 | # Fetch ALL SCA results from Wazuh Manager API (no pagination limits) |
| 607 | all_sca_results = await collect_sca_for_report( |
| 608 | db_session=db_session, |
| 609 | current_user=current_user, |
| 610 | customer_code=request.customer_code, |
| 611 | agent_name=request.agent_name, |
| 612 | policy_id=request.policy_id, |
| 613 | min_score=request.min_score, |
| 614 | max_score=request.max_score, |
| 615 | ) |
| 616 | |
| 617 | logger.info(f"Fetched {len(all_sca_results)} SCA policy results from Wazuh Manager for report") |
| 618 | |
| 619 | if not all_sca_results: |
| 620 | # If we have a report_id, update it to failed status |
| 621 | if report_id: |
| 622 | stmt = select(SCAReport).filter(SCAReport.id == report_id) |
| 623 | result = await db_session.execute(stmt) |
| 624 | report = result.scalars().first() |
| 625 | if report: |
| 626 | report.status = "failed" |
| 627 | report.error_message = "No SCA results found matching criteria" |
| 628 | await db_session.commit() |
| 629 | |
| 630 | return SCAReportGenerateResponse( |
| 631 | success=False, |
| 632 | message="No SCA results found matching the specified criteria", |
| 633 | error="No data to export", |
| 634 | ) |
| 635 | |
| 636 | # Generate CSV content |
| 637 | csv_buffer = io.StringIO() |
| 638 | csv_writer = csv.writer(csv_buffer) |
| 639 | |
| 640 | # Write headers |
| 641 | headers = [ |
| 642 | "Agent ID", |
| 643 | "Agent Name", |
| 644 | "Customer Code", |
| 645 | "Policy ID", |
| 646 | "Policy Name", |
| 647 | "Description", |
| 648 | "Total Checks", |
| 649 | "Passed", |
| 650 | "Failed", |
| 651 | "Invalid", |
| 652 | "Score", |
| 653 | "Start Scan", |
| 654 | "End Scan", |
| 655 | "References", |
| 656 | "Hash File", |
| 657 | ] |
| 658 | csv_writer.writerow(headers) |
| 659 | |
| 660 | # Count totals |
| 661 | total_policies = len(all_sca_results) |
| 662 | total_checks = sum(result.total_checks for result in all_sca_results) |
| 663 | passed_count = sum(result.pass_count for result in all_sca_results) |
| 664 | failed_count = sum(result.fail_count for result in all_sca_results) |
| 665 | invalid_count = sum(result.invalid_count for result in all_sca_results) |
| 666 | |
| 667 | # Write data rows |
| 668 | for result in all_sca_results: |
| 669 | try: |
| 670 | row = [ |
| 671 | result.agent_id, |
| 672 | result.agent_name, |
| 673 | result.customer_code or "", |
| 674 | result.policy_id, |
| 675 | result.policy_name, |
| 676 | result.description, |
| 677 | result.total_checks, |
| 678 | result.pass_count, |
| 679 | result.fail_count, |
| 680 | result.invalid_count, |
| 681 | result.score, |
| 682 | result.start_scan, |
| 683 | result.end_scan, |
| 684 | result.references or "", |
| 685 | result.hash_file or "", |
| 686 | ] |
| 687 | csv_writer.writerow(row) |
| 688 | |
| 689 | except Exception as row_error: |
| 690 | logger.error(f"Error processing SCA row: {row_error}") |
| 691 | continue |
| 692 | |
| 693 | # Get CSV content as bytes |
| 694 | csv_content = csv_buffer.getvalue().encode("utf-8") |
| 695 | csv_buffer.close() |
| 696 | |
| 697 | # Calculate file hash |
| 698 | file_hash = hashlib.sha256(csv_content).hexdigest() |
| 699 | |
| 700 | # Store in MinIO |
| 701 | minio_result = await store_file_in_minio( |
| 702 | file_content=csv_content, |
| 703 | bucket_name=bucket_name, |
| 704 | object_key=object_key, |
| 705 | content_type="text/csv", |
| 706 | ) |
| 707 | |
| 708 | if not minio_result["success"]: |
| 709 | # If we have a report_id, update it to failed status |
| 710 | if report_id: |
| 711 | stmt = select(SCAReport).filter(SCAReport.id == report_id) |
| 712 | result = await db_session.execute(stmt) |
| 713 | report = result.scalars().first() |
| 714 | if report: |
| 715 | report.status = "failed" |
| 716 | report.error_message = minio_result.get("error", "Unknown error") |
| 717 | await db_session.commit() |
| 718 | |
| 719 | return SCAReportGenerateResponse( |
| 720 | success=False, |
| 721 | message="Failed to store report in MinIO", |
| 722 | error=minio_result.get("error", "Unknown error"), |
| 723 | ) |
| 724 | |
| 725 | # Build filters JSON |
| 726 | filters = {} |
| 727 | if request.agent_name: |
| 728 | filters["agent_name"] = request.agent_name |
| 729 | if request.policy_id: |
| 730 | filters["policy_id"] = request.policy_id |
| 731 | if request.min_score is not None: |
| 732 | filters["min_score"] = request.min_score |
| 733 | if request.max_score is not None: |
| 734 | filters["max_score"] = request.max_score |
| 735 | |
| 736 | # Check if we're updating an existing report or creating a new one |
| 737 | if report_id: |
| 738 | # Update existing report (background task scenario) |
| 739 | stmt = select(SCAReport).filter(SCAReport.id == report_id) |
| 740 | result = await db_session.execute(stmt) |
| 741 | report_record = result.scalars().first() |
| 742 | |
| 743 | if report_record: |
| 744 | report_record.file_size = len(csv_content) |
| 745 | report_record.file_hash = file_hash |
| 746 | report_record.total_policies = total_policies |
| 747 | report_record.total_checks = total_checks |
| 748 | report_record.passed_count = passed_count |
| 749 | report_record.failed_count = failed_count |
| 750 | report_record.invalid_count = invalid_count |
| 751 | report_record.status = "completed" |
| 752 | report_record.error_message = None |
| 753 | |
| 754 | await db_session.commit() |
| 755 | await db_session.refresh(report_record) |
| 756 | |
| 757 | logger.info(f"Successfully updated SCA report: {report_name} (ID: {report_id})") |
| 758 | else: |
| 759 | logger.error(f"Report ID {report_id} not found for update") |
| 760 | return SCAReportGenerateResponse( |
| 761 | success=False, |
| 762 | message=f"Report ID {report_id} not found", |
| 763 | error="Report not found", |
| 764 | ) |
| 765 | else: |
| 766 | # Create new database record (synchronous scenario) |
| 767 | report_record = SCAReport( |
| 768 | report_name=report_name, |
| 769 | customer_code=request.customer_code, |
| 770 | bucket_name=bucket_name, |
| 771 | object_key=object_key, |
| 772 | file_name=file_name, |
| 773 | file_size=len(csv_content), |
| 774 | file_hash=file_hash, |
| 775 | generated_by=current_user.id, |
| 776 | filters_json=json.dumps(filters), |
| 777 | total_policies=total_policies, |
| 778 | total_checks=total_checks, |
| 779 | passed_count=passed_count, |
| 780 | failed_count=failed_count, |
| 781 | invalid_count=invalid_count, |
| 782 | status="completed", |
| 783 | ) |
| 784 | |
| 785 | db_session.add(report_record) |
| 786 | await db_session.commit() |
| 787 | await db_session.refresh(report_record) |
| 788 | |
| 789 | logger.info(f"Successfully generated SCA report: {report_name}") |
| 790 | |
| 791 | # Build response |
| 792 | report_response = SCAReportResponse( |
| 793 | id=report_record.id, |
| 794 | report_name=report_record.report_name, |
| 795 | customer_code=report_record.customer_code, |
| 796 | file_name=report_record.file_name, |
| 797 | file_size=report_record.file_size, |
| 798 | generated_at=report_record.generated_at, |
| 799 | generated_by=report_record.generated_by, |
| 800 | total_policies=report_record.total_policies, |
| 801 | total_checks=report_record.total_checks, |
| 802 | passed_count=report_record.passed_count, |
| 803 | failed_count=report_record.failed_count, |
| 804 | invalid_count=report_record.invalid_count, |
| 805 | filters_applied=json.loads(report_record.filters_json or "{}"), |
| 806 | status=report_record.status, |
| 807 | download_url=f"/api/v1/sca/reports/{report_record.id}/download", |
| 808 | ) |
| 809 | |
| 810 | return SCAReportGenerateResponse( |
| 811 | success=True, |
| 812 | message=f"Successfully generated report with {total_policies} SCA policy results", |
| 813 | report=report_response, |
| 814 | ) |
| 815 | |
| 816 | except Exception as e: |
| 817 | logger.error(f"Error generating SCA report: {e}") |
| 818 | |
| 819 | # If we have a report_id, update it to failed status |
| 820 | if report_id: |
| 821 | try: |
| 822 | stmt = select(SCAReport).filter(SCAReport.id == report_id) |
| 823 | result = await db_session.execute(stmt) |
| 824 | report = result.scalars().first() |
| 825 | if report: |
| 826 | report.status = "failed" |
| 827 | report.error_message = str(e) |
| 828 | await db_session.commit() |
| 829 | except Exception as update_error: |
| 830 | logger.error(f"Failed to update report status: {update_error}") |
| 831 | |
| 832 | return SCAReportGenerateResponse( |
| 833 | success=False, |
| 834 | message="Failed to generate SCA report", |
| 835 | error=str(e), |
| 836 | ) |
| 837 | |
| 838 | |
| 839 | async def list_sca_reports( |
| 840 | db_session: AsyncSession, |
| 841 | current_user: User, |
| 842 | customer_code: Optional[str] = None, |
| 843 | ) -> SCAReportListResponse: |
| 844 | """List available SCA reports""" |
| 845 | try: |
| 846 | # Get accessible customers |
| 847 | accessible_customers = await customer_access_handler.get_user_accessible_customers(current_user, db_session) |
| 848 | |
| 849 | # Build query |
| 850 | query = select(SCAReport).order_by(desc(SCAReport.generated_at)) |
| 851 | |
| 852 | # Apply customer filtering |
| 853 | if "*" not in accessible_customers: |
| 854 | query = query.filter(SCAReport.customer_code.in_(accessible_customers)) |
| 855 | |
| 856 | if customer_code: |
| 857 | if "*" not in accessible_customers and customer_code not in accessible_customers: |
| 858 | return SCAReportListResponse( |
| 859 | reports=[], |
| 860 | total_count=0, |
| 861 | success=True, |
| 862 | message=f"Access denied to customer {customer_code}", |
| 863 | ) |
| 864 | query = query.filter(SCAReport.customer_code == customer_code) |
| 865 | |
| 866 | result = await db_session.execute(query) |
| 867 | reports = result.scalars().all() |
| 868 | |
| 869 | report_list = [] |
| 870 | for report in reports: |
| 871 | report_response = SCAReportResponse( |
| 872 | id=report.id, |
| 873 | report_name=report.report_name, |
| 874 | customer_code=report.customer_code, |
| 875 | file_name=report.file_name, |
| 876 | file_size=report.file_size, |
| 877 | generated_at=report.generated_at, |
| 878 | generated_by=report.generated_by, |
| 879 | total_policies=report.total_policies, |
| 880 | total_checks=report.total_checks, |
| 881 | passed_count=report.passed_count, |
| 882 | failed_count=report.failed_count, |
| 883 | invalid_count=report.invalid_count, |
| 884 | filters_applied=json.loads(report.filters_json or "{}"), |
| 885 | status=report.status, |
| 886 | error_message=report.error_message, |
| 887 | download_url=f"/api/v1/sca/reports/{report.id}/download", |
| 888 | ) |
| 889 | report_list.append(report_response) |
| 890 | |
| 891 | return SCAReportListResponse( |
| 892 | reports=report_list, |
| 893 | total_count=len(report_list), |
| 894 | success=True, |
| 895 | message=f"Found {len(report_list)} SCA reports", |
| 896 | ) |
| 897 | |
| 898 | except Exception as e: |
| 899 | logger.error(f"Error listing SCA reports: {e}") |
| 900 | return SCAReportListResponse( |
| 901 | reports=[], |
| 902 | total_count=0, |
| 903 | success=False, |
| 904 | message=f"Failed to list reports: {e}", |
| 905 | ) |
| 906 | |
| 907 | |
| 908 | async def get_sca_report_download( |
| 909 | db_session: AsyncSession, |
| 910 | current_user: User, |
| 911 | report_id: int, |
| 912 | ) -> Dict[str, Any]: |
| 913 | """Get SCA report for download""" |
| 914 | try: |
| 915 | # Get report record |
| 916 | result = await db_session.execute(select(SCAReport).filter(SCAReport.id == report_id)) |
| 917 | report = result.scalars().first() |
| 918 | |
| 919 | if not report: |
| 920 | raise HTTPException(status_code=404, detail="Report not found") |
| 921 | |
| 922 | # Verify customer access |
| 923 | accessible_customers = await customer_access_handler.get_user_accessible_customers(current_user, db_session) |
| 924 | |
| 925 | if "*" not in accessible_customers and report.customer_code not in accessible_customers: |
| 926 | raise HTTPException(status_code=403, detail="Access denied to this report") |
| 927 | |
| 928 | # Retrieve file from MinIO |
| 929 | file_data = await retrieve_file_from_minio( |
| 930 | bucket_name=report.bucket_name, |
| 931 | object_key=report.object_key, |
| 932 | ) |
| 933 | |
| 934 | if not file_data["success"]: |
| 935 | raise HTTPException(status_code=500, detail="Failed to retrieve report file") |
| 936 | |
| 937 | return { |
| 938 | "file_content": file_data["file_content"], |
| 939 | "file_name": report.file_name, |
| 940 | "content_type": "text/csv", |
| 941 | } |
| 942 | |
| 943 | except HTTPException: |
| 944 | raise |
| 945 | except Exception as e: |
| 946 | logger.error(f"Error retrieving SCA report: {e}") |
| 947 | raise HTTPException(status_code=500, detail=f"Failed to retrieve report: {e}") |
| 948 | |
| 949 | |
| 950 | async def delete_sca_report( |
| 951 | db_session: AsyncSession, |
| 952 | current_user: User, |
| 953 | report_id: int, |
| 954 | ) -> Dict[str, Any]: |
| 955 | """ |
| 956 | Delete an SCA report and its associated file from MinIO. |
| 957 | |
| 958 | Args: |
| 959 | db_session: Database session |
| 960 | current_user: Current authenticated user |
| 961 | report_id: ID of the report to delete |
| 962 | |
| 963 | Returns: |
| 964 | Dict with success status and message |
| 965 | """ |
| 966 | try: |
| 967 | # Get report record |
| 968 | result = await db_session.execute(select(SCAReport).filter(SCAReport.id == report_id)) |
| 969 | report = result.scalars().first() |
| 970 | |
| 971 | if not report: |
| 972 | raise HTTPException(status_code=404, detail="Report not found") |
| 973 | |
| 974 | # Verify customer access |
| 975 | accessible_customers = await customer_access_handler.get_user_accessible_customers(current_user, db_session) |
| 976 | |
| 977 | if "*" not in accessible_customers and report.customer_code not in accessible_customers: |
| 978 | raise HTTPException(status_code=403, detail="Access denied to this report") |
| 979 | |
| 980 | # Delete file from MinIO |
| 981 | logger.info(f"Deleting SCA report file from MinIO: {report.bucket_name}/{report.object_key}") |
| 982 | |
| 983 | minio_result = await delete_file_from_minio( |
| 984 | bucket_name=report.bucket_name, |
| 985 | object_key=report.object_key, |
| 986 | ) |
| 987 | |
| 988 | if not minio_result["success"]: |
| 989 | logger.warning(f"Failed to delete file from MinIO: {minio_result.get('message')}") |
| 990 | # Continue with database deletion even if MinIO deletion fails |
| 991 | |
| 992 | # Delete database record |
| 993 | await db_session.delete(report) |
| 994 | await db_session.commit() |
| 995 | |
| 996 | logger.info(f"Successfully deleted SCA report: {report.report_name} (ID: {report_id})") |
| 997 | |
| 998 | return { |
| 999 | "success": True, |
| 1000 | "message": f"Successfully deleted report '{report.report_name}'", |
| 1001 | } |
| 1002 | |
| 1003 | except HTTPException: |
| 1004 | raise |
| 1005 | except Exception as e: |
| 1006 | logger.error(f"Error deleting SCA report: {e}") |
| 1007 | raise HTTPException(status_code=500, detail=f"Failed to delete report: {e}") |
| 1008 | |
| 1009 | |
| 1010 | async def stream_sca_for_all_agents( |
| 1011 | db_session: AsyncSession, |
| 1012 | customer_code: Optional[str] = None, |
| 1013 | agent_name: Optional[str] = None, |
| 1014 | policy_id: Optional[str] = None, |
| 1015 | policy_name: Optional[str] = None, |
| 1016 | min_score: Optional[int] = None, |
| 1017 | max_score: Optional[int] = None, |
| 1018 | max_concurrent_requests: int = DEFAULT_MAX_CONCURRENT_REQUESTS, |
| 1019 | ) -> AsyncGenerator[Dict[str, Any], None]: |
| 1020 | """ |
| 1021 | Stream SCA results for all agents as they are collected. |
| 1022 | |
| 1023 | Yields SSE-formatted events as results come in from each agent. |
| 1024 | |
| 1025 | Args: |
| 1026 | db_session: Database session to use |
| 1027 | customer_code: Optional customer code filter |
| 1028 | agent_name: Optional agent name filter |
| 1029 | policy_id: Optional policy ID filter |
| 1030 | policy_name: Optional policy name filter (partial matching) |
| 1031 | min_score: Optional minimum score filter |
| 1032 | max_score: Optional maximum score filter |
| 1033 | max_concurrent_requests: Maximum concurrent API requests |
| 1034 | |
| 1035 | Yields: |
| 1036 | Dict with 'event' type and 'data' payload |
| 1037 | """ |
| 1038 | try: |
| 1039 | # Get agents from database |
| 1040 | agents = await get_all_agents_from_db(db_session, customer_code) |
| 1041 | |
| 1042 | if not agents: |
| 1043 | yield { |
| 1044 | "event": "complete", |
| 1045 | "data": { |
| 1046 | "total_results": 0, |
| 1047 | "total_agents": 0, |
| 1048 | "message": "No agents found", |
| 1049 | }, |
| 1050 | } |
| 1051 | return |
| 1052 | |
| 1053 | # Filter agents by name if specified |
| 1054 | if agent_name: |
| 1055 | agents = [a for a in agents if a.hostname == agent_name] |
| 1056 | if not agents: |
| 1057 | yield { |
| 1058 | "event": "complete", |
| 1059 | "data": { |
| 1060 | "total_results": 0, |
| 1061 | "total_agents": 0, |
| 1062 | "message": f"No agents found matching hostname: {agent_name}", |
| 1063 | }, |
| 1064 | } |
| 1065 | return |
| 1066 | |
| 1067 | total_agents = len(agents) |
| 1068 | |
| 1069 | # Send start event |
| 1070 | yield { |
| 1071 | "event": "start", |
| 1072 | "data": { |
| 1073 | "total_agents": total_agents, |
| 1074 | "message": f"Starting SCA collection for {total_agents} agents...", |
| 1075 | }, |
| 1076 | } |
| 1077 | |
| 1078 | # Create semaphore for rate limiting |
| 1079 | semaphore = Semaphore(max_concurrent_requests) |
| 1080 | |
| 1081 | # Track statistics |
| 1082 | all_results: List[AgentScaOverviewItem] = [] |
| 1083 | processed_count = 0 |
| 1084 | successful_count = 0 |
| 1085 | failed_count = 0 |
| 1086 | |
| 1087 | # Create a queue to receive results as they complete |
| 1088 | result_queue: asyncio.Queue = asyncio.Queue() |
| 1089 | |
| 1090 | async def collect_and_queue(agent: Agents): |
| 1091 | """Collect SCA for an agent and put result in queue""" |
| 1092 | result = await collect_sca_for_single_agent( |
| 1093 | agent=agent, |
| 1094 | semaphore=semaphore, |
| 1095 | policy_id=policy_id, |
| 1096 | policy_name=policy_name, |
| 1097 | min_score=min_score, |
| 1098 | max_score=max_score, |
| 1099 | ) |
| 1100 | await result_queue.put((agent, result)) |
| 1101 | |
| 1102 | # Start all tasks |
| 1103 | tasks = [asyncio.create_task(collect_and_queue(agent)) for agent in agents] |
| 1104 | |
| 1105 | # Process results as they come in |
| 1106 | for _ in range(total_agents): |
| 1107 | try: |
| 1108 | # Wait for next result with timeout |
| 1109 | agent, results = await asyncio.wait_for(result_queue.get(), timeout=60.0) # 60 second timeout per agent |
| 1110 | |
| 1111 | processed_count += 1 |
| 1112 | |
| 1113 | if results: |
| 1114 | successful_count += 1 |
| 1115 | all_results.extend(results) |
| 1116 | |
| 1117 | # Yield agent results |
| 1118 | yield { |
| 1119 | "event": "agent_result", |
| 1120 | "data": { |
| 1121 | "agent_id": agent.agent_id, |
| 1122 | "agent_name": agent.hostname, |
| 1123 | "customer_code": agent.customer_code, |
| 1124 | "policy_count": len(results), |
| 1125 | "policies": [ |
| 1126 | { |
| 1127 | "policy_id": r.policy_id, |
| 1128 | "policy_name": r.policy_name, |
| 1129 | "description": r.description, |
| 1130 | "total_checks": r.total_checks, |
| 1131 | "pass_count": r.pass_count, |
| 1132 | "fail_count": r.fail_count, |
| 1133 | "invalid_count": r.invalid_count, |
| 1134 | "score": r.score, |
| 1135 | "start_scan": r.start_scan, |
| 1136 | "end_scan": r.end_scan, |
| 1137 | "references": r.references, |
| 1138 | "hash_file": r.hash_file, |
| 1139 | } |
| 1140 | for r in results |
| 1141 | ], |
| 1142 | }, |
| 1143 | } |
| 1144 | else: |
| 1145 | # Agent had no SCA data (not necessarily an error) |
| 1146 | yield { |
| 1147 | "event": "agent_empty", |
| 1148 | "data": { |
| 1149 | "agent_id": agent.agent_id, |
| 1150 | "agent_name": agent.hostname, |
| 1151 | "message": "No SCA data available", |
| 1152 | }, |
| 1153 | } |
| 1154 | |
| 1155 | # Yield progress update every 5 agents or on last agent |
| 1156 | if processed_count % 5 == 0 or processed_count == total_agents: |
| 1157 | yield { |
| 1158 | "event": "progress", |
| 1159 | "data": { |
| 1160 | "processed": processed_count, |
| 1161 | "total": total_agents, |
| 1162 | "successful": successful_count, |
| 1163 | "failed": failed_count, |
| 1164 | "results_so_far": len(all_results), |
| 1165 | "percent_complete": round((processed_count / total_agents) * 100, 1), |
| 1166 | }, |
| 1167 | } |
| 1168 | |
| 1169 | except asyncio.TimeoutError: |
| 1170 | failed_count += 1 |
| 1171 | processed_count += 1 |
| 1172 | yield { |
| 1173 | "event": "agent_error", |
| 1174 | "data": { |
| 1175 | "agent_id": "unknown", |
| 1176 | "message": "Timeout waiting for agent response", |
| 1177 | }, |
| 1178 | } |
| 1179 | except Exception as e: |
| 1180 | failed_count += 1 |
| 1181 | processed_count += 1 |
| 1182 | logger.error(f"Error processing agent result: {e}") |
| 1183 | yield { |
| 1184 | "event": "agent_error", |
| 1185 | "data": { |
| 1186 | "message": str(e), |
| 1187 | }, |
| 1188 | } |
| 1189 | |
| 1190 | # Wait for all tasks to complete (cleanup) |
| 1191 | await asyncio.gather(*tasks, return_exceptions=True) |
| 1192 | |
| 1193 | # Calculate final statistics |
| 1194 | unique_agents = set(item.agent_id for item in all_results) |
| 1195 | unique_policies = set(item.policy_id for item in all_results) |
| 1196 | |
| 1197 | total_checks = sum(item.total_checks for item in all_results) |
| 1198 | total_passes = sum(item.pass_count for item in all_results) |
| 1199 | total_fails = sum(item.fail_count for item in all_results) |
| 1200 | total_invalid = sum(item.invalid_count for item in all_results) |
| 1201 | |
| 1202 | average_score = sum(item.score for item in all_results) / len(all_results) if all_results else 0.0 |
| 1203 | |
| 1204 | # Yield completion event |
| 1205 | yield { |
| 1206 | "event": "complete", |
| 1207 | "data": { |
| 1208 | "total_results": len(all_results), |
| 1209 | "total_agents": len(unique_agents), |
| 1210 | "total_policies": len(unique_policies), |
| 1211 | "average_score": round(average_score, 2), |
| 1212 | "total_checks": total_checks, |
| 1213 | "total_passes": total_passes, |
| 1214 | "total_fails": total_fails, |
| 1215 | "total_invalid": total_invalid, |
| 1216 | "agents_processed": processed_count, |
| 1217 | "agents_successful": successful_count, |
| 1218 | "agents_failed": failed_count, |
| 1219 | "message": f"Completed SCA collection: {len(all_results)} results from {len(unique_agents)} agents", |
| 1220 | }, |
| 1221 | } |
| 1222 | |
| 1223 | except Exception as e: |
| 1224 | logger.error(f"Error in SCA streaming: {e}") |
| 1225 | yield { |
| 1226 | "event": "error", |
| 1227 | "data": { |
| 1228 | "error": str(e), |
| 1229 | "message": "Fatal error during SCA collection", |
| 1230 | }, |
| 1231 | } |
| 1232 | |
| 1233 | |
| 1234 | async def fetch_sca_policies_index() -> ScaPoliciesIndexResponse: |
| 1235 | """ |
| 1236 | Fetch the SCA policies index from the CoPilot-SCA public GitHub repository. |
| 1237 | |
| 1238 | Returns: |
| 1239 | ScaPoliciesIndexResponse with the list of available policies. |
| 1240 | """ |
| 1241 | try: |
| 1242 | async with httpx.AsyncClient(timeout=15.0) as client: |
| 1243 | response = await client.get(COPILOT_SCA_INDEX_URL) |
| 1244 | response.raise_for_status() |
| 1245 | |
| 1246 | data = response.json() |
| 1247 | |
| 1248 | policies = [ScaPolicyItem(**p) for p in data.get("policies", [])] |
| 1249 | |
| 1250 | return ScaPoliciesIndexResponse( |
| 1251 | version=data.get("version", "unknown"), |
| 1252 | last_updated=data.get("last_updated", "unknown"), |
| 1253 | policies=policies, |
| 1254 | success=True, |
| 1255 | message=f"Successfully fetched {len(policies)} available SCA policies", |
| 1256 | ) |
| 1257 | except httpx.HTTPStatusError as e: |
| 1258 | logger.error(f"HTTP error fetching SCA policies index: {e}") |
| 1259 | raise HTTPException( |
| 1260 | status_code=e.response.status_code, |
| 1261 | detail=f"Failed to fetch SCA policies index from GitHub: {e}", |
| 1262 | ) |
| 1263 | except Exception as e: |
| 1264 | logger.error(f"Error fetching SCA policies index: {e}") |
| 1265 | raise HTTPException(status_code=502, detail=f"Failed to fetch SCA policies index: {e}") |
| 1266 | |
| 1267 | |
| 1268 | async def fetch_sca_policy_content(policy_id: str) -> ScaPolicyContentResponse: |
| 1269 | """ |
| 1270 | Fetch the raw YAML content of a single SCA policy from the CoPilot-SCA |
| 1271 | public GitHub repository. |
| 1272 | |
| 1273 | The policy is looked up by its ``id`` field in the index, and the |
| 1274 | corresponding YAML file is downloaded from the raw content URL. |
| 1275 | |
| 1276 | Args: |
| 1277 | policy_id: The policy identifier (e.g. ``cis_apache_24_rpm``). |
| 1278 | |
| 1279 | Returns: |
| 1280 | ScaPolicyContentResponse with the YAML content. |
| 1281 | """ |
| 1282 | # First fetch the index to resolve the file path for the requested policy |
| 1283 | index_response = await fetch_sca_policies_index() |
| 1284 | |
| 1285 | policy = next((p for p in index_response.policies if p.id == policy_id), None) |
| 1286 | |
| 1287 | if policy is None: |
| 1288 | raise HTTPException( |
| 1289 | status_code=404, |
| 1290 | detail=f"SCA policy '{policy_id}' not found in the CoPilot-SCA repository index", |
| 1291 | ) |
| 1292 | |
| 1293 | file_url = f"{COPILOT_SCA_RAW_BASE}/{policy.file}" |
| 1294 | |
| 1295 | try: |
| 1296 | async with httpx.AsyncClient(timeout=15.0) as client: |
| 1297 | response = await client.get(file_url) |
| 1298 | response.raise_for_status() |
| 1299 | |
| 1300 | return ScaPolicyContentResponse( |
| 1301 | policy_id=policy.id, |
| 1302 | file_path=policy.file, |
| 1303 | content=response.text, |
| 1304 | success=True, |
| 1305 | message=f"Successfully fetched policy '{policy.name}'", |
| 1306 | ) |
| 1307 | except httpx.HTTPStatusError as e: |
| 1308 | logger.error(f"HTTP error fetching SCA policy content for {policy_id}: {e}") |
| 1309 | raise HTTPException( |
| 1310 | status_code=e.response.status_code, |
| 1311 | detail=f"Failed to fetch SCA policy file from GitHub: {e}", |
| 1312 | ) |
| 1313 | except Exception as e: |
| 1314 | logger.error(f"Error fetching SCA policy content for {policy_id}: {e}") |
| 1315 | raise HTTPException(status_code=502, detail=f"Failed to fetch SCA policy content: {e}") |
| 1316 | |
| 1317 | |
| 1318 | async def list_sca_package_registry() -> ScaPackageRegistryResponse: |
| 1319 | """ |
| 1320 | Return every entry in the SCA package registry so callers can see |
| 1321 | which application packages are tracked for SCA applicability. |
| 1322 | """ |
| 1323 | from app.agents.sca.models.sca_package_registry import SCA_PACKAGE_REGISTRY |
| 1324 | |
| 1325 | entries = [ |
| 1326 | ScaPackageRegistryItem( |
| 1327 | key=key, |
| 1328 | display_name=entry.display_name, |
| 1329 | sca_application=entry.sca_application, |
| 1330 | package_patterns=list(entry.package_patterns), |
| 1331 | ) |
| 1332 | for key, entry in SCA_PACKAGE_REGISTRY.items() |
| 1333 | ] |
| 1334 | |
| 1335 | return ScaPackageRegistryResponse( |
| 1336 | entries=entries, |
| 1337 | total=len(entries), |
| 1338 | success=True, |
| 1339 | message=f"Found {len(entries)} tracked SCA package categories", |
| 1340 | ) |
| 1341 | |
| 1342 | |
| 1343 | async def detect_agents_for_sca_package(registry_key: str) -> ScaPackageAgentsResponse: |
| 1344 | """ |
| 1345 | Given a registry key (e.g. ``apache``, ``mysql``), search the Wazuh |
| 1346 | Indexer for agents that have any of the associated packages installed, |
| 1347 | then cross-reference with available SCA policies for that application. |
| 1348 | """ |
| 1349 | from app.agents.sca.models.sca_package_registry import SCA_PACKAGE_REGISTRY |
| 1350 | |
| 1351 | entry = SCA_PACKAGE_REGISTRY.get(registry_key) |
| 1352 | if entry is None: |
| 1353 | raise HTTPException( |
| 1354 | status_code=404, |
| 1355 | detail=(f"Registry key '{registry_key}' not found. " f"Valid keys: {', '.join(SCA_PACKAGE_REGISTRY.keys())}"), |
| 1356 | ) |
| 1357 | |
| 1358 | # Build an OR query across all package patterns for this application |
| 1359 | should_clauses = [ |
| 1360 | {"wildcard": {"package.name": {"value": f"*{pattern}*", "case_insensitive": True}}} for pattern in entry.package_patterns |
| 1361 | ] |
| 1362 | |
| 1363 | query = {"query": {"bool": {"should": should_clauses, "minimum_should_match": 1}}} |
| 1364 | |
| 1365 | from app.connectors.wazuh_indexer.utils.universal import ( |
| 1366 | create_wazuh_indexer_client_async, |
| 1367 | ) |
| 1368 | |
| 1369 | es_client = await create_wazuh_indexer_client_async("Wazuh-Indexer") |
| 1370 | |
| 1371 | try: |
| 1372 | response = await es_client.search( |
| 1373 | index="wazuh-states-inventory-packages-*", |
| 1374 | body=query, |
| 1375 | size=10000, |
| 1376 | ) |
| 1377 | |
| 1378 | hits = response.get("hits", {}).get("hits", []) |
| 1379 | |
| 1380 | # De-duplicate by (agent_id, package_name) to avoid repeated entries |
| 1381 | seen = set() |
| 1382 | matches: list[AgentPackageMatch] = [] |
| 1383 | for hit in hits: |
| 1384 | src = hit.get("_source", {}) |
| 1385 | agent = src.get("agent", {}) |
| 1386 | pkg = src.get("package", {}) |
| 1387 | agent_id = agent.get("id") |
| 1388 | if agent_id in seen: |
| 1389 | continue |
| 1390 | seen.add(agent_id) |
| 1391 | matches.append( |
| 1392 | AgentPackageMatch( |
| 1393 | agent_id=agent.get("id"), |
| 1394 | agent_name=agent.get("name"), |
| 1395 | package_name=pkg.get("name"), |
| 1396 | package_version=pkg.get("version"), |
| 1397 | package_architecture=pkg.get("architecture"), |
| 1398 | ), |
| 1399 | ) |
| 1400 | except Exception as e: |
| 1401 | logger.error(f"Error detecting agents for SCA package '{registry_key}': {e}") |
| 1402 | raise HTTPException( |
| 1403 | status_code=500, |
| 1404 | detail=f"Failed to search packages in Wazuh Indexer: {e}", |
| 1405 | ) |
| 1406 | finally: |
| 1407 | await es_client.close() |
| 1408 | |
| 1409 | # Fetch applicable SCA policies for this application |
| 1410 | applicable_policies = [] |
| 1411 | try: |
| 1412 | index_resp = await fetch_sca_policies_index() |
| 1413 | applicable_policies = [p for p in index_resp.policies if p.application == entry.sca_application] |
| 1414 | except Exception as e: |
| 1415 | logger.warning(f"Could not fetch SCA policies index for cross-reference: {e}") |
| 1416 | |
| 1417 | return ScaPackageAgentsResponse( |
| 1418 | registry_key=registry_key, |
| 1419 | display_name=entry.display_name, |
| 1420 | sca_application=entry.sca_application, |
| 1421 | matched_agents=matches, |
| 1422 | total=len(matches), |
| 1423 | applicable_policies=applicable_policies, |
| 1424 | success=True, |
| 1425 | message=f"Found {len(matches)} agent-package combinations for '{entry.display_name}'", |
| 1426 | ) |