refactor: apply minimal code principles across backend modules

Author: utksh1

backend/secuscan/cache.py

@@ -1,6 +1,7 @@
 import json
 from typing import Any, Optional, Dict
 import time
+import heapq
 import logging
 
 from .config import settings
@@ -29,19 +30,20 @@ async def disconnect(self):
 
     def _sweep_expired(self):
         now = time.time()
-        keys = [k for k, exp in list(self._expires.items()) if exp <= now]
+        keys = [k for k, exp in self._expires.items() if exp <= now]
         for k in keys:
             self._data.pop(k, None)
             self._expires.pop(k, None)
             self._access_order.pop(k, None)
         if keys:
+            logger.debug("Swept %d expired cache entries", len(keys))
 
     def _evict_lru(self):
         if len(self._data) < self.max_entries:
             return
-        sorted_keys = sorted(self._access_order, key=lambda k: self._access_order[k])
         evict_count = max(1, int(self.max_entries * SWEEP_EVICT_FRACTION))
-        for k in sorted_keys[:evict_count]:
+        lru_keys = heapq.nsmallest(evict_count, self._access_order, key=self._access_order.get)
+        for k in lru_keys:
             self._data.pop(k, None)
             self._expires.pop(k, None)
             self._access_order.pop(k, None)

backend/secuscan/executor.py

@@ -1166,11 +1166,9 @@ async def _build_result_contract(
         structured_result["finding_groups"] = build_finding_groups(normalized_findings)
         structured_result["asset_summary"] = build_asset_summary(normalized_findings, asset_services)
         structured_result["scan_diff"] = build_scan_diff(normalized_findings, previous_findings)
-        severity_counts: Dict[str, int] = {}
-        for f in normalized_findings:
-            sev = str(f.get("severity", "info")).lower()
-            severity_counts[sev] = severity_counts.get(sev, 0) + 1
-        structured_result["severity_counts"] = severity_counts
+        from collections import Counter
+        severity_counts = Counter(str(f.get("severity", "info")).lower() for f in normalized_findings)
+        structured_result["severity_counts"] = dict(severity_counts)
         structured_result["count"] = len(normalized_findings)
         return structured_result, previous_findings, asset_services
 
@@ -1299,20 +1297,104 @@ async def _upsert_findings_and_report(self, db, task_id: str, owner_id: str, plu
         )
         findings_data: List[Dict[str, Any]] = []
         async with db.transaction():
+            insert_sql = """INSERT INTO findings (
+                id, owner_id, task_id, plugin_id, title, category, severity, target, description,
+                remediation, proof, cvss, cve, metadata_json, discovered_at, exploitability,
+                confidence, validated, validation_method, confidence_reason, finding_kind,
+                finding_group_id, asset_id, first_seen_at, last_seen_at, occurrence_count,
+                corroborating_sources_json, evidence_count, analyst_status, retest_status,
+                evidence_json, asset_refs_json, service_fingerprint, cpe, references_json,
+                asset_exposure, risk_score, risk_factors_json
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(owner_id, finding_group_id) DO UPDATE SET
+                occurrence_count = occurrence_count + 1,
+                last_seen_at = excluded.last_seen_at,
+                evidence_count = excluded.evidence_count,
+                corroborating_sources_json = excluded.corroborating_sources_json"""
+            
+            batch_data = []
             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,
-                    )
+                finding_id = generate_id("finding")
+                discovered = datetime.now(timezone.utc)
+                target_value = str(finding.get("target") or target)
+                metadata = finding.get("metadata", {}) if isinstance(finding.get("metadata"), dict) else {}
+                evidence = finding.get("evidence", []) if isinstance(finding.get("evidence"), list) else []
+                asset_refs = finding.get("asset_refs", []) if isinstance(finding.get("asset_refs"), list) else []
+                exploitability = finding.get("exploitability")
+                asset_exposure = finding.get("asset_exposure")
+                confidence = finding.get("confidence")
+                finding_group_id = self._compute_finding_group_id(
+                    owner_id=owner_id,
+                    target=target_value,
+                    title=finding["title"],
+                    category=finding["category"],
+                    severity=finding["severity"],
+                    host=finding.get("host"),
+                    port=finding.get("port"),
+                    path=finding.get("path"),
+                    cve=finding.get("cve"),
+                )
+                references = finding.get("references", []) if isinstance(finding.get("references"), list) else []
+                corroborating_sources = finding.get("corroborating_sources", []) if isinstance(finding.get("corroborating_sources"), list) else []
+                first_seen_at = str(finding.get("first_seen_at") or to_utc_iso(discovered))
+                last_seen_at = str(finding.get("last_seen_at") or to_utc_iso(discovered))
+                occurrence_count = int(finding.get("occurrence_count") or 1)
+                evidence_count = int(finding.get("evidence_count") or len(evidence))
+                risk_score = compute_risk_score(
+                    severity=finding["severity"],
+                    exploitability=exploitability,
+                    asset_exposure=asset_exposure,
+                    discovered_at=discovered,
+                    confidence=confidence,
                 )
