From b9041012de88d31abb03c0ae1ad6138629983d0f Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Fri, 22 May 2026 19:38:03 +0200 Subject: [PATCH] feat: persist imported DMARC reports --- backend/app/api/api_v1/endpoints/domains.py | 47 +++- backend/app/api/api_v1/endpoints/imap.py | 22 +- .../app/api/api_v1/endpoints/mail_sources.py | 1 + backend/app/api/api_v1/endpoints/reports.py | 45 +++- backend/app/main.py | 54 ++-- backend/app/services/gmail_client.py | 10 +- backend/app/services/imap_client.py | 14 +- backend/app/services/report_persistence.py | 236 ++++++++++++++++++ backend/app/tests/test_gmail_client.py | 22 ++ backend/app/tests/test_imap_client.py | 16 +- backend/app/tests/test_reports_api.py | 59 ++++- docs/development/roadmap.md | 23 +- docs/milestones.md | 17 +- docs/todo.md | 8 +- 14 files changed, 492 insertions(+), 82 deletions(-) create mode 100644 backend/app/services/report_persistence.py diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py index c91d8a3..eb1488a 100644 --- a/backend/app/api/api_v1/endpoints/domains.py +++ b/backend/app/api/api_v1/endpoints/domains.py @@ -15,6 +15,10 @@ from app.services.dns_resolver import ( extract_dmarc_policy, get_default_provider, ) +from app.services.report_persistence import ( + delete_persisted_domain, + hydrate_report_store_from_db, +) from app.services.report_store import ReportStore logger = logging.getLogger(__name__) @@ -146,9 +150,7 @@ def _get_domain_selectors_from_db(db: Session, domain_name: str) -> List[str]: return [] -def _get_domain_selectors_map_from_db( - db: Session, domain_names: List[str] -) -> Dict[str, List[str]]: +def _get_domain_selectors_map_from_db(db: Session, domain_names: List[str]) -> Dict[str, List[str]]: """Return manually configured DKIM selectors for all requested domains.""" if not domain_names: return {} @@ -157,11 +159,7 @@ def _get_domain_selectors_map_from_db( selectors_by_domain: Dict[str, List[str]] = {} for index in range(0, len(unique_names), DOMAIN_SELECTOR_LOOKUP_CHUNK_SIZE): chunk = unique_names[index : index + DOMAIN_SELECTOR_LOOKUP_CHUNK_SIZE] - rows = ( - db.query(Domain.name, Domain.dkim_selectors) - .filter(Domain.name.in_(chunk)) - .all() - ) + rows = db.query(Domain.name, Domain.dkim_selectors).filter(Domain.name.in_(chunk)).all() for name, selectors in rows: selectors_by_domain[name] = [ selector.strip() for selector in (selectors or "").split(",") if selector.strip() @@ -180,6 +178,7 @@ async def get_domains_summary(db: Session = Depends(get_db)): blocking the page load. """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() summaries = store.get_all_domain_summaries() @@ -256,12 +255,13 @@ async def get_domains_summary(db: Session = Depends(get_db)): @router.get("/domains", response_model=List[DomainResponse]) -async def read_domains(): +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() summaries = store.get_all_domain_summaries() @@ -281,11 +281,12 @@ async def read_domains(): @router.get("/domains/{domain_name}", response_model=DomainResponse) -async def read_domain(domain_name: str): +async def read_domain(domain_name: str, db: Session = Depends(get_db)): """ Get statistics for a specific domain. """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() if domain_name not in domains: @@ -309,11 +310,15 @@ async def read_domain(domain_name: str): @router.get("/{domain_id}/stats", response_model=DomainStatsResponse) -async def get_domain_stats(domain_id: str = Path(..., title="The domain ID or name")): +async def get_domain_stats( + domain_id: str = Path(..., title="The domain ID or name"), + db: Session = Depends(get_db), +): """ Get detailed statistics for a specific domain """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() # For Milestone 1, domain_id is simply the domain name @@ -351,6 +356,7 @@ async def get_domain_dns_records( selectors used as a final fallback. """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() if domain_id not in domains: @@ -380,11 +386,13 @@ async def get_domain_dns_records( async def get_domain_reports( domain_id: str = Path(..., title="The domain ID or name"), limit: int = Query(10, title="Maximum number of reports to return"), + db: Session = Depends(get_db), ): """ Get recent DMARC reports for a specific domain, along with compliance timeline """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() if domain_id not in domains: @@ -493,12 +501,14 @@ async def _safe_ptr_lookup(provider: Any, ip: str, timeout: float = 3.0) -> Opti async def get_domain_sources( domain_id: str = Path(..., title="The domain ID or name"), days: int = Query(30, title="Number of days to look back"), + db: Session = Depends(get_db), ): """ Get sending sources for a specific domain, including reverse-DNS hostnames and SPF fix hints for sources that fail authentication. """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() if domain_id not in domains: @@ -546,6 +556,7 @@ async def get_domain_selectors( DMARC reports, read-only). """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) if domain_id not in store.get_domains(): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, @@ -571,6 +582,7 @@ async def add_domain_selector( any received DMARC report. """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) if domain_id not in store.get_domains(): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, @@ -627,12 +639,16 @@ async def delete_domain_selector( @router.delete("/{domain_id}", status_code=status.HTTP_204_NO_CONTENT) -async def delete_domain(domain_id: str = Path(..., title="The domain ID or name")): +async def delete_domain( + domain_id: str = Path(..., title="The domain ID or name"), + db: Session = Depends(get_db), +): """ Delete a domain and all associated data. This performs a full cleanup of all reports and records related to this domain. """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() if domain_id not in domains: @@ -642,7 +658,10 @@ async def delete_domain(domain_id: str = Path(..., title="The domain ID or name" ) # Perform deletion with cleanup - deleted = store.delete_domain_with_cleanup(domain_id) + deleted_from_db = delete_persisted_domain(db, domain_id) + if deleted_from_db: + db.commit() + deleted = store.delete_domain_with_cleanup(domain_id) or deleted_from_db if not deleted: raise HTTPException( @@ -660,6 +679,7 @@ async def search_domains( policy: Optional[str] = Query(None, title="Filter by DMARC policy"), page: int = Query(1, title="Page number", ge=1), limit: int = Query(10, title="Number of domains per page", ge=1, le=100), + db: Session = Depends(get_db), ): """ Search domains with filtering and pagination. @@ -672,6 +692,7 @@ async def search_domains( limit: Number of domains per page (max 100) """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() summaries = store.get_all_domain_summaries() diff --git a/backend/app/api/api_v1/endpoints/imap.py b/backend/app/api/api_v1/endpoints/imap.py index f5cf803..6686d0e 100644 --- a/backend/app/api/api_v1/endpoints/imap.py +++ b/backend/app/api/api_v1/endpoints/imap.py @@ -4,7 +4,9 @@ from typing import Any, Dict, Optional from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException from pydantic import BaseModel +from sqlalchemy.orm import Session +from app.core.database import SessionLocal, get_db from app.core.security import require_admin_auth from app.services.imap_client import IMAPClient @@ -22,6 +24,20 @@ class IMAPTestRequest(BaseModel): ssl: bool = True +def _fetch_imap_reports_background(days: int, delete_emails: bool) -> None: + """Fetch IMAP reports with a standalone DB session for background imports.""" + db = SessionLocal() + try: + imap_client = IMAPClient(delete_emails=delete_emails, db=db) + imap_client.fetch_reports(days=days) + db.commit() + except Exception: + db.rollback() + raise + finally: + db.close() + + @router.post("/test-connection") async def test_imap_connection( request: IMAPTestRequest, @@ -57,6 +73,7 @@ async def test_imap_connection( async def fetch_imap_reports( background_tasks: BackgroundTasks, _auth: dict = Depends(require_admin_auth), + db: Session = Depends(get_db), days: int = 7, delete_emails: bool = False, ) -> Dict[str, Any]: @@ -69,11 +86,11 @@ async def fetch_imap_reports( if days < 1 or days > 365: raise HTTPException(status_code=400, detail="Days parameter must be between 1 and 365") - imap_client = IMAPClient(delete_emails=delete_emails) + imap_client = IMAPClient(delete_emails=delete_emails, db=db) # Run in background if it might take a while if days > 14: - background_tasks.add_task(imap_client.fetch_reports, days) + background_tasks.add_task(_fetch_imap_reports_background, days, delete_emails) return { "success": True, "message": f"Background task started to fetch {days} days of reports", @@ -83,6 +100,7 @@ async def fetch_imap_reports( # Otherwise run immediately try: results = imap_client.fetch_reports(days=days) + db.commit() return { "success": results["success"], diff --git a/backend/app/api/api_v1/endpoints/mail_sources.py b/backend/app/api/api_v1/endpoints/mail_sources.py index 31a4c6e..d3b54f9 100644 --- a/backend/app/api/api_v1/endpoints/mail_sources.py +++ b/backend/app/api/api_v1/endpoints/mail_sources.py @@ -679,6 +679,7 @@ async def gmail_fetch_reports( access_token=source.gmail_access_token, refresh_token=source.gmail_refresh_token or "", already_ingested_ids=already, + db=db, ) started_at = datetime.utcnow() diff --git a/backend/app/api/api_v1/endpoints/reports.py b/backend/app/api/api_v1/endpoints/reports.py index 6d6da0b..1498f30 100644 --- a/backend/app/api/api_v1/endpoints/reports.py +++ b/backend/app/api/api_v1/endpoints/reports.py @@ -1,10 +1,18 @@ import logging from typing import Any, Dict, List, Optional -from fastapi import APIRouter, File, HTTPException, UploadFile, status +from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status from pydantic import BaseModel +from sqlalchemy.orm import Session +from app.core.database import get_db from app.services.dmarc_parser import DMARCParser +from app.services.report_persistence import ( + delete_persisted_report, + hydrate_report_store_from_db, + report_exists, + save_parsed_report, +) from app.services.report_store import ReportStore from app.utils.domain_validator import DomainValidationError, validate_domain @@ -157,7 +165,7 @@ class PaginatedReportResponse(BaseModel): @router.post("/upload", response_model=UploadResponse) -async def upload_report(file: UploadFile = File(...)): +async def upload_report(file: UploadFile = File(...), db: Session = Depends(get_db)): """ Upload and process a DMARC aggregate report file (XML, ZIP, or GZIP) @@ -195,7 +203,9 @@ async def upload_report(file: UploadFile = File(...)): # Check for duplicate report before storing store = ReportStore.get_instance() report_id = report.get("report_id", "") - if report_id and store.has_report(domain, report_id): + if report_id and ( + store.has_report(domain, report_id) or report_exists(db, domain, report_id) + ): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=( @@ -205,6 +215,8 @@ async def upload_report(file: UploadFile = File(...)): ) # Store the report + save_parsed_report(db, report) + db.commit() store.add_report(report) processed_records = report.get("summary", {}).get("total_count", 0) @@ -231,11 +243,12 @@ async def upload_report(file: UploadFile = File(...)): @router.get("", response_model=List[AllReportsItem]) -async def get_all_reports(): +async def get_all_reports(db: Session = Depends(get_db)): """ Get all DMARC reports across all domains, sorted by end_date descending. """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) domains = store.get_domains() all_reports: List[AllReportsItem] = [] @@ -267,20 +280,22 @@ async def get_all_reports(): @router.get("/domains", response_model=List[str]) -async def get_domains(): +async def get_domains(db: Session = Depends(get_db)): """ Get list of all domains with reports """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) return store.get_domains() @router.get("/domain/{domain}/summary", response_model=DomainSummary) -async def get_domain_summary(domain: str): +async def get_domain_summary(domain: str, db: Session = Depends(get_db)): """ Get summary statistics for a specific domain """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) summary = store.get_domain_summary(domain) if not summary: @@ -292,22 +307,24 @@ async def get_domain_summary(domain: str): @router.get("/summary", response_model=List[DomainSummary]) -async def get_all_summaries(): +async def get_all_summaries(db: Session = Depends(get_db)): """ Get summary statistics for all domains """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) all_summaries = store.get_all_domain_summaries() return [DomainSummary(domain=domain, **summary) for domain, summary in all_summaries.items()] @router.get("/domain/{domain}/reports", response_model=List[ReportSummary]) -async def get_domain_reports(domain: str): +async def get_domain_reports(domain: str, db: Session = Depends(get_db)): """ Get all reports for a specific domain """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) reports = store.get_domain_reports(domain) if not reports: @@ -336,6 +353,7 @@ async def get_domain_reports_paginated( page_size: int = 10, sort_by: str = "end_date", sort_order: str = "desc", + db: Session = Depends(get_db), ): """ Get paginated reports for a specific domain with sorting options @@ -348,6 +366,7 @@ async def get_domain_reports_paginated( sort_order: Sort order (asc or desc) """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) all_reports = store.get_domain_reports(domain) if not all_reports: @@ -404,7 +423,7 @@ class DeleteReportResponse(BaseModel): "/domain/{domain}/reports/{report_id}", response_model=DeleteReportResponse, ) -async def delete_report(domain: str, report_id: str): +async def delete_report(domain: str, report_id: str, db: Session = Depends(get_db)): """ Delete a single DMARC report for a domain. @@ -412,7 +431,10 @@ async def delete_report(domain: str, report_id: str): that aggregated numbers remain accurate after deletion. """ store = ReportStore.get_instance() - deleted = store.delete_report(domain, report_id) + deleted_from_db = delete_persisted_report(db, domain, report_id) + if deleted_from_db: + db.commit() + deleted = store.delete_report(domain, report_id) or deleted_from_db if not deleted: raise HTTPException( @@ -473,11 +495,12 @@ class ReportDetail(BaseModel): @router.get("/{report_id}", response_model=ReportDetail) -async def get_report_by_id(report_id: str): +async def get_report_by_id(report_id: str, db: Session = Depends(get_db)): """ Get full details for a single DMARC report by its report ID. """ store = ReportStore.get_instance() + hydrate_report_store_from_db(db, store) report = store.get_report_by_id(report_id) if report is None: diff --git a/backend/app/main.py b/backend/app/main.py index 05b6199..a7f8424 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -24,6 +24,7 @@ from app.models.mail_source import MailSource # noqa: F401 – ensure table is 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 # Set up logging @@ -40,19 +41,20 @@ def _poll_single_imap_source(source: MailSource) -> None: """Fetch DMARC reports for a single IMAP mail source and update its last_checked timestamp.""" global last_check_time # pylint: disable=global-statement - imap_client = IMAPClient( - server=source.server, - port=source.port or 993, - username=source.username, - password=source.password, - delete_emails=False, - ) - started_at = datetime.utcnow() - results = imap_client.fetch_reports(days=9999) - db = SessionLocal() try: src = db.query(MailSource).get(source.id) + poll_source = src or source + imap_client = IMAPClient( + server=poll_source.server, + port=poll_source.port or 993, + username=poll_source.username, + password=poll_source.password, + delete_emails=False, + db=db, + ) + started_at = datetime.utcnow() + results = imap_client.fetch_reports(days=9999) if src: src.last_checked = datetime.utcnow() record_import_attempt(db, src, results, started_at=started_at, trigger="scheduled") @@ -90,21 +92,22 @@ def _poll_single_gmail_source(source: MailSource) -> None: ) return - already = GmailClient.load_ingested_ids(source.gmail_ingested_ids) - client = GmailClient( - client_id=source.gmail_client_id or "", - client_secret=source.gmail_client_secret or "", - access_token=source.gmail_access_token, - refresh_token=source.gmail_refresh_token or "", - already_ingested_ids=already, - ) - - started_at = datetime.utcnow() - results = client.fetch_reports() - db = SessionLocal() try: src = db.query(MailSource).get(source.id) + poll_source = src or source + already = GmailClient.load_ingested_ids(poll_source.gmail_ingested_ids) + client = GmailClient( + client_id=poll_source.gmail_client_id or "", + client_secret=poll_source.gmail_client_secret or "", + access_token=poll_source.gmail_access_token, + refresh_token=poll_source.gmail_refresh_token or "", + already_ingested_ids=already, + db=db, + ) + + started_at = datetime.utcnow() + results = client.fetch_reports() if src: if results.get("new_ingested_ids"): all_ids = list(dict.fromkeys(already + results["new_ingested_ids"])) @@ -431,6 +434,11 @@ async def domains(request: Request): async def domain_details(request: Request, domain_id: str): """View detailed reports for a specific domain""" store = ReportStore.get_instance() + db = SessionLocal() + try: + hydrate_report_store_from_db(db, store) + finally: + db.close() known_domains = store.get_domains() if domain_id not in known_domains: @@ -522,6 +530,7 @@ def _trigger_poll_imap_source(source: MailSource, db) -> dict: username=source.username, password=source.password, delete_emails=False, + db=db, ) started_at = datetime.utcnow() results = imap_client.fetch_reports(days=7) @@ -550,6 +559,7 @@ def _trigger_poll_gmail_source(source: MailSource, db) -> dict: access_token=source.gmail_access_token, refresh_token=source.gmail_refresh_token or "", already_ingested_ids=already, + db=db, ) started_at = datetime.utcnow() results = gmail_client.fetch_reports() diff --git a/backend/app/services/gmail_client.py b/backend/app/services/gmail_client.py index 34fb157..63cf941 100644 --- a/backend/app/services/gmail_client.py +++ b/backend/app/services/gmail_client.py @@ -21,6 +21,7 @@ from googleapiclient.discovery import build from googleapiclient.errors import HttpError from app.services.dmarc_parser import DMARCParser +from app.services.report_persistence import report_exists, save_parsed_report from app.services.report_store import ReportStore logger = logging.getLogger(__name__) @@ -74,12 +75,14 @@ class GmailClient: access_token: str, refresh_token: str, already_ingested_ids: Optional[List[str]] = None, + db: Any = None, ): self.client_id = client_id self.client_secret = client_secret self._initial_access_token = access_token self.already_ingested_ids: List[str] = list(already_ingested_ids or []) self.report_store = ReportStore.get_instance() + self.db = db self.credentials = Credentials( token=access_token, @@ -337,10 +340,15 @@ class GmailClient: """Store a parsed report unless that domain/report ID is already present.""" domain = report.get("domain", "unknown") report_id = report.get("report_id", "") - if report_id and self.report_store.has_report(domain, report_id): + if report_id and ( + self.report_store.has_report(domain, report_id) + or (self.db is not None and report_exists(self.db, domain, report_id)) + ): logger.info("Skipping duplicate DMARC report %s for %s", report_id, domain) return False + if self.db is not None: + save_parsed_report(self.db, report) self.report_store.add_report(report) return True diff --git a/backend/app/services/imap_client.py b/backend/app/services/imap_client.py index 5c0b91c..34d68f1 100644 --- a/backend/app/services/imap_client.py +++ b/backend/app/services/imap_client.py @@ -7,6 +7,7 @@ from typing import Any, Dict, Tuple from app.core.config import get_settings from app.services.dmarc_parser import DMARCParser +from app.services.report_persistence import report_exists, save_parsed_report from app.services.report_store import ReportStore # Setup logger @@ -25,6 +26,7 @@ class IMAPClient: username: str = None, password: str = None, delete_emails: bool = False, + db: Any = None, ): """ Initialize the IMAP client with credentials @@ -35,6 +37,7 @@ class IMAPClient: username: IMAP username (if None, uses settings) password: IMAP password (if None, uses settings) delete_emails: Whether to delete emails after processing (default: False) + db: Optional SQLAlchemy session used to persist imported reports """ settings = get_settings() @@ -43,6 +46,7 @@ class IMAPClient: self.username = username or settings.IMAP_USERNAME self.password = password or settings.IMAP_PASSWORD self.delete_emails = delete_emails + self.db = db self.report_store = ReportStore.get_instance() @@ -389,7 +393,13 @@ class IMAPClient: domain = report.get("domain", "unknown") report_id = report.get("report_id", "") - if report_id and self.report_store.has_report(domain, report_id): + if report_id and ( + self.report_store.has_report(domain, report_id) + or ( + self.db is not None + and report_exists(self.db, domain, report_id) + ) + ): logger.info( "Skipping duplicate DMARC report %s for %s", report_id, @@ -402,6 +412,8 @@ class IMAPClient: continue # Add the report to the store + if self.db is not None: + save_parsed_report(self.db, report) self.report_store.add_report(report) reports_found += 1 diff --git a/backend/app/services/report_persistence.py b/backend/app/services/report_persistence.py new file mode 100644 index 0000000..6a6099e --- /dev/null +++ b/backend/app/services/report_persistence.py @@ -0,0 +1,236 @@ +import json +from datetime import datetime +from typing import Any, Dict, List, Optional + +from sqlalchemy.orm import Session, selectinload + +from app.models.domain import Domain +from app.models.report import DMARCReport, ReportRecord +from app.services.report_store import ReportStore + + +def _parse_timestamp(value: Any) -> int: + """Return a Unix timestamp from an int-like or ISO date value.""" + if value in (None, ""): + return 0 + if isinstance(value, (int, float)): + return int(value) + try: + return int(value) + except (TypeError, ValueError): + pass + try: + return int(datetime.fromisoformat(str(value)).timestamp()) + except (TypeError, ValueError): + return 0 + + +def _iso_from_timestamp(value: int) -> str: + if not value: + return "" + return datetime.fromtimestamp(value).isoformat() + + +def _loads_json_list(value: Optional[str]) -> Optional[List[Dict[str, Any]]]: + if not value: + return None + try: + decoded = json.loads(value) + except (json.JSONDecodeError, TypeError): + return None + return decoded if isinstance(decoded, list) else None + + +def _policy_parts(report: Dict[str, Any]) -> Dict[str, Any]: + policy = report.get("policy") or {} + if isinstance(policy, str): + return {"p": policy, "sp": "", "pct": "100"} + if not isinstance(policy, dict): + return {"p": "none", "sp": "", "pct": "100"} + return { + "p": policy.get("p", "none"), + "sp": policy.get("sp", ""), + "pct": str(policy.get("pct", "100")), + "adkim": policy.get("adkim") or report.get("adkim"), + "aspf": policy.get("aspf") or report.get("aspf"), + } + + +def report_exists(db: Session, domain_name: str, report_id: str) -> bool: + """Return True when the domain/report ID pair is already persisted.""" + if not report_id: + return False + return ( + db.query(DMARCReport.id) + .join(Domain, DMARCReport.domain_id == Domain.id) + .filter(Domain.name == domain_name, DMARCReport.report_id == report_id) + .first() + is not None + ) + + +def save_parsed_report(db: Session, report: Dict[str, Any]) -> tuple[DMARCReport, bool]: + """Persist a parsed DMARC report and its records. + + Returns ``(row, created)``. The caller owns the transaction and should + commit after all related work has completed. + """ + domain_name = report.get("domain") or "unknown" + report_id = report.get("report_id") or "" + policy = _policy_parts(report) + + domain = db.query(Domain).filter(Domain.name == domain_name).first() + if domain is None: + domain = Domain(name=domain_name, dmarc_policy=policy["p"]) + db.add(domain) + db.flush() + elif policy.get("p"): + domain.dmarc_policy = policy["p"] + + existing = ( + db.query(DMARCReport) + .filter(DMARCReport.domain_id == domain.id, DMARCReport.report_id == report_id) + .first() + ) + if existing is not None: + return existing, False + + begin_ts = _parse_timestamp(report.get("begin_timestamp") or report.get("begin_date")) + end_ts = _parse_timestamp(report.get("end_timestamp") or report.get("end_date")) + pct = _parse_timestamp(policy.get("pct")) or 100 + + db_report = DMARCReport( + domain_id=domain.id, + report_id=report_id, + org_name=report.get("org_name") or "", + begin_date=begin_ts, + end_date=end_ts, + source_email=report.get("email") or report.get("source_email"), + policy=policy["p"], + subdomain_policy=policy.get("sp") or None, + adkim=policy.get("adkim") or None, + aspf=policy.get("aspf") or None, + percentage=pct, + ) + db.add(db_report) + db.flush() + + for record in report.get("records", []): + db.add( + ReportRecord( + report_id=db_report.id, + source_ip=record.get("source_ip") or "unknown", + count=int(record.get("count") or 0), + disposition=record.get("disposition") or "none", + dkim=record.get("dkim_result") or record.get("dkim") or "unknown", + spf=record.get("spf_result") or record.get("spf") or "unknown", + header_from=record.get("header_from"), + envelope_from=record.get("envelope_from"), + dkim_auth_details=( + json.dumps(record.get("dkim")) if isinstance(record.get("dkim"), list) else None + ), + spf_auth_details=( + json.dumps(record.get("spf")) if isinstance(record.get("spf"), list) else None + ), + ) + ) + + return db_report, True + + +def persisted_report_to_dict(report: DMARCReport) -> Dict[str, Any]: + """Convert persisted report rows into the parsed-report shape used by the UI.""" + records: List[Dict[str, Any]] = [] + total_count = 0 + passed_count = 0 + + for record in report.records: + count = int(record.count or 0) + dkim_result = record.dkim or "unknown" + spf_result = record.spf or "unknown" + total_count += count + if dkim_result == "pass" or spf_result == "pass": + passed_count += count + + records.append( + { + "source_ip": record.source_ip, + "count": count, + "disposition": record.disposition or "none", + "dkim_result": dkim_result, + "spf_result": spf_result, + "header_from": record.header_from or "", + "dkim": _loads_json_list(record.dkim_auth_details), + "spf": _loads_json_list(record.spf_auth_details), + } + ) + + failed_count = total_count - passed_count + pass_rate = round(passed_count / total_count * 100, 1) if total_count > 0 else 0.0 + return { + "domain": report.domain.name if report.domain else "unknown", + "report_id": report.report_id, + "org_name": report.org_name, + "email": report.source_email or "", + "begin_date": _iso_from_timestamp(report.begin_date), + "end_date": _iso_from_timestamp(report.end_date), + "begin_timestamp": report.begin_date, + "end_timestamp": report.end_date, + "policy": { + "p": report.policy or "none", + "sp": report.subdomain_policy or "", + "pct": str(report.percentage or 100), + }, + "records": records, + "summary": { + "total_count": total_count, + "passed_count": passed_count, + "failed_count": failed_count, + "pass_rate": pass_rate, + }, + } + + +def hydrate_report_store_from_db(db: Session, store: ReportStore | None = None) -> int: + """Load persisted reports into ReportStore when the database has report rows.""" + report_count = db.query(DMARCReport.id).count() + if report_count == 0: + return 0 + + store = store or ReportStore.get_instance() + store.clear() + reports = ( + db.query(DMARCReport) + .options( + selectinload(DMARCReport.domain), + selectinload(DMARCReport.records), + ) + .order_by(DMARCReport.end_date.desc()) + .all() + ) + for report in reports: + store.add_report(persisted_report_to_dict(report)) + return len(reports) + + +def delete_persisted_report(db: Session, domain_name: str, report_id: str) -> bool: + """Delete a persisted report by domain/report ID.""" + report = ( + db.query(DMARCReport) + .join(Domain, DMARCReport.domain_id == Domain.id) + .filter(Domain.name == domain_name, DMARCReport.report_id == report_id) + .first() + ) + if report is None: + return False + db.delete(report) + return True + + +def delete_persisted_domain(db: Session, domain_name: str) -> bool: + """Delete a domain row and cascaded report data.""" + domain = db.query(Domain).filter(Domain.name == domain_name).first() + if domain is None: + return False + db.delete(domain) + return True diff --git a/backend/app/tests/test_gmail_client.py b/backend/app/tests/test_gmail_client.py index c1e2c52..724aab3 100644 --- a/backend/app/tests/test_gmail_client.py +++ b/backend/app/tests/test_gmail_client.py @@ -19,6 +19,7 @@ from unittest.mock import MagicMock, patch import pytest +from app.models.report import DMARCReport from app.services.gmail_client import GmailClient from app.services.report_store import ReportStore from app.tests.test_data import SAMPLE_XML @@ -32,6 +33,7 @@ def _make_client( access_token: str = "acc", refresh_token: str = "ref", already_ingested: Optional[list] = None, + db=None, ) -> GmailClient: """Instantiate a GmailClient with real Credentials mocked out.""" with patch("app.services.gmail_client.Credentials") as mock_creds_class: @@ -46,6 +48,7 @@ def _make_client( access_token=access_token, refresh_token=refresh_token, already_ingested_ids=already_ingested or [], + db=db, ) # Expose the mock so tests can manipulate it client._mock_creds = mock_creds # type: ignore[attr-defined] @@ -484,6 +487,25 @@ class TestProcessAttachments: assert stats["reports_found"] == 1 assert "example.com" in client.report_store.get_domains() + def test_google_style_zip_attachment_is_persisted(self, db_session): + """Gmail imports write parsed DMARC reports to the database when a DB is provided.""" + client = _make_client(db=db_session) + raw = _make_raw_email( + [ + { + "filename": "google.com!example.com!1597449600!1597535999.zip", + "content": _zip_xml(), + } + ] + ) + msg = email_mod.message_from_bytes(raw) + stats = {"reports_found": 0, "errors": []} + + count = client._process_attachments(msg, stats) + + assert count == 1 + assert db_session.query(DMARCReport).filter_by(report_id="123456789").count() == 1 + def test_duplicate_report_is_skipped(self): """Repeated imports of the same domain/report ID should not inflate totals.""" client = _make_client() diff --git a/backend/app/tests/test_imap_client.py b/backend/app/tests/test_imap_client.py index 34c9040..602c9d7 100644 --- a/backend/app/tests/test_imap_client.py +++ b/backend/app/tests/test_imap_client.py @@ -16,6 +16,7 @@ from zipfile import ZipFile import pytest +from app.models.report import DMARCReport from app.services.imap_client import IMAPClient from app.services.report_store import ReportStore @@ -411,7 +412,7 @@ class TestHasDmarcAttachments: class TestProcessAttachments: - def _make_client(self): + def _make_client(self, db=None): with patch("app.services.imap_client.get_settings") as mock_settings: mock_settings.return_value = MagicMock( IMAP_SERVER="imap.example.com", @@ -419,7 +420,7 @@ class TestProcessAttachments: IMAP_USERNAME="u", IMAP_PASSWORD="p", ) - return IMAPClient() + return IMAPClient(db=db) def test_processes_xml_attachment(self): client = self._make_client() @@ -438,6 +439,17 @@ class TestProcessAttachments: count = client._process_attachments(msg) assert count == 1 + def test_processes_xml_attachment_persists_report(self, db_session): + client = self._make_client(db=db_session) + msg = email.message_from_bytes( + _make_email_with_attachment("report.xml", MINIMAL_DMARC_XML, "application/xml") + ) + + count = client._process_attachments(msg) + + assert count == 1 + assert db_session.query(DMARCReport).filter_by(report_id="abc-123").count() == 1 + def test_bad_attachment_does_not_raise(self): client = self._make_client() msg = email.message_from_bytes(_make_email_with_attachment("report.xml", b"not xml at all")) diff --git a/backend/app/tests/test_reports_api.py b/backend/app/tests/test_reports_api.py index 11f04f6..10adeb5 100644 --- a/backend/app/tests/test_reports_api.py +++ b/backend/app/tests/test_reports_api.py @@ -3,6 +3,8 @@ import zipfile from fastapi.testclient import TestClient +from app.models.report import DMARCReport, ReportRecord +from app.services.report_store import ReportStore from app.tests.test_data import SAMPLE_XML @@ -27,6 +29,41 @@ def test_upload_report_success(client: TestClient): assert data["domain"] == "example.com" +def test_upload_persists_report_rows(client: TestClient, db_session): + """Uploaded reports are written to the durable report tables.""" + zip_bytes = _make_zip(SAMPLE_XML) + response = client.post( + "/api/v1/reports/upload", + files={"file": ("report.zip", zip_bytes, "application/zip")}, + ) + assert response.status_code == 200 + + report = db_session.query(DMARCReport).filter_by(report_id="123456789").one() + assert report.org_name == "google.com" + assert report.domain.name == "example.com" + assert db_session.query(ReportRecord).filter_by(report_id=report.id).count() == 1 + + +def test_report_reads_hydrate_from_persisted_rows(client: TestClient): + """Report read APIs rebuild the in-memory projection from the database.""" + zip_bytes = _make_zip(SAMPLE_XML) + response = client.post( + "/api/v1/reports/upload", + files={"file": ("report.zip", zip_bytes, "application/zip")}, + ) + assert response.status_code == 200 + + ReportStore.get_instance().clear() + + domains = client.get("/api/v1/reports/domains") + assert domains.status_code == 200 + assert domains.json() == ["example.com"] + + detail = client.get("/api/v1/reports/123456789") + assert detail.status_code == 200 + assert detail.json()["summary"]["total_count"] == 2 + + def test_upload_populates_domains_list(client: TestClient): """After uploading a report, the domain appears in the reports/domains endpoint.""" zip_bytes = _make_zip(SAMPLE_XML) @@ -89,7 +126,26 @@ def test_duplicate_upload_returns_409(client: TestClient): assert "already been uploaded" in second.json()["detail"].lower() -def test_delete_report_success(client: TestClient): +def test_duplicate_upload_checks_persisted_rows(client: TestClient): + """Duplicate detection still works when the in-memory store is empty.""" + zip_bytes = _make_zip(SAMPLE_XML) + + first = client.post( + "/api/v1/reports/upload", + files={"file": ("report.zip", zip_bytes, "application/zip")}, + ) + assert first.status_code == 200 + + ReportStore.get_instance().clear() + + second = client.post( + "/api/v1/reports/upload", + files={"file": ("report.zip", zip_bytes, "application/zip")}, + ) + assert second.status_code == 409 + + +def test_delete_report_success(client: TestClient, db_session): """Deleting an existing report returns 200 and removes it from the store.""" zip_bytes = _make_zip(SAMPLE_XML) client.post( @@ -105,6 +161,7 @@ def test_delete_report_success(client: TestClient): assert response.status_code == 200 data = response.json() assert data["success"] is True + assert db_session.query(DMARCReport).filter_by(report_id="123456789").count() == 0 # Domain should be gone now assert client.get("/api/v1/reports/domain/example.com/summary").status_code == 404 diff --git a/docs/development/roadmap.md b/docs/development/roadmap.md index 4a89769..820d9c8 100644 --- a/docs/development/roadmap.md +++ b/docs/development/roadmap.md @@ -24,24 +24,13 @@ Recently improved: - Gmail and IMAP imports now skip duplicate domain/report IDs. - Tests cover Google-style DMARC ZIP attachment imports. - Mail source imports now create sanitized import-history records for manual and scheduled polls. +- Parsed upload, Gmail, and IMAP reports are now persisted to `dmarc_reports` and `report_records`. +- Report/domain API reads can hydrate the dashboard projection from persisted data after restart. -Important gap: -- Parsed DMARC report data is still served primarily from the in-memory `ReportStore`. The database schema exists, but report upload/import paths and dashboard read paths must be completed before the persistence milestone can be called done. +Implementation note: +- The legacy `ReportStore` remains as a projection layer for existing report/dashboard code, but durable report data now lives in the database. -## Active Milestone: Finish Report Persistence - -Objective: complete the database-backed report storage promised by Milestone 3. - -Priority tasks: -- Add a report persistence service that converts parsed DMARC report dicts into `Domain`, `DMARCReport`, and `ReportRecord` rows. -- Load or query persisted reports for dashboard, domain, and report endpoints. -- Add duplicate report detection against the database. -- Keep tests covering upload, Gmail import, IMAP import, and restart-style reload behavior. - -Quality bar: -- Uploading or importing a report survives application restart and remains visible in report/domain endpoints. - -## Next Milestone: Reporting Quality and Import Confidence +## Active Milestone: Reporting Quality and Import Confidence Objective: make mailbox imports auditable and make report totals trustworthy. @@ -57,7 +46,7 @@ Quality bar: - Parse failures must be visible and actionable. - The user should be able to tell whether a mail source is healthy without reading logs. -## Following Milestone: Meaningful Reports +## Next Milestone: Meaningful Reports Objective: turn parsed DMARC data into administrator-friendly reports. diff --git a/docs/milestones.md b/docs/milestones.md index 04b8cb2..0fe76f7 100644 --- a/docs/milestones.md +++ b/docs/milestones.md @@ -29,24 +29,25 @@ Delivered: - Duplicate report protection for both Gmail and IMAP imports. - Background polling and manual poll hooks for configured mail sources. -## Milestone 3: Database Foundation, Domain Management, and Auth Foundation - In Progress +## Milestone 3: Database Foundation, Domain Management, and Auth Foundation - Complete -Status: In progress +Status: Complete Delivered: - SQLAlchemy models and Alembic migrations. - SQLite/PostgreSQL-compatible database configuration. - Domain management APIs and UI. - Report and source database models. +- Database-backed persistence for uploaded DMARC reports. +- Database-backed persistence for Gmail and IMAP imported reports. +- Duplicate report detection against persisted report data. +- Report/domain API reads can hydrate their dashboard projection from persisted reports after restart. - Settings and mail source persistence. - Logto-based auth integration plus an explicit local development auth-disabled mode. - Security middleware, safer default secret generation, and security-focused tests. -Remaining before this milestone is complete: -- Persist parsed DMARC reports and report records through the database-backed models. -- Load or query persisted reports after restart so dashboards do not depend on process memory. -- Move report/domain summary endpoints from the in-memory `ReportStore` to database queries. -- Keep duplicate report detection consistent across upload, IMAP, and Gmail after persistence is enabled. +Implementation note: +- The existing `ReportStore` remains as a compatibility projection for dashboard/report code, but persisted database rows are now the durable source for uploads and mailbox imports. ## Milestone 4: Reporting Quality and Import Confidence - In Progress @@ -59,9 +60,9 @@ Recently delivered: - Gmail/IMAP imports skip duplicate report IDs to avoid inflated totals. - Tests now cover a real Google-style ZIP attachment path rather than only mocked parser behavior. - Mail source imports now persist sanitized import-history records for manual and scheduled polls. +- Uploaded, Gmail-imported, and IMAP-imported reports are now persisted to report/record tables and can be reloaded into report/domain views. Next tasks: -- Finish Milestone 3 report persistence before expanding report features. - Add per-import result details: skipped duplicates, parse failures, unsupported attachments, and imported report IDs. - Add a UI import history view for each mail source. - Add mailbox search controls for date range/backfill without requiring code changes. diff --git a/docs/todo.md b/docs/todo.md index 6631dd1..854b20d 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -101,14 +101,14 @@ This file tracks the specific implementation tasks for each milestone of the DMA - [x] Implement data access layer ### Model Migration -- [ ] Convert report ingestion and dashboard reads from in-memory storage to database-backed storage +- [x] Convert report ingestion and dashboard reads from in-memory storage to database-backed storage - [x] Create Domain table - [x] Create AggregateReport table - [x] Create ReportRecord table for sender details - [x] Implement relationships between models -- [ ] Persist parsed upload reports to `dmarc_reports` and `report_records` -- [ ] Persist parsed Gmail/IMAP reports to `dmarc_reports` and `report_records` -- [ ] Load/query persisted reports after app restart +- [x] Persist parsed upload reports to `dmarc_reports` and `report_records` +- [x] Persist parsed Gmail/IMAP reports to `dmarc_reports` and `report_records` +- [x] Load/query persisted reports after app restart ### Domain Management - [x] Create UI for adding/editing domains