diff --git a/backend/app/tests/test_forensic_parser.py b/backend/app/tests/test_forensic_parser.py index 041317c..7f8519f 100644 --- a/backend/app/tests/test_forensic_parser.py +++ b/backend/app/tests/test_forensic_parser.py @@ -1,6 +1,8 @@ import email -from app.services.forensic_parser import ForensicParser +import pytest + +from app.services.forensic_parser import ForensicParser, MAX_FORENSIC_REPORT_SIZE SAMPLE_FORENSIC_EMAIL = b"""\ @@ -69,9 +71,97 @@ def test_parse_forensic_email_redacts_and_extracts_failure_fields(): def test_non_forensic_email_is_rejected(): content = b"From: sender@example.com\r\nSubject: hello\r\n\r\nplain email" - try: + with pytest.raises(ValueError, match="forensic"): ForensicParser.parse_bytes(content) - except ValueError as exc: - assert "forensic" in str(exc) - else: - raise AssertionError("Expected parser to reject non-forensic email") + + +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_falls_back_to_dkim_domain_and_content_hash(): + content = SAMPLE_FORENSIC_EMAIL.replace(b"Message-ID: \n", b"") + content = content.replace(b"Reported-Domain: example.com\n", b"DKIM-Domain: fallback.test\n") + content = content.replace(b"Message-ID: \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 +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 +To: Dave +Subject: Forwarded failure sample +Message-ID: + +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)) diff --git a/backend/app/tests/test_forensics_api.py b/backend/app/tests/test_forensics_api.py index 6e3058b..65f9068 100644 --- a/backend/app/tests/test_forensics_api.py +++ b/backend/app/tests/test_forensics_api.py @@ -1,4 +1,10 @@ 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 @@ -54,3 +60,48 @@ def test_upload_forensic_report_rejects_aggregate_xml(authed_client): ) 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_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_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"} + + 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 diff --git a/backend/app/tests/test_gmail_client.py b/backend/app/tests/test_gmail_client.py index 14e0eed..5c96273 100644 --- a/backend/app/tests/test_gmail_client.py +++ b/backend/app/tests/test_gmail_client.py @@ -465,6 +465,48 @@ class TestProcessMessage: 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 is False + 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 is False + assert stats["duplicate_forensic_reports"] == 1 + assert stats["details"][1]["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 is False + assert stats["errors"] + assert stats["details"][0]["reason"] == "forensic_parse_failed" + # =========================================================================== # _process_attachments diff --git a/backend/app/tests/test_imap_client.py b/backend/app/tests/test_imap_client.py index 4ab69ac..c95b06f 100644 --- a/backend/app/tests/test_imap_client.py +++ b/backend/app/tests/test_imap_client.py @@ -640,6 +640,46 @@ class TestProcessSingleEmail: 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_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()