"""Detector registry — holds all registered leak detectors and runs scans.
Per-company enable/disable + overrides live in Company.settings_json under
the 'leak_detectors' key:
settings_json['leak_detectors'] = {
'<detector_id>': {
'enabled': bool,
'severity_override': 'low'|'medium'|'high'|'critical'|None,
'threshold_override': dict|None,
},
...
}
Detectors absent from settings fall back to detector.enabled_by_default.
"""
from __future__ import annotations
import logging
from typing import Any, Dict, List, Optional
from .base import BaseDetector, VALID_SEVERITIES
logger = logging.getLogger(__name__)
SETTINGS_KEY = "leak_detectors"
def get_detector_settings(company) -> Dict[str, Any]:
"""Return the leak_detectors settings dict from a Company row."""
settings = company.settings_json or {}
detectors = settings.get(SETTINGS_KEY) or {}
return detectors if isinstance(detectors, dict) else {}
class DetectorRegistry:
"""Registry of all available leak detectors."""
def __init__(self) -> None:
self._detectors: Dict[str, BaseDetector] = {}
# -- Registration -----------------------------------------------------------
def register(self, detector: BaseDetector) -> None:
if not detector.id:
raise ValueError("Detector must define a non-empty id")
if detector.id in self._detectors:
raise ValueError(f"Duplicate detector id: {detector.id}")
self._detectors[detector.id] = detector
# -- Introspection ----------------------------------------------------------
def list_detectors(self) -> List[BaseDetector]:
"""All registered detectors (stable order by id)."""
return [self._detectors[k] for k in sorted(self._detectors)]
def get(self, detector_id: str) -> Optional[BaseDetector]:
return self._detectors.get(detector_id)
def describe(self, company=None) -> List[Dict[str, Any]]:
"""Serializable detector list, merged with company settings if given."""
settings = get_detector_settings(company) if company is not None else {}
out = []
for det in self.list_detectors():
conf = settings.get(det.id) or {}
out.append({
"id": det.id,
"name": det.name,
"description": det.description,
"connector": det.connector,
"enabled_by_default": det.enabled_by_default,
"default_params": det.default_params or {},
"enabled": bool(conf.get("enabled", det.enabled_by_default)),
"severity_override": conf.get("severity_override"),
"threshold_override": conf.get("threshold_override"),
})
return out
# -- Enablement -------------------------------------------------------------
def get_enabled(self, company_id: str) -> List[BaseDetector]:
"""Detectors enabled for a company (settings_json override aware)."""
from app.models import db, Company
company = db.session.get(Company, company_id)
if company is None:
return []
settings = get_detector_settings(company)
enabled = []
for det in self.list_detectors():
conf = settings.get(det.id) or {}
if bool(conf.get("enabled", det.enabled_by_default)):
enabled.append(det)
return enabled
# -- Execution ---------------------------------------------------------------
def run_all(
self,
company_id: str,
detector_ids: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""Run enabled detectors for a company and upsert results.
detector_ids: optional whitelist — only run these (still respecting
per-company enablement). Returns a summary dict.
"""
from app.models import db, Company
company = db.session.get(Company, company_id)
if company is None:
return {
"scanned": 0, "new_leaks": 0, "updated_leaks": 0,
"already_resolved": 0, "leaks": [], "errors": ["Company not found"],
}
settings = get_detector_settings(company)
detectors = self.get_enabled(company_id)
if detector_ids is not None:
wanted = set(detector_ids)
detectors = [d for d in detectors if d.id in wanted]
new_leaks = 0
updated_leaks = 0
already_resolved = 0
leaks: List[Dict[str, Any]] = []
errors: List[str] = []
for det in detectors:
conf = settings.get(det.id) or {}
severity_override = conf.get("severity_override")
try:
candidates = det.check(company_id) or []
except Exception as exc: # never let one detector kill the scan
logger.exception(
"Detector %s failed for company %s", det.id, company_id
)
errors.append(f"{det.id}: {exc}")
continue
for candidate in candidates:
if severity_override in VALID_SEVERITIES:
candidate.severity = severity_override
try:
status, leak = det._upsert(company_id, candidate)
except Exception as exc:
logger.exception(
"Upsert failed for detector %s / company %s",
det.id, company_id,
)
errors.append(f"{det.id}: upsert failed: {exc}")
continue
if status == "updated":
updated_leaks += 1
else:
new_leaks += 1
if status == "created_after_resolved":
already_resolved += 1
leaks.append({
"detector_id": leak.detector_id,
"source": leak.source,
"description": leak.description,
"estimated_loss": leak.estimated_loss,
"severity": leak.severity,
"status": "updated" if status == "updated" else "new",
})
try:
db.session.commit()
except Exception as exc:
logger.exception("Leak scan commit failed for company %s", company_id)
db.session.rollback()
errors.append(f"commit failed: {exc}")
new_leaks = updated_leaks = already_resolved = 0
leaks = []
# ids only available post-commit
return {
"scanned": len(detectors),
"new_leaks": new_leaks,
"updated_leaks": updated_leaks,
"already_resolved": already_resolved,
"leaks": leaks,
"errors": errors,
}
# Global singleton registry
registry = DetectorRegistry()