Merge pull request #174 from christianlouis/codex/m10-forensic-foundation
[codex] Add forensic report parsing and storage
This commit is contained in:
Christian Krakau-Louis
@@ -63,9 +63,10 @@ have no working implementation in the codebase yet.
|
||||
|
||||
### Forensic Reports (RFC 6591)
|
||||
- **Documented in**: README.md ("Forensic Reports: Analyze failure samples (RFC 6591 support)")
|
||||
- **Current state**: The DMARC parser (`backend/app/services/dmarc_parser.py`) only
|
||||
handles aggregate reports. There is no forensic report parsing, UI, or storage.
|
||||
- [ ] Forensic report parsing
|
||||
- **Current state**: Aggregate and forensic reports are now parsed separately. Forensic
|
||||
reports are stored in dedicated database rows and surfaced through authenticated APIs
|
||||
without affecting aggregate compliance statistics.
|
||||
- [x] Forensic report parsing
|
||||
- [ ] Failure sample analysis
|
||||
- [ ] PII redaction options
|
||||
- [ ] Detailed authentication failure views
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
"""add forensic reports
|
||||
|
||||
Revision ID: d1e2f3a4b5c6
|
||||
Revises: c0d1e2f3a4b5
|
||||
Create Date: 2026-05-23 00:00:00.000000
|
||||
|
||||
"""
|
||||
|
||||
from typing import Sequence, Union
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "d1e2f3a4b5c6"
|
||||
down_revision: Union[str, Sequence[str], None] = "c0d1e2f3a4b5"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Create DMARC forensic/failure report storage."""
|
||||
op.create_table(
|
||||
"forensic_reports",
|
||||
sa.Column("id", sa.Integer(), nullable=False),
|
||||
sa.Column("domain_id", sa.Integer(), nullable=True),
|
||||
sa.Column("report_id", sa.String(), nullable=False),
|
||||
sa.Column("source_email", sa.String(), nullable=True),
|
||||
sa.Column("feedback_type", sa.String(), nullable=True),
|
||||
sa.Column("user_agent", sa.String(), nullable=True),
|
||||
sa.Column("version", sa.String(), nullable=True),
|
||||
sa.Column("reported_domain", sa.String(), nullable=True),
|
||||
sa.Column("source_ip", sa.String(), nullable=True),
|
||||
sa.Column("auth_failure", sa.String(), nullable=True),
|
||||
sa.Column("delivery_result", sa.String(), nullable=True),
|
||||
sa.Column("arrival_date", sa.DateTime(), nullable=True),
|
||||
sa.Column("authentication_results", sa.Text(), nullable=True),
|
||||
sa.Column("original_mail_from", sa.String(), nullable=True),
|
||||
sa.Column("original_from", sa.String(), nullable=True),
|
||||
sa.Column("original_to", sa.String(), nullable=True),
|
||||
sa.Column("original_subject", sa.String(), nullable=True),
|
||||
sa.Column("original_message_id", sa.String(), nullable=True),
|
||||
sa.Column("original_date", sa.String(), nullable=True),
|
||||
sa.Column("feedback_headers", sa.Text(), nullable=True),
|
||||
sa.Column("processed_at", sa.DateTime(), nullable=True),
|
||||
sa.ForeignKeyConstraint(["domain_id"], ["domains.id"]),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
sa.UniqueConstraint("report_id", name="uq_forensic_reports_report_id"),
|
||||
)
|
||||
op.create_index(op.f("ix_forensic_reports_id"), "forensic_reports", ["id"])
|
||||
op.create_index(op.f("ix_forensic_reports_domain_id"), "forensic_reports", ["domain_id"])
|
||||
op.create_index(op.f("ix_forensic_reports_report_id"), "forensic_reports", ["report_id"])
|
||||
op.create_index(
|
||||
op.f("ix_forensic_reports_feedback_type"), "forensic_reports", ["feedback_type"]
|
||||
)
|
||||
op.create_index(
|
||||
op.f("ix_forensic_reports_reported_domain"), "forensic_reports", ["reported_domain"]
|
||||
)
|
||||
op.create_index(op.f("ix_forensic_reports_source_ip"), "forensic_reports", ["source_ip"])
|
||||
op.create_index(op.f("ix_forensic_reports_auth_failure"), "forensic_reports", ["auth_failure"])
|
||||
op.create_index(op.f("ix_forensic_reports_arrival_date"), "forensic_reports", ["arrival_date"])
|
||||
op.create_index(op.f("ix_forensic_reports_processed_at"), "forensic_reports", ["processed_at"])
|
||||
op.create_index(
|
||||
"ix_forensic_reports_domain_arrival",
|
||||
"forensic_reports",
|
||||
["domain_id", "arrival_date"],
|
||||
)
|
||||
op.create_index(
|
||||
"ix_forensic_reports_failure_source",
|
||||
"forensic_reports",
|
||||
["auth_failure", "source_ip"],
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Drop DMARC forensic/failure report storage."""
|
||||
op.drop_index("ix_forensic_reports_failure_source", table_name="forensic_reports")
|
||||
op.drop_index("ix_forensic_reports_domain_arrival", table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_processed_at"), table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_arrival_date"), table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_auth_failure"), table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_source_ip"), table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_reported_domain"), table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_feedback_type"), table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_report_id"), table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_domain_id"), table_name="forensic_reports")
|
||||
op.drop_index(op.f("ix_forensic_reports_id"), table_name="forensic_reports")
|
||||
op.drop_table("forensic_reports")
|
||||
@@ -3,6 +3,7 @@ from fastapi import APIRouter
|
||||
from app.api.api_v1.endpoints import (
|
||||
auth,
|
||||
domains,
|
||||
forensics,
|
||||
health,
|
||||
imap,
|
||||
mail_sources,
|
||||
@@ -20,6 +21,7 @@ api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
|
||||
api_router.include_router(health.router, tags=["health"])
|
||||
api_router.include_router(domains.router, prefix="/domains", tags=["domains"])
|
||||
api_router.include_router(reports.router, prefix="/reports", tags=["reports"])
|
||||
api_router.include_router(forensics.router, prefix="/forensics", tags=["forensics"])
|
||||
api_router.include_router(setup.router, prefix="/setup", tags=["setup"])
|
||||
api_router.include_router(imap.router, prefix="/imap", tags=["imap"])
|
||||
api_router.include_router(stats.router, prefix="/stats", tags=["stats"])
|
||||
|
||||
@@ -0,0 +1,177 @@
|
||||
import logging
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy.orm import Session, selectinload
|
||||
|
||||
from app.core.database import get_db
|
||||
from app.core.security import require_admin_auth
|
||||
from app.models.domain import Domain
|
||||
from app.models.report import ForensicReport
|
||||
from app.services.forensic_parser import ForensicParser, MAX_FORENSIC_REPORT_SIZE
|
||||
from app.services.forensic_persistence import (
|
||||
forensic_report_exists,
|
||||
forensic_report_to_dict,
|
||||
save_forensic_report,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
class ForensicReportResponse(BaseModel):
|
||||
id: int
|
||||
report_id: str
|
||||
domain: Optional[str] = None
|
||||
reported_domain: Optional[str] = None
|
||||
source_email: Optional[str] = None
|
||||
feedback_type: Optional[str] = None
|
||||
user_agent: Optional[str] = None
|
||||
version: Optional[str] = None
|
||||
source_ip: Optional[str] = None
|
||||
auth_failure: Optional[str] = None
|
||||
delivery_result: Optional[str] = None
|
||||
arrival_date: Optional[str] = None
|
||||
authentication_results: Optional[str] = None
|
||||
original_mail_from: Optional[str] = None
|
||||
original_from: Optional[str] = None
|
||||
original_to: Optional[str] = None
|
||||
original_subject: Optional[str] = None
|
||||
original_message_id: Optional[str] = None
|
||||
original_date: Optional[str] = None
|
||||
feedback_headers: Dict[str, Any] = Field(default_factory=dict)
|
||||
processed_at: Optional[str] = None
|
||||
|
||||
|
||||
class ForensicListResponse(BaseModel):
|
||||
total: int
|
||||
page: int
|
||||
page_size: int
|
||||
total_pages: int
|
||||
reports: List[ForensicReportResponse]
|
||||
|
||||
|
||||
class ForensicUploadResponse(BaseModel):
|
||||
success: bool
|
||||
report_id: str
|
||||
domain: Optional[str] = None
|
||||
message: str
|
||||
duplicate: bool = False
|
||||
|
||||
|
||||
def _validate_upload(file: UploadFile, content: bytes) -> None:
|
||||
filename = file.filename or ""
|
||||
if not filename:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Filename is required")
|
||||
if len(content) == 0:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="File is empty")
|
||||
if len(content) > MAX_FORENSIC_REPORT_SIZE:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="File too large"
|
||||
)
|
||||
if not filename.lower().endswith((".eml", ".txt")):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Invalid file type. Upload a forensic report email as .eml or .txt.",
|
||||
)
|
||||
|
||||
|
||||
@router.post("/upload", response_model=ForensicUploadResponse)
|
||||
async def upload_forensic_report(
|
||||
file: UploadFile = File(...),
|
||||
db: Session = Depends(get_db),
|
||||
_auth: dict = Depends(require_admin_auth),
|
||||
):
|
||||
"""Upload and store a DMARC forensic/failure report email."""
|
||||
try:
|
||||
content = await file.read()
|
||||
_validate_upload(file, content)
|
||||
parsed = ForensicParser.parse_bytes(content)
|
||||
if forensic_report_exists(db, parsed["report_id"]):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail="Forensic report has already been uploaded.",
|
||||
)
|
||||
|
||||
row, created = save_forensic_report(db, parsed)
|
||||
if not created:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail="Forensic report has already been uploaded.",
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(row)
|
||||
return ForensicUploadResponse(
|
||||
success=True,
|
||||
report_id=row.report_id,
|
||||
domain=row.reported_domain,
|
||||
message="Forensic report processed successfully.",
|
||||
)
|
||||
except HTTPException:
|
||||
raise
|
||||
except ValueError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Invalid forensic report format.",
|
||||
) from exc
|
||||
except Exception as exc: # pylint: disable=broad-exception-caught
|
||||
logger.error("Unexpected forensic upload failure for %s: %s", file.filename, exc)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="Error processing forensic report.",
|
||||
) from exc
|
||||
|
||||
|
||||
@router.get("", response_model=ForensicListResponse)
|
||||
async def list_forensic_reports(
|
||||
domain: Optional[str] = Query(default=None),
|
||||
page: int = Query(default=1, ge=1),
|
||||
page_size: int = Query(default=50, ge=1, le=200),
|
||||
db: Session = Depends(get_db),
|
||||
_auth: dict = Depends(require_admin_auth),
|
||||
):
|
||||
"""List stored forensic reports, newest first."""
|
||||
query = db.query(ForensicReport).options(selectinload(ForensicReport.domain))
|
||||
if domain:
|
||||
normalized = domain.lower()
|
||||
query = query.outerjoin(Domain).filter(
|
||||
(Domain.name == normalized) | (ForensicReport.reported_domain == normalized)
|
||||
)
|
||||
|
||||
total = query.count()
|
||||
rows = (
|
||||
query.order_by(ForensicReport.arrival_date.desc().nullslast(), ForensicReport.id.desc())
|
||||
.offset((page - 1) * page_size)
|
||||
.limit(page_size)
|
||||
.all()
|
||||
)
|
||||
total_pages = (total + page_size - 1) // page_size if total else 0
|
||||
return ForensicListResponse(
|
||||
total=total,
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
total_pages=total_pages,
|
||||
reports=[ForensicReportResponse(**forensic_report_to_dict(row)) for row in rows],
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{report_id}", response_model=ForensicReportResponse)
|
||||
async def get_forensic_report(
|
||||
report_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
_auth: dict = Depends(require_admin_auth),
|
||||
):
|
||||
"""Return one stored forensic report by numeric ID."""
|
||||
row = (
|
||||
db.query(ForensicReport)
|
||||
.options(selectinload(ForensicReport.domain))
|
||||
.filter(ForensicReport.id == report_id)
|
||||
.first()
|
||||
)
|
||||
if row is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND, detail="Forensic report not found"
|
||||
)
|
||||
return ForensicReportResponse(**forensic_report_to_dict(row))
|
||||
@@ -108,6 +108,8 @@ async def fetch_imap_reports(
|
||||
"success": results["success"],
|
||||
"processed_emails": results["processed"],
|
||||
"reports_found": results["reports_found"],
|
||||
"forensic_reports_found": int(results.get("forensic_reports_found", 0)),
|
||||
"duplicate_forensic_reports": int(results.get("duplicate_forensic_reports", 0)),
|
||||
"new_domains": results["new_domains"],
|
||||
"errors": results["errors"] if "errors" in results and results["errors"] else None,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
|
||||
@@ -392,6 +392,8 @@ def _fetch_response(source: MailSource, results: Dict[str, Any]) -> Dict[str, An
|
||||
"processed": int(results.get("processed", 0)),
|
||||
"reports_found": int(results.get("reports_found", 0)),
|
||||
"duplicate_reports": int(results.get("duplicate_reports", 0)),
|
||||
"forensic_reports_found": int(results.get("forensic_reports_found", 0)),
|
||||
"duplicate_forensic_reports": int(results.get("duplicate_forensic_reports", 0)),
|
||||
"new_domains": [str(d) for d in results.get("new_domains", [])],
|
||||
"error_count": len(results.get("errors", [])),
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
@@ -555,10 +557,12 @@ async def fetch_mail_source(
|
||||
source = _get_source_or_404(source_id, db)
|
||||
results = _fetch_source(source, db, days)
|
||||
logger.info(
|
||||
"Manual fetch for source id=%d: processed=%d reports_found=%d duplicates=%d",
|
||||
"Manual fetch for source id=%d: processed=%d reports_found=%d "
|
||||
"forensic_reports_found=%d duplicates=%d",
|
||||
int(source_id),
|
||||
int(results.get("processed", 0)),
|
||||
int(results.get("reports_found", 0)),
|
||||
int(results.get("forensic_reports_found", 0)),
|
||||
int(results.get("duplicate_reports", 0)),
|
||||
)
|
||||
for err in results.get("errors", []):
|
||||
@@ -980,10 +984,11 @@ async def gmail_fetch_reports(
|
||||
db.commit()
|
||||
|
||||
logger.info(
|
||||
"Gmail fetch for source id=%d: processed=%d reports_found=%d",
|
||||
"Gmail fetch for source id=%d: processed=%d reports_found=%d forensic_reports_found=%d",
|
||||
int(source_id),
|
||||
int(results.get("processed", 0)),
|
||||
int(results.get("reports_found", 0)),
|
||||
int(results.get("forensic_reports_found", 0)),
|
||||
)
|
||||
|
||||
for err in results.get("errors", []):
|
||||
@@ -997,6 +1002,8 @@ async def gmail_fetch_reports(
|
||||
"success": bool(results.get("success", False)),
|
||||
"processed": int(results.get("processed", 0)),
|
||||
"reports_found": int(results.get("reports_found", 0)),
|
||||
"forensic_reports_found": int(results.get("forensic_reports_found", 0)),
|
||||
"duplicate_forensic_reports": int(results.get("duplicate_forensic_reports", 0)),
|
||||
"new_domains": [str(d) for d in results.get("new_domains", [])],
|
||||
"error_count": len(results.get("errors", [])),
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
|
||||
+10
-2
@@ -80,10 +80,12 @@ def _poll_single_imap_source(source: MailSource) -> None:
|
||||
|
||||
if results["success"]:
|
||||
logger.info(
|
||||
"IMAP polling (source id=%d): %s emails processed, %s reports found",
|
||||
"IMAP polling (source id=%d): %s emails processed, %s aggregate reports found, "
|
||||
"%s forensic reports found",
|
||||
source.id,
|
||||
results["processed"],
|
||||
results["reports_found"],
|
||||
results.get("forensic_reports_found", 0),
|
||||
)
|
||||
if results["new_domains"]:
|
||||
logger.info("New domains found: %s", ", ".join(results["new_domains"]))
|
||||
@@ -143,10 +145,12 @@ def _poll_single_gmail_source(source: MailSource) -> None:
|
||||
|
||||
if results["success"]:
|
||||
logger.info(
|
||||
"Gmail polling (source id=%d): %s emails processed, %s reports found",
|
||||
"Gmail polling (source id=%d): %s emails processed, %s aggregate reports found, "
|
||||
"%s forensic reports found",
|
||||
source.id,
|
||||
results["processed"],
|
||||
results["reports_found"],
|
||||
results.get("forensic_reports_found", 0),
|
||||
)
|
||||
if results["new_domains"]:
|
||||
logger.info("New domains found: %s", ", ".join(results["new_domains"]))
|
||||
@@ -616,6 +620,8 @@ def _trigger_poll_imap_source(source: MailSource, db, days: int = 7) -> dict:
|
||||
"success": results["success"],
|
||||
"processed": results.get("processed", 0),
|
||||
"reports_found": results.get("reports_found", 0),
|
||||
"forensic_reports_found": results.get("forensic_reports_found", 0),
|
||||
"duplicate_forensic_reports": results.get("duplicate_forensic_reports", 0),
|
||||
"new_domains": results.get("new_domains", []),
|
||||
}
|
||||
|
||||
@@ -654,6 +660,8 @@ def _trigger_poll_gmail_source(source: MailSource, db) -> dict:
|
||||
"success": results["success"],
|
||||
"processed": results.get("processed", 0),
|
||||
"reports_found": results.get("reports_found", 0),
|
||||
"forensic_reports_found": results.get("forensic_reports_found", 0),
|
||||
"duplicate_forensic_reports": results.get("duplicate_forensic_reports", 0),
|
||||
"new_domains": results.get("new_domains", []),
|
||||
}
|
||||
|
||||
|
||||
@@ -31,6 +31,9 @@ class Domain(Base):
|
||||
|
||||
# Relationships
|
||||
reports = relationship("DMARCReport", back_populates="domain", cascade="all, delete-orphan")
|
||||
forensic_reports = relationship(
|
||||
"ForensicReport", back_populates="domain", cascade="all, delete-orphan"
|
||||
)
|
||||
user_domains = relationship("UserDomain", back_populates="domain", cascade="all, delete-orphan")
|
||||
|
||||
# Indexes for common queries
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import Column, DateTime, ForeignKey, Index, Integer, String, Text
|
||||
from sqlalchemy import Column, DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from app.core.database import Base
|
||||
@@ -90,3 +90,50 @@ class ReportRecord(Base):
|
||||
|
||||
def __repr__(self):
|
||||
return f"<ReportRecord {self.id} ({self.source_ip})>"
|
||||
|
||||
|
||||
class ForensicReport(Base):
|
||||
"""DMARC forensic/failure report model (RFC 6591 / ARF)."""
|
||||
|
||||
__tablename__ = "forensic_reports"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
domain_id = Column(Integer, ForeignKey("domains.id"), nullable=True, index=True)
|
||||
|
||||
# Report metadata
|
||||
report_id = Column(String, nullable=False, index=True)
|
||||
source_email = Column(String, nullable=True)
|
||||
feedback_type = Column(String, nullable=True, index=True)
|
||||
user_agent = Column(String, nullable=True)
|
||||
version = Column(String, nullable=True)
|
||||
|
||||
# DMARC failure fields
|
||||
reported_domain = Column(String, nullable=True, index=True)
|
||||
source_ip = Column(String, nullable=True, index=True)
|
||||
auth_failure = Column(String, nullable=True, index=True)
|
||||
delivery_result = Column(String, nullable=True)
|
||||
arrival_date = Column(DateTime, nullable=True, index=True)
|
||||
authentication_results = Column(Text, nullable=True)
|
||||
|
||||
# Redacted original-message metadata. Never store original body content here.
|
||||
original_mail_from = Column(String, nullable=True)
|
||||
original_from = Column(String, nullable=True)
|
||||
original_to = Column(String, nullable=True)
|
||||
original_subject = Column(String, nullable=True)
|
||||
original_message_id = Column(String, nullable=True)
|
||||
original_date = Column(String, nullable=True)
|
||||
|
||||
# Sanitized parser details for operators/debugging.
|
||||
feedback_headers = Column(Text, nullable=True)
|
||||
processed_at = Column(DateTime, default=datetime.utcnow, index=True)
|
||||
|
||||
domain = relationship("Domain", back_populates="forensic_reports")
|
||||
|
||||
__table_args__ = (
|
||||
UniqueConstraint("report_id", name="uq_forensic_reports_report_id"),
|
||||
Index("ix_forensic_reports_domain_arrival", "domain_id", "arrival_date"),
|
||||
Index("ix_forensic_reports_failure_source", "auth_failure", "source_ip"),
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<ForensicReport {self.report_id}>"
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
from email import message_from_bytes
|
||||
from email.message import Message
|
||||
from email.parser import Parser
|
||||
from email.utils import getaddresses, parsedate_to_datetime
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
|
||||
MAX_FORENSIC_REPORT_SIZE = 10 * 1024 * 1024
|
||||
|
||||
_EMAIL_RE = re.compile(r"\b([A-Z0-9._%+-]{1,64})@([A-Z0-9.-]+\.[A-Z]{2,})\b", re.IGNORECASE)
|
||||
_LONG_TOKEN_RE = re.compile(r"\b[A-Za-z0-9_./+=-]{28,}\b")
|
||||
|
||||
|
||||
def _coerce_text(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, bytes):
|
||||
return value.decode("utf-8", errors="replace")
|
||||
return str(value)
|
||||
|
||||
|
||||
def _clean(value: Any, *, redact: bool = True) -> str:
|
||||
text = " ".join(_coerce_text(value).replace("\r", " ").replace("\n", " ").split())
|
||||
return redact_text(text) if redact else text
|
||||
|
||||
|
||||
def redact_text(value: str) -> str:
|
||||
"""Redact email local-parts and long opaque tokens from forensic metadata."""
|
||||
|
||||
def _redact_email(match: re.Match[str]) -> str:
|
||||
local = match.group(1)
|
||||
domain = match.group(2)
|
||||
prefix = local[:2] if len(local) > 2 else local[:1]
|
||||
return f"{prefix}***@{domain.lower()}"
|
||||
|
||||
redacted = _EMAIL_RE.sub(_redact_email, value)
|
||||
return _LONG_TOKEN_RE.sub("[redacted-token]", redacted)
|
||||
|
||||
|
||||
def _header(msg: Optional[Message], name: str, *, redact: bool = True) -> str:
|
||||
return _clean(msg.get(name, "") if msg is not None else "", redact=redact)
|
||||
|
||||
|
||||
def _payload_text(part: Message) -> str:
|
||||
payload = part.get_payload(decode=True)
|
||||
if payload is not None:
|
||||
charset = part.get_content_charset() or "utf-8"
|
||||
return payload.decode(charset, errors="replace")
|
||||
payload_value = part.get_payload()
|
||||
if isinstance(payload_value, list):
|
||||
return ""
|
||||
return _coerce_text(payload_value)
|
||||
|
||||
|
||||
def _message_part_payload(part: Message) -> Optional[Message]:
|
||||
payload = part.get_payload()
|
||||
if isinstance(payload, list) and payload:
|
||||
return payload[0]
|
||||
return None
|
||||
|
||||
|
||||
def _parse_feedback_headers(text: str) -> Message:
|
||||
return Parser().parsestr(text or "")
|
||||
|
||||
|
||||
def _domain_from_address(value: str) -> str:
|
||||
addresses = getaddresses([value])
|
||||
for _, addr in addresses:
|
||||
if "@" in addr:
|
||||
return addr.rsplit("@", 1)[-1].lower()
|
||||
return ""
|
||||
|
||||
|
||||
def _parse_datetime(value: str) -> Optional[datetime]:
|
||||
if not value:
|
||||
return None
|
||||
try:
|
||||
parsed = parsedate_to_datetime(value)
|
||||
except (TypeError, ValueError, IndexError):
|
||||
return None
|
||||
if parsed is None:
|
||||
return None
|
||||
if parsed.tzinfo is not None:
|
||||
parsed = parsed.astimezone(timezone.utc)
|
||||
return parsed.replace(tzinfo=None)
|
||||
|
||||
|
||||
def _message_id_hash(value: str) -> str:
|
||||
cleaned = _clean(value, redact=False)
|
||||
if not cleaned:
|
||||
return ""
|
||||
return hashlib.sha256(cleaned.encode("utf-8")).hexdigest()[:24]
|
||||
|
||||
|
||||
class ForensicParser:
|
||||
"""Parse DMARC forensic/failure report emails without retaining message bodies."""
|
||||
|
||||
@staticmethod
|
||||
def is_forensic_report(msg: Message) -> bool:
|
||||
if msg.get_content_type() == "multipart/report":
|
||||
report_type = (msg.get_param("report-type") or "").lower()
|
||||
if report_type == "feedback-report":
|
||||
return True
|
||||
|
||||
for part in msg.walk():
|
||||
content_type = part.get_content_type().lower()
|
||||
if content_type == "message/feedback-report":
|
||||
return True
|
||||
if content_type == "text/rfc822-headers" and "dmarc" in _payload_text(part).lower():
|
||||
return True
|
||||
|
||||
subject = _header(msg, "Subject", redact=False).lower()
|
||||
return "dmarc" in subject and any(
|
||||
term in subject for term in ("failure", "forensic", "ruf")
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def parse_bytes(
|
||||
cls, content: bytes, *, message_id_hint: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
if len(content) > MAX_FORENSIC_REPORT_SIZE:
|
||||
raise ValueError("Forensic report is too large")
|
||||
if not content:
|
||||
raise ValueError("Forensic report is empty")
|
||||
|
||||
msg = message_from_bytes(content)
|
||||
if not cls.is_forensic_report(msg):
|
||||
raise ValueError("Email is not a DMARC forensic report")
|
||||
|
||||
feedback = None
|
||||
original_headers = None
|
||||
|
||||
for part in msg.walk():
|
||||
content_type = part.get_content_type().lower()
|
||||
if content_type == "message/feedback-report":
|
||||
feedback = _message_part_payload(part) or _parse_feedback_headers(
|
||||
_payload_text(part)
|
||||
)
|
||||
elif content_type == "text/rfc822-headers":
|
||||
original_headers = _parse_feedback_headers(_payload_text(part))
|
||||
elif content_type == "message/rfc822" and original_headers is None:
|
||||
original_headers = _message_part_payload(part)
|
||||
|
||||
feedback = feedback or msg
|
||||
reported_domain = (
|
||||
_header(feedback, "Reported-Domain", redact=False)
|
||||
or _header(feedback, "DKIM-Domain", redact=False)
|
||||
or _domain_from_address(_header(feedback, "Original-Mail-From", redact=False))
|
||||
or _domain_from_address(_header(original_headers, "From", redact=False))
|
||||
).lower()
|
||||
source_ip = _header(feedback, "Source-IP", redact=False)
|
||||
auth_failure = _header(feedback, "Auth-Failure", redact=False)
|
||||
original_message_id = _header(original_headers, "Message-ID", redact=False)
|
||||
top_message_id = _header(msg, "Message-ID", redact=False)
|
||||
|
||||
report_id = (
|
||||
_clean(message_id_hint, redact=False)
|
||||
or _message_id_hash(top_message_id)
|
||||
or _message_id_hash(original_message_id)
|
||||
or hashlib.sha256(content).hexdigest()[:24]
|
||||
)
|
||||
if not report_id.startswith("ruf-"):
|
||||
report_id = f"ruf-{report_id}"
|
||||
|
||||
source_email = _header(msg, "From")
|
||||
arrival_date = _parse_datetime(_header(feedback, "Arrival-Date", redact=False))
|
||||
|
||||
details = {
|
||||
"identity_alignment": _header(feedback, "Identity-Alignment", redact=False),
|
||||
"dkim_domain": _header(feedback, "DKIM-Domain", redact=False),
|
||||
"spf_dns": _header(feedback, "SPF-DNS", redact=False),
|
||||
"reported_uri": _header(feedback, "Reported-URI"),
|
||||
}
|
||||
details = {key: value for key, value in details.items() if value}
|
||||
|
||||
return {
|
||||
"report_id": report_id,
|
||||
"source_email": source_email,
|
||||
"feedback_type": _header(feedback, "Feedback-Type", redact=False) or "auth-failure",
|
||||
"user_agent": _header(feedback, "User-Agent"),
|
||||
"version": _header(feedback, "Version", redact=False),
|
||||
"reported_domain": reported_domain,
|
||||
"source_ip": source_ip,
|
||||
"auth_failure": auth_failure,
|
||||
"delivery_result": _header(feedback, "Delivery-Result", redact=False),
|
||||
"arrival_date": arrival_date,
|
||||
"authentication_results": _header(feedback, "Authentication-Results"),
|
||||
"original_mail_from": _header(feedback, "Original-Mail-From"),
|
||||
"original_from": _header(original_headers, "From"),
|
||||
"original_to": _header(original_headers, "To"),
|
||||
"original_subject": _header(original_headers, "Subject"),
|
||||
"original_message_id": _message_id_hash(original_message_id),
|
||||
"original_date": _header(original_headers, "Date", redact=False),
|
||||
"feedback_headers": json.dumps(details, sort_keys=True) if details else None,
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
import json
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.domain import Domain
|
||||
from app.models.report import ForensicReport
|
||||
from app.utils.domain_validator import DomainValidationError, validate_domain
|
||||
|
||||
|
||||
def forensic_report_exists(db: Session, report_id: str) -> bool:
|
||||
"""Return True when a forensic report ID is already persisted."""
|
||||
if not str(report_id or "").strip():
|
||||
return False
|
||||
return (
|
||||
db.query(ForensicReport.id).filter(ForensicReport.report_id == report_id).first()
|
||||
is not None
|
||||
)
|
||||
|
||||
|
||||
def _domain_for_report(db: Session, domain_name: Optional[str]) -> Optional[Domain]:
|
||||
if not domain_name:
|
||||
return None
|
||||
normalized = domain_name.lower().strip(".")
|
||||
is_valid, _, error_code = validate_domain(normalized, check_dns=False)
|
||||
if not is_valid and error_code != DomainValidationError.DNS_RESOLUTION_FAILED:
|
||||
return None
|
||||
|
||||
domain = db.query(Domain).filter(Domain.name == normalized).first()
|
||||
if domain is None:
|
||||
domain = Domain(name=normalized)
|
||||
db.add(domain)
|
||||
db.flush()
|
||||
return domain
|
||||
|
||||
|
||||
def save_forensic_report(db: Session, report: Dict[str, Any]) -> tuple[ForensicReport, bool]:
|
||||
"""Persist a parsed forensic report.
|
||||
|
||||
Returns ``(row, created)``. The caller owns the transaction and should
|
||||
commit after related work has completed.
|
||||
"""
|
||||
report_id = str(report.get("report_id") or "").strip()
|
||||
if not report_id:
|
||||
raise ValueError("Forensic report_id is required")
|
||||
|
||||
existing = db.query(ForensicReport).filter(ForensicReport.report_id == report_id).first()
|
||||
if existing is not None:
|
||||
return existing, False
|
||||
|
||||
domain = _domain_for_report(db, report.get("reported_domain"))
|
||||
feedback_headers = report.get("feedback_headers")
|
||||
if isinstance(feedback_headers, dict):
|
||||
feedback_headers = json.dumps(feedback_headers, sort_keys=True)
|
||||
|
||||
row = ForensicReport(
|
||||
domain_id=domain.id if domain else None,
|
||||
report_id=report_id,
|
||||
source_email=report.get("source_email"),
|
||||
feedback_type=report.get("feedback_type"),
|
||||
user_agent=report.get("user_agent"),
|
||||
version=report.get("version"),
|
||||
reported_domain=report.get("reported_domain"),
|
||||
source_ip=report.get("source_ip"),
|
||||
auth_failure=report.get("auth_failure"),
|
||||
delivery_result=report.get("delivery_result"),
|
||||
arrival_date=report.get("arrival_date"),
|
||||
authentication_results=report.get("authentication_results"),
|
||||
original_mail_from=report.get("original_mail_from"),
|
||||
original_from=report.get("original_from"),
|
||||
original_to=report.get("original_to"),
|
||||
original_subject=report.get("original_subject"),
|
||||
original_message_id=report.get("original_message_id"),
|
||||
original_date=report.get("original_date"),
|
||||
feedback_headers=feedback_headers,
|
||||
)
|
||||
db.add(row)
|
||||
try:
|
||||
db.flush()
|
||||
except IntegrityError:
|
||||
db.rollback()
|
||||
existing = db.query(ForensicReport).filter(ForensicReport.report_id == report_id).first()
|
||||
if existing is not None:
|
||||
return existing, False
|
||||
raise
|
||||
return row, True
|
||||
|
||||
|
||||
def forensic_report_to_dict(row: ForensicReport) -> Dict[str, Any]:
|
||||
"""Convert a forensic report row to an API-safe dictionary."""
|
||||
feedback_headers = {}
|
||||
if row.feedback_headers:
|
||||
try:
|
||||
feedback_headers = json.loads(row.feedback_headers)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
feedback_headers = {}
|
||||
|
||||
return {
|
||||
"id": row.id,
|
||||
"report_id": row.report_id,
|
||||
"domain": row.domain.name if row.domain else row.reported_domain,
|
||||
"reported_domain": row.reported_domain,
|
||||
"source_email": row.source_email,
|
||||
"feedback_type": row.feedback_type,
|
||||
"user_agent": row.user_agent,
|
||||
"version": row.version,
|
||||
"source_ip": row.source_ip,
|
||||
"auth_failure": row.auth_failure,
|
||||
"delivery_result": row.delivery_result,
|
||||
"arrival_date": row.arrival_date.isoformat() if row.arrival_date else None,
|
||||
"authentication_results": row.authentication_results,
|
||||
"original_mail_from": row.original_mail_from,
|
||||
"original_from": row.original_from,
|
||||
"original_to": row.original_to,
|
||||
"original_subject": row.original_subject,
|
||||
"original_message_id": row.original_message_id,
|
||||
"original_date": row.original_date,
|
||||
"feedback_headers": feedback_headers,
|
||||
"processed_at": row.processed_at.isoformat() if row.processed_at else None,
|
||||
}
|
||||
@@ -21,6 +21,8 @@ from googleapiclient.discovery import build
|
||||
from googleapiclient.errors import HttpError
|
||||
|
||||
from app.services.dmarc_parser import DMARCParser
|
||||
from app.services.forensic_parser import ForensicParser
|
||||
from app.services.forensic_persistence import forensic_report_exists, save_forensic_report
|
||||
from app.services.report_persistence import report_exists, save_parsed_report
|
||||
from app.services.report_store import ReportStore
|
||||
|
||||
@@ -47,8 +49,8 @@ GMAIL_SCOPES = [
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
DMARC_GMAIL_QUERY = (
|
||||
"has:attachment "
|
||||
"(filename:zip OR filename:gz OR filename:xml) "
|
||||
"((has:attachment (filename:zip OR filename:gz OR filename:xml)) "
|
||||
'OR subject:"DMARC failure" OR subject:"failure report" OR subject:forensic OR subject:ruf) '
|
||||
"(subject:dmarc OR subject:report OR subject:rua OR subject:submitter "
|
||||
'OR subject:"aggregate report" OR subject:"domain report" '
|
||||
'OR subject:"report domain" OR from:dmarc OR from:dmarc-noreply '
|
||||
@@ -57,6 +59,7 @@ DMARC_GMAIL_QUERY = (
|
||||
|
||||
# How many message results to fetch per API page
|
||||
_PAGE_SIZE = 100
|
||||
RETRYABLE_MESSAGE_FAILURE = -1
|
||||
|
||||
|
||||
class GmailClient:
|
||||
@@ -210,7 +213,9 @@ class GmailClient:
|
||||
"success": True,
|
||||
"processed": 0,
|
||||
"reports_found": 0,
|
||||
"forensic_reports_found": 0,
|
||||
"duplicate_reports": 0,
|
||||
"duplicate_forensic_reports": 0,
|
||||
"new_domains": [],
|
||||
"errors": [],
|
||||
"new_ingested_ids": [],
|
||||
@@ -243,11 +248,9 @@ class GmailClient:
|
||||
|
||||
stats["processed"] += 1
|
||||
found = self._process_message(service, msg_id, stats)
|
||||
if found > 0:
|
||||
stats["new_ingested_ids"].append(msg_id)
|
||||
self.already_ingested_ids.append(msg_id)
|
||||
else:
|
||||
# Track it anyway so we don't re-examine it next run
|
||||
if found >= 0:
|
||||
# Track it even when no report is found so we don't re-examine
|
||||
# unrelated messages on every poll. Retryable failures return -1.
|
||||
stats["new_ingested_ids"].append(msg_id)
|
||||
self.already_ingested_ids.append(msg_id)
|
||||
|
||||
@@ -302,7 +305,9 @@ class GmailClient:
|
||||
@staticmethod
|
||||
def _append_detail(stats: dict, **detail: str) -> None:
|
||||
"""Append a compact attachment/message outcome to the import stats."""
|
||||
stats.setdefault("details", []).append({key: value for key, value in detail.items() if value})
|
||||
stats.setdefault("details", []).append(
|
||||
{key: value for key, value in detail.items() if value}
|
||||
)
|
||||
|
||||
def _process_message(self, service, msg_id: str, stats: dict) -> int:
|
||||
"""
|
||||
@@ -327,6 +332,8 @@ class GmailClient:
|
||||
|
||||
raw_bytes = base64.urlsafe_b64decode(msg_data.get("raw", ""))
|
||||
msg = email.message_from_bytes(raw_bytes)
|
||||
if ForensicParser.is_forensic_report(msg):
|
||||
return self._process_forensic_message(raw_bytes, stats, message_id=msg_id)
|
||||
return self._process_attachments(msg, stats, message_id=msg_id)
|
||||
|
||||
@staticmethod
|
||||
@@ -370,6 +377,78 @@ class GmailClient:
|
||||
self.report_store.add_report(report)
|
||||
return True
|
||||
|
||||
def _process_forensic_message(
|
||||
self,
|
||||
raw_bytes: bytes,
|
||||
stats: dict,
|
||||
message_id: Optional[str] = None,
|
||||
) -> int:
|
||||
"""Parse and persist one DMARC forensic report message."""
|
||||
try:
|
||||
report = ForensicParser.parse_bytes(raw_bytes, message_id_hint=message_id)
|
||||
report_id = str(report.get("report_id", ""))
|
||||
domain = str(report.get("reported_domain") or "unknown")
|
||||
|
||||
if self.db is None:
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="skipped",
|
||||
reason="forensic_report_requires_database",
|
||||
message_id=message_id,
|
||||
domain=domain,
|
||||
report_id=report_id,
|
||||
)
|
||||
return RETRYABLE_MESSAGE_FAILURE
|
||||
|
||||
if forensic_report_exists(self.db, report_id):
|
||||
stats["duplicate_forensic_reports"] = stats.get("duplicate_forensic_reports", 0) + 1
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="duplicate",
|
||||
reason="duplicate_forensic_report",
|
||||
message_id=message_id,
|
||||
domain=domain,
|
||||
report_id=report_id,
|
||||
)
|
||||
return 0
|
||||
|
||||
_row, created = save_forensic_report(self.db, report)
|
||||
if not created:
|
||||
stats["duplicate_forensic_reports"] = stats.get("duplicate_forensic_reports", 0) + 1
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="duplicate",
|
||||
reason="duplicate_forensic_report",
|
||||
message_id=message_id,
|
||||
domain=domain,
|
||||
report_id=report_id,
|
||||
)
|
||||
return 0
|
||||
|
||||
stats["forensic_reports_found"] = stats.get("forensic_reports_found", 0) + 1
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="imported",
|
||||
reason="forensic_report",
|
||||
message_id=message_id,
|
||||
domain=domain,
|
||||
report_id=report_id,
|
||||
)
|
||||
return 1
|
||||
except Exception as exc: # pylint: disable=broad-exception-caught
|
||||
logger.error("Failed to parse Gmail forensic report %s: %s", message_id, exc)
|
||||
stats.setdefault("errors", []).append(
|
||||
f"Failed to parse forensic report {message_id}: {exc}"
|
||||
)
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="error",
|
||||
reason="forensic_parse_failed",
|
||||
message_id=message_id,
|
||||
error=str(exc),
|
||||
)
|
||||
return RETRYABLE_MESSAGE_FAILURE
|
||||
|
||||
def _process_attachments(
|
||||
self,
|
||||
msg: email.message.Message,
|
||||
|
||||
@@ -7,6 +7,8 @@ from typing import Any, Dict, Optional, Tuple
|
||||
|
||||
from app.core.config import get_settings
|
||||
from app.services.dmarc_parser import DMARCParser
|
||||
from app.services.forensic_parser import ForensicParser
|
||||
from app.services.forensic_persistence import forensic_report_exists, save_forensic_report
|
||||
from app.services.report_persistence import report_exists, save_parsed_report
|
||||
from app.services.report_store import ReportStore
|
||||
|
||||
@@ -202,6 +204,19 @@ class IMAPClient:
|
||||
raw_email = msg_data[0][1]
|
||||
msg = email.message_from_bytes(raw_email)
|
||||
|
||||
if ForensicParser.is_forensic_report(msg):
|
||||
imported = self._process_forensic_email(
|
||||
raw_email,
|
||||
stats=stats,
|
||||
message_id=message_id,
|
||||
)
|
||||
mail.store(email_id, "+FLAGS", "\\Seen")
|
||||
if self.delete_emails and imported:
|
||||
mail.store(email_id, "+FLAGS", "\\Deleted")
|
||||
stats["deleted"] = stats.get("deleted", 0) + 1
|
||||
stats["processed"] += 1
|
||||
return
|
||||
|
||||
if self._is_dmarc_report_email(msg):
|
||||
reports_found = self._process_attachments(msg, stats, message_id=message_id)
|
||||
stats["reports_found"] += reports_found
|
||||
@@ -243,8 +258,10 @@ class IMAPClient:
|
||||
"success": True,
|
||||
"processed": 0,
|
||||
"reports_found": 0,
|
||||
"forensic_reports_found": 0,
|
||||
"deleted": 0,
|
||||
"duplicate_reports": 0,
|
||||
"duplicate_forensic_reports": 0,
|
||||
"new_domains": [],
|
||||
"errors": [],
|
||||
"details": [],
|
||||
@@ -472,6 +489,84 @@ class IMAPClient:
|
||||
)
|
||||
return True
|
||||
|
||||
def _process_forensic_email(
|
||||
self,
|
||||
raw_email: bytes,
|
||||
*,
|
||||
stats: Optional[Dict[str, Any]],
|
||||
message_id: Optional[str],
|
||||
) -> bool:
|
||||
try:
|
||||
report = ForensicParser.parse_bytes(raw_email)
|
||||
report_id = str(report.get("report_id", ""))
|
||||
domain = str(report.get("reported_domain") or "unknown")
|
||||
|
||||
if self.db is not None and forensic_report_exists(self.db, report_id):
|
||||
if stats is not None:
|
||||
stats["duplicate_forensic_reports"] = (
|
||||
stats.get("duplicate_forensic_reports", 0) + 1
|
||||
)
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="duplicate",
|
||||
reason="duplicate_forensic_report",
|
||||
message_id=message_id,
|
||||
domain=domain,
|
||||
report_id=report_id,
|
||||
)
|
||||
return False
|
||||
|
||||
if self.db is None:
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="skipped",
|
||||
reason="forensic_report_requires_database",
|
||||
message_id=message_id,
|
||||
domain=domain,
|
||||
report_id=report_id,
|
||||
)
|
||||
return False
|
||||
|
||||
_row, created = save_forensic_report(self.db, report)
|
||||
if created:
|
||||
if stats is not None:
|
||||
stats["forensic_reports_found"] = stats.get("forensic_reports_found", 0) + 1
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="imported",
|
||||
reason="forensic_report",
|
||||
message_id=message_id,
|
||||
domain=domain,
|
||||
report_id=report_id,
|
||||
)
|
||||
return True
|
||||
|
||||
if stats is not None:
|
||||
stats["duplicate_forensic_reports"] = stats.get("duplicate_forensic_reports", 0) + 1
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="duplicate",
|
||||
reason="duplicate_forensic_report",
|
||||
message_id=message_id,
|
||||
domain=domain,
|
||||
report_id=report_id,
|
||||
)
|
||||
return False
|
||||
except Exception as exc: # pylint: disable=broad-exception-caught
|
||||
logger.error("Error processing forensic report email %s: %s", message_id, exc)
|
||||
if stats is not None:
|
||||
stats.setdefault("errors", []).append(
|
||||
f"Failed to parse forensic report {message_id}: {exc}"
|
||||
)
|
||||
self._append_detail(
|
||||
stats,
|
||||
status="error",
|
||||
reason="forensic_parse_failed",
|
||||
message_id=message_id,
|
||||
error=str(exc),
|
||||
)
|
||||
return False
|
||||
|
||||
def _process_dmarc_attachment(
|
||||
self,
|
||||
part: email.message.Message,
|
||||
|
||||
@@ -0,0 +1,249 @@
|
||||
import email
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from app.services import forensic_parser as forensic_parser_module
|
||||
from app.services.forensic_parser import (
|
||||
ForensicParser,
|
||||
MAX_FORENSIC_REPORT_SIZE,
|
||||
_coerce_text,
|
||||
_domain_from_address,
|
||||
_message_part_payload,
|
||||
_payload_text,
|
||||
)
|
||||
|
||||
|
||||
SAMPLE_FORENSIC_EMAIL = b"""\
|
||||
From: DMARC Reporter <dmarc-reports@example.net>
|
||||
To: postmaster@example.com
|
||||
Subject: DMARC Failure Report for example.com
|
||||
Message-ID: <report-1@example.net>
|
||||
MIME-Version: 1.0
|
||||
Content-Type: multipart/report; report-type=feedback-report; boundary="ruf-boundary"
|
||||
|
||||
--ruf-boundary
|
||||
Content-Type: text/plain; charset=utf-8
|
||||
|
||||
This is a DMARC failure report.
|
||||
|
||||
--ruf-boundary
|
||||
Content-Type: message/feedback-report
|
||||
|
||||
Feedback-Type: auth-failure
|
||||
User-Agent: Example Reporter
|
||||
Version: 1
|
||||
Original-Mail-From: alice@example.com
|
||||
Arrival-Date: Fri, 22 May 2026 10:15:00 +0000
|
||||
Source-IP: 203.0.113.8
|
||||
Reported-Domain: example.com
|
||||
Authentication-Results: mx.example.net; dkim=fail header.d=example.com; spf=pass
|
||||
Auth-Failure: dkim
|
||||
Delivery-Result: reject
|
||||
|
||||
--ruf-boundary
|
||||
Content-Type: text/rfc822-headers
|
||||
|
||||
From: Alice Sender <alice@example.com>
|
||||
To: Bob Receiver <bob@example.net>
|
||||
Subject: Customer renewal token abcdefghijklmnopqrstuvwxyz123456
|
||||
Message-ID: <original-message@example.com>
|
||||
Date: Fri, 22 May 2026 10:14:55 +0000
|
||||
|
||||
--ruf-boundary--
|
||||
"""
|
||||
|
||||
|
||||
def test_detects_forensic_report_email():
|
||||
msg = email.message_from_bytes(SAMPLE_FORENSIC_EMAIL)
|
||||
|
||||
assert ForensicParser.is_forensic_report(msg) is True
|
||||
|
||||
|
||||
def test_parse_forensic_email_redacts_and_extracts_failure_fields():
|
||||
parsed = ForensicParser.parse_bytes(SAMPLE_FORENSIC_EMAIL)
|
||||
|
||||
assert parsed["report_id"].startswith("ruf-")
|
||||
assert parsed["reported_domain"] == "example.com"
|
||||
assert parsed["source_ip"] == "203.0.113.8"
|
||||
assert parsed["auth_failure"] == "dkim"
|
||||
assert parsed["delivery_result"] == "reject"
|
||||
assert parsed["arrival_date"].year == 2026
|
||||
assert parsed["original_mail_from"] == "al***@example.com"
|
||||
assert "al***@example.com" in parsed["original_from"]
|
||||
assert "bo***@example.net" in parsed["original_to"]
|
||||
assert "[redacted-token]" in parsed["original_subject"]
|
||||
assert parsed["original_message_id"]
|
||||
assert "original-message@example.com" not in parsed["original_message_id"]
|
||||
|
||||
|
||||
def test_non_forensic_email_is_rejected():
|
||||
content = b"From: sender@example.com\r\nSubject: hello\r\n\r\nplain email"
|
||||
|
||||
with pytest.raises(ValueError, match="forensic"):
|
||||
ForensicParser.parse_bytes(content)
|
||||
|
||||
|
||||
def test_parse_forensic_email_uses_explicit_message_hint():
|
||||
parsed = ForensicParser.parse_bytes(SAMPLE_FORENSIC_EMAIL, message_id_hint="gmail-message-1")
|
||||
|
||||
assert parsed["report_id"] == "ruf-gmail-message-1"
|
||||
|
||||
|
||||
def test_parse_forensic_email_handles_invalid_dates():
|
||||
content = SAMPLE_FORENSIC_EMAIL.replace(
|
||||
b"Arrival-Date: Fri, 22 May 2026 10:15:00 +0000",
|
||||
b"Arrival-Date: not a real date",
|
||||
)
|
||||
|
||||
parsed = ForensicParser.parse_bytes(content)
|
||||
|
||||
assert parsed["arrival_date"] is None
|
||||
|
||||
|
||||
def test_parse_forensic_email_normalizes_offset_dates_to_utc():
|
||||
content = SAMPLE_FORENSIC_EMAIL.replace(
|
||||
b"Arrival-Date: Fri, 22 May 2026 10:15:00 +0000",
|
||||
b"Arrival-Date: Fri, 22 May 2026 12:15:00 +0200",
|
||||
)
|
||||
|
||||
parsed = ForensicParser.parse_bytes(content)
|
||||
|
||||
assert parsed["arrival_date"].hour == 10
|
||||
assert parsed["arrival_date"].tzinfo is None
|
||||
|
||||
|
||||
def test_parse_forensic_email_falls_back_to_dkim_domain_and_content_hash():
|
||||
content = SAMPLE_FORENSIC_EMAIL.replace(b"Message-ID: <report-1@example.net>\n", b"")
|
||||
content = content.replace(b"Reported-Domain: example.com\n", b"DKIM-Domain: fallback.test\n")
|
||||
content = content.replace(b"Message-ID: <original-message@example.com>\n", b"")
|
||||
|
||||
parsed = ForensicParser.parse_bytes(content)
|
||||
|
||||
assert parsed["reported_domain"] == "fallback.test"
|
||||
assert parsed["report_id"].startswith("ruf-")
|
||||
assert parsed["original_message_id"] == ""
|
||||
assert parsed["feedback_headers"]
|
||||
|
||||
|
||||
def test_parse_forensic_email_extracts_message_rfc822_headers():
|
||||
content = b"""\
|
||||
From: DMARC Reporter <dmarc-reports@example.net>
|
||||
Subject: DMARC forensic report
|
||||
MIME-Version: 1.0
|
||||
Content-Type: multipart/report; report-type=feedback-report; boundary="ruf-boundary"
|
||||
|
||||
--ruf-boundary
|
||||
Content-Type: message/feedback-report
|
||||
|
||||
Feedback-Type: auth-failure
|
||||
Original-Mail-From: sender@fallback.test
|
||||
Source-IP: 203.0.113.9
|
||||
|
||||
--ruf-boundary
|
||||
Content-Type: message/rfc822
|
||||
|
||||
From: Carol <carol@fallback.test>
|
||||
To: Dave <dave@example.net>
|
||||
Subject: Forwarded failure sample
|
||||
Message-ID: <forwarded@example.test>
|
||||
|
||||
body is ignored
|
||||
--ruf-boundary--
|
||||
"""
|
||||
|
||||
parsed = ForensicParser.parse_bytes(content)
|
||||
|
||||
assert parsed["reported_domain"] == "fallback.test"
|
||||
assert "ca***@fallback.test" in parsed["original_from"]
|
||||
assert parsed["original_subject"] == "Forwarded failure sample"
|
||||
|
||||
|
||||
def test_is_forensic_report_detects_headers_and_subject_fallbacks():
|
||||
header_only = email.message_from_bytes(
|
||||
b"""\
|
||||
Subject: Delivery report
|
||||
MIME-Version: 1.0
|
||||
Content-Type: multipart/mixed; boundary="b"
|
||||
|
||||
--b
|
||||
Content-Type: text/rfc822-headers
|
||||
|
||||
Authentication-Results: mx; dmarc=fail
|
||||
|
||||
--b--
|
||||
"""
|
||||
)
|
||||
subject_only = email.message_from_bytes(b"Subject: DMARC RUF failure\r\n\r\nbody")
|
||||
|
||||
assert ForensicParser.is_forensic_report(header_only) is True
|
||||
assert ForensicParser.is_forensic_report(subject_only) is True
|
||||
|
||||
|
||||
def test_parse_forensic_email_rejects_empty_and_oversized_payloads():
|
||||
with pytest.raises(ValueError, match="empty"):
|
||||
ForensicParser.parse_bytes(b"")
|
||||
|
||||
with pytest.raises(ValueError, match="too large"):
|
||||
ForensicParser.parse_bytes(b"x" * (MAX_FORENSIC_REPORT_SIZE + 1))
|
||||
|
||||
|
||||
def test_forensic_parser_helper_fallbacks(monkeypatch):
|
||||
multipart = email.message_from_bytes(
|
||||
b"Subject: container\r\nContent-Type: multipart/mixed; boundary=x\r\n\r\n--x--"
|
||||
)
|
||||
message = email.message_from_bytes(b"Subject: plain\r\n\r\nbody")
|
||||
|
||||
assert _coerce_text(b"hello") == "hello"
|
||||
assert _payload_text(multipart) == ""
|
||||
assert _message_part_payload(message) is None
|
||||
assert _domain_from_address("not-an-address") == ""
|
||||
|
||||
part = MagicMock()
|
||||
part.get_payload.side_effect = [None, ["nested"]]
|
||||
assert _payload_text(part) == ""
|
||||
|
||||
part = MagicMock()
|
||||
part.get_payload.side_effect = [None, "plain text"]
|
||||
assert _payload_text(part) == "plain text"
|
||||
|
||||
monkeypatch.setattr(forensic_parser_module, "parsedate_to_datetime", lambda _value: None)
|
||||
assert forensic_parser_module._parse_datetime("Fri, 22 May 2026 10:15:00 +0000") is None
|
||||
|
||||
|
||||
def test_is_forensic_report_detects_feedback_part_without_report_container():
|
||||
msg = email.message_from_bytes(
|
||||
b"""\
|
||||
Subject: Delivery notice
|
||||
MIME-Version: 1.0
|
||||
Content-Type: multipart/mixed; boundary="b"
|
||||
|
||||
--b
|
||||
Content-Type: message/feedback-report
|
||||
|
||||
Feedback-Type: auth-failure
|
||||
|
||||
--b--
|
||||
"""
|
||||
)
|
||||
|
||||
assert ForensicParser.is_forensic_report(msg) is True
|
||||
|
||||
|
||||
def test_is_forensic_report_ignores_non_dmarc_header_parts():
|
||||
msg = email.message_from_bytes(
|
||||
b"""\
|
||||
Subject: DMARC forensic notice
|
||||
MIME-Version: 1.0
|
||||
Content-Type: multipart/mixed; boundary="b"
|
||||
|
||||
--b
|
||||
Content-Type: text/rfc822-headers
|
||||
|
||||
Authentication-Results: mx; spf=pass
|
||||
|
||||
--b--
|
||||
"""
|
||||
)
|
||||
|
||||
assert ForensicParser.is_forensic_report(msg) is True
|
||||
@@ -0,0 +1,187 @@
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from app.api.api_v1.endpoints import forensics as forensics_endpoint
|
||||
from app.models.report import ForensicReport
|
||||
from app.services.forensic_parser import ForensicParser
|
||||
from app.services.forensic_persistence import (
|
||||
forensic_report_exists,
|
||||
forensic_report_to_dict,
|
||||
save_forensic_report,
|
||||
)
|
||||
from app.tests.test_forensic_parser import SAMPLE_FORENSIC_EMAIL
|
||||
|
||||
|
||||
def test_upload_forensic_report_persists_redacted_metadata(authed_client, db_session):
|
||||
response = authed_client.post(
|
||||
"/api/v1/forensics/upload",
|
||||
files={"file": ("report.eml", SAMPLE_FORENSIC_EMAIL, "message/rfc822")},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert data["domain"] == "example.com"
|
||||
assert db_session.query(ForensicReport).count() == 1
|
||||
report = db_session.query(ForensicReport).one()
|
||||
assert report.original_mail_from == "al***@example.com"
|
||||
assert "original-message@example.com" not in report.original_message_id
|
||||
assert not report.original_message_id.startswith("<")
|
||||
|
||||
|
||||
def test_upload_forensic_report_rejects_duplicates(authed_client):
|
||||
files = {"file": ("report.eml", SAMPLE_FORENSIC_EMAIL, "message/rfc822")}
|
||||
assert authed_client.post("/api/v1/forensics/upload", files=files).status_code == 200
|
||||
|
||||
response = authed_client.post(
|
||||
"/api/v1/forensics/upload",
|
||||
files={"file": ("report.eml", SAMPLE_FORENSIC_EMAIL, "message/rfc822")},
|
||||
)
|
||||
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
def test_list_and_detail_forensic_reports(authed_client):
|
||||
authed_client.post(
|
||||
"/api/v1/forensics/upload",
|
||||
files={"file": ("report.eml", SAMPLE_FORENSIC_EMAIL, "message/rfc822")},
|
||||
)
|
||||
|
||||
list_response = authed_client.get("/api/v1/forensics?domain=example.com")
|
||||
assert list_response.status_code == 200
|
||||
list_data = list_response.json()
|
||||
assert list_data["total"] == 1
|
||||
item = list_data["reports"][0]
|
||||
assert item["source_ip"] == "203.0.113.8"
|
||||
assert item["auth_failure"] == "dkim"
|
||||
assert item["original_message_id"] != "<original-message@example.com>"
|
||||
|
||||
detail_response = authed_client.get(f"/api/v1/forensics/{item['id']}")
|
||||
assert detail_response.status_code == 200
|
||||
assert detail_response.json()["reported_domain"] == "example.com"
|
||||
|
||||
|
||||
def test_upload_forensic_report_rejects_aggregate_xml(authed_client):
|
||||
response = authed_client.post(
|
||||
"/api/v1/forensics/upload",
|
||||
files={"file": ("report.xml", b"<feedback />", "application/xml")},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_upload_forensic_report_rejects_empty_file(authed_client):
|
||||
response = authed_client.post(
|
||||
"/api/v1/forensics/upload",
|
||||
files={"file": ("report.eml", b"", "message/rfc822")},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_validate_upload_rejects_missing_name_and_large_file():
|
||||
missing_name = type("Upload", (), {"filename": ""})()
|
||||
too_large = type("Upload", (), {"filename": "report.eml"})()
|
||||
|
||||
with pytest.raises(HTTPException) as missing:
|
||||
forensics_endpoint._validate_upload(missing_name, b"content")
|
||||
assert missing.value.status_code == 400
|
||||
|
||||
with pytest.raises(HTTPException) as large:
|
||||
forensics_endpoint._validate_upload(
|
||||
too_large,
|
||||
b"x" * (forensics_endpoint.MAX_FORENSIC_REPORT_SIZE + 1),
|
||||
)
|
||||
assert large.value.status_code == 413
|
||||
|
||||
|
||||
def test_upload_forensic_report_rejects_invalid_email(authed_client):
|
||||
response = authed_client.post(
|
||||
"/api/v1/forensics/upload",
|
||||
files={"file": ("report.eml", b"Subject: hello\r\n\r\nbody", "message/rfc822")},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_upload_forensic_report_handles_save_duplicate_race(authed_client, monkeypatch):
|
||||
parsed = ForensicParser.parse_bytes(SAMPLE_FORENSIC_EMAIL)
|
||||
row = ForensicReport(report_id=parsed["report_id"], reported_domain=parsed["reported_domain"])
|
||||
monkeypatch.setattr(forensics_endpoint, "forensic_report_exists", lambda *_args: False)
|
||||
monkeypatch.setattr(
|
||||
forensics_endpoint,
|
||||
"save_forensic_report",
|
||||
lambda *_args: (row, False),
|
||||
)
|
||||
|
||||
response = authed_client.post(
|
||||
"/api/v1/forensics/upload",
|
||||
files={"file": ("report.eml", SAMPLE_FORENSIC_EMAIL, "message/rfc822")},
|
||||
)
|
||||
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
def test_upload_forensic_report_unexpected_error_returns_500(authed_client, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
forensics_endpoint.ForensicParser,
|
||||
"parse_bytes",
|
||||
lambda _content: (_ for _ in ()).throw(RuntimeError("boom")),
|
||||
)
|
||||
|
||||
response = authed_client.post(
|
||||
"/api/v1/forensics/upload",
|
||||
files={"file": ("report.eml", SAMPLE_FORENSIC_EMAIL, "message/rfc822")},
|
||||
)
|
||||
|
||||
assert response.status_code == 500
|
||||
|
||||
|
||||
def test_forensic_detail_returns_404(authed_client):
|
||||
response = authed_client.get("/api/v1/forensics/999")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_save_forensic_report_duplicate_and_invalid_domain_paths(db_session):
|
||||
parsed = ForensicParser.parse_bytes(SAMPLE_FORENSIC_EMAIL)
|
||||
parsed["feedback_headers"] = {"identity_alignment": "dkim"}
|
||||
first, created = save_forensic_report(db_session, parsed)
|
||||
second, duplicate_created = save_forensic_report(db_session, parsed)
|
||||
|
||||
assert created is True
|
||||
assert duplicate_created is False
|
||||
assert first.id == second.id
|
||||
assert forensic_report_exists(db_session, parsed["report_id"]) is True
|
||||
assert forensic_report_exists(db_session, "") is False
|
||||
assert forensic_report_to_dict(first)["feedback_headers"] == {"identity_alignment": "dkim"}
|
||||
|
||||
first.feedback_headers = "{not-json"
|
||||
assert forensic_report_to_dict(first)["feedback_headers"] == {}
|
||||
|
||||
missing_id = dict(parsed)
|
||||
missing_id["report_id"] = " "
|
||||
with pytest.raises(ValueError, match="report_id"):
|
||||
save_forensic_report(db_session, missing_id)
|
||||
|
||||
invalid = dict(parsed)
|
||||
invalid["report_id"] = "ruf-invalid-domain"
|
||||
invalid["reported_domain"] = "bad domain"
|
||||
row, invalid_created = save_forensic_report(db_session, invalid)
|
||||
assert invalid_created is True
|
||||
assert row.domain_id is None
|
||||
|
||||
|
||||
def test_save_forensic_report_reraises_unexpected_integrity_errors(db_session, monkeypatch):
|
||||
parsed = ForensicParser.parse_bytes(SAMPLE_FORENSIC_EMAIL)
|
||||
parsed["report_id"] = "ruf-race-without-existing-row"
|
||||
parsed["reported_domain"] = ""
|
||||
|
||||
def raise_integrity_error():
|
||||
raise IntegrityError("insert", {}, Exception("unique"))
|
||||
|
||||
monkeypatch.setattr(db_session, "flush", raise_integrity_error)
|
||||
|
||||
with pytest.raises(IntegrityError):
|
||||
save_forensic_report(db_session, parsed)
|
||||
@@ -19,10 +19,11 @@ from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.models.report import DMARCReport
|
||||
from app.services.gmail_client import GmailClient
|
||||
from app.models.report import DMARCReport, ForensicReport
|
||||
from app.services.gmail_client import GmailClient, RETRYABLE_MESSAGE_FAILURE
|
||||
from app.services.report_store import ReportStore
|
||||
from app.tests.test_data import SAMPLE_XML
|
||||
from app.tests.test_forensic_parser import SAMPLE_FORENSIC_EMAIL
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
@@ -442,6 +443,88 @@ class TestProcessMessage:
|
||||
assert stats["details"][0]["status"] == "error"
|
||||
assert stats["details"][0]["message_id"] == "bad-id"
|
||||
|
||||
def test_forensic_report_is_processed_separately(self, db_session):
|
||||
client = _make_client(db=db_session)
|
||||
service = MagicMock()
|
||||
service.users.return_value.messages.return_value.get.return_value.execute.return_value = {
|
||||
"raw": _b64_raw(SAMPLE_FORENSIC_EMAIL)
|
||||
}
|
||||
|
||||
stats = {
|
||||
"reports_found": 0,
|
||||
"forensic_reports_found": 0,
|
||||
"duplicate_forensic_reports": 0,
|
||||
"errors": [],
|
||||
}
|
||||
count = client._process_message(service, "msg-forensic", stats)
|
||||
|
||||
assert count == 1
|
||||
assert stats["reports_found"] == 0
|
||||
assert stats["forensic_reports_found"] == 1
|
||||
assert stats["details"][0]["reason"] == "forensic_report"
|
||||
assert db_session.query(DMARCReport).count() == 0
|
||||
assert db_session.query(ForensicReport).count() == 1
|
||||
|
||||
def test_forensic_report_without_database_is_skipped(self):
|
||||
client = _make_client()
|
||||
stats = {"forensic_reports_found": 0, "duplicate_forensic_reports": 0, "errors": []}
|
||||
|
||||
imported = client._process_forensic_message(
|
||||
SAMPLE_FORENSIC_EMAIL,
|
||||
stats,
|
||||
message_id="msg-forensic",
|
||||
)
|
||||
|
||||
assert imported == RETRYABLE_MESSAGE_FAILURE
|
||||
assert stats["details"][0]["reason"] == "forensic_report_requires_database"
|
||||
|
||||
def test_duplicate_forensic_report_is_skipped(self, db_session):
|
||||
client = _make_client(db=db_session)
|
||||
stats = {"forensic_reports_found": 0, "duplicate_forensic_reports": 0, "errors": []}
|
||||
|
||||
assert client._process_forensic_message(SAMPLE_FORENSIC_EMAIL, stats, message_id="msg-1")
|
||||
imported = client._process_forensic_message(
|
||||
SAMPLE_FORENSIC_EMAIL,
|
||||
stats,
|
||||
message_id="msg-1",
|
||||
)
|
||||
|
||||
assert imported == 0
|
||||
assert stats["duplicate_forensic_reports"] == 1
|
||||
assert stats["details"][1]["status"] == "duplicate"
|
||||
|
||||
def test_forensic_save_duplicate_result_is_skipped(self, db_session):
|
||||
client = _make_client(db=db_session)
|
||||
stats = {"forensic_reports_found": 0, "duplicate_forensic_reports": 0, "errors": []}
|
||||
|
||||
with (
|
||||
patch("app.services.gmail_client.forensic_report_exists", return_value=False),
|
||||
patch("app.services.gmail_client.save_forensic_report", return_value=(None, False)),
|
||||
):
|
||||
imported = client._process_forensic_message(
|
||||
SAMPLE_FORENSIC_EMAIL,
|
||||
stats,
|
||||
message_id="msg-1",
|
||||
)
|
||||
|
||||
assert imported == 0
|
||||
assert stats["duplicate_forensic_reports"] == 1
|
||||
assert stats["details"][0]["status"] == "duplicate"
|
||||
|
||||
def test_forensic_parse_error_adds_error_detail(self, db_session):
|
||||
client = _make_client(db=db_session)
|
||||
stats = {"errors": []}
|
||||
|
||||
imported = client._process_forensic_message(
|
||||
b"Subject: not forensic\r\n\r\nbody",
|
||||
stats,
|
||||
message_id="bad",
|
||||
)
|
||||
|
||||
assert imported == RETRYABLE_MESSAGE_FAILURE
|
||||
assert stats["errors"]
|
||||
assert stats["details"][0]["reason"] == "forensic_parse_failed"
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# _process_attachments
|
||||
@@ -678,6 +761,19 @@ class TestFetchReports:
|
||||
assert "id1" in result["new_ingested_ids"]
|
||||
assert "id2" in result["new_ingested_ids"]
|
||||
|
||||
def test_retryable_processing_failure_is_not_marked_ingested(self):
|
||||
client = _make_client()
|
||||
mock_service = MagicMock()
|
||||
with (
|
||||
patch.object(client, "_build_service", return_value=mock_service),
|
||||
patch.object(client, "_list_dmarc_message_ids", return_value=["id1"]),
|
||||
patch.object(client, "_process_message", return_value=RETRYABLE_MESSAGE_FAILURE),
|
||||
):
|
||||
result = client.fetch_reports()
|
||||
|
||||
assert result["new_ingested_ids"] == []
|
||||
assert client.already_ingested_ids == []
|
||||
|
||||
def test_reports_new_domains(self):
|
||||
"""fetch_reports should report domains that appear after ingestion."""
|
||||
from app.services.report_store import ReportStore
|
||||
@@ -717,11 +813,17 @@ class TestFetchReports:
|
||||
|
||||
|
||||
class TestIngestedIdHelpers:
|
||||
def test_load_empty_json_text_returns_empty_list(self):
|
||||
assert GmailClient.load_ingested_ids("") == []
|
||||
|
||||
def test_load_non_list_json_returns_no_error(self):
|
||||
# Valid JSON but not a list – should gracefully not raise
|
||||
result = GmailClient.load_ingested_ids('{"key": "value"}')
|
||||
assert result is not None # no crash
|
||||
|
||||
def test_load_invalid_json_returns_empty_list(self):
|
||||
assert GmailClient.load_ingested_ids("{not-json") == []
|
||||
|
||||
def test_dump_preserves_order(self):
|
||||
ids = ["z", "a", "m"]
|
||||
dumped = GmailClient.dump_ingested_ids(ids)
|
||||
|
||||
@@ -17,9 +17,10 @@ from zipfile import ZipFile
|
||||
|
||||
import pytest
|
||||
|
||||
from app.models.report import DMARCReport
|
||||
from app.models.report import DMARCReport, ForensicReport
|
||||
from app.services.imap_client import IMAPClient
|
||||
from app.services.report_store import ReportStore
|
||||
from app.tests.test_forensic_parser import SAMPLE_FORENSIC_EMAIL
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
@@ -587,7 +588,7 @@ class TestProcessAttachments:
|
||||
|
||||
|
||||
class TestProcessSingleEmail:
|
||||
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",
|
||||
@@ -595,7 +596,7 @@ class TestProcessSingleEmail:
|
||||
IMAP_USERNAME="u",
|
||||
IMAP_PASSWORD="p",
|
||||
)
|
||||
return IMAPClient()
|
||||
return IMAPClient(db=db)
|
||||
|
||||
def test_processes_valid_dmarc_email(self):
|
||||
client = self._make_client()
|
||||
@@ -616,6 +617,88 @@ class TestProcessSingleEmail:
|
||||
assert stats["reports_found"] == 1
|
||||
assert stats["details"][0]["status"] == "imported"
|
||||
|
||||
def test_processes_forensic_report_without_aggregate_count(self, db_session):
|
||||
ReportStore.get_instance().clear()
|
||||
client = self._make_client(db=db_session)
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.fetch.return_value = ("OK", [(b"1", SAMPLE_FORENSIC_EMAIL)])
|
||||
mock_mail.store.return_value = ("OK", None)
|
||||
|
||||
stats = {
|
||||
"processed": 0,
|
||||
"reports_found": 0,
|
||||
"forensic_reports_found": 0,
|
||||
"deleted": 0,
|
||||
"errors": [],
|
||||
}
|
||||
client._process_single_email(mock_mail, b"1", stats)
|
||||
|
||||
assert stats["processed"] == 1
|
||||
assert stats["reports_found"] == 0
|
||||
assert stats["forensic_reports_found"] == 1
|
||||
assert stats["details"][0]["reason"] == "forensic_report"
|
||||
assert db_session.query(DMARCReport).count() == 0
|
||||
assert db_session.query(ForensicReport).count() == 1
|
||||
assert ReportStore.get_instance().get_domains() == []
|
||||
|
||||
def test_forensic_report_without_database_is_skipped(self):
|
||||
client = self._make_client()
|
||||
|
||||
stats = {"processed": 0, "reports_found": 0, "errors": []}
|
||||
imported = client._process_forensic_email(
|
||||
SAMPLE_FORENSIC_EMAIL,
|
||||
stats=stats,
|
||||
message_id="msg-1",
|
||||
)
|
||||
|
||||
assert imported is False
|
||||
assert stats["details"][0]["reason"] == "forensic_report_requires_database"
|
||||
|
||||
def test_duplicate_forensic_report_adds_detail(self, db_session):
|
||||
client = self._make_client(db=db_session)
|
||||
stats = {"forensic_reports_found": 0, "duplicate_forensic_reports": 0, "errors": []}
|
||||
|
||||
assert client._process_forensic_email(SAMPLE_FORENSIC_EMAIL, stats=stats, message_id="1")
|
||||
imported = client._process_forensic_email(
|
||||
SAMPLE_FORENSIC_EMAIL, stats=stats, message_id="2"
|
||||
)
|
||||
|
||||
assert imported is False
|
||||
assert stats["duplicate_forensic_reports"] == 1
|
||||
assert stats["details"][1]["status"] == "duplicate"
|
||||
|
||||
def test_forensic_save_duplicate_result_adds_detail(self, db_session):
|
||||
client = self._make_client(db=db_session)
|
||||
stats = {"forensic_reports_found": 0, "duplicate_forensic_reports": 0, "errors": []}
|
||||
|
||||
with (
|
||||
patch("app.services.imap_client.forensic_report_exists", return_value=False),
|
||||
patch("app.services.imap_client.save_forensic_report", return_value=(None, False)),
|
||||
):
|
||||
imported = client._process_forensic_email(
|
||||
SAMPLE_FORENSIC_EMAIL,
|
||||
stats=stats,
|
||||
message_id="msg-1",
|
||||
)
|
||||
|
||||
assert imported is False
|
||||
assert stats["duplicate_forensic_reports"] == 1
|
||||
assert stats["details"][0]["status"] == "duplicate"
|
||||
|
||||
def test_forensic_parse_error_adds_error_detail(self, db_session):
|
||||
client = self._make_client(db=db_session)
|
||||
stats = {"errors": []}
|
||||
|
||||
imported = client._process_forensic_email(
|
||||
b"Subject: not forensic\r\n\r\nbody",
|
||||
stats=stats,
|
||||
message_id="bad",
|
||||
)
|
||||
|
||||
assert imported is False
|
||||
assert stats["errors"]
|
||||
assert stats["details"][0]["reason"] == "forensic_parse_failed"
|
||||
|
||||
def test_fetch_error_skips_email(self):
|
||||
client = self._make_client()
|
||||
mock_mail = MagicMock()
|
||||
|
||||
+7
-2
@@ -163,15 +163,20 @@ Exit criteria:
|
||||
|
||||
## Milestone 10: Forensic Report Support
|
||||
|
||||
Status: Backlog
|
||||
Status: In Progress
|
||||
|
||||
Goal: support DMARC RUF/forensic reports for individual failure investigation.
|
||||
|
||||
Planned:
|
||||
Delivered:
|
||||
- Detect forensic report messages.
|
||||
- Parse safe metadata from ARF/attached email formats.
|
||||
- Store minimal incident details with privacy controls.
|
||||
- Keep forensic reports out of aggregate report statistics and ReportStore rollups.
|
||||
- Expose authenticated forensic upload/list/detail APIs.
|
||||
|
||||
Planned:
|
||||
- Add a dedicated forensic report view.
|
||||
- Add configurable redaction controls and richer failure investigation workflows.
|
||||
|
||||
Exit criteria:
|
||||
- A security analyst can inspect individual failure reports without mixing them into aggregate statistics.
|
||||
|
||||
Reference in New Issue
Block a user