diff --git a/backend/app/api/api_v1/endpoints/settings.py b/backend/app/api/api_v1/endpoints/settings.py
index 785b3f5..0872a8a 100644
--- a/backend/app/api/api_v1/endpoints/settings.py
+++ b/backend/app/api/api_v1/endpoints/settings.py
@@ -21,6 +21,7 @@ from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import require_admin_auth
from app.models.setting import Setting
+from app.services.alert_rules import evaluate_alert_rules, send_current_alerts
from app.services.notifications import send_notification
router = APIRouter()
@@ -127,6 +128,55 @@ SETTING_DEFAULTS: List[Dict[str, Any]] = [
"value_type": "string",
"category": "notifications",
},
+ {
+ "key": "notifications.alert_new_sources_enabled",
+ "value": "true",
+ "description": "Alert when a new sending source appears in recent DMARC reports",
+ "value_type": "boolean",
+ "category": "notifications",
+ },
+ {
+ "key": "notifications.alert_compliance_drop_enabled",
+ "value": "true",
+ "description": "Alert when DMARC compliance drops by the configured percentage points",
+ "value_type": "boolean",
+ "category": "notifications",
+ },
+ {
+ "key": "notifications.alert_compliance_drop_points",
+ "value": "10",
+ "description": "Minimum compliance-rate drop, in percentage points, before alerting",
+ "value_type": "integer",
+ "category": "notifications",
+ },
+ {
+ "key": "notifications.alert_failure_threshold_enabled",
+ "value": "true",
+ "description": "Alert when DMARC failures exceed the configured daily threshold",
+ "value_type": "boolean",
+ "category": "notifications",
+ },
+ {
+ "key": "notifications.alert_failure_threshold_count",
+ "value": "100",
+ "description": "Minimum failed message count in the last day before alerting",
+ "value_type": "integer",
+ "category": "notifications",
+ },
+ {
+ "key": "notifications.alert_missing_reports_enabled",
+ "value": "true",
+ "description": "Alert when a monitored domain has not received recent DMARC reports",
+ "value_type": "boolean",
+ "category": "notifications",
+ },
+ {
+ "key": "notifications.alert_missing_reports_days",
+ "value": "2",
+ "description": "Number of days without DMARC reports before alerting",
+ "value_type": "integer",
+ "category": "notifications",
+ },
]
# Keys whose values should be redacted in GET responses (treated as secrets)
@@ -216,6 +266,19 @@ class NotificationTestResponse(BaseModel):
error: Optional[str] = None
+class AlertRulesResponse(BaseModel):
+ """Current alert-rule evaluation response."""
+
+ alerts: List[Dict[str, Any]]
+
+
+class AlertNotificationResponse(BaseModel):
+ """Alert-rule evaluation plus notification delivery status."""
+
+ alerts: List[Dict[str, Any]]
+ notification: Dict[str, Any]
+
+
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@@ -261,6 +324,33 @@ async def test_notification_settings(
return result.to_dict()
+@router.get("/notifications/alerts", response_model=AlertRulesResponse)
+async def evaluate_notification_alerts(
+ db: Session = Depends(get_db),
+ _auth: dict = Depends(require_admin_auth),
+) -> AlertRulesResponse:
+ """Evaluate enabled notification alert rules against current DMARC data."""
+ _seed_defaults(db)
+ return {"alerts": evaluate_alert_rules(db)}
+
+
+@router.post("/notifications/alerts/send", response_model=AlertNotificationResponse)
+async def send_notification_alerts(
+ db: Session = Depends(get_db),
+ _auth: dict = Depends(require_admin_auth),
+) -> AlertNotificationResponse:
+ """Evaluate current alert rules and send a notification summary when needed."""
+ _seed_defaults(db)
+ result = send_current_alerts(db)
+ notification = result["notification"]
+ if result["alerts"] and not notification.get("success"):
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=result,
+ )
+ return result
+
+
@router.get("/{key:path}", response_model=SettingResponse)
async def get_setting(
key: str,
diff --git a/backend/app/services/alert_rules.py b/backend/app/services/alert_rules.py
new file mode 100644
index 0000000..829f969
--- /dev/null
+++ b/backend/app/services/alert_rules.py
@@ -0,0 +1,256 @@
+"""Alert rule evaluation for DMARC report data."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy import case, func
+from sqlalchemy.orm import Session
+
+from app.models.domain import Domain
+from app.models.report import DMARCReport, ReportRecord
+from app.models.setting import Setting
+from app.services.notifications import NotificationResult, send_notification
+
+
+def _truthy(value: Optional[str], default: bool = True) -> bool:
+ if value in (None, ""):
+ return default
+ return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _int_setting(value: Optional[str], default: int) -> int:
+ try:
+ return int(value or default)
+ except (TypeError, ValueError):
+ return default
+
+
+def _settings(db: Session) -> Dict[str, Optional[str]]:
+ rows = db.query(Setting).filter(Setting.category == "notifications").all()
+ return {row.key: row.value for row in rows}
+
+
+def _days_ago_ts(days: int) -> int:
+ return int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp())
+
+
+def _new_source_alerts(db: Session, window_days: int = 7) -> List[Dict[str, Any]]:
+ cutoff_ts = _days_ago_ts(window_days)
+ previous_sources = {
+ (row.domain, row.source_ip)
+ for row in (
+ db.query(Domain.name.label("domain"), ReportRecord.source_ip.label("source_ip"))
+ .join(DMARCReport, DMARCReport.domain_id == Domain.id)
+ .join(ReportRecord, ReportRecord.report_id == DMARCReport.id)
+ .filter(DMARCReport.begin_date < cutoff_ts)
+ .distinct()
+ .all()
+ )
+ }
+
+ current_sources = (
+ db.query(
+ Domain.name.label("domain"),
+ ReportRecord.source_ip.label("source_ip"),
+ func.sum(ReportRecord.count).label("message_count"),
+ )
+ .join(DMARCReport, DMARCReport.domain_id == Domain.id)
+ .join(ReportRecord, ReportRecord.report_id == DMARCReport.id)
+ .filter(DMARCReport.begin_date >= cutoff_ts)
+ .group_by(Domain.name, ReportRecord.source_ip)
+ .order_by(func.sum(ReportRecord.count).desc())
+ .all()
+ )
+
+ alerts = []
+ for row in current_sources:
+ if (row.domain, row.source_ip) in previous_sources:
+ continue
+ count = int(row.message_count or 0)
+ alerts.append(
+ {
+ "rule": "new_sender_source",
+ "severity": "warning",
+ "domain": row.domain,
+ "source_ip": row.source_ip,
+ "message_count": count,
+ "title": "New sending source",
+ "detail": f"{row.source_ip} first appeared for {row.domain} with {count} messages.",
+ }
+ )
+ return alerts
+
+
+def _failure_threshold_alerts(
+ db: Session, threshold: int, window_days: int = 1
+) -> List[Dict[str, Any]]:
+ cutoff_ts = _days_ago_ts(window_days)
+ rows = (
+ db.query(
+ Domain.name.label("domain"),
+ func.sum(ReportRecord.count).label("failed_messages"),
+ )
+ .join(DMARCReport, DMARCReport.domain_id == Domain.id)
+ .join(ReportRecord, ReportRecord.report_id == DMARCReport.id)
+ .filter(DMARCReport.begin_date >= cutoff_ts)
+ .filter(func.coalesce(ReportRecord.dkim, "fail") != "pass")
+ .filter(func.coalesce(ReportRecord.spf, "fail") != "pass")
+ .group_by(Domain.name)
+ .having(func.sum(ReportRecord.count) >= threshold)
+ .all()
+ )
+ return [
+ {
+ "rule": "dmarc_failures_above_threshold",
+ "severity": "error",
+ "domain": row.domain,
+ "failed_messages": int(row.failed_messages or 0),
+ "threshold": threshold,
+ "title": "DMARC failures above threshold",
+ "detail": (
+ f"{row.domain} had {int(row.failed_messages or 0)} DMARC failures "
+ f"in the last {window_days} day(s)."
+ ),
+ }
+ for row in rows
+ ]
+
+
+def _missing_report_alerts(db: Session, missing_days: int) -> List[Dict[str, Any]]:
+ cutoff_ts = _days_ago_ts(missing_days)
+ latest_rows = (
+ db.query(
+ Domain.name.label("domain"),
+ func.max(DMARCReport.end_date).label("last_report_ts"),
+ )
+ .outerjoin(DMARCReport, DMARCReport.domain_id == Domain.id)
+ .group_by(Domain.id, Domain.name)
+ .all()
+ )
+ alerts = []
+ for row in latest_rows:
+ last_report_ts = int(row.last_report_ts or 0)
+ if last_report_ts >= cutoff_ts:
+ continue
+ alerts.append(
+ {
+ "rule": "missing_reports",
+ "severity": "warning",
+ "domain": row.domain,
+ "missing_days": missing_days,
+ "last_report_at": (
+ datetime.fromtimestamp(last_report_ts, tz=timezone.utc).isoformat()
+ if last_report_ts
+ else None
+ ),
+ "title": "Missing DMARC reports",
+ "detail": f"{row.domain} has no DMARC report in the last {missing_days} day(s).",
+ }
+ )
+ return alerts
+
+
+def _compliance_drop_alerts(
+ db: Session, drop_points: int, window_days: int = 2
+) -> List[Dict[str, Any]]:
+ cutoff_ts = _days_ago_ts(window_days)
+ rows = (
+ db.query(
+ Domain.name.label("domain"),
+ DMARCReport.begin_date.label("begin_date"),
+ func.sum(ReportRecord.count).label("total"),
+ func.sum(
+ case(
+ (
+ (ReportRecord.dkim == "pass") | (ReportRecord.spf == "pass"),
+ ReportRecord.count,
+ ),
+ else_=0,
+ )
+ ).label("passed"),
+ )
+ .join(DMARCReport, DMARCReport.domain_id == Domain.id)
+ .join(ReportRecord, ReportRecord.report_id == DMARCReport.id)
+ .filter(DMARCReport.begin_date >= cutoff_ts)
+ .group_by(Domain.name, DMARCReport.begin_date)
+ .order_by(Domain.name, DMARCReport.begin_date)
+ .all()
+ )
+
+ by_domain: Dict[str, List[Dict[str, float]]] = {}
+ for row in rows:
+ total = int(row.total or 0)
+ passed = int(row.passed or 0)
+ rate = round((passed / total) * 100, 1) if total else 0.0
+ by_domain.setdefault(row.domain, []).append({"date": row.begin_date, "rate": rate})
+
+ alerts = []
+ for domain, points in by_domain.items():
+ if len(points) < 2:
+ continue
+ previous = points[-2]
+ current = points[-1]
+ drop = round(previous["rate"] - current["rate"], 1)
+ if drop < drop_points:
+ continue
+ alerts.append(
+ {
+ "rule": "compliance_drop",
+ "severity": "error" if drop >= 25 else "warning",
+ "domain": domain,
+ "previous_rate": previous["rate"],
+ "current_rate": current["rate"],
+ "drop": drop,
+ "threshold": drop_points,
+ "title": "Compliance dropped",
+ "detail": f"{domain} compliance fell by {drop} percentage points.",
+ }
+ )
+ return alerts
+
+
+def evaluate_alert_rules(db: Session) -> List[Dict[str, Any]]:
+ """Evaluate all enabled alert rules against persisted DMARC data."""
+ settings = _settings(db)
+ alerts: List[Dict[str, Any]] = []
+
+ if _truthy(settings.get("notifications.alert_new_sources_enabled")):
+ alerts.extend(_new_source_alerts(db))
+
+ if _truthy(settings.get("notifications.alert_failure_threshold_enabled")):
+ threshold = _int_setting(settings.get("notifications.alert_failure_threshold_count"), 100)
+ alerts.extend(_failure_threshold_alerts(db, threshold))
+
+ if _truthy(settings.get("notifications.alert_missing_reports_enabled")):
+ missing_days = _int_setting(settings.get("notifications.alert_missing_reports_days"), 2)
+ alerts.extend(_missing_report_alerts(db, missing_days))
+
+ if _truthy(settings.get("notifications.alert_compliance_drop_enabled")):
+ drop_points = _int_setting(settings.get("notifications.alert_compliance_drop_points"), 10)
+ alerts.extend(_compliance_drop_alerts(db, drop_points))
+
+ return alerts
+
+
+def send_current_alerts(db: Session) -> Dict[str, Any]:
+ """Evaluate current alert rules and send one summary notification when needed."""
+ alerts = evaluate_alert_rules(db)
+ if not alerts:
+ return {
+ "alerts": [],
+ "notification": NotificationResult(
+ success=True, skipped=True, message="No alerts."
+ ).to_dict(),
+ }
+
+ lines = [f"{alert['title']}: {alert['detail']}" for alert in alerts[:10]]
+ if len(alerts) > 10:
+ lines.append(f"...and {len(alerts) - 10} more alert(s).")
+ result = send_notification(
+ db,
+ title=f"DMARQ alert summary: {len(alerts)} alert(s)",
+ body="\n".join(lines),
+ )
+ return {"alerts": alerts, "notification": result.to_dict()}
diff --git a/backend/app/templates/settings.html b/backend/app/templates/settings.html
index 2b3e1f8..eeea592 100644
--- a/backend/app/templates/settings.html
+++ b/backend/app/templates/settings.html
@@ -235,7 +235,100 @@
+
+
+
Alert Rules
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+