diff --git a/TODO.md b/TODO.md index 8eda2a7..1fce31f 100644 --- a/TODO.md +++ b/TODO.md @@ -68,7 +68,7 @@ have no working implementation in the codebase yet. without affecting aggregate compliance statistics. Operators can configure forensic email-address and token redaction under Settings. - [x] Forensic report parsing -- [ ] Failure sample analysis +- [x] Failure sample analysis - [x] PII redaction options - [x] Detailed authentication failure views diff --git a/backend/app/api/api_v1/endpoints/forensics.py b/backend/app/api/api_v1/endpoints/forensics.py index 9b768e0..a805269 100644 --- a/backend/app/api/api_v1/endpoints/forensics.py +++ b/backend/app/api/api_v1/endpoints/forensics.py @@ -9,6 +9,7 @@ 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_analysis import analyze_forensic_report, summarize_forensic_samples from app.services.forensic_parser import ForensicParser, MAX_FORENSIC_REPORT_SIZE from app.services.forensic_persistence import ( forensic_report_exists, @@ -22,6 +23,45 @@ logger = logging.getLogger(__name__) router = APIRouter() +class ForensicSampleAnalysisResponse(BaseModel): + id: int + report_id: str + domain: Optional[str] = None + source_ip: Optional[str] = None + auth_failure: str + delivery_result: Optional[str] = None + priority: str + diagnosis: str + recommendations: List[str] = Field(default_factory=list) + signals: List[str] = Field(default_factory=list) + authentication_results: Dict[str, str] = Field(default_factory=dict) + dkim_domain: Optional[str] = None + mail_from_domain: Optional[str] = None + privacy_note: str + + +class ForensicAnalysisGroupResponse(BaseModel): + key: str + domain: str + source_ip: str + auth_failure: str + delivery_result: str + count: int + priority: str + latest_arrival: Optional[str] = None + diagnosis: str + recommendations: List[str] = Field(default_factory=list) + + +class ForensicAnalysisResponse(BaseModel): + total: int + priority_counts: Dict[str, int] = Field(default_factory=dict) + failure_counts: Dict[str, int] = Field(default_factory=dict) + result_counts: Dict[str, int] = Field(default_factory=dict) + groups: List[ForensicAnalysisGroupResponse] = Field(default_factory=list) + samples: List[ForensicSampleAnalysisResponse] = Field(default_factory=list) + + class ForensicReportResponse(BaseModel): id: int report_id: str @@ -44,6 +84,7 @@ class ForensicReportResponse(BaseModel): original_date: Optional[str] = None feedback_headers: Dict[str, Any] = Field(default_factory=dict) processed_at: Optional[str] = None + analysis: Optional[ForensicSampleAnalysisResponse] = None class ForensicListResponse(BaseModel): @@ -79,6 +120,35 @@ def _validate_upload(file: UploadFile, content: bytes) -> None: ) +def _filtered_forensic_query( + db: Session, + *, + domain: Optional[str] = None, + source_ip: Optional[str] = None, + auth_failure: Optional[str] = None, + delivery_result: Optional[str] = None, +): + 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) + ) + if source_ip: + query = query.filter(ForensicReport.source_ip == source_ip.strip()) + if auth_failure: + query = query.filter(ForensicReport.auth_failure == auth_failure.strip().lower()) + if delivery_result: + query = query.filter(ForensicReport.delivery_result == delivery_result.strip().lower()) + return query + + +def _response_for_row(row: ForensicReport, redaction_policy) -> ForensicReportResponse: + data = forensic_report_to_dict(row, redaction_policy=redaction_policy) + data["analysis"] = analyze_forensic_report(row) + return ForensicReportResponse(**data) + + @router.post("/upload", response_model=ForensicUploadResponse) async def upload_forensic_report( file: UploadFile = File(...), @@ -138,19 +208,13 @@ async def list_forensic_reports( _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) - ) - if source_ip: - query = query.filter(ForensicReport.source_ip == source_ip.strip()) - if auth_failure: - query = query.filter(ForensicReport.auth_failure == auth_failure.strip().lower()) - if delivery_result: - query = query.filter(ForensicReport.delivery_result == delivery_result.strip().lower()) - + query = _filtered_forensic_query( + db, + domain=domain, + source_ip=source_ip, + auth_failure=auth_failure, + delivery_result=delivery_result, + ) total = query.count() rows = ( query.order_by(ForensicReport.arrival_date.desc().nullslast(), ForensicReport.id.desc()) @@ -165,15 +229,36 @@ async def list_forensic_reports( page=page, page_size=page_size, total_pages=total_pages, - reports=[ - ForensicReportResponse( - **forensic_report_to_dict(row, redaction_policy=redaction_policy) - ) - for row in rows - ], + reports=[_response_for_row(row, redaction_policy) for row in rows], ) +@router.get("/analysis", response_model=ForensicAnalysisResponse) +async def analyze_forensic_reports( + domain: Optional[str] = Query(default=None), + source_ip: Optional[str] = Query(default=None), + auth_failure: Optional[str] = Query(default=None), + delivery_result: Optional[str] = Query(default=None), + page_size: int = Query(default=200, ge=1, le=500), + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +): + """Summarize stored forensic samples into operator investigation groups.""" + query = _filtered_forensic_query( + db, + domain=domain, + source_ip=source_ip, + auth_failure=auth_failure, + delivery_result=delivery_result, + ) + rows = ( + query.order_by(ForensicReport.arrival_date.desc().nullslast(), ForensicReport.id.desc()) + .limit(page_size) + .all() + ) + return ForensicAnalysisResponse(**summarize_forensic_samples(rows)) + + @router.get("/{report_id}", response_model=ForensicReportResponse) async def get_forensic_report( report_id: int, @@ -192,4 +277,4 @@ async def get_forensic_report( status_code=status.HTTP_404_NOT_FOUND, detail="Forensic report not found" ) redaction_policy = get_forensic_redaction_policy(db) - return ForensicReportResponse(**forensic_report_to_dict(row, redaction_policy=redaction_policy)) + return _response_for_row(row, redaction_policy) diff --git a/backend/app/services/forensic_analysis.py b/backend/app/services/forensic_analysis.py new file mode 100644 index 0000000..f709198 --- /dev/null +++ b/backend/app/services/forensic_analysis.py @@ -0,0 +1,271 @@ +import json +import re +from collections import Counter +from datetime import datetime +from typing import Any, Dict, Iterable, List, Optional, Tuple + +from app.models.report import ForensicReport + + +AUTH_RESULT_PATTERN = re.compile(r"\b(dkim|spf|dmarc)=([a-zA-Z0-9_-]+)", re.IGNORECASE) +HEADER_DOMAIN_PATTERN = re.compile(r"\bheader\.d=([^;\s]+)", re.IGNORECASE) +MAILFROM_DOMAIN_PATTERN = re.compile(r"\bsmtp\.mailfrom=([^;\s]+)", re.IGNORECASE) +PRIORITY_ORDER = {"high": 3, "medium": 2, "low": 1} + + +def _clean_value(value: Any) -> str: + return str(value or "").strip() + + +def _normalize(value: Any) -> str: + return _clean_value(value).lower() + + +def _feedback_headers(row: ForensicReport) -> Dict[str, Any]: + if not row.feedback_headers: + return {} + try: + parsed = json.loads(row.feedback_headers) + except (json.JSONDecodeError, TypeError): + return {} + return parsed if isinstance(parsed, dict) else {} + + +def _parse_authentication_results(value: str) -> Dict[str, str]: + results: Dict[str, str] = {} + for mechanism, result in AUTH_RESULT_PATTERN.findall(value or ""): + results[mechanism.lower()] = result.lower() + return results + + +def _first_match(pattern: re.Pattern[str], value: str) -> str: + match = pattern.search(value or "") + return match.group(1).lower().strip(".,") if match else "" + + +def _failure_kind(row: ForensicReport, auth_results: Dict[str, str]) -> str: + reported = _normalize(row.auth_failure) + if reported in {"dkim", "spf", "dmarc", "both"}: + return reported + failed = {name for name, result in auth_results.items() if result in {"fail", "softfail"}} + if {"dkim", "spf"}.issubset(failed): + return "both" + for mechanism in ("dmarc", "dkim", "spf"): + if mechanism in failed: + return mechanism + return reported or "unknown" + + +def _priority(row: ForensicReport, failure_kind: str) -> str: + delivery = _normalize(row.delivery_result) + if delivery in {"reject", "quarantine"}: + return "high" + if failure_kind in {"both", "dmarc"}: + return "high" + if failure_kind in {"dkim", "spf"}: + return "medium" + return "low" + + +def _diagnosis(failure_kind: str, auth_results: Dict[str, str], delivery_result: str) -> str: + delivery = _normalize(delivery_result) + rejected = delivery in {"reject", "quarantine"} + suffix = " The receiver enforced the failure." if rejected else "" + if failure_kind == "both": + return "Both DKIM and SPF failed, so DMARC could not find an aligned pass." + suffix + if failure_kind == "dmarc": + return "DMARC failed after the receiver evaluated DKIM and SPF alignment." + suffix + if failure_kind == "dkim": + if auth_results.get("spf") == "pass": + return "DKIM failed while SPF passed; focus on DKIM signing and alignment." + suffix + return "DKIM failed for the reported message sample." + suffix + if failure_kind == "spf": + if auth_results.get("dkim") == "pass": + return ( + "SPF failed while DKIM passed; focus on SPF authorization and alignment." + suffix + ) + return "SPF failed for the reported message sample." + suffix + return "The receiver reported an authentication failure, but did not include a clear mechanism." + + +def _recommendations( + failure_kind: str, + auth_results: Dict[str, str], + source_ip: str, + reported_domain: str, +) -> List[str]: + actions: List[str] = [] + if failure_kind in {"dkim", "both", "dmarc"}: + actions.append( + "Confirm the sending system signs mail with a DKIM domain aligned to the visible From domain." + ) + actions.append( + "Check recent DKIM key, selector, and canonicalization changes for this sender." + ) + if failure_kind in {"spf", "both", "dmarc"}: + actions.append( + "Verify the source IP or provider include is authorized in the domain SPF record." + ) + actions.append( + "Review forwarding paths, because forwarding commonly breaks SPF while preserving DKIM." + ) + if auth_results.get("spf") == "pass" and failure_kind == "dkim": + actions.append( + "If SPF is aligned and passing, this may be a DKIM-only repair rather than a sender authorization issue." + ) + if auth_results.get("dkim") == "pass" and failure_kind == "spf": + actions.append( + "If DKIM is aligned and passing, treat SPF repair as lower risk before changing DMARC policy." + ) + if source_ip: + actions.append( + f"Compare {source_ip} with known mail sources for {reported_domain or 'this domain'}." + ) + actions.append( + "Keep using redacted forensic metadata; do not import or retain message bodies for this investigation." + ) + return actions + + +def _signals( + row: ForensicReport, + feedback_headers: Dict[str, Any], + auth_results: Dict[str, str], + header_domain: str, + mailfrom_domain: str, +) -> List[str]: + signals = [] + if row.source_ip: + signals.append(f"Source IP: {row.source_ip}") + if row.reported_domain: + signals.append(f"Reported domain: {row.reported_domain}") + if row.auth_failure: + signals.append(f"Failure: {row.auth_failure}") + if row.delivery_result: + signals.append(f"Delivery result: {row.delivery_result}") + if header_domain: + signals.append(f"DKIM header domain: {header_domain}") + if mailfrom_domain: + signals.append(f"SPF mail-from domain: {mailfrom_domain}") + identity_alignment = _clean_value(feedback_headers.get("identity_alignment")) + if identity_alignment: + signals.append(f"Identity alignment: {identity_alignment}") + for mechanism, result in sorted(auth_results.items()): + signals.append(f"{mechanism.upper()} result: {result}") + return signals + + +def analyze_forensic_report(row: ForensicReport) -> Dict[str, Any]: + """Build a privacy-preserving operator analysis for one forensic sample.""" + feedback_headers = _feedback_headers(row) + auth_results = _parse_authentication_results(row.authentication_results or "") + header_domain = _first_match( + HEADER_DOMAIN_PATTERN, row.authentication_results or "" + ) or _normalize(feedback_headers.get("dkim_domain")) + mailfrom_domain = _first_match(MAILFROM_DOMAIN_PATTERN, row.authentication_results or "") + failure_kind = _failure_kind(row, auth_results) + priority = _priority(row, failure_kind) + reported_domain = _clean_value(row.reported_domain or (row.domain.name if row.domain else "")) + source_ip = _clean_value(row.source_ip) + + return { + "id": row.id, + "report_id": row.report_id, + "domain": reported_domain, + "source_ip": source_ip, + "auth_failure": failure_kind, + "delivery_result": _clean_value(row.delivery_result), + "priority": priority, + "diagnosis": _diagnosis(failure_kind, auth_results, row.delivery_result or ""), + "recommendations": _recommendations( + failure_kind, + auth_results, + source_ip, + reported_domain, + ), + "signals": _signals(row, feedback_headers, auth_results, header_domain, mailfrom_domain), + "authentication_results": auth_results, + "dkim_domain": header_domain, + "mail_from_domain": mailfrom_domain, + "privacy_note": "Analysis uses redacted headers and metadata only; message bodies are not stored.", + } + + +def _group_key(row: ForensicReport) -> Tuple[str, str, str, str]: + return ( + _clean_value(row.reported_domain or (row.domain.name if row.domain else "")) or "unknown", + _clean_value(row.source_ip) or "unknown", + _normalize(row.auth_failure) or "unknown", + _normalize(row.delivery_result) or "unknown", + ) + + +def _latest(left: Optional[datetime], right: Optional[datetime]) -> Optional[datetime]: + if left is None: + return right + if right is None: + return left + return max(left, right) + + +def summarize_forensic_samples(rows: Iterable[ForensicReport]) -> Dict[str, Any]: + """Summarize forensic samples into investigation groups and top examples.""" + reports = list(rows) + analyses = [analyze_forensic_report(row) for row in reports] + priority_counts = Counter(item["priority"] for item in analyses) + failure_counts = Counter(item["auth_failure"] for item in analyses) + result_counts = Counter(_normalize(row.delivery_result) or "unknown" for row in reports) + grouped: Dict[Tuple[str, str, str, str], Dict[str, Any]] = {} + + for row, analysis in zip(reports, analyses): + key = _group_key(row) + group = grouped.setdefault( + key, + { + "key": "|".join(key), + "domain": key[0], + "source_ip": key[1], + "auth_failure": analysis["auth_failure"], + "delivery_result": key[3], + "count": 0, + "priority": analysis["priority"], + "latest_arrival": None, + "diagnosis": analysis["diagnosis"], + "recommendations": analysis["recommendations"][:3], + }, + ) + group["count"] += 1 + group["latest_arrival"] = _latest( + group["latest_arrival"], row.arrival_date or row.processed_at + ) + if PRIORITY_ORDER[analysis["priority"]] > PRIORITY_ORDER[group["priority"]]: + group["priority"] = analysis["priority"] + group["diagnosis"] = analysis["diagnosis"] + group["recommendations"] = analysis["recommendations"][:3] + + groups = sorted( + grouped.values(), + key=lambda item: ( + PRIORITY_ORDER[item["priority"]], + item["count"], + item["latest_arrival"] or datetime.min, + ), + reverse=True, + ) + for group in groups: + if group["latest_arrival"] is not None: + group["latest_arrival"] = group["latest_arrival"].isoformat() + + samples = sorted( + analyses, + key=lambda item: (PRIORITY_ORDER[item["priority"]], item["id"] or 0), + reverse=True, + ) + return { + "total": len(reports), + "priority_counts": dict(priority_counts), + "failure_counts": dict(failure_counts), + "result_counts": dict(result_counts), + "groups": groups, + "samples": samples, + } diff --git a/backend/app/templates/forensic_report_detail.html b/backend/app/templates/forensic_report_detail.html index 117ccbe..558b7af 100644 --- a/backend/app/templates/forensic_report_detail.html +++ b/backend/app/templates/forensic_report_detail.html @@ -61,6 +61,42 @@ {% endcall %} + {% call card() %} + {% call card_header() %} +
+
+ {% call card_title() %}Failure Sample Analysis{% endcall %} + {% call card_description() %}Privacy-preserving diagnosis from the redacted failure sample{% endcall %} +
+ +
+ {% endcall %} + {% call card_content() %} +
+

+
+
+

Recommended Actions

+
    + +
+
+
+

Signals

+
    + +
+
+
+

+
+ {% endcall %} + {% endcall %} + {% call card() %} {% call card_header() %} {% call card_title() %}Message Identity{% endcall %} @@ -160,6 +196,11 @@ function forensicReportDetailApp(reportId) { labelize(value) { return String(value || '').replaceAll('_', ' ').replace(/\b\w/g, (char) => char.toUpperCase()); }, + priorityClass(priority) { + if (priority === 'high') return 'badge-error'; + if (priority === 'medium') return 'badge-warning'; + return 'badge-ghost'; + }, }; } diff --git a/backend/app/templates/forensic_reports.html b/backend/app/templates/forensic_reports.html index cba5097..a873fee 100644 --- a/backend/app/templates/forensic_reports.html +++ b/backend/app/templates/forensic_reports.html @@ -101,6 +101,49 @@ {% endcall %} + {% call card() %} + {% call card_header() %} +
+
+ {% call card_title() %}Sample Analysis{% endcall %} + {% call card_description() %}Grouped investigation hints from redacted forensic metadata{% endcall %} +
+
+ High + Medium +
+
+ {% endcall %} + {% call card_content() %} + +
+ +
+ {% endcall %} + {% endcall %} + {% call card() %} {% call card_header() %}
@@ -173,6 +216,7 @@ function forensicReportsApp() { uploadError: false, selectedFile: null, reports: [], + analysis: { groups: [], priority_counts: {}, failure_counts: {}, result_counts: {}, samples: [] }, domainOptions: [], total: 0, filters: { @@ -220,6 +264,7 @@ function forensicReportsApp() { const data = await response.json(); this.reports = data.reports || []; this.total = data.total || 0; + await this.fetchAnalysis(params); if (!this.filters.domain) { this.domainOptions = [...new Set(this.reports.map((report) => report.domain || report.reported_domain).filter(Boolean))].sort(); } @@ -229,6 +274,11 @@ function forensicReportsApp() { this.loading = false; } }, + async fetchAnalysis(params) { + const response = await fetch(`/api/v1/forensics/analysis?${params.toString()}`); + if (!response.ok) throw new Error('Unable to analyze forensic reports'); + this.analysis = await response.json(); + }, async uploadReport() { if (!this.selectedFile) return; this.uploading = true; @@ -262,6 +312,11 @@ function forensicReportsApp() { const date = new Date(value); return Number.isNaN(date.getTime()) ? value : date.toLocaleString(); }, + priorityClass(priority) { + if (priority === 'high') return 'badge-error'; + if (priority === 'medium') return 'badge-warning'; + return 'badge-ghost'; + }, }; } diff --git a/backend/app/tests/test_forensics_api.py b/backend/app/tests/test_forensics_api.py index dc06592..30aa9e9 100644 --- a/backend/app/tests/test_forensics_api.py +++ b/backend/app/tests/test_forensics_api.py @@ -90,6 +90,54 @@ def test_list_forensic_reports_filters_failure_fields(authed_client, db_session) assert data["reports"][0]["report_id"] == "ruf-spf-filter-test" +def test_forensic_analysis_groups_failure_samples(authed_client, db_session): + dkim = ForensicParser.parse_bytes(SAMPLE_FORENSIC_EMAIL) + spf = dict(dkim) + spf.update( + { + "report_id": "ruf-spf-analysis-test", + "reported_domain": "example.com", + "source_ip": "198.51.100.23", + "auth_failure": "spf", + "delivery_result": "quarantine", + "authentication_results": ( + "mx.example.net; dkim=pass header.d=example.com; " + "spf=fail smtp.mailfrom=example.com; dmarc=fail" + ), + } + ) + save_forensic_report(db_session, dkim) + save_forensic_report(db_session, spf) + db_session.commit() + + response = authed_client.get("/api/v1/forensics/analysis?domain=example.com") + + assert response.status_code == 200 + data = response.json() + assert data["total"] == 2 + assert data["priority_counts"]["high"] == 2 + assert data["failure_counts"]["dkim"] == 1 + assert data["failure_counts"]["spf"] == 1 + assert data["groups"][0]["priority"] == "high" + assert "redacted headers and metadata only" in data["samples"][0]["privacy_note"] + assert any("SPF" in action for action in data["samples"][0]["recommendations"]) + + +def test_forensic_report_responses_include_sample_analysis(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") + + assert list_response.status_code == 200 + item = list_response.json()["reports"][0] + assert item["analysis"]["priority"] == "high" + assert "DKIM" in item["analysis"]["diagnosis"] + assert item["original_subject"] not in item["analysis"]["signals"] + + def test_forensic_api_applies_configured_redaction_policy(authed_client, db_session): authed_client.post( "/api/v1/forensics/upload", @@ -216,6 +264,7 @@ def test_forensic_html_pages_render(): assert "Authentication Failures" in list_response.text assert detail_response.status_code == 200 assert "Forensic Investigation" in detail_response.text + assert "Failure Sample Analysis" in detail_response.text def test_save_forensic_report_duplicate_and_invalid_domain_paths(db_session): diff --git a/docs/milestones.md b/docs/milestones.md index 42d078d..6623e62 100644 --- a/docs/milestones.md +++ b/docs/milestones.md @@ -175,9 +175,7 @@ Delivered: - Keep forensic reports out of aggregate report statistics and ReportStore rollups. - Expose authenticated forensic upload/list/detail APIs. - Provide dedicated forensic report list/detail views for authentication failure investigation. - -Planned: -- Add richer failure investigation workflows. +- Add privacy-preserving failure sample analysis with grouped causes, priorities, signals, and recommended actions. Exit criteria: - A security analyst can inspect individual failure reports without mixing them into aggregate statistics.