diff options
| -rw-r--r-- | abusectl/report.py | 180 | ||||
| -rw-r--r-- | tests/test_report.py | 219 |
2 files changed, 399 insertions, 0 deletions
diff --git a/abusectl/report.py b/abusectl/report.py index a3d017e..36bbbff 100644 --- a/abusectl/report.py +++ b/abusectl/report.py @@ -28,6 +28,11 @@ itself is a module that can disclose it in a code path nobody reviewed. import hashlib +from email.header import Header +from email.headerregistry import Address +from email.message import EmailMessage +from email.policy import SMTP + VERSION = "0.1.0" @@ -702,3 +707,178 @@ def feedback_fields(manifest: dict, destination: dict) -> list[tuple[str, str]]: fields.append(("Reported-Uri", url)) return [(name, _field_value(value)) for name, value in fields] + + +# The header names the third part may carry, and the reason it is a list here +# rather than a trust in the manifest. +# +# parse.report_headers() already produces a whitelist, so in the normal path +# every name arriving here is one of these and this filter changes nothing. +# The manifest is a FILE THE USER EDITS, though, and the first property's +# structural argument is that report cannot disclose what it was never given. +# A hand-edited manifest is precisely a way it CAN be given something: a user +# pasting a header block back in, or a case written by a future parse.py whose +# whitelist grew, hands this module a "To:" pair and the argument no longer +# holds by construction. +# +# So the same whitelist is applied a second time at the point of publication. +# That is a deliberate exception to "two places to remember is how the fourth +# property leaked", and the trade runs the other way here: this is not a +# second DECISION about what may be disclosed, it is the same decision +# enforced where the disclosure actually happens. parse.py still decides; this +# refuses to publish anything it did not decide for. A duplicated whitelist +# that drifts loses a header from a report, which review sees. A missing one +# publishes a recipient address to the attacker, which nobody sees. +# +# Transcribed from the spec's "Where the headers come from" whitelist, which +# is the same list parse._REPORT_HEADERS holds. Matched case-insensitively, +# because a header name is case-insensitive by RFC 5322 and a hand-edited +# manifest will not have preserved anyone's capitalisation. +_PUBLISHABLE_HEADERS = frozenset({ + "received", "from", "subject", "date", "message-id", "reply-to", + "return-path", "authentication-results", "received-spf", + "mime-version", "content-type", +}) + + +def _header_line(name: str, value: str) -> str: + """One line of the third part, safe to emit even when the value is not. + + A header VALUE here is attacker-supplied free text. The spec keeps + Subject and the From display name deliberately, because they are what + lets a desk recognise a campaign, and it says plainly that the whitelist + governs WHICH headers travel and never what is inside one. A Subject + carrying a newline therefore reaches this function, and emitting it + verbatim forges a header line inside a part whose entire content is read + as headers: "Subject: evil\\nFrom: forged@attacker.invalid" becomes two + headers, the second of which a desk reads as something the message + declared. That is the Task 5 injection one part further along, and it is + worse here, because the forged line is grammatical where a forged x-arf + field is merely present. + + The answer differs from _field_value()'s percent-encoding, and the + difference is the audience. This part's content IS rfc822 headers, so the + encoding a reader of it already knows is RFC 2047, not URL escaping. An + encoded word neutralises the break by turning it into an RFC 5322 FOLD: + the value continues on a continuation line, a parser unfolds it back to + one header whose value is the original text including the character that + was there, and nothing new appears in the header list. Percent-encoding + would also be safe and would read as a bug to a mail parser, which is the + one audience this part has. + + Encoded ONLY when a value actually carries something unsafe. RFC 2047 + encodes indiscriminately, so applying it to every header would render an + ordinary Subject as "=?utf-8?q?Your_account?=" and cost the desk the + legibility this part exists for. The condition is the same _UNSAFE set + the machine part uses, minus "%": "%" is in that set because + percent-encoding must be a true inverse, and nothing here percent-encodes, + so a literal "%" in a Subject is just a character. + + Non-strings are coerced for the reason _field_value() coerces them: a + manifest is hand-edited, JSON has numbers, and a report that raises + produces nothing at all. + """ + text = str(value) + if any(ch in text for ch in _UNSAFE if ch != "%"): + # maxlinelen leaves room for "Name: " on the first line; the exact + # number only affects where a fold lands, never what unfolds back. + return f"{name}: {Header(text, 'utf-8', maxlinelen=64).encode()}" + return f"{name}: {text}" + + +def _headers_part(manifest: dict) -> str | None: + """The third part's body, or None when there is nothing to publish. + + Returns None rather than an empty string so build() can OMIT the part. + An empty text/rfc822-headers is a positive claim that the message + declared no headers, which 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 or one a user emptied by hand. Same rule as + feedback_fields() omitting Source rather than emitting it empty, and the + same rule as the config's absent-not-empty-string. + + Pairs, not a dict, matching what case.load() hands back: JSON has no + tuple, so these arrive as lists and both shapes destructure identically. + """ + lines = [ + _header_line(name, value) + for name, value in (manifest.get("headers") or []) + if str(name).lower() in _PUBLISHABLE_HEADERS + ] + if not lines: + return None + return "\n".join(lines) + "\n" + + +def build(manifest: dict, destination: dict, identity: dict) -> str: + """Assemble one destination's report as an RFC 5965 MIME document. + + Three parts: what a human reads, what a parser reads, and the headers. + + THE ORIGINAL MESSAGE IS NOT ATTACHED, and there is no message/rfc822 + part. source.eml carries every identifier the first property exists to + keep out: To, Cc, Delivered-To, unredacted URLs whose query and path + segments encode the recipient, the user's own Message-IDs and maildir + paths. An abuse desk forwards a report to the abused customer, who for a + phishing domain may be the attacker, and URLhaus is a public feed. RFC + 5965 provides text/rfc822-headers for exactly the case where the full + message cannot be included, so this is the standard's own answer rather + than a deviation from it. + + The FROM is built with email.headerregistry.Address rather than by + formatting a string. A reporting identity legitimately contains a comma, + "Example Consulting, Ltd" being the obvious one, and a comma is the + address-list separator: f"{name} <{email}>" then parses back as TWO + addresses, the first of which is a bogus addr-spec with no domain. A desk + replying to the report replies to that, and the reply reaches nobody. + Address quotes the display name when it needs quoting and RFC 2047-encodes + it when it is not ASCII, which are the two cases a consultant's org name + actually hits. + + Each subpart's type is set AFTER its content, which is the opposite of + what looks right and was checked rather than assumed: set_content() + REPLACES the Content-Type it derived from the payload, so setting the type + first leaves all three parts as text/plain. Setting it afterwards keeps + the transfer encoding and charset set_content() chose, which is what makes + a non-ASCII header value in the third part survive as base64 rather than + as a malformed 7bit line. + + The multipart boundary is chosen by the generator at SERIALISATION time, + after it has seen every payload, so a body containing something shaped + like a delimiter cannot collide with the real one; that is verified + against a payload carrying a literal "--===============0==" line. + """ + message = EmailMessage(policy=SMTP) + message["From"] = Address(str(identity.get("name", "")), + addr_spec=str(identity["email"])) + message["To"] = destination["target"] + # The case id is generated by case.py from a date and random hex, so it is + # not attacker-supplied and needs no escaping; it is the string the user + # greps for when a desk replies. + message["Subject"] = ( + f"Abuse report: phishing infrastructure, case {manifest['case_id']}" + ) + message.make_mixed() + message.set_type("multipart/report") + message.set_param("report-type", "feedback-report") + + human = EmailMessage(policy=SMTP) + human.set_content(text_part(manifest, destination, identity)) + message.attach(human) + + machine = EmailMessage(policy=SMTP) + machine.set_content( + "\n".join(f"{name}: {value}" + for name, value in feedback_fields(manifest, destination)) + + "\n") + machine.set_type("message/feedback-report") + message.attach(machine) + + body = _headers_part(manifest) + if body is not None: + headers = EmailMessage(policy=SMTP) + headers.set_content(body) + headers.set_type("text/rfc822-headers") + message.attach(headers) + + return message.as_string() diff --git a/tests/test_report.py b/tests/test_report.py index f414841..6a340c9 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -1,4 +1,6 @@ import copy +import email +import email.policy import json import unittest @@ -1377,5 +1379,222 @@ 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) + + if __name__ == "__main__": unittest.main() |