diff options
Diffstat (limited to 'tests/test_report.py')
| -rw-r--r-- | tests/test_report.py | 672 |
1 files changed, 661 insertions, 11 deletions
diff --git a/tests/test_report.py b/tests/test_report.py index f414841..c8c39f8 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -1,8 +1,14 @@ import copy +import email +import email.policy +import hashlib import json +import re +import tempfile import unittest +from pathlib import Path -from abusectl import report +from abusectl import parse, report class Grouping(unittest.TestCase): @@ -634,6 +640,32 @@ class TextPart(unittest.TestCase): self.assertLessEqual(len(line), 72, line) self.assertIn(long_subject, report.unwrap(text)) + def test_an_identity_missing_a_name_or_org_still_builds(self): + """Each [reporter] key is individually skippable. + + config drops a skipped one rather than storing "", so a partial + identity is the normal shape here, not a malformed one. Subscripting + it raised KeyError on a case that had parsed perfectly. + """ + destination = report.email_destinations(MANIFEST["contacts"])[0] + + text = report.text_part(MANIFEST, destination, + {"email": "reporter@example.org"}) + self.assertIn("Reported by: <reporter@example.org>", text) + self.assertNotIn(",", report.unwrap(text).split("Reported by:")[1] + .splitlines()[0]) + + text = report.text_part(MANIFEST, destination, + {"name": "A Reporter", + "email": "reporter@example.org"}) + self.assertIn("Reported by: A Reporter <reporter@example.org>", text) + + text = report.text_part(MANIFEST, destination, + {"org": "Example Consulting", + "email": "reporter@example.org"}) + self.assertIn("Reported by: Example Consulting " + "<reporter@example.org>", text) + def test_a_long_reporter_identity_is_not_truncated(self): """The identity is the one thing disclosed deliberately. @@ -929,19 +961,51 @@ class FeedbackPart(unittest.TestCase): self.lookup = dict(self.fields) def test_the_three_rfc5965_required_fields_are_present(self): - self.assertEqual(self.lookup["Feedback-Type"], "abuse") + self.assertEqual(self.lookup["Feedback-Type"], "fraud") self.assertEqual(self.lookup["Version"], "1") self.assertTrue(self.lookup["User-Agent"].startswith("abusectl/")) - def test_the_xarf_report_type_is_phishing(self): - self.assertEqual(self.lookup["Report-Type"], "phishing") + def test_the_feedback_type_is_the_registered_value_naming_phishing(self): + """Checked against the IANA MARF registry, not a worked example. + + RFC 5965 registers fraud as "indicates some kind of fraud or + phishing activity" and abuse as "unsolicited email or some other + kind of email abuse". This tool reports phishing. + """ + self.assertEqual(self.lookup["Feedback-Type"], "fraud") + + def test_no_unregistered_field_is_emitted(self): + """Every name here appears in the IANA MARF registry. + + Report-Type: phishing used to be emitted and is NOT registered. It + was harmless, since RFC 5965 section 6 makes ignoring an unknown + field a MUST, but that same section requires an extension field be + registered, and a desk should be able to look up every field in a + document this tool sends. + """ + registered = { + "Arrival-Date", "Auth-Failure", "Authentication-Results", + "Delivery-Result", "DKIM-ADSP-DNS", "DKIM-Canonicalized-Body", + "DKIM-Canonicalized-Header", "DKIM-Domain", "DKIM-Identity", + "DKIM-Selector", "DKIM-Selector-DNS", "Feedback-Type", + "Identity-Alignment", "Incidents", "Original-Mail-From", + "Original-Rcpt-To", "Received-Date", "Reported-Domain", + "Reported-URI", "Reporting-MTA", "Source-IP", "Source-Port", + "SPF-DNS", "User-Agent", "Version", + } + # Source is x-arf's, deliberately: RFC 5965 has no field for "this + # host is the thing being reported", which is why the envelope + # carries x-arf fields at all. It is the one exception and is named + # here so a NEW unregistered field cannot slip in unnoticed. + emitted = {name for name, _ in self.fields} + self.assertEqual(emitted - registered - {"Source"}, set()) def test_source_is_the_primary_indicator(self): self.assertEqual(self.lookup["Source"], "203.0.113.42") self.assertEqual(self.lookup["Source-IP"], "203.0.113.42") def test_every_url_appears_as_a_reported_uri(self): - uris = [value for name, value in self.fields if name == "Reported-Uri"] + uris = [value for name, value in self.fields if name == "Reported-URI"] self.assertEqual( uris, ["http://login.sender.invalid/verify?id=REDACTED"] ) @@ -1016,11 +1080,11 @@ class FeedbackPart(unittest.TestCase): seen: dict[str, int] = {} for name, _ in fields: seen[name] = seen.get(name, 0) + 1 - repeatable = {"Reported-Uri", "Reported-Domain"} + repeatable = {"Reported-URI", "Reported-Domain"} for name, count in seen.items(): if name not in repeatable: self.assertEqual(count, 1, f"{name} appeared {count} times") - self.assertEqual(seen["Reported-Uri"], 2) + self.assertEqual(seen["Reported-URI"], 2) self.assertEqual(seen["Reported-Domain"], 2) def test_an_ipv6_indicator_fills_source_ip_too(self): @@ -1054,7 +1118,7 @@ class FeedbackPart(unittest.TestCase): self.assertNotIn("Source", lookup) self.assertNotIn("Source-IP", lookup) # The envelope is still well formed: a desk gets a valid part. - self.assertEqual(lookup["Feedback-Type"], "abuse") + self.assertEqual(lookup["Feedback-Type"], "fraud") self.assertEqual(lookup["Version"], "1") def test_a_type_this_module_does_not_place_is_not_invented_into_one(self): @@ -1101,7 +1165,7 @@ class FeedbackPart(unittest.TestCase): "target": "abuse@host.invalid", "body": None, "status": "pending"} lookup = dict(report.feedback_fields(MANIFEST, destination)) - self.assertEqual(lookup["Feedback-Type"], "abuse") + self.assertEqual(lookup["Feedback-Type"], "fraud") def test_arrival_date_is_not_taken_from_the_senders_date_header(self): """RFC 5965: Arrival-Date is when the generating ADMD's MTA received @@ -1157,7 +1221,7 @@ class FeedbackInjection(unittest.TestCase): destination = report.email_destinations(manifest["contacts"])[0] fields = report.feedback_fields(manifest, destination) for name, out in fields: - if name == "Reported-Uri": + if name == "Reported-URI": return out return None @@ -1354,7 +1418,7 @@ class FeedbackInjection(unittest.TestCase): fields = report.feedback_fields(manifest, destination) self.assertEqual( {name for name, _ in fields}, - {"Feedback-Type", "User-Agent", "Version", "Report-Type"}) + {"Feedback-Type", "User-Agent", "Version"}) def test_a_non_string_value_does_not_crash_the_report(self): """A manifest is edited by hand and JSON has numbers. @@ -1377,5 +1441,591 @@ class FeedbackInjection(unittest.TestCase): self.assertEqual(lookup["Source"], "42") +class Document(unittest.TestCase): + def setUp(self): + self.destination = report.email_destinations(MANIFEST["contacts"])[0] + self.raw = report.build(MANIFEST, self.destination, IDENTITY) + self.parsed = email.message_from_string( + self.raw, policy=email.policy.default + ) + + def test_it_is_a_feedback_report_with_three_parts(self): + self.assertEqual(self.parsed.get_content_type(), "multipart/report") + self.assertEqual(self.parsed.get_param("report-type"), + "feedback-report") + parts = list(self.parsed.iter_parts()) + self.assertEqual( + [part.get_content_type() for part in parts], + ["text/plain", "message/feedback-report", "text/rfc822-headers"], + ) + + def test_the_envelope_is_addressed_and_identified(self): + self.assertEqual(self.parsed["To"], "abuse@host.invalid") + self.assertIn("reporter@example.org", self.parsed["From"]) + self.assertTrue(self.parsed["Subject"]) + + def test_the_source_message_is_never_attached(self): + self.assertNotIn("message/rfc822", self.raw) + + def test_the_headers_part_carries_what_the_manifest_declared(self): + """The third part is the whitelisted headers, unmangled. + + Asserted by PARSING the part as headers rather than by looking for + substrings, because what a desk does with this part is parse it. + """ + headers = self._headers_of(self.parsed) + self.assertEqual(headers.keys(), ["From", "Subject", "Date"]) + self.assertEqual(headers["Subject"], + "Your account requires verification") + self.assertEqual(headers["Date"], "Mon, 07 Sep 2026 09:12:40 +0000") + # The From is asserted twice over: on the wire text, which is what + # is actually published, and on the address a desk would act on. + # A structured header re-renders the display name's quoting on + # read-back, so only the raw text pins what was written. + self.assertIn('From: "Example Bank" <phish@sender.invalid>', self.raw) + self.assertEqual([a.addr_spec for a in headers["From"].addresses], + ["phish@sender.invalid"]) + + @staticmethod + def _headers_of(parsed): + """Parse the third part's body the way a desk's parser would.""" + body = list(parsed.iter_parts())[2].get_content() + return email.message_from_string(body, policy=email.policy.default) + + +class DocumentHeaders(unittest.TestCase): + """The third part: what a hand-edited manifest can and cannot publish.""" + + def _build(self, headers): + manifest = copy.deepcopy(MANIFEST) + manifest["headers"] = headers + destination = report.email_destinations(manifest["contacts"])[0] + return report.build(manifest, destination, IDENTITY) + + def _part(self, raw, index=2): + parsed = email.message_from_string(raw, policy=email.policy.default) + return list(parsed.iter_parts())[index] + + def test_a_recipient_header_in_the_manifest_is_not_published(self): + """A manifest is a FILE THE USER EDITS, so it can carry a "To". + + parse.report_headers() would never produce one, which is exactly + why asserting on its output proves nothing: the property has to + survive a manifest nobody generated. The victim of getting this + wrong is the recipient, whose address reaches the abuse desk and + through it the attacker. + """ + raw = self._build([ + ("To", "victim@example.org"), + ("Cc", "other@example.org"), + ("Delivered-To", "victim@example.org"), + ("X-Original-To", "victim@example.org"), + ("Subject", "kept"), + ]) + body = self._part(raw).get_content() + self.assertNotIn("victim", raw) + self.assertNotIn("other@example.org", raw) + published = email.message_from_string(body, + policy=email.policy.default) + self.assertEqual(published.keys(), ["Subject"]) + + def test_a_newline_in_a_value_cannot_forge_a_header(self): + """Subject is attacker-controlled free text and is kept deliberately. + + Emitted verbatim it forges a header line in a part whose entire + content is read as headers. The value must survive intact and the + header list must not grow. + """ + raw = self._build([ + ("Subject", "lure\nFrom: forged@attacker.invalid"), + ]) + published = email.message_from_string( + self._part(raw).get_content(), policy=email.policy.default) + self.assertEqual(published.keys(), ["Subject"]) + self.assertEqual(published["Subject"], + "lure From: forged@attacker.invalid") + self.assertEqual(published["From"], None) + + def test_every_break_character_is_neutralised(self): + """Not only LF. Python's own parsers break on more than RFC 5322 does. + + One header in, one header out, for each character in turn. + """ + for ch in ("\r", "\n", "\r\n", "\v", "\f", "\x1c", "\x1d", "\x1e", + "\x85", "
", "
"): + with self.subTest(ch=repr(ch)): + raw = self._build([("Subject", f"a{ch}From: forged@x.invalid")]) + published = email.message_from_string( + self._part(raw).get_content(), + policy=email.policy.default) + self.assertEqual(published.keys(), ["Subject"]) + self.assertEqual(published["From"], None) + + def test_an_ordinary_value_is_left_legible(self): + """RFC 2047 is applied only when it is needed. + + Encoding every header would render an ordinary Subject as + "=?utf-8?q?..." and cost the desk the legibility this part is for. + """ + raw = self._build([("Subject", "Your account requires verification")]) + # Asserted on the THIRD PART's own text, not on the whole document. + # The text part prints the same header under "Message as declared", + # so a whole-document substring passes even when this part is + # entirely encoded, which a mutation confirmed. + body = self._part(raw).get_content() + self.assertEqual(body.splitlines(), + ["Subject: Your account requires verification"]) + self.assertNotIn("=?utf-8?", body) + + def test_a_case_with_no_headers_omits_the_part(self): + """An empty third part claims the message declared no headers. + + That is never true of a real message. Absent says the report + carries none, which is the truth for a case parsed before the + headers block existed. + """ + for headers in ([], None): + with self.subTest(headers=headers): + raw = self._build(headers) + parsed = email.message_from_string( + raw, policy=email.policy.default) + self.assertEqual( + [p.get_content_type() for p in parsed.iter_parts()], + ["text/plain", "message/feedback-report"]) + self.assertNotIn("rfc822-headers", raw) + + def test_a_manifest_carrying_only_unpublishable_headers_omits_the_part( + self): + """The filter must not leave an empty part behind either.""" + raw = self._build([("To", "victim@example.org")]) + parsed = email.message_from_string(raw, policy=email.policy.default) + self.assertEqual( + [p.get_content_type() for p in parsed.iter_parts()], + ["text/plain", "message/feedback-report"]) + self.assertNotIn("victim", raw) + + def test_a_header_name_is_matched_case_insensitively(self): + """A header name is case-insensitive, and a hand edit will not match. + + Both directions matter: a lowercased "subject" must still publish, + and an uppercased "TO" must still be refused. + """ + raw = self._build([("subject", "lower"), ("SUBJECT", "upper"), + ("Subject", "mixed"), ("TO", "victim@example.org")]) + published = email.message_from_string( + self._part(raw).get_content(), policy=email.policy.default) + # All three spellings publish. The whitelist is stored lowercase, so + # a case-SENSITIVE comparison would still admit "subject" and the + # assertion would pass while the other two vanished; a mutation + # found exactly that. The capitalised spellings are what pin it. + self.assertEqual(published.keys(), ["subject", "SUBJECT", "Subject"]) + self.assertNotIn("victim", raw) + + +class DocumentIdentity(unittest.TestCase): + """The From header: the one identifier disclosed deliberately.""" + + def _from(self, identity): + destination = report.email_destinations(MANIFEST["contacts"])[0] + raw = report.build(MANIFEST, destination, identity) + parsed = email.message_from_string(raw, policy=email.policy.default) + return parsed["From"].addresses + + def test_a_comma_in_the_org_name_does_not_split_the_address(self): + """"Example Consulting, Ltd" is a legitimate name, and a comma is + the address-list separator. + + Formatted into an f-string it parses back as TWO addresses, the + first a bogus addr-spec with no domain, and a desk replying to the + report replies to nobody. + """ + addresses = self._from({"name": "Example Consulting, Ltd", + "org": "Example Consulting", + "email": "reporter@example.org"}) + self.assertEqual(len(addresses), 1) + self.assertEqual(addresses[0].addr_spec, "reporter@example.org") + self.assertEqual(addresses[0].display_name, "Example Consulting, Ltd") + + def test_awkward_names_still_yield_one_reachable_address(self): + for name in ('A "Quoted" Reporter', "Angle <brackets>", "Dänilo Ü", + "Back\\slash", "semi;colon", "at@sign"): + with self.subTest(name=name): + addresses = self._from({"name": name, "org": "o", + "email": "reporter@example.org"}) + self.assertEqual(len(addresses), 1) + self.assertEqual(addresses[0].addr_spec, + "reporter@example.org") + self.assertEqual(addresses[0].display_name, name) + + +class Freeze(unittest.TestCase): + """A case that has reached a desk is an evidence record, not a draft.""" + + def _frozen(self, marker): + manifest = copy.deepcopy(MANIFEST) + manifest["frozen"] = marker + return manifest + + def test_a_frozen_case_refuses(self): + manifest = self._frozen({"at": "2026-09-07T10:00:00Z", + "by": "abusedb"}) + with self.assertRaises(report.Frozen) as caught: + report.check_regenerable(manifest, modified=[]) + self.assertIn("abusedb", str(caught.exception)) + + def test_a_frozen_case_refuses_even_when_forced(self): + manifest = self._frozen({"at": "2026-09-07T10:00:00Z", + "by": "abusedb"}) + with self.assertRaises(report.Frozen): + report.check_regenerable(manifest, modified=[], force=True) + + def test_a_frozen_case_refuses_before_the_modified_check(self): + """Frozen wins over force for EVERY shape of the second argument. + + The two rules are not independent: an edited body on a frozen case + must still refuse, and must refuse as Frozen rather than as + Modified, because Modified is the one --force can clear. + """ + manifest = self._frozen({"at": "2026-09-07T10:00:00Z", + "by": "abusedb"}) + with self.assertRaises(report.Frozen): + report.check_regenerable(manifest, modified=["email-1"]) + with self.assertRaises(report.Frozen): + report.check_regenerable(manifest, modified=["email-1"], + force=True) + + def test_a_present_but_empty_marker_still_freezes(self): + """Present means frozen. The marker is write-once and monotonic, so + a shape that is present and says nothing useful is a half-written or + corrupted record, and the safe direction is to refuse. Truthiness is + the wrong test: `{}` is present and falsy, and treating it as absent + silently unfreezes a case a desk may already hold. + """ + for marker in ({}, "", 0, []): + with self.subTest(marker=marker): + with self.assertRaises(report.Frozen): + report.check_regenerable(self._frozen(marker), + modified=[], force=True) + + def test_a_malformed_marker_refuses_rather_than_crashing(self): + """A marker that is not a mapping is still a refusal, not an + AttributeError. cli turns Frozen into a message and a traceback into + a bug report. + """ + for marker in ("2026-09-07T10:00:00Z", 1, ["abusedb"]): + with self.subTest(marker=marker): + with self.assertRaises(report.Frozen): + report.check_regenerable(self._frozen(marker), + modified=[]) + + def test_an_explicit_null_marker_is_absent(self): + """JSON has no way to omit a key it wrote as null, and null is the + one present value that unambiguously carries no record. + """ + manifest = self._frozen(None) + report.check_regenerable(manifest, modified=[]) + + def test_a_modified_body_refuses_without_force(self): + with self.assertRaises(report.Modified) as caught: + report.check_regenerable(MANIFEST, modified=["email-1"]) + self.assertIn("email-1", str(caught.exception)) + + def test_a_modified_body_is_allowed_with_force(self): + report.check_regenerable(MANIFEST, modified=["email-1"], force=True) + + def test_an_untouched_case_regenerates(self): + report.check_regenerable(MANIFEST, modified=[]) + + def test_it_does_not_mutate_the_manifest(self): + """It is a check, not a step. case.py is the only writer.""" + manifest = copy.deepcopy(MANIFEST) + before = copy.deepcopy(manifest) + report.check_regenerable(manifest, modified=[]) + self.assertEqual(manifest, before) + + +class Hashes(unittest.TestCase): + def test_the_hash_detects_a_changed_body(self): + first = report.body_hash("one") + self.assertNotEqual(first, report.body_hash("two")) + self.assertEqual(first, report.body_hash("one")) + + def test_it_is_sha256_of_the_utf8_bytes(self): + """The encoding is pinned, not incidental: the hash is compared + against a body read back off disk, and for a sent destination it is + the record of what was disclosed. + """ + text = "a body with a non-ascii org name: Örg\n" + self.assertEqual( + report.body_hash(text), + hashlib.sha256(text.encode("utf-8")).hexdigest(), + ) + + def test_it_hashes_a_real_generated_body(self): + destination = report.email_destinations(MANIFEST["contacts"])[0] + text = report.build(MANIFEST, destination, IDENTITY) + self.assertEqual(len(report.body_hash(text)), 64) + + +class Writing(unittest.TestCase): + """generate() writing bodies into a case directory. + + A body is a byte-exact RFC 5322 document, not a text file the platform + may reflow, which is what most of these tests are really about. + """ + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = Path(self.tmp.name) + (self.path / "bodies").mkdir() + self.addCleanup(self.tmp.cleanup) + + def _generate(self, manifest=None, **kwargs): + if manifest is None: + manifest = copy.deepcopy(MANIFEST) + return report.generate(manifest, self.path, IDENTITY, **kwargs) + + def test_it_writes_one_body_per_destination_and_records_the_hash(self): + manifest = self._generate() + destination = manifest["destinations"][0] + body = self.path / destination["body"] + self.assertTrue(body.exists()) + self.assertEqual( + destination["body_sha256"], + report.body_hash(body.read_bytes().decode("utf-8")), + ) + + def test_the_body_path_is_named_for_the_destination_id(self): + """The id names the file, because the edit check pairs the two.""" + manifest = self._generate() + destination = manifest["destinations"][0] + self.assertEqual(destination["body"], + f"bodies/{destination['id']}.xarf") + + def test_the_recorded_hash_is_the_hash_of_the_bytes_on_disk(self): + """The round trip must be byte-exact, and this is the test that + says so for every destination rather than the first. + + A report is CRLF-delimited per RFC 5322 while its feedback part + carries bare LFs, so a text-mode write can translate the bare LFs + and a text-mode read collapses every CRLF. Either one makes the + recorded hash disagree with the file, which reads as a hand edit + and, for a sent destination, is a false record of what was + disclosed. + """ + manifest = self._generate() + for destination in manifest["destinations"]: + with self.subTest(destination=destination["id"]): + raw = (self.path / destination["body"]).read_bytes() + self.assertEqual(destination["body_sha256"], + hashlib.sha256(raw).hexdigest()) + + def test_the_written_body_keeps_its_crlf_line_endings(self): + """Verified against the document, not assumed from the policy.""" + manifest = self._generate() + raw = (self.path / manifest["destinations"][0]["body"]).read_bytes() + self.assertIn(b"\r\n", raw) + # The header block is pure CRLF: a bare LF there would mean the + # write translated line endings, which is the corruption a + # text-mode write causes on a platform whose linesep is CRLF. + header, _, _ = raw.partition(b"\r\n\r\n") + self.assertNotIn(b"\n", header.replace(b"\r\n", b"")) + # And the bare LFs inside the feedback part survive untranslated, + # which is the same defect seen from the other side. + self.assertIn(b"Feedback-Type: fraud\nUser-Agent:", raw) + + def test_an_untouched_case_regenerates_without_complaint(self): + """The regression that the plan's read_text() caused. + + Nothing was edited, so a second run must simply rewrite the + bodies. Under a lossy round trip every body reads as hand-edited + and the second run refuses, which makes report unusable twice. + """ + manifest = self._generate() + again = report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + self.assertEqual([d["id"] for d in again["destinations"]], + [d["id"] for d in manifest["destinations"]]) + self.assertEqual(list((self.path / "bodies").glob("*.orig")), []) + + def test_it_records_the_unreportable_indicators(self): + manifest = copy.deepcopy(MANIFEST) + manifest["contacts"] = manifest["contacts"] + [ + {"iocs": ["ioc-4"], "query": "x.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + ] + result = self._generate(manifest) + self.assertEqual( + result["unreportable"], + [{"ioc": "ioc-4", "reason": "no abuse role published"}], + ) + + def test_a_modified_body_is_backed_up_before_being_overwritten(self): + manifest = self._generate() + destination_id = manifest["destinations"][0]["id"] + body = self.path / manifest["destinations"][0]["body"] + body.write_bytes(b"hand edited during review\r\n") + + with self.assertRaises(report.Modified): + report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + + report.generate(copy.deepcopy(manifest), self.path, IDENTITY, + force=True) + backups = list((self.path / "bodies").glob("*.orig")) + self.assertEqual(len(backups), 1) + # Pin the NAME, not just the count: a backup written beside the + # wrong destination still globs as one file. The stamp sits + # between the extension and .orig so the id stays readable. + self.assertTrue( + re.fullmatch(rf"{destination_id}\.xarf\.\d{{8}}T\d{{6}}Z\.orig", + backups[0].name), + backups[0].name, + ) + self.assertEqual(backups[0].read_bytes(), + b"hand edited during review\r\n") + # And the body itself was replaced rather than left as the edit. + self.assertNotEqual(body.read_bytes(), + b"hand edited during review\r\n") + + def test_a_backup_preserves_the_edited_bytes_exactly(self): + """The backup is the user's twenty minutes of review, so it is + copied verbatim rather than through a text-mode round trip. + """ + manifest = self._generate() + body = self.path / manifest["destinations"][0]["body"] + edited = "line one\r\nline two\nnon-ascii: Örg\r\n" + body.write_bytes(edited.encode("utf-8")) + report.generate(copy.deepcopy(manifest), self.path, IDENTITY, + force=True) + backups = list((self.path / "bodies").glob("*.orig")) + self.assertEqual(backups[0].read_bytes(), edited.encode("utf-8")) + + def test_a_deleted_body_regenerates_without_complaint(self): + manifest = self._generate() + (self.path / manifest["destinations"][0]["body"]).unlink() + again = report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + self.assertTrue((self.path / again["destinations"][0]["body"]).exists()) + + def test_a_frozen_case_writes_nothing_at_all(self): + """A refusal must leave the bodies directory exactly as it was. + + check_regenerable runs after the scan and before every write, so + no body is rewritten and no backup is renamed. Asserting the + directory's full state is what makes that ordering testable: a + partial write on a frozen case corrupts an evidence record a desk + already holds. + """ + manifest = self._generate() + before = {p.name: p.read_bytes() + for p in (self.path / "bodies").iterdir()} + manifest["frozen"] = {"at": "2026-09-07T10:00:00Z", "by": "abusedb"} + + with self.assertRaises(report.Frozen): + report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + with self.assertRaises(report.Frozen): + report.generate(copy.deepcopy(manifest), self.path, IDENTITY, + force=True) + + after = {p.name: p.read_bytes() + for p in (self.path / "bodies").iterdir()} + self.assertEqual(after, before) + + def test_a_modified_refusal_writes_nothing_at_all(self): + """Same ordering, the other refusal. Without --force an edited + body must survive untouched, including every body that was NOT + edited: a partial rewrite would discard part of a review. + """ + manifest = self._generate() + body = self.path / manifest["destinations"][0]["body"] + body.write_bytes(b"hand edited during review\r\n") + before = {p.name: p.read_bytes() + for p in (self.path / "bodies").iterdir()} + + with self.assertRaises(report.Modified): + report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + + after = {p.name: p.read_bytes() + for p in (self.path / "bodies").iterdir()} + self.assertEqual(after, before) + + def test_it_refuses_a_case_directory_that_does_not_exist(self): + """generate does not conjure a case. case.py creates a case, and a + missing directory means a wrong path rather than a first run, so + creating the tree here would scatter bodies into an empty dir. + """ + with self.assertRaises(FileNotFoundError): + report.generate(copy.deepcopy(MANIFEST), self.path / "absent", + IDENTITY) + + def test_no_recipient_identifier_reaches_a_written_body(self): + """Property 1, checked against the bytes that land on disk. + + The manifest is given a recipient address in every place a body + draws from, and none of it may appear in a file. + """ + manifest = copy.deepcopy(MANIFEST) + secret = "victim@example.org" + manifest["headers"] = manifest["headers"] + [ + ("To", f"Victim <{secret}>"), + ("Delivered-To", secret), + ] + manifest["iocs"] = manifest["iocs"] + [ + {"id": "ioc-3", "type": "url", + "value": f"http://kit.invalid/verify?e={secret}", + "origin": "body"}, + ] + result = self._generate(manifest) + for destination in result["destinations"]: + raw = (self.path / destination["body"]).read_bytes() + with self.subTest(destination=destination["id"]): + self.assertNotIn(secret.encode("utf-8"), raw) + self.assertNotIn(b"victim", raw.lower()) + + def test_it_returns_the_manifest_rather_than_saving_it(self): + """case.py is the only writer of a manifest.""" + result = self._generate() + self.assertNotIn("manifest.json", + [p.name for p in self.path.iterdir()]) + self.assertIn("destinations", result) + + if __name__ == "__main__": unittest.main() + + +class PersonalisedSubject(unittest.TestCase): + """The documented limit, pinned so a later change has to face it. + + A phishing kit personalises the Subject with the recipient's local part. + The whitelist governs WHICH headers travel, never what is inside one, so + that local part is published. The spec accepts this deliberately: Subject + is what lets a desk recognise a campaign, and filtering free text is the + judgement-shaped problem AGENTS.md names as the origin of every leak this + project has had. + + Measured, not theoretical: a sweep of 92 real messages found 14 carrying + the recipient's local part in the Subject, and none in the From display + name. This test exists so that number cannot change silently. + """ + + def _headers(self): + raw = (Path(__file__).parent + / "fixtures" / "personalised-subject.eml").read_bytes() + return dict(parse.report_headers(raw, trusted=["192.0.2.0/24"])) + + def test_the_envelope_recipient_is_cut_from_the_received_line(self): + # The protection that DOES hold: our own relay wrote the "for" + # clause, and it is cut before the line is ever stored. + received = self._headers()["Received"] + self.assertNotIn("alicejones", received) + self.assertNotIn("@example.org", received) + + def test_no_published_header_carries_the_full_recipient_address(self): + published = " ".join(self._headers().values()) + self.assertNotIn("alicejones@example.org", published) + + def test_the_subject_still_carries_the_local_part(self): + # Deliberate. If this ever fails because Subject was filtered, the + # spec's "considered and not built" section is what to read first: + # the fix is a redaction rule with its own tests, not a passthrough + # quietly turned into a filter. + self.assertIn("alicejones", self._headers()["Subject"]) |