+                risk_factors = compute_risk_factors(
+                    severity=finding["severity"],
+                    exploitability=exploitability,
+                    asset_exposure=asset_exposure,
+                    discovered_at=discovered,
+                    confidence=confidence,
+                    risk_score=risk_score,
+                )
+                
+                batch_data.append((
+                    finding_id, owner_id, task_id, plugin_id, finding["title"], finding["category"],
+                    finding["severity"], target_value, finding["description"], finding.get("remediation", ""),
+                    finding.get("proof"), finding.get("cvss"), finding.get("cve"), json.dumps(metadata),
+                    to_utc_iso(discovered), exploitability, confidence, 1 if finding.get("validated") else 0,
+                    finding.get("validation_method"), finding.get("confidence_reason"),
+                    str(finding.get("finding_kind") or "observation"), finding_group_id, finding.get("asset_id"),
+                    first_seen_at, last_seen_at, occurrence_count, json.dumps(corroborating_sources),
+                    evidence_count, str(finding.get("analyst_status") or "new"),
+                    str(finding.get("retest_status") or "not_requested"), json.dumps(evidence),
+                    json.dumps(asset_refs), finding.get("service_fingerprint"), finding.get("cpe"),
+                    json.dumps(references), asset_exposure, risk_score, json.dumps(risk_factors)
+                ))
+                
+                findings_data.append({
+                    **finding,
+                    "id": finding_id,
+                    "plugin_id": plugin_id,
+                    "target": target_value,
+                    "discovered_at": to_utc_iso(discovered),
+                    "metadata": metadata,
+                    "evidence": evidence,
+                    "asset_refs": asset_refs,
+                    "references": references,
+                    "corroborating_sources": corroborating_sources,
+                    "first_seen_at": first_seen_at,
+                    "last_seen_at": last_seen_at,
+                    "occurrence_count": occurrence_count,
+                    "evidence_count": evidence_count,
+                    "risk_score": risk_score,
+                    "risk_factors": risk_factors,
+                })
+            
+            if batch_data:
+                await db.executemany(insert_sql, batch_data)
 
             structured_result["findings"] = findings_data
