diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_report.py | 455 |
1 files changed, 455 insertions, 0 deletions
diff --git a/tests/test_report.py b/tests/test_report.py index 3b13d11..f414841 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -922,5 +922,460 @@ class BackslashRoundTrip(unittest.TestCase): self.assertIn("Generated by abusectl.", report.unwrap(text)) +class FeedbackPart(unittest.TestCase): + def setUp(self): + destination = report.email_destinations(MANIFEST["contacts"])[0] + self.fields = report.feedback_fields(MANIFEST, destination) + 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["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_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"] + self.assertEqual( + uris, ["http://login.sender.invalid/verify?id=REDACTED"] + ) + + def test_a_destination_with_no_ip_omits_source_ip(self): + contacts = [{"iocs": ["ioc-2"], "query": "sender.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}] + destination = report.email_destinations(contacts)[0] + lookup = dict(report.feedback_fields(MANIFEST, destination)) + self.assertNotIn("Source-IP", lookup) + self.assertEqual( + lookup["Source"], "http://login.sender.invalid/verify?id=REDACTED" + ) + + # --- the parts the plan got wrong ------------------------------------- + + def _fields_for(self, iocs: list[dict]) -> list[tuple[str, str]]: + """Render the machine part for a hand-built IOC list. + + Every IOC reaches one destination, so the field list is exactly what + those indicators produce and nothing is filtered out behind the test. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = iocs + manifest["contacts"] = [ + {"iocs": [entry["id"] for entry in iocs], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + return report.feedback_fields(manifest, destination) + + def test_source_ip_is_emitted_at_most_once(self): + """RFC 5965 says Source-IP appears "once maximum". + + The plan emitted one per IP. A strict parser meeting a repeated + single-occurrence field either rejects the part or keeps whichever + occurrence it saw last, so the field a repeat was meant to add is + the field that displaces the primary one. Every IP still travels, + in the text part and in Reported-Uri's sibling below. + """ + fields = self._fields_for([ + {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", + "origin": "received-chain", "confidence": "boundary-hop"}, + {"id": "ioc-2", "type": "ipv4", "value": "203.0.113.43", + "origin": "received-chain"}, + ]) + ips = [v for n, v in fields if n == "Source-IP"] + self.assertEqual(ips, ["203.0.113.42"]) + self.assertEqual(dict(fields)["Source"], "203.0.113.42") + + def test_no_field_appears_twice_unless_the_rfc_allows_it(self): + """The invariant behind the test above, stated once for every field. + + Reported-Uri and Reported-Domain are "any number of times"; every + other field this module emits is once-maximum. Asserting only on + Source-IP would let the next repeated field ship unnoticed. + """ + fields = self._fields_for([ + {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", + "origin": "received-chain"}, + {"id": "ioc-2", "type": "ipv6", "value": "2001:db8::1", + "origin": "received-chain"}, + {"id": "ioc-3", "type": "url", "value": "http://a.invalid/x", + "origin": "body"}, + {"id": "ioc-4", "type": "url", "value": "http://b.invalid/y", + "origin": "body"}, + {"id": "ioc-5", "type": "domain", "value": "a.invalid", + "origin": "header-from"}, + {"id": "ioc-6", "type": "domain", "value": "b.invalid", + "origin": "header-reply_to"}, + ]) + seen: dict[str, int] = {} + for name, _ in fields: + seen[name] = seen.get(name, 0) + 1 + 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-Domain"], 2) + + def test_an_ipv6_indicator_fills_source_ip_too(self): + """"ipv6" is a distinct type string from parse.iocs(). + + A branch testing only for "ipv4" drops every IPv6 sender, and the + given tests use IPv4 throughout so none of them would notice. + """ + fields = dict(self._fields_for([ + {"id": "ioc-1", "type": "ipv6", "value": "2001:db8::1", + "origin": "received-chain"}, + ])) + self.assertEqual(fields["Source"], "2001:db8::1") + self.assertEqual(fields["Source-IP"], "2001:db8::1") + + def test_a_destination_with_no_typed_indicator_omits_source(self): + """An empty Source is worse than an absent one. + + "Source:" with nothing after it asserts that the thing being + reported is the empty string. A 5965 parser reading a present-but- + empty field has been told a value; reading no field it has been + told nothing, which is the truth. Only sha256 and observation + indicators reach a desk here, and both belong in the text part. + """ + fields = self._fields_for([ + {"id": "ioc-1", "type": "observation", + "value": "display-name-carries-address", + "origin": "display-name-from"}, + ]) + lookup = dict(fields) + 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["Version"], "1") + + def test_a_type_this_module_does_not_place_is_not_invented_into_one(self): + """sha256 and observation have no 5965 or x-arf field. + + Neither is a Source, a Reported-Uri or a Reported-Domain, and + forcing one into the nearest-looking field would tell a desk that a + file hash is a URI. They travel in the human part, which is where a + desk reads what an attachment was. + """ + fields = self._fields_for([ + {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", + "origin": "received-chain"}, + {"id": "ioc-2", "type": "sha256", "value": "a" * 64, + "origin": "attachment", "filename": "invoice.zip"}, + {"id": "ioc-3", "type": "observation", + "value": "display-name-carries-address", + "origin": "display-name-from"}, + ]) + blob = repr(fields) + self.assertNotIn("a" * 64, blob) + self.assertNotIn("display-name-carries-address", blob) + self.assertEqual(dict(fields)["Source"], "203.0.113.42") + + def test_an_ioc_id_the_manifest_lacks_is_skipped_not_raised(self): + """A manifest is a file the user edits, so the two can disagree. + + text_part() already tolerates this; the machine part indexed with + destination["iocs"] straight into a dict would raise instead, and + the two parts of one document must not disagree about whether the + case can be reported at all. + """ + destination = {"id": "email-x", "kind": "email", + "target": "abuse@host.invalid", + "iocs": ["ioc-1", "ioc-404"], "body": None, + "status": "pending"} + lookup = dict(report.feedback_fields(MANIFEST, destination)) + self.assertEqual(lookup["Source"], "203.0.113.42") + self.assertNotIn("ioc-404", repr(lookup)) + + def test_a_destination_with_no_iocs_key_still_renders(self): + """destination.get("iocs"), not destination["iocs"].""" + destination = {"id": "email-x", "kind": "email", + "target": "abuse@host.invalid", "body": None, + "status": "pending"} + lookup = dict(report.feedback_fields(MANIFEST, destination)) + self.assertEqual(lookup["Feedback-Type"], "abuse") + + 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 + the message. The Date header is when the SENDER CLAIMS it was sent. + + The plan copied Date into Arrival-Date. On a phishing message that + header is attacker-controlled free text, so the report would assert + as our own observation a timestamp the attacker chose, and a desk + correlating it against their own logs would look in the wrong place + or find nothing and discount the report. + + The honest source is the boundary Received hop's own timestamp, + which parse.report_headers() already publishes. Parsing one is a + date parser this task does not need, so the field is OMITTED: 5965 + makes it optional, and an absent optional field misstates nothing. + """ + lookup = dict(self.fields) + self.assertNotIn("Arrival-Date", lookup) + self.assertNotIn("Mon, 07 Sep 2026 09:12:40 +0000", repr(lookup)) + + +class FeedbackInjection(unittest.TestCase): + """A field value carrying a line break forges a field in the report. + + This is not hypothetical and it is not stopped upstream. redact.py + URL-DECODES a redirector's destination parameter to recover it as an + indicator, so a message body carrying + + http://r.invalid/go?next=http%3A%2F%2Fa.invalid%2Fx%0AFeedback-Type... + + produces, through parse.iocs() on a real .eml, an IOC whose value is + "http://a.invalid/x\\nFeedback-Type: not-abuse". Emitted verbatim, the + abuse desk's parser reads a Feedback-Type this tool never asserted, on a + report that carries the reporter's identity. That is an attacker writing + fields into mail sent under our name. + + The answer here is to PERCENT-ENCODE the control characters rather than + to drop the indicator or strip them. Dropping loses a real redirect + target; stripping silently rewrites an indicator into a different one a + desk would then act on. Percent-encoding is the URL's own native + encoding, is exactly reversible, and leaves the value visibly altered + rather than quietly wrong. + """ + + def _value_out(self, value: str) -> str | None: + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [{"id": "ioc-1", "type": "url", "value": value, + "origin": "redirect-target"}] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + fields = report.feedback_fields(manifest, destination) + for name, out in fields: + if name == "Reported-Uri": + return out + return None + + def _assert_no_break(self, fields: list[tuple[str, str]]) -> None: + for name, value in fields: + for bad in ("\r", "\n", "
", "
", "\v", "\f", + "\x1c", "\x1d", "\x1e", "\x85"): + self.assertNotIn(bad, name) + self.assertNotIn(bad, value) + + def test_a_newline_in_a_value_cannot_forge_a_field(self): + out = self._value_out("http://a.invalid/x\nFeedback-Type: not-abuse") + self.assertNotIn("\n", out) + self.assertIn("%0A", out) + # The forged field name must not survive as a line of its own, but + # the text of the indicator is still legible and reversible. + self.assertEqual(out, + "http://a.invalid/x%0AFeedback-Type: not-abuse") + + def test_the_real_parse_output_that_makes_this_reachable(self): + """End to end from an .eml, not from a hand-written IOC. + + A test that only feeds feedback_fields() a crafted string proves the + encoder works; it does not prove the encoder is needed. This runs + the actual redirector through parse.iocs() so the fixture and the + defence cannot drift apart. + """ + from abusectl import parse + raw = ( + "Received: from evil.invalid ([203.0.113.9]) by mx.example.org; " + "Mon, 07 Sep 2026 09:12:40 +0000\r\n" + "From: <phish@sender.invalid>\r\n" + "Subject: verify\r\n" + "Date: Mon, 07 Sep 2026 09:12:40 +0000\r\n" + "Content-Type: text/plain\r\n\r\n" + "http://r.invalid/go?next=http%3A%2F%2Fa.invalid%2Fx%0A" + "Feedback-Type%3A%20not-abuse\r\n" + ).encode() + iocs = parse.iocs(raw, trusted=["192.0.2.0/24"]) + injected = [e for e in iocs if "\n" in e["value"]] + self.assertTrue(injected, "the injection vector itself has changed") + + manifest = {"format": 1, "iocs": iocs, "headers": [], "auth": {}} + destination = {"id": "email-x", "kind": "email", + "target": "abuse@host.invalid", + "iocs": [e["id"] for e in iocs], "body": None, + "status": "pending"} + fields = report.feedback_fields(manifest, destination) + self._assert_no_break(fields) + + def test_every_line_breaking_shape_is_neutralised(self): + """The adversarial sweep, not a handful of cases. + + U+2028 and U+2029 are in here because Python's own email module + raises on them: str.splitlines() treats them as breaks, so a value + carrying one would make the whole document fail to assemble in + Task 6 rather than merely render oddly. + """ + breaks = ["\n", "\r", "\r\n", "\n\r", "
", "
", + "\v", "\f", "\x1c", "\x1d", "\x1e", "\x85"] + shapes = [] + for brk in breaks: + shapes += [ + brk, + "http://a.invalid/x" + brk, + brk + "http://a.invalid/x", + "http://a.invalid/x" + brk + "Feedback-Type: not-abuse", + "http://a.invalid/" + brk * 3 + "Source: 192.0.2.1", + ] + for value in shapes: + with self.subTest(value=repr(value)): + out = self._value_out(value) + self.assertIsNotNone(out) + for bad in breaks: + if len(bad) == 1: + self.assertNotIn(bad, out) + + def test_a_value_that_is_only_a_newline_still_yields_a_field(self): + """It must not become an empty value or vanish silently.""" + out = self._value_out("\n") + self.assertEqual(out, "%0A") + + def test_encoding_is_reversible_so_the_indicator_is_not_misstated(self): + """The property that makes encoding honest rather than a strip. + + A desk, or a later submit path, must be able to recover exactly what + the message declared. Stripping the character would pass every + assertion above and hand the desk a DIFFERENT URL. + """ + from urllib.parse import unquote + for value in ("http://a.invalid/x\nFeedback-Type: not-abuse", + "http://a.invalid/\r\n\r\n", + "http://a.invalid/x
y"): + with self.subTest(value=repr(value)): + self.assertEqual(unquote(self._value_out(value)), value) + + def test_a_literal_percent_is_encoded_so_the_reversal_is_unambiguous(self): + """Without this, "%0A" typed by the attacker decodes to a newline. + + A redacted URL legitimately contains percent signs, and an encoder + that leaves them alone produces text that unquote() turns into the + very control character the encoder existed to remove. The reversal + must be a true inverse or it is a second injection one step later. + """ + from urllib.parse import unquote + value = "http://a.invalid/x?a=%0AFeedback-Type: not-abuse" + out = self._value_out(value) + self.assertNotIn("\n", out) + self.assertEqual(unquote(out), value) + + def test_an_injected_field_name_in_a_domain_is_neutralised_too(self): + """Reported-Domain and Source take the same path as Reported-Uri. + + The defence must not live in one branch. That is the exact shape of + the fourth property's three leaks: validation applied per branch + gets forgotten on the next branch. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "domain", + "value": "a.invalid\nSource: 192.0.2.1", + "origin": "header-from"}, + {"id": "ioc-2", "type": "ipv4", + "value": "203.0.113.42\nVersion: 9", + "origin": "received-chain"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1", "ioc-2"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + fields = report.feedback_fields(manifest, destination) + self._assert_no_break(fields) + self.assertEqual(dict(fields)["Version"], "1") + lookup = dict(fields) + self.assertEqual(lookup["Source"], "203.0.113.42%0AVersion: 9") + self.assertEqual(lookup["Reported-Domain"], + "a.invalid%0ASource: 192.0.2.1") + + def test_the_rendered_part_survives_pythons_own_header_setter(self): + """The end the whole defence is for: Task 6 assembles with email. + + EmailMessage raises ValueError on a header value containing a break, + so an unencoded value does not merely render oddly, it aborts the + document. Asserting through the real setter is what makes this a + test of the outcome rather than of my own notion of a break. + """ + from email.message import EmailMessage + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "url", + "value": "http://a.invalid/x\r\nFeedback-Type: not-abuse
z", + "origin": "redirect-target"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + part = EmailMessage() + for name, value in report.feedback_fields(manifest, destination): + part[name] = value + rendered = part.as_string() + # The property is per LINE, not per substring: the attacker's text + # may legitimately appear INSIDE a value, and asserting it absent + # would forbid reporting a URL that merely contains the words. What + # must not exist is a line a parser reads as a field of its own. + names = [line.split(":", 1)[0] for line in rendered.splitlines() + if line and not line[0].isspace() and ":" in line] + self.assertEqual(names.count("Feedback-Type"), 1) + self.assertEqual( + [n for n in names if n == "Feedback-Type"], ["Feedback-Type"]) + for line in rendered.splitlines(): + self.assertNotEqual(line.strip(), "Feedback-Type: not-abuse") + + def test_a_field_name_is_never_taken_from_data(self): + """Names are literals in this module, so no input can invent one. + + Pinned because the obvious "generalise it" refactor is a table + mapping an IOC's own type string to a field name, and a manifest is + a file the user edits: a type of "x: y\\nFeedback-Type" would then + BE a field name. The set is closed on purpose. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "url\nFeedback-Type", "value": "x", + "origin": "body"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + fields = report.feedback_fields(manifest, destination) + self.assertEqual( + {name for name, _ in fields}, + {"Feedback-Type", "User-Agent", "Version", "Report-Type"}) + + def test_a_non_string_value_does_not_crash_the_report(self): + """A manifest is edited by hand and JSON has numbers. + + Not a security property, but a report that raises produces nothing + at all, and this is the one module standing between a reviewed case + and a sent mail. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "ipv4", "value": 42, + "origin": "received-chain"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + lookup = dict(report.feedback_fields(manifest, destination)) + self.assertEqual(lookup["Source"], "42") + + if __name__ == "__main__": unittest.main() |