from __future__ import annotations import json import threading from datetime import date, datetime, timedelta, timezone import os from pathlib import Path from types import SimpleNamespace from fastapi import Depends, FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from sqlalchemy import desc, func, select from sqlalchemy.orm import Session, selectinload from app import __version__ from app.auth import require_admin_csrf, require_dashboard_auth, require_homepage_token from app.config import Settings, configure_logging, get_settings from app.db import database_ok, get_db, init_db, session_scope from app.homepage import domain_homepage_summary, domain_metrics, homepage_summary, latest_summary, resolve_date_range, traffic_distribution from app.inbox_locks import InboxRunLease, inbox_run_locks from app.jobs import import_jobs from app.models import Alert, DailyStat, InboxStatus, LLMReport, Record, Report, SkippedReportPayload, utcnow from app.scheduler import generate_open_posture_summaries, scheduler_ok, start_scheduler from app.schemas import BacklogRequest, ProcessNowRequest from app.message_processor import process_inbox from app.validation import parse_positive_int_ids settings = get_settings() configure_logging(settings) init_db() app = FastAPI( title=settings.app.name, version=__version__, docs_url=None, redoc_url=None, openapi_url=None, ) templates = Jinja2Templates(directory="app/templates") Path("app/static").mkdir(parents=True, exist_ok=True) app.mount("/static", StaticFiles(directory="app/static"), name="static") dashboard_auth = [Depends(require_dashboard_auth)] dashboard_post_auth = [Depends(require_dashboard_auth), Depends(require_admin_csrf)] def _format_display_datetime(value, fallback: str = "never") -> str: if not value: return fallback if isinstance(value, str): parsed = _parse_dt(value) if parsed: value = parsed else: return value if isinstance(value, datetime): return value.strftime("%d/%m/%Y %H:%M:%S") if isinstance(value, date): return value.strftime("%d/%m/%Y") return str(value) def _format_display_date(value, fallback: str = "never") -> str: if not value: return fallback if isinstance(value, str): parsed = _parse_dt(value) if parsed: return parsed.strftime("%d/%m/%Y") try: return date.fromisoformat(value).strftime("%d/%m/%Y") except ValueError: return value if isinstance(value, datetime): return value.strftime("%d/%m/%Y") if isinstance(value, date): return value.strftime("%d/%m/%Y") return str(value) templates.env.filters["fmt_dt"] = _format_display_datetime templates.env.filters["fmt_date"] = _format_display_date @app.on_event("startup") def _startup() -> None: start_scheduler(settings) @app.get("/health") def health(): if not database_ok(): raise HTTPException(status_code=500, detail={"status": "error", "database": "failed"}) return {"status": "ok", "database": "ok", "scheduler": "ok" if scheduler_ok() else "stopped", "version": __version__} @app.get("/", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) def index(request: Request, session: Session = Depends(get_db)): data = homepage_summary(session, period="all") domains = session.execute(select(Report.domain).distinct().order_by(Report.domain)).scalars().all() alerts_total = session.scalar(select(func.count(Alert.id)).where(Alert.status == "open")) or 0 alerts = _alert_views(session.execute(select(Alert).where(Alert.status == "open")).scalars().all(), session)[:5] traffic = traffic_distribution(session, period="all") return templates.TemplateResponse( "index.html", { "request": request, "data": data, "domains": domains, "alerts": alerts, "alerts_total": alerts_total, "traffic": traffic, "traffic_label": f'{data["scope_label"]} ยท All domains', }, ) def _parse_dt(value: str | None) -> datetime | None: if not value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: return parsed.replace(tzinfo=timezone.utc) return parsed def _alert_details(alert: Alert) -> dict: try: return json.loads(alert.details_json or "{}") except json.JSONDecodeError: return {} def _alert_report_time(alert: Alert) -> datetime: details = _alert_details(alert) date_range = details.get("date_range") if isinstance(details.get("date_range"), dict) else {} return ( _parse_dt(date_range.get("end")) or _parse_dt(date_range.get("begin")) or _parse_dt(alert.last_seen_at.isoformat() if alert.last_seen_at else None) or _parse_dt(alert.created_at.isoformat() if alert.created_at else None) or datetime.now(timezone.utc) ) def _severity_class(severity: str) -> str: return { "critical": "critical", "warning": "warning", "info": "info", }.get(severity, "info") def _infer_alert_report_details(session: Session | None, alert: Alert, details: dict) -> dict: date_range = details.get("date_range") if isinstance(details.get("date_range"), dict) else {} if date_range.get("begin") or date_range.get("end") or not session: return details source_ip = details.get("source_ip") if not source_ip: parts = alert.fingerprint.split(":", 2) source_ip = parts[2] if len(parts) == 3 and parts[2] != "global" else None report = None is_aggregate = alert.type in { "sudden_unknown_failure_spike", } if source_ip and not is_aggregate: report = session.scalar( select(Report) .join(Record) .where(Report.domain == alert.domain, Record.source_ip == source_ip) .order_by(desc(func.coalesce(Report.date_end, Report.date_begin, Report.created_at))) .limit(1) ) if not report: report = session.scalar( select(Report) .where(Report.domain == alert.domain) .order_by(desc(func.coalesce(Report.date_end, Report.date_begin, Report.created_at))) .limit(1) ) if not report: return details enriched = dict(details) enriched["source_ip"] = source_ip if not is_aggregate: enriched["report_db_id"] = report.id enriched["date_range"] = { "begin": report.date_begin.isoformat() if report.date_begin else None, "end": report.date_end.isoformat() if report.date_end else None, } return enriched def _alert_view(alert: Alert, session: Session | None = None) -> SimpleNamespace: details = _alert_details(alert) details = _infer_alert_report_details(session, alert, details) date_range = details.get("date_range") if isinstance(details.get("date_range"), dict) else {} report_db_id = details.get("report_db_id") report_db_ids = details.get("report_db_ids") if isinstance(details.get("report_db_ids"), list) else [] if not report_db_id and isinstance(details.get("report_db_ids"), list) and details["report_db_ids"]: report_db_id = details["report_db_ids"][-1] if alert.type in {"sudden_unknown_failure_spike"}: report_db_id = None report_time = ( _parse_dt(date_range.get("end")) or _parse_dt(date_range.get("begin")) or _alert_report_time(alert) ) return SimpleNamespace( id=alert.id, fingerprint=alert.fingerprint, inbox_id=alert.inbox_id, domain=alert.domain, severity=alert.severity, severity_class=_severity_class(alert.severity), type=alert.type, title=alert.title, summary=alert.summary, details_json=alert.details_json, llm_summary=alert.llm_summary, llm_risk=alert.llm_risk, llm_recommended_action=alert.llm_recommended_action, status=alert.status, first_seen_at=alert.first_seen_at, last_seen_at=alert.last_seen_at, created_at=alert.created_at, updated_at=alert.updated_at, report_start=_parse_dt(date_range.get("begin")), report_end=_parse_dt(date_range.get("end")), report_time=report_time, report_db_id=report_db_id, report_db_ids=report_db_ids, source_ip=details.get("source_ip"), source_history=_source_history(session, alert.domain, details.get("source_ip"), alert.type, report_db_id) if session else None, ) def _alert_views(alerts: list[Alert], session: Session | None = None) -> list[SimpleNamespace]: return sorted((_alert_view(alert, session) for alert in alerts), key=lambda item: item.report_time, reverse=True) def _prompt_settings(settings: Settings) -> list[SimpleNamespace]: items = [ ("System", settings.llm.system_prompt_path), ("Alert Explanation", settings.llm.alert_prompt_path), ("Posture Digest", settings.llm.digest_prompt_path), ("Weekly Summary", settings.llm.weekly_prompt_path), ] prompts = [] for label, path in items: prompt_path = Path(path) try: content = prompt_path.read_text(encoding="utf-8") if prompt_path.exists() else "" except OSError: content = "" prompts.append(SimpleNamespace(label=label, path=path, exists=prompt_path.exists(), content=content)) return prompts def _domain_trend(session: Session, domain: str) -> list[SimpleNamespace]: rows = session.execute(select(Record, Report).join(Report).where(Report.domain == domain)).all() by_day: dict[date, dict[str, int]] = {} for record, report in rows: stamp = report.date_end or report.date_begin or report.created_at day = stamp.date() bucket = by_day.setdefault(day, {"total": 0, "pass": 0, "fail": 0}) bucket["total"] += record.count if record.dmarc_pass: bucket["pass"] += record.count else: bucket["fail"] += record.count return [ SimpleNamespace(date=day, total_messages=data["total"], dmarc_pass_count=data["pass"], dmarc_fail_count=data["fail"]) for day, data in sorted(by_day.items(), reverse=True) ] def _domain_sources(session: Session, domain: str) -> list[SimpleNamespace]: rows = session.execute( select(Record) .options(selectinload(Record.auth_results)) .join(Report) .where(Report.domain == domain) ).scalars().all() sources: dict[str, dict[str, object]] = {} for record in rows: source = sources.setdefault( record.source_ip, {"source_ip": record.source_ip, "count": 0, "pass": 0, "fail": 0, "known": None, "dkim": set()}, ) source["count"] = int(source["count"]) + record.count source["pass"] = int(source["pass"]) + (record.count if record.dmarc_pass else 0) source["fail"] = int(source["fail"]) + (0 if record.dmarc_pass else record.count) if record.known_sender_name: source["known"] = record.known_sender_name for auth in record.auth_results: if auth.auth_type == "dkim" and auth.domain: source["dkim"].add(auth.domain) return [ SimpleNamespace( source_ip=str(item["source_ip"]), count=int(item["count"]), pass_count=int(item["pass"]), fail_count=int(item["fail"]), known_sender_name=item["known"], dkim_domains=", ".join(sorted(item["dkim"])) or "none reported", dmarc_pass=int(item["pass"]) >= int(item["fail"]), ) for item in sorted(sources.values(), key=lambda entry: int(entry["count"]), reverse=True) ] def _source_history(session: Session, domain: str, source_ip: str | None, alert_type: str, report_db_id: int | None) -> str | None: if not source_ip: return None rows = session.execute( select( Report.id, Record.count, Record.dmarc_pass, func.coalesce(Report.date_end, Report.date_begin, Report.created_at), ) .select_from(Record) .join(Report) .where(Report.domain == domain, Record.source_ip == source_ip) ).all() if not rows: return None by_report: dict[int, dict[str, object]] = {} for report_id, count, dmarc_pass, stamp in rows: parsed = _parse_dt(stamp.isoformat() if hasattr(stamp, "isoformat") else stamp) item = by_report.setdefault(report_id, {"time": parsed, "messages": 0, "failed": 0}) item["messages"] = int(item["messages"]) + int(count or 0) item["failed"] = int(item["failed"]) + (0 if dmarc_pass else int(count or 0)) reports = sorted(by_report.items(), key=lambda item: item[1]["time"] or datetime.min.replace(tzinfo=timezone.utc)) report_count = len(reports) total_messages = sum(int(item["messages"]) for _, item in reports) failed_messages = sum(int(item["failed"]) for _, item in reports) first_day = reports[0][1]["time"] last_day = reports[-1][1]["time"] linked_time = by_report.get(report_db_id, {}).get("time") if report_db_id else first_day linked_messages = int(by_report.get(report_db_id, {}).get("messages", 0)) if report_db_id else int(reports[0][1]["messages"]) later_reports = [ item for item in reports if linked_time and item[1]["time"] and item[1]["time"] > linked_time ] if alert_type in {"new_unknown_source", "dkim_authenticated_relay", "new_authenticated_source", "new_spf_authenticated_source"}: noun = "relay path" if alert_type == "dkim_authenticated_relay" else "source" if later_reports: later_failed = sum(int(item["failed"]) for _, item in later_reports) return f"First observed {noun}: {linked_messages} messages in the source report. Later seen in {len(later_reports)} more reports; {later_failed} failed messages afterward." return f"First observed {noun}: {linked_messages or total_messages} messages in 1 report." if report_count <= 1: return f"First seen source: {total_messages} messages in 1 report." date_text = f" since {_format_display_date(first_day)}" if first_day else "" last_text = f"; latest {_format_display_date(last_day)}" if last_day else "" return f"Repeat offender: seen in {report_count} reports{date_text}{last_text}; {failed_messages} failed messages." def _record_auth_tooltip(record: Record, auth_type: str) -> str: items = [] for auth in record.auth_results: if auth.auth_type != auth_type: continue parts = [] if auth.domain: parts.append(f"domain={auth.domain}") if auth.selector: parts.append(f"selector={auth.selector}") if auth.scope: parts.append(f"scope={auth.scope}") if auth.result: parts.append(f"result={auth.result}") items.append(", ".join(parts) if parts else f"{auth_type.upper()} result without reported domain") return "; ".join(items) if items else f"No {auth_type.upper()} auth result domain reported." @app.get("/domains/{domain}", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) def domain_page(domain: str, request: Request, source_page: int = 1, alert_page: int = 1, report_page: int = 1, trend_page: int = 1, session: Session = Depends(get_db)): metrics = domain_metrics(session, domain) all_stats = _domain_trend(session, domain) trend_page_size = 14 trend_page = max(1, trend_page) trend_total = len(all_stats) stats = all_stats[(trend_page - 1) * trend_page_size : trend_page * trend_page_size] all_sources = _domain_sources(session, domain) source_page_size = 25 source_page = max(1, source_page) source_total = len(all_sources) records = all_sources[(source_page - 1) * source_page_size : source_page * source_page_size] alert_page_size = 10 alert_page = max(1, alert_page) alert_total = session.scalar(select(func.count(Alert.id)).where(Alert.domain == domain, Alert.status == "open")) or 0 all_alerts = _alert_views(session.execute(select(Alert).where(Alert.domain == domain, Alert.status == "open")).scalars().all(), session) alerts = all_alerts[(alert_page - 1) * alert_page_size : alert_page * alert_page_size] report_page_size = 20 report_page = max(1, report_page) report_total = session.scalar(select(func.count(Report.id)).where(Report.domain == domain)) or 0 reports = session.execute( select(Report) .where(Report.domain == domain) .order_by(desc(func.coalesce(Report.date_end, Report.date_begin, Report.created_at))) .offset((report_page - 1) * report_page_size) .limit(report_page_size) ).scalars().all() reporters = session.execute( select(Report.org_name, func.count(Report.id)).where(Report.domain == domain).group_by(Report.org_name).order_by(desc(func.count(Report.id))).limit(10) ).all() dispositions = session.execute( select(Record.disposition, func.sum(Record.count)).join(Report).where(Report.domain == domain).group_by(Record.disposition) ).all() known_unknown = session.execute( select(Record.is_known_sender, func.sum(Record.count)).join(Report).where(Report.domain == domain).group_by(Record.is_known_sender) ).all() return templates.TemplateResponse( "domain.html", { "request": request, "domain": domain, "metrics": metrics, "stats": stats, "trend_page": trend_page, "trend_page_size": trend_page_size, "trend_total": trend_total, "records": records, "source_page": source_page, "source_page_size": source_page_size, "source_total": source_total, "alerts": alerts, "alert_page": alert_page, "alert_page_size": alert_page_size, "alert_total": alert_total, "reports": reports, "report_page": report_page, "report_page_size": report_page_size, "report_total": report_total, "reporters": reporters, "dispositions": dispositions, "known_unknown": known_unknown, "summary": latest_summary(session, domain), }, ) @app.get("/reports/{report_id}", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) def report_page(report_id: int, request: Request, session: Session = Depends(get_db)): report = session.scalar(select(Report).options(selectinload(Report.records).selectinload(Record.auth_results)).where(Report.id == report_id)) if not report: raise HTTPException(status_code=404) for record in report.records: record.dkim_auth_tooltip = _record_auth_tooltip(record, "dkim") record.spf_auth_tooltip = _record_auth_tooltip(record, "spf") domain_alerts = session.execute(select(Alert).where(Alert.domain == report.domain)).scalars().all() related_alerts = [] for view in _alert_views(domain_alerts, session): same_report = view.report_db_id == report.id or report.id in view.report_db_ids if same_report: related_alerts.append(view) return templates.TemplateResponse("report.html", {"request": request, "report": report, "alerts": related_alerts}) @app.get("/alerts", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) def alerts_page( request: Request, page: int = 1, domain: str | None = None, alert_type: str | None = None, severity: str | None = None, status: str | None = "open", date_from: str | None = None, date_to: str | None = None, session: Session = Depends(get_db), ): page = max(1, page) page_size = 25 stmt = select(Alert) count_stmt = select(func.count(Alert.id)) filters = [] if domain: filters.append(Alert.domain == domain) if alert_type: filters.append(Alert.type == alert_type) if severity: filters.append(Alert.severity == severity) if status: filters.append(Alert.status == status) for item in filters: stmt = stmt.where(item) count_stmt = count_stmt.where(item) filtered_alerts = _alert_views(session.execute(stmt).scalars().all(), session) start = _parse_dt(f"{date_from}T00:00:00+00:00") if date_from else None end = _parse_dt(f"{date_to}T23:59:59+00:00") if date_to else None if start: filtered_alerts = [alert for alert in filtered_alerts if alert.report_time >= start] if end: filtered_alerts = [alert for alert in filtered_alerts if alert.report_time <= end] total = len(filtered_alerts) alerts = filtered_alerts[(page - 1) * page_size : page * page_size] domains = session.execute(select(Alert.domain).distinct().order_by(Alert.domain)).scalars().all() type_stmt = select(Alert.type).distinct().order_by(Alert.type) if domain: type_stmt = type_stmt.where(Alert.domain == domain) if status: type_stmt = type_stmt.where(Alert.status == status) alert_types = session.execute(type_stmt).scalars().all() severity_stmt = select(Alert.severity).distinct().order_by(Alert.severity) if domain: severity_stmt = severity_stmt.where(Alert.domain == domain) if status: severity_stmt = severity_stmt.where(Alert.status == status) severities = session.execute(severity_stmt).scalars().all() return templates.TemplateResponse( "alerts.html", { "request": request, "alerts": alerts, "domains": domains, "alert_types": alert_types, "severities": severities, "page": page, "page_size": page_size, "total": total, "selected_domain": domain or "", "selected_type": alert_type or "", "selected_severity": severity or "", "selected_status": status or "", "selected_date_from": date_from or "", "selected_date_to": date_to or "", }, ) @app.get("/inboxes", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) def inboxes_page(request: Request, session: Session = Depends(get_db)): statuses = { status.inbox_id: status for status in session.execute(select(InboxStatus).order_by(InboxStatus.inbox_id)).scalars().all() } inboxes = [] for configured in settings.inboxes: status = statuses.pop(configured.id, None) inboxes.append( SimpleNamespace( inbox_id=configured.id, label=configured.label, domain=configured.domain, folder=configured.folder, recipient=configured.recipient, enabled=configured.enabled, last_check_at=status.last_check_at if status else None, last_success_at=status.last_success_at if status else None, last_error=status.last_error if status else None, last_new_messages=status.last_new_messages if status else 0, last_reports_imported=status.last_reports_imported if status else 0, ) ) inboxes.extend(statuses.values()) jobs = {} for job in import_jobs.list(): jobs.setdefault(job.inbox_id, job.to_dict()) inbox_ids = [inbox.inbox_id for inbox in inboxes] skipped_payloads = {inbox_id: [] for inbox_id in inbox_ids} if inbox_ids: skipped_rows = session.execute( select(SkippedReportPayload) .where(SkippedReportPayload.inbox_id.in_(inbox_ids)) .order_by(desc(SkippedReportPayload.created_at), desc(SkippedReportPayload.id)) .limit(500) ).scalars().all() for row in skipped_rows: skipped_payloads.setdefault(row.inbox_id, []).append(row) return templates.TemplateResponse( "inboxes.html", {"request": request, "inboxes": inboxes, "jobs": jobs, "skipped_payloads": skipped_payloads}, ) def _inbox_status_payload(inbox_id: str, session: Session) -> dict: try: configured = settings.get_inbox(inbox_id) except KeyError: raise HTTPException(status_code=404, detail=f"Unknown inbox: {inbox_id}") from None status = session.scalar(select(InboxStatus).where(InboxStatus.inbox_id == inbox_id)) return { "inbox_id": configured.id, "label": configured.label, "domain": configured.domain, "folder": configured.folder, "recipient": configured.recipient, "enabled": configured.enabled, "last_check_at": status.last_check_at.isoformat() if status and status.last_check_at else None, "last_success_at": status.last_success_at.isoformat() if status and status.last_success_at else None, "last_error_at": status.last_error_at.isoformat() if status and status.last_error_at else None, "last_error": status.last_error if status else None, "last_new_messages": status.last_new_messages if status else 0, "last_reports_imported": status.last_reports_imported if status else 0, } @app.get("/settings", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) def settings_page(request: Request): env_status = { settings.security.dashboard_username_env: bool(os.getenv(settings.security.dashboard_username_env)), settings.security.dashboard_password_env: bool(os.getenv(settings.security.dashboard_password_env)), settings.security.homepage_token_env: bool(os.getenv(settings.security.homepage_token_env)), settings.llm.api_key_env: bool(os.getenv(settings.llm.api_key_env)), } for inbox in settings.inboxes: env_status[inbox.username_env] = bool(os.getenv(inbox.username_env)) env_status[inbox.password_env] = bool(os.getenv(inbox.password_env)) if settings.alerts.email.enabled: email = settings.alerts.email for name in [ email.smtp_host_env, email.smtp_port_env, email.smtp_user_env, email.smtp_password_env, email.from_env, email.to_env, ]: env_status[name] = bool(os.getenv(name)) return templates.TemplateResponse( "settings.html", { "request": request, "settings": settings, "env_status": env_status, "config_path": os.getenv("DMARC_SENTINEL_CONFIG") or "config/config.yml", "prompts": _prompt_settings(settings), }, ) @app.get("/api/homepage", dependencies=[Depends(require_homepage_token)]) def api_homepage(session: Session = Depends(get_db)): return homepage_summary(session) @app.get("/api/homepage/{domain}", dependencies=[Depends(require_homepage_token)]) def api_homepage_domain(domain: str, session: Session = Depends(get_db)): return domain_homepage_summary(session, domain) @app.get("/api/domains", dependencies=[Depends(require_dashboard_auth)]) def api_domains(session: Session = Depends(get_db)): return {"domains": session.execute(select(Report.domain).distinct()).scalars().all()} def _overview_payload(session: Session, period: str = "all", domain: str | None = None, date_from: str | None = None, date_to: str | None = None) -> dict: data = homepage_summary(session, period=period, domain=domain or None, date_from=date_from, date_to=date_to) traffic = traffic_distribution(session, period=period, domain=domain or None, date_from=date_from, date_to=date_to) return { "period": period, "period_label": data["scope_label"], "domain": domain, "metrics": data, "buckets": traffic, } @app.get("/api/overview", dependencies=[Depends(require_dashboard_auth)]) def api_overview(period: str = "all", domain: str | None = None, date_from: str | None = None, date_to: str | None = None, session: Session = Depends(get_db)): return _overview_payload(session, period=period, domain=domain, date_from=date_from, date_to=date_to) @app.get("/api/traffic", dependencies=[Depends(require_dashboard_auth)]) def api_traffic(period: str = "all", domain: str | None = None, date_from: str | None = None, date_to: str | None = None, session: Session = Depends(get_db)): payload = _overview_payload(session, period=period, domain=domain, date_from=date_from, date_to=date_to) return {key: payload[key] for key in ["period", "period_label", "domain", "buckets"]} def _latest_report_day(session: Session) -> date | None: latest = session.scalar(select(func.max(func.coalesce(Report.date_end, Report.date_begin, Report.created_at)))) if isinstance(latest, str): latest = _parse_dt(latest) return latest.date() if latest else None @app.post("/api/admin/scheduler/daily-summary", dependencies=dashboard_post_auth) def api_generate_daily_summary(session: Session = Depends(get_db)): if not settings.llm.generate_daily_summary: raise HTTPException(status_code=400, detail="Daily LLM summaries are disabled.") target_day = _latest_report_day(session) if not target_day: raise HTTPException(status_code=400, detail="No reports have been imported yet.") generate_open_posture_summaries(settings, force=True) summary = latest_summary(session) return {"ok": True, "target_day": target_day.isoformat(), "summary": summary} @app.get("/api/domains/{domain}/summary", dependencies=[Depends(require_dashboard_auth)]) def api_domain_summary(domain: str, session: Session = Depends(get_db)): return domain_homepage_summary(session, domain) @app.get("/api/domains/{domain}/reports", dependencies=[Depends(require_dashboard_auth)]) def api_domain_reports(domain: str, session: Session = Depends(get_db)): reports = session.execute( select(Report) .where(Report.domain == domain) .order_by(desc(func.coalesce(Report.date_end, Report.date_begin, Report.created_at))) .limit(100) ).scalars().all() return {"reports": [{"id": r.id, "org_name": r.org_name, "report_id": r.report_id, "date_begin": r.date_begin, "date_end": r.date_end} for r in reports]} @app.get("/api/domains/{domain}/sources", dependencies=[Depends(require_dashboard_auth)]) def api_domain_sources(domain: str, session: Session = Depends(get_db)): rows = session.execute(select(Record.source_ip, func.sum(Record.count), func.max(Record.is_known_sender)).join(Report).where(Report.domain == domain).group_by(Record.source_ip)).all() return {"sources": [{"source_ip": ip, "count": count, "known": bool(known)} for ip, count, known in rows]} @app.get("/api/reports/{report_id}", dependencies=[Depends(require_dashboard_auth)]) def api_report(report_id: int, session: Session = Depends(get_db)): report = session.scalar(select(Report).options(selectinload(Report.records)).where(Report.id == report_id)) if not report: raise HTTPException(status_code=404) return { "id": report.id, "domain": report.domain, "org_name": report.org_name, "report_id": report.report_id, "records": [ { "source_ip": row.source_ip, "count": row.count, "spf_aligned": row.spf_aligned, "dkim_aligned": row.dkim_aligned, "dmarc_pass": row.dmarc_pass, "known_sender": row.known_sender_name, "disposition": row.disposition, } for row in report.records ], } @app.get("/api/alerts", dependencies=[Depends(require_dashboard_auth)]) def api_alerts(session: Session = Depends(get_db)): alerts = _alert_views(session.execute(select(Alert)).scalars().all(), session) return {"alerts": [{"id": a.id, "severity": a.severity, "type": a.type, "title": a.title, "status": a.status, "llm_summary": a.llm_summary} for a in alerts]} def _set_alert_status(alert_id: int, status: str, session: Session) -> dict: alert = session.get(Alert, alert_id) if not alert: raise HTTPException(status_code=404) alert.status = status session.commit() return {"ok": True, "status": status} def _copy_summary_to_job(job_id: str, summary) -> None: def mutate(job): job.status = "running" job.scanned_messages = summary.scanned_messages job.processed_messages = summary.processed_messages job.candidate_messages = summary.candidate_messages job.valid_reports_imported = summary.valid_reports_imported job.duplicate_messages_skipped = summary.duplicate_messages_skipped job.duplicate_reports_skipped = summary.duplicate_reports_skipped job.failed_messages = summary.failed_messages job.rejected_messages = summary.rejected_messages job.records_imported = summary.records_imported job.alerts_created = summary.alerts_created job.duplicate_report_samples = summary.duplicate_report_samples import_jobs.update(job_id, mutate) def _run_import_job(job_id: str, action: str, body: ProcessNowRequest | BacklogRequest) -> None: lease: InboxRunLease | None = None try: inbox = settings.get_inbox(body.inbox_id) lease = inbox_run_locks.acquire(inbox.id, blocking=True) def mark_running(job): job.status = "running" import_jobs.update(job_id, mark_running) try: with session_scope() as session: if action == "backlog": assert isinstance(body, BacklogRequest) summary = process_inbox( session, settings, inbox, folder=body.folder or inbox.folder, mode="backlog", since=body.since, before=body.before, limit=body.limit, dry_run=body.dry_run, reprocess=body.reprocess, mark_seen=body.mark_seen, progress_callback=lambda item: _copy_summary_to_job(job_id, item), ) else: assert isinstance(body, ProcessNowRequest) summary = process_inbox( session, settings, inbox, mode=body.mode, limit=body.limit, progress_callback=lambda item: _copy_summary_to_job(job_id, item), ) _copy_summary_to_job(job_id, summary) finally: lease.release() lease = None def mark_done(job): job.status = "succeeded" job.completed_at = utcnow() import_jobs.update(job_id, mark_done) except Exception as exc: error = str(exc) def mark_failed(job): job.status = "failed" job.error = error job.completed_at = utcnow() import_jobs.update(job_id, mark_failed) finally: if lease: lease.release() def _start_import_job(action: str, body: ProcessNowRequest | BacklogRequest) -> dict: try: settings.get_inbox(body.inbox_id) except KeyError: raise HTTPException(status_code=404, detail=f"Unknown inbox: {body.inbox_id}") from None active = import_jobs.active_for_inbox(body.inbox_id) if active: return active.to_dict() try: job = import_jobs.create(body.inbox_id, action) thread = threading.Thread(target=_run_import_job, args=(job.id, action, body), daemon=True) thread.start() return job.to_dict() except Exception: raise @app.post("/api/alerts/{alert_id}/ack", dependencies=dashboard_post_auth) def api_alert_ack(alert_id: int, session: Session = Depends(get_db)): return _set_alert_status(alert_id, "acknowledged", session) @app.post("/api/alerts/{alert_id}/resolve", dependencies=dashboard_post_auth) def api_alert_resolve(alert_id: int, session: Session = Depends(get_db)): return _set_alert_status(alert_id, "resolved", session) @app.post("/api/alerts/{alert_id}/reopen", dependencies=dashboard_post_auth) def api_alert_reopen(alert_id: int, session: Session = Depends(get_db)): return _set_alert_status(alert_id, "open", session) @app.post("/api/alerts/bulk", dependencies=dashboard_post_auth) async def api_alert_bulk(request: Request, session: Session = Depends(get_db)): try: payload = await request.json() except json.JSONDecodeError: raise HTTPException(status_code=400, detail="Request body must be valid JSON") from None if not isinstance(payload, dict): raise HTTPException(status_code=400, detail="Request body must be a JSON object") ids = parse_positive_int_ids(payload.get("ids", [])) status = payload.get("status") if status not in {"open", "acknowledged", "resolved"}: raise HTTPException(status_code=400, detail="Invalid alert status") updated = 0 if ids: alerts = session.execute(select(Alert).where(Alert.id.in_(ids))).scalars().all() for alert in alerts: alert.status = status updated += 1 session.commit() return {"ok": True, "status": status, "updated": updated} @app.post("/api/admin/import-jobs/process-now", dependencies=dashboard_post_auth) def api_start_process_now(body: ProcessNowRequest): return _start_import_job("process-now", body) @app.post("/api/admin/import-jobs/backlog", dependencies=dashboard_post_auth) def api_start_backlog(body: BacklogRequest): return _start_import_job("backlog", body) @app.get("/api/admin/import-jobs", dependencies=[Depends(require_dashboard_auth)]) def api_import_jobs(inbox_id: str | None = None): return {"jobs": [job.to_dict() for job in import_jobs.list(inbox_id)]} @app.get("/api/admin/import-jobs/{job_id}", dependencies=[Depends(require_dashboard_auth)]) def api_import_job(job_id: str): job = import_jobs.get(job_id) if not job: raise HTTPException(status_code=404) return job.to_dict() @app.get("/api/admin/inboxes/{inbox_id}/status", dependencies=[Depends(require_dashboard_auth)]) def api_inbox_status(inbox_id: str, session: Session = Depends(get_db)): return _inbox_status_payload(inbox_id, session)