diff --git a/.gitignore b/.gitignore index 700d3a3..8f54720 100644 --- a/.gitignore +++ b/.gitignore @@ -130,6 +130,7 @@ celerybeat.pid # Environments .env .venv +.pipcache/ env/ venv/ ENV/ @@ -216,4 +217,4 @@ temp/ *.backup # Docker override -docker-compose.override.yml \ No newline at end of file +docker-compose.override.yml diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py index 0607251..a6b64c1 100644 --- a/backend/app/api/api_v1/endpoints/domains.py +++ b/backend/app/api/api_v1/endpoints/domains.py @@ -9,9 +9,11 @@ from typing import Any, Dict, List, Optional from fastapi import APIRouter, Depends, HTTPException, Path, Query, status from fastapi.responses import Response from pydantic import BaseModel, Field +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core.database import get_db +from app.core.security import require_admin_auth from app.models.domain import Domain from app.services.cloudflare_dns import ( analyze_dns_records, @@ -32,6 +34,7 @@ from app.services.report_persistence import ( hydrate_report_store_from_db, ) from app.services.report_store import ReportStore +from app.utils.domain_validator import validate_domain_config logger = logging.getLogger(__name__) @@ -55,6 +58,14 @@ class DomainResponse(DomainBase): compliance_rate: float = 0.0 +class DomainCreate(BaseModel): + """Payload for creating a monitored domain.""" + + name: str + description: Optional[str] = None + dkim_selectors: Optional[List[str]] = None + + class DomainStatsResponse(BaseModel): """Domain statistics for the domain details page""" @@ -283,6 +294,22 @@ def _policy_enforcement_suggestions( return [] +def _normalize_domain_name(name: str) -> str: + return name.strip().strip(".").lower() + + +def _domain_names_for_summary(db: Session, store: ReportStore) -> List[str]: + report_domains = store.get_domains() + stored_domains = [ + name + for (name,) in db.query(Domain.name) + .filter(Domain.active == True) # noqa: E712 + .order_by(Domain.name) + .all() + ] + return list(dict.fromkeys(stored_domains + report_domains)) + + @router.get("/summary", response_model=DomainSummaryResponse) async def get_domains_summary(db: Session = Depends(get_db)): """ @@ -295,7 +322,7 @@ async def get_domains_summary(db: Session = Depends(get_db)): """ store = ReportStore.get_instance() hydrate_report_store_from_db(db, store) - domains = store.get_domains() + domains = _domain_names_for_summary(db, store) summaries = store.get_all_domain_summaries() # Perform DNS checks for all domains, reusing fresh cached results. @@ -385,19 +412,28 @@ async def get_domains_summary(db: Session = Depends(get_db)): async def read_domains(db: Session = Depends(get_db)): """ Retrieve domains with their statistics. - For Milestone 1, this simply returns domains from the in-memory store. """ store = ReportStore.get_instance() hydrate_report_store_from_db(db, store) - domains = store.get_domains() + domains = _domain_names_for_summary(db, store) summaries = store.get_all_domain_summaries() + stored = { + domain.name: domain + for domain in db.query(Domain).filter(Domain.name.in_(domains)).all() + } result = [] for domain_name in domains: summary = summaries.get(domain_name, {}) + stored_domain = stored.get(domain_name) domain_response = DomainResponse( name=domain_name, - policy=summary.get("policy", "unknown"), + description=stored_domain.description if stored_domain else None, + policy=( + summary.get("policy") + or (stored_domain.dmarc_policy if stored_domain else None) + or "unknown" + ), reports_count=summary.get("reports_processed", 0), emails_count=summary.get("total_count", 0), compliance_rate=summary.get("compliance_rate", 0.0), @@ -407,6 +443,58 @@ async def read_domains(db: Session = Depends(get_db)): return result +@router.post("/domains", response_model=DomainResponse, status_code=status.HTTP_201_CREATED) +async def create_domain( + payload: DomainCreate, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +): + """Create a monitored domain before any DMARC reports have arrived.""" + name = _normalize_domain_name(payload.name) + validation = validate_domain_config( + {"name": name, "description": payload.description or ""} + ) + if not validation["valid"]: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=validation["errors"], + ) + existing = db.query(Domain).filter(Domain.name == name).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Domain is already monitored", + ) + + selectors = ",".join( + selector.strip() + for selector in payload.dkim_selectors or [] + if selector and selector.strip() + ) + domain = Domain( + name=name, + description=payload.description, + dkim_selectors=selectors or None, + active=True, + verified=False, + ) + db.add(domain) + try: + db.commit() + except IntegrityError as exc: + db.rollback() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Domain is already monitored", + ) from exc + db.refresh(domain) + return DomainResponse( + name=domain.name, + description=domain.description, + policy=domain.dmarc_policy or "unknown", + ) + + @router.get("/domains/{domain_name}", response_model=DomainResponse) async def read_domain(domain_name: str, db: Session = Depends(get_db)): """ @@ -415,8 +503,9 @@ async def read_domain(domain_name: str, db: Session = Depends(get_db)): store = ReportStore.get_instance() hydrate_report_store_from_db(db, store) domains = store.get_domains() + stored_domain = db.query(Domain).filter(Domain.name == domain_name).first() - if domain_name not in domains: + if domain_name not in domains and stored_domain is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Domain not found", @@ -426,7 +515,12 @@ async def read_domain(domain_name: str, db: Session = Depends(get_db)): return DomainResponse( name=domain_name, - policy=summary.get("policy", "unknown"), + description=stored_domain.description if stored_domain else None, + policy=( + summary.get("policy") + or (stored_domain.dmarc_policy if stored_domain else None) + or "unknown" + ), reports_count=summary.get("reports_processed", 0), emails_count=summary.get("total_count", 0), compliance_rate=summary.get("compliance_rate", 0.0), diff --git a/backend/app/api/api_v1/endpoints/health.py b/backend/app/api/api_v1/endpoints/health.py index a721120..e9c73a3 100644 --- a/backend/app/api/api_v1/endpoints/health.py +++ b/backend/app/api/api_v1/endpoints/health.py @@ -1,6 +1,14 @@ -from fastapi import APIRouter +from fastapi import APIRouter, Depends +from sqlalchemy import func, text +from sqlalchemy.orm import Session from app.api.api_v1.endpoints.setup import setup_status +from app.core.database import get_db +from app.core.security import require_admin_auth +from app.models.mail_source import MailSource +from app.models.mail_source_import import MailSourceImport +from app.models.report import DMARCReport +from app.services.runtime_status import get_scheduler_status router = APIRouter() @@ -17,3 +25,94 @@ async def health_check(): "service": "dmarq", "is_setup_complete": setup_status["is_setup_complete"], } + + +def _iso(value): + return value.isoformat() if value else None + + +@router.get("/health/operations", status_code=200) +async def operations_health( + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +): + """Return operational health details for the web health page.""" + database = {"ok": True, "detail": "Connected"} + try: + db.execute(text("SELECT 1")) + except Exception as exc: # pylint: disable=broad-exception-caught + database = {"ok": False, "detail": str(exc)} + + enabled_sources = 0 + total_sources = 0 + report_count = 0 + latest_report = None + latest_import = None + latest_successful_import = None + if database["ok"]: + enabled_sources = ( + db.query(func.count(MailSource.id)) + .filter(MailSource.enabled == True) # noqa: E712 + .scalar() + ) + total_sources = db.query(func.count(MailSource.id)).scalar() + report_count = db.query(func.count(DMARCReport.id)).scalar() + latest_report = db.query(func.max(DMARCReport.processed_at)).scalar() + latest_import = ( + db.query(MailSourceImport) + .order_by(MailSourceImport.finished_at.desc(), MailSourceImport.id.desc()) + .first() + ) + latest_successful_import = ( + db.query(MailSourceImport) + .filter(MailSourceImport.status.in_(["success", "warning"])) + .order_by(MailSourceImport.finished_at.desc(), MailSourceImport.id.desc()) + .first() + ) + + scheduler = get_scheduler_status() + status = "ok" + checks = [] + if not database["ok"]: + status = "degraded" + checks.append("Database connectivity failed.") + if total_sources and enabled_sources == 0: + status = "degraded" + checks.append("All mail sources are disabled.") + if scheduler.get("last_error"): + status = "degraded" + checks.append("The scheduler reported a recent error.") + + return { + "status": status, + "service": "dmarq", + "database": database, + "scheduler": { + **scheduler, + "enabled_sources": int(enabled_sources or 0), + "total_sources": int(total_sources or 0), + }, + "imports": { + "latest": { + "status": latest_import.status, + "trigger": latest_import.trigger, + "reports_found": latest_import.reports_found, + "finished_at": _iso(latest_import.finished_at), + } + if latest_import + else None, + "latest_successful": { + "status": latest_successful_import.status, + "trigger": latest_successful_import.trigger, + "reports_found": latest_successful_import.reports_found, + "finished_at": _iso(latest_successful_import.finished_at), + } + if latest_successful_import + else None, + }, + "reports": { + "count": int(report_count or 0), + "latest_processed_at": _iso(latest_report), + }, + "checks": checks, + } diff --git a/backend/app/api/api_v1/endpoints/setup.py b/backend/app/api/api_v1/endpoints/setup.py index 11b3bf1..502a30a 100644 --- a/backend/app/api/api_v1/endpoints/setup.py +++ b/backend/app/api/api_v1/endpoints/setup.py @@ -7,6 +7,8 @@ from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import api_key_header, require_admin_auth, security_bearer +from app.models.domain import Domain +from app.models.mail_source import MailSource from app.models.setting import Setting router = APIRouter() @@ -89,6 +91,9 @@ class SetupStatusResponse(BaseModel): is_setup_complete: bool app_name: str + total_domains: int = 0 + total_mail_sources: int = 0 + enabled_mail_sources: int = 0 class AdminSetupRequest(BaseModel): @@ -125,6 +130,11 @@ async def get_setup_status(db: Session = Depends(get_db)): return SetupStatusResponse( is_setup_complete=current_status["is_setup_complete"], app_name=current_status["app_name"], + total_domains=db.query(Domain.id).count(), + total_mail_sources=db.query(MailSource.id).count(), + enabled_mail_sources=db.query(MailSource.id) + .filter(MailSource.enabled == True) # noqa: E712 + .count(), ) diff --git a/backend/app/main.py b/backend/app/main.py index bfc004d..cec4dcd 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -2,6 +2,7 @@ import asyncio import logging import os from datetime import datetime +from typing import List, Optional from fastapi import Depends, FastAPI, Query, Request from fastapi.middleware.cors import CORSMiddleware @@ -24,12 +25,20 @@ from app.core.security import add_api_key, generate_api_key, require_admin_auth from app.core.startup_checks import run_startup_checks from app.middleware.auth import AuthRedirectMiddleware from app.middleware.security import SecurityHeadersMiddleware +from app.models.domain import Domain from app.models.mail_source import MailSource # noqa: F401 – ensure table is registered from app.services.gmail_client import GmailClient from app.services.imap_client import IMAPClient from app.services.import_history import record_import_attempt from app.services.report_persistence import hydrate_report_store_from_db from app.services.report_store import ReportStore +from app.services.runtime_status import ( + mark_scheduler_cycle_started, + mark_scheduler_error, + mark_scheduler_started, + mark_scheduler_stopped, + mark_scheduler_success, +) from app.services.summary_notifications import send_due_scheduled_summaries # Set up logging @@ -203,7 +212,7 @@ def _send_due_summary_notifications() -> None: def _next_sleep_seconds( - min_sleep: int = 60, enabled_sources: list[MailSource] | None = None + min_sleep: int = 60, enabled_sources: Optional[List[MailSource]] = None ) -> int: """Return how many seconds to sleep until the next polling cycle.""" try: @@ -226,11 +235,14 @@ async def scheduled_imap_polling(): try: while True: logger.info("Starting scheduled IMAP polling for DMARC reports") + mark_scheduler_cycle_started() try: enabled_sources = _poll_all_enabled_sources() _send_due_summary_notifications() + mark_scheduler_success() except Exception as e: # pylint: disable=broad-exception-caught logger.error("Error in IMAP polling task: %s", str(e)) + mark_scheduler_error(e) enabled_sources = None try: @@ -243,6 +255,7 @@ async def scheduled_imap_polling(): except asyncio.CancelledError: logger.info("IMAP polling task cancelled") + mark_scheduler_stopped() def _migrate_imap_env_vars_to_db() -> None: @@ -407,6 +420,7 @@ def create_app() -> FastAPI: # Start background polling task (iterates over DB-enabled mail sources) logger.info("Starting IMAP polling background task") + mark_scheduler_started() background_task = asyncio.create_task(scheduled_imap_polling()) @application.on_event("shutdown") @@ -419,6 +433,7 @@ def create_app() -> FastAPI: await background_task except asyncio.CancelledError: pass + mark_scheduler_stopped() return application @@ -484,17 +499,18 @@ async def domain_details(request: Request, domain_id: str): db = SessionLocal() try: hydrate_report_store_from_db(db, store) + stored_domain = db.query(Domain).filter(Domain.name == domain_id).first() finally: db.close() known_domains = store.get_domains() - if domain_id not in known_domains: + if domain_id not in known_domains and stored_domain is None: # Domain not found, redirect to domains list return templates.TemplateResponse( request, "domains.html", {"error": f"Domain {domain_id} not found"} ) - domain_summary = store.get_domain_summary(domain_id) + domain_summary = store.get_domain_summary(domain_id) if domain_id in known_domains else {} return templates.TemplateResponse( request, @@ -503,8 +519,12 @@ async def domain_details(request: Request, domain_id: str): "domain_id": domain_id, "domain": { "name": domain_id, - "description": "", # Add description if available - "policy": domain_summary.get("policy", "unknown"), + "description": stored_domain.description if stored_domain else "", + "policy": ( + domain_summary.get("policy") + or (stored_domain.dmarc_policy if stored_domain else None) + or "unknown" + ), }, }, ) @@ -550,6 +570,11 @@ async def mail_sources_page(request: Request): return templates.TemplateResponse(request, "mail_sources.html") +@app.get("/operations", response_class=HTMLResponse) +async def operations_page(request: Request): + return templates.TemplateResponse(request, "operations.html") + + @app.get("/upload", response_class=HTMLResponse) async def upload_page(request: Request): return templates.TemplateResponse(request, "upload.html") diff --git a/backend/app/models/dns_cache.py b/backend/app/models/dns_cache.py index fcb992b..c230d6e 100644 --- a/backend/app/models/dns_cache.py +++ b/backend/app/models/dns_cache.py @@ -1,4 +1,4 @@ -from datetime import UTC, datetime +from datetime import datetime, timezone from sqlalchemy import Boolean, Column, DateTime, Index, Integer, String, Text, UniqueConstraint @@ -6,7 +6,7 @@ from app.core.database import Base def _utcnow_naive() -> datetime: - return datetime.now(UTC).replace(tzinfo=None) + return datetime.now(timezone.utc).replace(tzinfo=None) class DNSCache(Base): diff --git a/backend/app/services/cloudflare_dns.py b/backend/app/services/cloudflare_dns.py index 75833fa..b7e0465 100644 --- a/backend/app/services/cloudflare_dns.py +++ b/backend/app/services/cloudflare_dns.py @@ -5,7 +5,7 @@ from __future__ import annotations import hashlib import json from dataclasses import dataclass -from datetime import UTC, datetime +from datetime import datetime, timezone from typing import Any, Dict, List, Optional from sqlalchemy.orm import Session @@ -33,7 +33,7 @@ class CloudflareCredentials: def _utcnow_naive() -> datetime: - return datetime.now(UTC).replace(tzinfo=None) + return datetime.now(timezone.utc).replace(tzinfo=None) def _plain_setting_value(db: Session, key: str) -> Optional[str]: diff --git a/backend/app/services/dns_cache.py b/backend/app/services/dns_cache.py index 70b0943..042f045 100644 --- a/backend/app/services/dns_cache.py +++ b/backend/app/services/dns_cache.py @@ -5,7 +5,7 @@ from __future__ import annotations import hashlib import json from dataclasses import asdict -from datetime import UTC, datetime, timedelta +from datetime import datetime, timedelta, timezone from typing import List, Tuple from sqlalchemy.orm import Session @@ -17,7 +17,7 @@ DEFAULT_DNS_CACHE_TTL_SECONDS = 900 def _utcnow_naive() -> datetime: - return datetime.now(UTC).replace(tzinfo=None) + return datetime.now(timezone.utc).replace(tzinfo=None) def _selectors_key(selectors: List[str]) -> str: diff --git a/backend/app/services/imap_client.py b/backend/app/services/imap_client.py index 99e31da..d83ce35 100644 --- a/backend/app/services/imap_client.py +++ b/backend/app/services/imap_client.py @@ -392,7 +392,7 @@ class IMAPClient: ) @staticmethod - def _append_detail(stats: dict | None, **detail: str) -> None: + def _append_detail(stats: Optional[Dict[str, Any]], **detail: str) -> None: """Append a compact attachment/message outcome to the import stats.""" if stats is None: return @@ -405,8 +405,8 @@ class IMAPClient: report: Dict[str, Any], *, filename: str, - stats: dict | None, - message_id: str | None, + stats: Optional[Dict[str, Any]], + message_id: Optional[str], ) -> bool: domain = report.get("domain", "unknown") report_id = report.get("report_id", "") @@ -445,8 +445,8 @@ class IMAPClient: part: email.message.Message, *, filename: str, - stats: dict | None, - message_id: str | None, + stats: Optional[Dict[str, Any]], + message_id: Optional[str], ) -> bool: try: content = part.get_payload(decode=True) @@ -487,8 +487,8 @@ class IMAPClient: def _process_attachments( self, msg: email.message.Message, - stats: dict | None = None, - message_id: str | None = None, + stats: Optional[Dict[str, Any]] = None, + message_id: Optional[str] = None, ) -> int: """ Process email attachments that might be DMARC reports diff --git a/backend/app/services/report_persistence.py b/backend/app/services/report_persistence.py index 2b5769a..d386c83 100644 --- a/backend/app/services/report_persistence.py +++ b/backend/app/services/report_persistence.py @@ -191,7 +191,7 @@ def persisted_report_to_dict(report: DMARCReport) -> Dict[str, Any]: } -def hydrate_report_store_from_db(db: Session, store: ReportStore | None = None) -> int: +def hydrate_report_store_from_db(db: Session, store: Optional[ReportStore] = None) -> int: """Load persisted reports into ReportStore when the database has report rows.""" report_count = db.query(DMARCReport.id).count() if report_count == 0: diff --git a/backend/app/services/runtime_status.py b/backend/app/services/runtime_status.py new file mode 100644 index 0000000..17a4956 --- /dev/null +++ b/backend/app/services/runtime_status.py @@ -0,0 +1,65 @@ +"""Small in-process status tracker for scheduler health.""" + +from datetime import datetime +from typing import Any, Dict, Optional + + +_scheduler_state: Dict[str, Any] = { + "running": False, + "started_at": None, + "stopped_at": None, + "last_cycle_started_at": None, + "last_success_at": None, + "last_error_at": None, + "last_error": None, +} + + +def _iso(value: Optional[datetime]) -> Optional[str]: + return value.isoformat() if value else None + + +def mark_scheduler_started() -> None: + now = datetime.utcnow() + _scheduler_state.update( + { + "running": True, + "started_at": now, + "stopped_at": None, + "last_error": None, + } + ) + + +def mark_scheduler_stopped() -> None: + _scheduler_state.update({"running": False, "stopped_at": datetime.utcnow()}) + + +def mark_scheduler_cycle_started() -> None: + _scheduler_state["last_cycle_started_at"] = datetime.utcnow() + + +def mark_scheduler_success() -> None: + _scheduler_state.update({"last_success_at": datetime.utcnow(), "last_error": None}) + + +def mark_scheduler_error(error: object) -> None: + text = str(error) + _scheduler_state.update( + { + "last_error_at": datetime.utcnow(), + "last_error": text[:300], + } + ) + + +def get_scheduler_status() -> Dict[str, Any]: + return { + "running": bool(_scheduler_state["running"]), + "started_at": _iso(_scheduler_state["started_at"]), + "stopped_at": _iso(_scheduler_state["stopped_at"]), + "last_cycle_started_at": _iso(_scheduler_state["last_cycle_started_at"]), + "last_success_at": _iso(_scheduler_state["last_success_at"]), + "last_error_at": _iso(_scheduler_state["last_error_at"]), + "last_error": _scheduler_state["last_error"], + } diff --git a/backend/app/templates/domains.html b/backend/app/templates/domains.html index 1152f0f..4f19726 100644 --- a/backend/app/templates/domains.html +++ b/backend/app/templates/domains.html @@ -21,9 +21,9 @@ {% call card_header() %}