feat(alerts): realtime alert notifications over SMTP + in-app (#25)

core/api_routers/rules.py

@@ -12,8 +12,30 @@
 
 from core.db import get_db
 from core.db.models import Rule, Alert
-from core.api_schemas.rules import RuleCreate, RuleUpdate, RuleResponse, AlertResponse
+from core.api_schemas.rules import (
+    RuleCreate, RuleUpdate, RuleResponse, AlertCreate, AlertResponse,
+)
 from core.auth import require_permission
+from core.services.notification_service import AlertNotifier
+
+SDK_ALERT_RULE_NAME = "SDK Alerts"
+
+
+def _get_or_create_sdk_rule(db: Session) -> Rule:
+    '''owning rule for alerts raised directly via the SDK (alerts.rule_id is required)'''
+    rule = db.query(Rule).filter(Rule.name == SDK_ALERT_RULE_NAME).first()
+    if rule is None:
+        rule = Rule(
+            name=SDK_ALERT_RULE_NAME,
+            description="System rule for alerts raised directly through the OpenUBA SDK.",
+            rule_type="single-fire",
+            condition="sdk",
+            enabled=True,
+            severity="medium",
+        )
+        db.add(rule)
+        db.flush()
+    return rule
 
 router = APIRouter()
 logger = logging.getLogger(__name__)
@@ -119,6 +141,62 @@ async def delete_rule(
     logger.info(f"deleted rule: {rule_id}")
 
 
+@router.post("/alerts", response_model=AlertResponse, status_code=201)
+async def create_alert(
+    alert_data: AlertCreate,
+    db: Session = Depends(get_db),
+    current_user: dict = Depends(require_permission("rules", "write"))
+):
+    '''
+    raise an alert directly (e.g. from a model via the OpenUBA SDK).
+
+    if `notify` is set, realtime notifications (SMTP email + in-app) are
+    dispatched best-effort, reusing the same delivery path as rule-fired alerts.
+    '''
+    if alert_data.rule_id is not None:
+        rule = db.query(Rule).filter(Rule.id == alert_data.rule_id).first()
+        if rule is None:
+            raise HTTPException(status_code=404, detail="rule not found")
+    else:
+        rule = _get_or_create_sdk_rule(db)
+
+    context = dict(alert_data.context or {})
+    context.setdefault("source", "sdk")
+
+    alert = Alert(
+        rule_id=rule.id,
+        severity=alert_data.severity,
+        message=alert_data.message,
+        entity_id=alert_data.entity_id,
+        entity_type=alert_data.entity_type or "user",
+        alert_context=context,
+        acknowledged=False,
+    )
+    db.add(alert)
+    db.flush()
+
+    if alert_data.notify:
+        try:
+            AlertNotifier().notify(
+                db,
+                rule_name=rule.name,
+                severity=alert.severity,
+                message=alert.message,
+                entity_id=alert.entity_id or "unknown",
+                entity_type=alert.entity_type or "user",
+                alert_id=str(alert.id),
+                recipients=alert_data.recipients,
+                context=context,
+            )
+        except Exception as e:
+            logger.error(f"alert notification dispatch failed: {e}")
+
+    db.commit()
+    db.refresh(alert)
+    logger.info(f"alert created via api: rule={rule.name} severity={alert.severity}")
+    return alert
+
+
 @router.get("/alerts", response_model=List[AlertResponse])
 async def list_alerts(
     severity: Optional[str] = Query(None, pattern="^(critical|high|medium|low)$"),

core/api_routers/settings.py

@@ -17,7 +17,7 @@
 router = APIRouter()
 logger = logging.getLogger(__name__)
 
-VALID_INTEGRATION_TYPES = {"ollama", "openai", "claude", "gemini", "elasticsearch", "spark"}
+VALID_INTEGRATION_TYPES = {"ollama", "openai", "claude", "gemini", "elasticsearch", "spark", "smtp"}
 
 
 class IntegrationConfigUpdate(BaseModel):
@@ -183,6 +183,8 @@ async def test_integration(
             return await _test_elasticsearch(config)
         elif integration_type == "spark":
             return await _test_spark(config)
+        elif integration_type == "smtp":
+            return await _test_smtp(config)
     except Exception as e:
         logger.error(f"test {integration_type} failed: {e}")
         return {"status": "error", "message": str(e)}
@@ -283,10 +285,42 @@ async def _test_spark(config: dict) -> dict:
         return {"status": "error", "message": f"HTTP {resp.status_code}"}
 
 
+async def _test_smtp(config: dict) -> dict:
+    '''verify SMTP connectivity + auth by opening a session and issuing NOOP'''
+    import smtplib
+    import ssl
+
+    host = config.get("host", "")
+    if not host:
+        return {"status": "error", "message": "host not configured"}
+    port = int(config.get("port") or 587)
+    use_ssl = bool(config.get("use_ssl", False))
+    use_tls = bool(config.get("use_tls", True))
+    username = config.get("username", "")
+    password = config.get("password", "")
+
+    try:
+        if use_ssl:
+            server: smtplib.SMTP = smtplib.SMTP_SSL(
+                host, port, timeout=10, context=ssl.create_default_context()
+            )
+        else:
+            server = smtplib.SMTP(host, port, timeout=10)
+            if use_tls:
+                server.starttls(context=ssl.create_default_context())
+        with server:
+            if username and password:
+                server.login(username, password)
+            server.noop()
+        return {"status": "connected"}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
 def _mask_sensitive_fields(config: dict) -> dict:
-    '''mask api keys in config for safe display'''
+    '''mask api keys / passwords in config for safe display'''
     masked = dict(config)
-    for key in ("api_key",):
+    for key in ("api_key", "password"):
         if key in masked and masked[key]:
             val = str(masked[key])
             if len(val) > 8:

core/api_schemas/rules.py

@@ -62,6 +62,17 @@ class Config:
         from_attributes = True
 
 
+class AlertCreate(BaseModel):
+    severity: str = Field(default="medium", pattern="^(critical|high|medium|low)$")
+    message: str = Field(..., min_length=1)
+    entity_id: Optional[str] = None
+    entity_type: Optional[str] = "user"
+    rule_id: Optional[UUID] = None
+    context: Optional[dict] = None
+    notify: bool = False
+    recipients: Optional[Union[str, List[str]]] = None
+
+
 class AlertResponse(BaseModel):
     id: UUID
     rule_id: UUID

core/services/notification_service.py

@@ -0,0 +1,218 @@
+'''
+Copyright 2019-Present The OpenUBA Platform Authors
+notification service — realtime alert delivery over SMTP (email) and in-app.
+
+Config is read from the integration_settings table (integration_type="smtp"),
+mirroring how ChatService loads LLM provider config, and falls back to
+environment variables. Nothing here raises to the caller: delivery is
+best-effort so a mail outage can never block alert creation.
+'''
+
+import logging
+import os
+import smtplib
+import ssl
+from email.message import EmailMessage
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from core.db import get_db_context
+from core.db.models import Notification
+
+logger = logging.getLogger(__name__)
+
+# actions (from the rule-canvas alert node) that should send notifications
+NOTIFY_ACTIONS = {"notify", "fire_alert_and_notify", "notify_and_open_case"}
+
+
+def _as_recipient_list(value: Any) -> List[str]:
+    '''normalize a recipients value (list or comma/space string) to a clean list'''
+    if not value:
+        return []
+    if isinstance(value, str):
+        parts = value.replace(";", ",").replace(" ", ",").split(",")
+    elif isinstance(value, (list, tuple, set)):
+        parts = list(value)
+    else:
+        return []
+    return [str(p).strip() for p in parts if str(p).strip()]
+
+
+class EmailService:
+    '''
+    minimal SMTP sender. loads config from integration_settings (type="smtp"),
+    falls back to SMTP_* env vars. send() never raises — returns True/False.
+    '''
+
+    def _load_smtp_config(self) -> Optional[Dict[str, Any]]:
+        '''return smtp config dict if enabled+configured, else None'''
+        config: Dict[str, Any] = {}
+        try:
+            with get_db_context() as db:
+                row = db.execute(text(
+                    "SELECT config, enabled FROM integration_settings "
+                    "WHERE integration_type = 'smtp'"
+                )).fetchone()
+            if row and row[1]:  # enabled
+                config = dict(row[0]) if row[0] else {}
+        except Exception as e:
+            logger.warning(f"failed to load smtp config from db: {e}")
+
+        # env fallback for any field not set in the db row
+        host = config.get("host") or os.environ.get("SMTP_HOST", "")
+        if not host:
+            return None
+
+        def _flag(cfg_key: str, env_key: str, default: bool) -> bool:
+            if cfg_key in config:
+                return bool(config[cfg_key])
+            env = os.environ.get(env_key)
+            if env is None:
+                return default
+            return env.strip().lower() in ("1", "true", "yes", "on")
+
+        return {
+            "host": host,
+            "port": int(config.get("port") or os.environ.get("SMTP_PORT", 587)),
+            "username": config.get("username") or os.environ.get("SMTP_USERNAME", ""),
+            "password": config.get("password") or os.environ.get("SMTP_PASSWORD", ""),
+            "from_addr": (
+                config.get("from_addr")
+                or config.get("from")
+                or os.environ.get("SMTP_FROM")
+                or config.get("username")
+                or os.environ.get("SMTP_USERNAME", "openuba@localhost")
+            ),
+            "use_tls": _flag("use_tls", "SMTP_USE_TLS", True),
+            "use_ssl": _flag("use_ssl", "SMTP_USE_SSL", False),
+            "default_recipients": _as_recipient_list(
+                config.get("default_recipients")
+                or os.environ.get("SMTP_DEFAULT_RECIPIENTS", "")
+            ),
+        }
+
+    def is_configured(self) -> bool:
+        return self._load_smtp_config() is not None
+
+    def default_recipients(self) -> List[str]:
+        cfg = self._load_smtp_config()
+        return cfg["default_recipients"] if cfg else []
+
+    def send(
+        self,
+        to: List[str],
+        subject: str,
+        body_text: str,
+        body_html: Optional[str] = None,
+        timeout: int = 10,
+    ) -> bool:
+        '''send an email. returns True on success, False otherwise (never raises).'''
+        recipients = _as_recipient_list(to)
+        if not recipients:
+            logger.debug("email send skipped: no recipients")
+            return False
+
+        cfg = self._load_smtp_config()
+        if not cfg:
+            logger.debug("email send skipped: smtp not configured/enabled")
+            return False
+
+        msg = EmailMessage()
+        msg["Subject"] = subject
+        msg["From"] = cfg["from_addr"]
+        msg["To"] = ", ".join(recipients)
+        msg.set_content(body_text)
+        if body_html:
+            msg.add_alternative(body_html, subtype="html")
+
+        try:
+            if cfg["use_ssl"]:
+                context = ssl.create_default_context()
+                server: smtplib.SMTP = smtplib.SMTP_SSL(
+                    cfg["host"], cfg["port"], timeout=timeout, context=context
+                )
+            else:
+                server = smtplib.SMTP(cfg["host"], cfg["port"], timeout=timeout)
+                if cfg["use_tls"]:
+                    server.starttls(context=ssl.create_default_context())
+            with server:
+                if cfg["username"] and cfg["password"]:
+                    server.login(cfg["username"], cfg["password"])
+                server.send_message(msg)
+            logger.info(f"alert email sent to {len(recipients)} recipient(s)")
+            return True
+        except Exception as e:
+            logger.error(f"failed to send alert email: {e}")
+            return False
+
+
+class AlertNotifier:
+    '''
+    orchestrates realtime delivery for a fired alert: SMTP email (to the rule's
+    recipients or the configured default list) plus an in-app Notification for
+    every active user. Best-effort — failures are logged, never raised.
+    '''
+
+    def __init__(self, email_service: Optional[EmailService] = None):
+        self.email = email_service or EmailService()
+
+    @staticmethod
+    def should_notify(action: Optional[str]) -> bool:
+        return bool(action) and action in NOTIFY_ACTIONS
+
+    def notify(
+        self,
+        db: Session,
+        *,
+        rule_name: str,
+        severity: str,
+        message: str,
+        entity_id: str,
+        entity_type: str,
+        alert_id: Optional[str] = None,
+        recipients: Optional[Any] = None,
+        context: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        '''deliver email + in-app notifications for one alert. returns a summary.'''
+        subject = f"[OpenUBA] {severity.upper()} alert: {rule_name}"
+        lines = [
+            f"Rule: {rule_name}",
+            f"Severity: {severity}",
+            f"Entity: {entity_type}/{entity_id}",
+            f"Message: {message}",
+        ]
+        if context:
+            if context.get("risk_score") is not None:
+                lines.append(f"Risk score: {context.get('risk_score')}")
+            if context.get("anomaly_type"):
+                lines.append(f"Anomaly type: {context.get('anomaly_type')}")
+        body_text = "\n".join(lines)
+
+        summary = {"email_sent": False, "in_app_created": 0}
+
+        # 1) email delivery (rule recipients override the configured default list)
+        to = _as_recipient_list(recipients) or self.email.default_recipients()
+        if to:
+            summary["email_sent"] = self.email.send(to, subject, body_text)
+
+        # 2) in-app notifications for every active user (reuses the notifications
+        #    center already wired into the frontend + notifications router)
+        try:
+            user_rows = db.execute(text(
+                "SELECT id FROM users WHERE is_active = true"
+            )).fetchall()
+            for (user_id,) in user_rows:
+                db.add(Notification(
+                    user_id=user_id,
+                    title=subject,
+                    message=message,
+                    type="alert",
+                    link="/alerts",
+                ))
+            summary["in_app_created"] = len(user_rows)
+        except Exception as e:
+            logger.error(f"failed to create in-app notifications: {e}")
+
+        return summary