-            structured_result["severity_counts"] = self._build_severity_counts(findings_data)
+            from collections import Counter
+            structured_result["severity_counts"] = dict(Counter(str(f.get("severity", "info")).lower() for f in 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)

backend/secuscan/finding_intelligence.py

@@ -196,16 +196,12 @@ def _dedupe_evidence(items: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
     unique: List[Dict[str, Any]] = []
     seen = set()
     for item in items:
-        key = json.dumps(
-            {
-                "type": item.get("type"),
-                "label": item.get("label"),
-                "value": item.get("value"),
-                "artifact_ref": item.get("artifact_ref"),
-                "source": item.get("source"),
-            },
-            sort_keys=True,
-            default=str,
+        key = (
+            item.get("type"),
+            item.get("label"),
+            str(item.get("value")),
+            item.get("artifact_ref"),
+            item.get("source"),
         )
         if key in seen:
             continue
@@ -372,11 +368,18 @@ async def normalize_and_correlate_findings(
         staged_item["corroborating_sources"] = sorted({str(s).strip() for s in [*staged_item.get("corroborating_sources", [] if str(s).strip()}), *sources])
         staged_item["metadata"].update({key: value for key, value in (finding.get("metadata") or {}).items() if value not in ("", None, [], {})})
 
+    all_group_ids = list(staged.keys())
+    previous_map = {}
+    if all_group_ids:
+        previous_rows = await db.fetchall(
+            f"SELECT * FROM findings WHERE owner_id = ? AND finding_group_id IN ({','.join('?' * len(all_group_ids))})",
+            (owner_id, *all_group_ids)
+        )
+        previous_map = {row["finding_group_id"]: row for row in previous_rows}
+    
     normalized: List[Dict[str, Any]] = []
     for finding_group_id, finding in staged.items():
-        previous = await db.fetchone(
-            (owner_id, finding_group_id),
-        )
+        previous = previous_map.get(finding_group_id)
         prior_sources = []
         if previous and previous.get("corroborating_sources_json"):
             try:
@@ -414,7 +417,7 @@ async def normalize_and_correlate_findings(
 
     if settings.triage_engine_enabled and settings.triage_engine_api_key:
         try:
-            triage_engine.triage_findings(
+            await triage_engine.triage_findings_async(
                 normalized,
                 model=settings.triage_engine_model,
                 api_key=settings.triage_engine_api_key,

backend/secuscan/ratelimit.py

@@ -1,6 +1,6 @@
-from collections import defaultdict
+from collections import defaultdict, deque
 from datetime import datetime, timedelta
-from typing import Tuple, Dict, List
+from typing import Tuple, Dict
 import asyncio
 
 from fastapi import Request, Response, HTTPException
@@ -9,7 +9,7 @@
 
 class RateLimiter:
     def __init__(self):
-        self.task_history: Dict[str, List[datetime]] = defaultdict(list)
+        self.task_history: Dict[str, deque] = defaultdict(lambda: deque())
         self.lock = asyncio.Lock()
 
     async def can_execute(
@@ -24,28 +24,24 @@ async def can_execute(
             now = datetime.now()
             hour_ago = now - timedelta(hours=1)
 
-            # Clean old entries for this bucket
-            self.task_history[bucket] = [
-                ts for ts in self.task_history[bucket]
-                if ts > hour_ago
-            ]
+            history = self.task_history[bucket]
+            while history and history[0] <= hour_ago:
+                history.popleft()
 
-            recent_count = len(self.task_history[bucket])
+            recent_count = len(history)
 
             if recent_count >= max_per_hour:
                 return False, f"Rate limit exceeded: {recent_count}/{max_per_hour} per hour"
 
-            # Record this execution
-            self.task_history[bucket].append(now)
+            history.append(now)
             return True, ""
 
     async def reset(self, plugin_id: str = None):
         async with self.lock:
             if plugin_id:
-                # Remove every bucket that ends with :<plugin_id>
                 keys_to_clear = [k for k in self.task_history if k.endswith(f":{plugin_id}")]
                 for k in keys_to_clear:
-                    self.task_history[k] = []
+                    self.task_history[k].clear()
             else:
                 self.task_history.clear()
 

backend/secuscan/routes.py

@@ -1045,8 +1045,6 @@ async def get_findings(
     page: int = Query(1, ge=1),
     per_page: int = Query(50, ge=1, le=200),
 ):
-    # Return the caller's vulnerability findings with pagination.
-
     async def build():
         db = await get_db()
         offset = (page - 1) * per_page
@@ -1060,22 +1058,21 @@ async def build():
         )
         total = total_row["count"] if total_row else 0
         findings = deserialize_finding_rows(rows)
-        # Build finding_groups from *all* findings so group counts remain accurate
-        # regardless of which page is being viewed.
-        all_rows = await db.fetchall(
-            "SELECT * FROM findings WHERE owner_id = ? ORDER BY discovered_at DESC",
+        
+        group_rows = await db.fetchall(
+            "SELECT finding_group_id, COUNT(*) as count FROM findings WHERE owner_id = ? GROUP BY finding_group_id",
             (owner,),
         )
-        all_findings = deserialize_finding_rows(all_rows)
+        finding_groups = [{"id": row["finding_group_id"], "count": row["count"]} for row in group_rows]
+        
         return {
             "findings": findings,
-            "finding_groups": build_finding_groups(all_findings),
+            "finding_groups": finding_groups,
             "total": total,
             "page": page,
             "per_page": per_page,
         }
 
-    # Cache key includes pagination params so different pages do not collide.
     return await get_or_set_cached(f"findings:list:{owner}:page={page}:per_page={per_page}", build)