diff --git a/backend/secuscan/executor.py b/backend/secuscan/executor.py index 2d99769f..fb7f84b9 100644 --- a/backend/secuscan/executor.py +++ b/backend/secuscan/executor.py @@ -1468,25 +1468,25 @@ async def _upsert_findings_and_report(self, db, task_id: str, owner_id: str, plu result=parsed, ) findings_data: List[Dict[str, Any]] = [] - for finding in structured_result.get("findings", []): - findings_data.append( - await self._persist_finding( - db, - owner_id=owner_id, - task_id=task_id, - plugin_id=plugin_id, - target=target, - finding=finding, + async with db.transaction(): + for finding in structured_result.get("findings", []): + findings_data.append( + await self._persist_finding( + db, + owner_id=owner_id, + task_id=task_id, + plugin_id=plugin_id, + target=target, + finding=finding, + ) ) - ) - structured_result["findings"] = findings_data - structured_result["severity_counts"] = self._build_severity_counts(findings_data) - structured_result["finding_groups"] = build_finding_groups(findings_data) - structured_result["asset_summary"] = build_asset_summary(findings_data, asset_services) - structured_result["scan_diff"] = build_scan_diff(findings_data, previous_findings) + structured_result["findings"] = findings_data + structured_result["severity_counts"] = self._build_severity_counts(findings_data) + structured_result["finding_groups"] = build_finding_groups(findings_data) + structured_result["asset_summary"] = build_asset_summary(findings_data, asset_services) + structured_result["scan_diff"] = build_scan_diff(findings_data, previous_findings) - async with db.transaction(): await db.execute( "UPDATE tasks SET structured_json = ? WHERE id = ?", (json.dumps(structured_result), task_id) @@ -1534,25 +1534,25 @@ async def _upsert_findings_and_report_from_scanner(self, db, task_id: str, owner result=result, ) findings_data: List[Dict[str, Any]] = [] - for finding in structured_result.get("findings", []): - findings_data.append( - await self._persist_finding( - db, - owner_id=owner_id, - task_id=task_id, - plugin_id=plugin_id, - target=target, - finding=finding, + async with db.transaction(): + for finding in structured_result.get("findings", []): + findings_data.append( + await self._persist_finding( + db, + owner_id=owner_id, + task_id=task_id, + plugin_id=plugin_id, + target=target, + finding=finding, + ) ) - ) - structured_result["findings"] = findings_data - structured_result["severity_counts"] = self._build_severity_counts(findings_data) - structured_result["finding_groups"] = build_finding_groups(findings_data) - structured_result["asset_summary"] = build_asset_summary(findings_data, asset_services) - structured_result["scan_diff"] = build_scan_diff(findings_data, previous_findings) + structured_result["findings"] = findings_data + structured_result["severity_counts"] = self._build_severity_counts(findings_data) + structured_result["finding_groups"] = build_finding_groups(findings_data) + structured_result["asset_summary"] = build_asset_summary(findings_data, asset_services) + structured_result["scan_diff"] = build_scan_diff(findings_data, previous_findings) - async with db.transaction(): await db.execute( "UPDATE tasks SET structured_json = ? WHERE id = ?", (json.dumps(structured_result), task_id) diff --git a/backend/secuscan/notification_service.py b/backend/secuscan/notification_service.py index 54145f9f..f21701c9 100644 --- a/backend/secuscan/notification_service.py +++ b/backend/secuscan/notification_service.py @@ -629,6 +629,84 @@ async def deliver_via_rule( ) +async def _deliver_rule_batch( + db: Database, + rule: Dict[str, Any], + findings: List[Dict[str, Any]], +) -> List[DeliveryResult]: + """Deliver a batched notification for one rule across multiple findings. + + Sends a single webhook per (rule, task) containing an array of findings, + then records individual delivery records in a batch insert. + """ + results: List[DeliveryResult] = [] + channel = str(rule.get("channel_type", "")).lower() + target = str(rule.get("target_url_or_email", "")) + + if channel == NotificationChannelType.WEBHOOK.value: + payloads = [build_alert_payload(finding, rule) for finding in findings] + batch_payload = { + "event": "finding.alert.batch", + "rule": { + "id": rule.get("id"), + "name": rule.get("name"), + "severity_threshold": rule.get("severity_threshold"), + "channel_type": rule.get("channel_type"), + }, + "findings": payloads, + "count": len(payloads), + } + + async def _send_and_record(): + ok, error = await send_webhook(target, batch_payload) + status = ( + NotificationDeliveryStatus.SUCCESS if ok else NotificationDeliveryStatus.FAILED + ) + for finding in findings: + await record_delivery( + db, str(rule["id"]), str(finding["id"]), status, error, + ) + return [ + DeliveryResult( + rule_id=str(rule["id"]), + finding_id=str(finding["id"]), + status=status, + error_message=error, + ) + for finding in findings + ] + + results = await _send_and_record() + elif channel == NotificationChannelType.EMAIL.value: + for finding in findings: + payload = build_alert_payload(finding, rule) + ok, error = await send_email(target, payload) + status = ( + NotificationDeliveryStatus.SUCCESS if ok else NotificationDeliveryStatus.FAILED + ) + await record_delivery(db, str(rule["id"]), str(finding["id"]), status, error) + results.append( + DeliveryResult( + rule_id=str(rule["id"]), + finding_id=str(finding["id"]), + status=status, + error_message=error, + ) + ) + else: + for finding in findings: + results.append( + DeliveryResult( + rule_id=str(rule["id"]), + finding_id=str(finding["id"]), + status=NotificationDeliveryStatus.FAILED, + error_message=f"Unsupported channel type: {channel}", + ) + ) + + return results + + async def process_finding_notifications( db: Database, finding_id: str, @@ -651,14 +729,64 @@ async def process_task_notifications( db: Database, task_id: str, ) -> List[DeliveryResult]: - """Evaluate notifications for every finding produced by a task.""" + """Evaluate notifications for every finding produced by a task. + + Batches all findings and rules upfront (2 queries), then evaluates + (finding, rule) pairs in-memory to eliminate N+1 query cascading. + Delivery history records are batched and webhook payloads are + deduplicated per (rule, task). + """ findings = await db.fetchall( - "SELECT id FROM findings WHERE task_id = ? ORDER BY discovered_at ASC", + "SELECT * FROM findings WHERE task_id = ? ORDER BY discovered_at ASC", (task_id,), ) + if not findings: + return [] + + rules = await db.fetchall( + "SELECT * FROM notification_rules WHERE is_active = 1 ORDER BY created_at ASC" + ) + if not rules: + return [] + + existing_deliveries = await db.fetchall( + """ + SELECT nh.rule_id, nh.finding_id + FROM notification_history nh + JOIN findings f ON f.id = nh.finding_id + WHERE f.task_id = ? AND nh.status = ? + """, + (task_id, NotificationDeliveryStatus.SUCCESS.value), + ) + already_delivered = {(row["rule_id"], row["finding_id"]) for row in existing_deliveries} + + rule_map: Dict[str, Dict[str, Any]] = {str(r["id"]): r for r in rules} + pending_by_rule: Dict[str, List[Dict[str, Any]]] = {} + + for finding in findings: + finding_id = str(finding["id"]) + for rule in rules: + rule_id = str(rule["id"]) + + if not bool(rule.get("is_active")): + continue + + if not severity_meets_threshold( + str(finding.get("severity", "info")), + str(rule.get("severity_threshold", "info")), + ): + continue + + if (rule_id, finding_id) in already_delivered: + continue + + pending_by_rule.setdefault(rule_id, []).append(finding) + results: List[DeliveryResult] = [] - for row in findings: - results.extend(await process_finding_notifications(db, str(row["id"]))) + for rule_id, pending_findings in pending_by_rule.items(): + rule = rule_map[rule_id] + results.extend(await _deliver_rule_batch(db, rule, pending_findings)) + return results