gitea-admin/gh-christianlouis-dmarq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #99 from christianlouis/codex/report-persistence

Browse Source 
feat: persist imported DMARC reports
This commit is contained in:
Christian Krakau-Louis
2026-05-22 19:38:31 +02:00
committed by GitHub
parent 9630944460 b9041012de
commit 40f7d52df2
14 changed files with 492 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/api_v1/endpoints/domains.py
+34 -13
View File
@@ -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()
backend/app/api/api_v1/endpoints/imap.py
+20 -2
View File
@@ -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"],
backend/app/api/api_v1/endpoints/mail_sources.py
+1
View File
@@ -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()
backend/app/api/api_v1/endpoints/reports.py
+34 -11
View File
@@ -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:
backend/app/main.py
+32 -22
View File
@@ -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()
backend/app/services/gmail_client.py
+9 -1
View File
@@ -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
backend/app/services/imap_client.py
+13 -1
View File
@@ -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
backend/app/services/report_persistence.py
+236
View File
@@ -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
backend/app/tests/test_gmail_client.py
+22
View File
@@ -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()
backend/app/tests/test_imap_client.py
+14 -2
View File
@@ -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"))
backend/app/tests/test_reports_api.py
+58 -1
View File
@@ -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
docs/development/roadmap.md
+6 -17
View File
@@ -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.
docs/milestones.md
+9 -8
View File
@@ -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.
docs/todo.md
+4 -4
View File
@@ -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