From f17a12003563330b1a57c245ebbd435e97aa73cc Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Wed, 9 Sep 2026 19:10:02 +0200 Subject: feat: group abuse contacts into one destination per address Two contacts can resolve to the same desk, an IP and a domain at one hoster being the common case, and grouping per contact would send that desk two mails about one incident. The domain half of an address is folded to lowercase before grouping, since it is case-insensitive by definition; the local part is left verbatim, because only the receiving host knows whether it folds, and folding two desks a host distinguishes would silently drop one. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xj1ayFRSUQ2u7cwb3S4axE --- tests/test_report.py | 143 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 143 insertions(+) create mode 100644 tests/test_report.py (limited to 'tests') diff --git a/tests/test_report.py b/tests/test_report.py new file mode 100644 index 0000000..6a0dc51 --- /dev/null +++ b/tests/test_report.py @@ -0,0 +1,143 @@ +import unittest + +from abusectl import report + + +class Grouping(unittest.TestCase): + def test_two_contacts_at_one_address_become_one_destination(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["target"], "abuse@host.invalid") + self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"]) + + def test_a_contact_with_two_addresses_reaches_both_desks(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["a@host.invalid", "b@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual( + sorted(d["target"] for d in destinations), + ["a@host.invalid", "b@host.invalid"], + ) + + def test_a_contact_with_no_address_creates_no_destination(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + ] + self.assertEqual(report.email_destinations(contacts), []) + + def test_destinations_carry_stable_ids_and_pending_status(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(contacts)[0] + self.assertEqual(destination["id"], "email-1") + self.assertEqual(destination["kind"], "email") + self.assertEqual(destination["status"], "pending") + + def test_ids_are_numbered_per_destination_not_per_contact(self): + """A skipped contact must not leave a hole in the numbering. + + The obvious implementation enumerates the contacts, and a contact + with no abuse address then burns an id: the desks come back as + "email-2" and "email-4", which reads to a reviewer as two reports + having gone missing. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + {"iocs": ["ioc-2"], "query": "198.51.100.7", + "abuse": ["a@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-3"], "query": "198.51.100.8", + "abuse": ["b@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual([d["id"] for d in destinations], + ["email-1", "email-2"]) + self.assertEqual([d["target"] for d in destinations], + ["a@host.invalid", "b@host.invalid"]) + + def test_one_desk_listed_twice_by_one_contact_is_one_destination(self): + """A duplicate in a contact's own abuse list must not duplicate a desk. + + RDAP jCards are attacker-adjacent data: an entity can publish the + same address in two vcard rows, and one destination per ADDRESS is + the rule regardless of how many rows produced it. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid", "abuse@host.invalid"], + "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["iocs"], ["ioc-1"]) + + def test_one_desk_spelled_with_two_domain_cases_is_one_destination(self): + """A domain is case-insensitive, so two spellings are one desk.""" + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@Host.Invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"]) + self.assertEqual(destinations[0]["target"], "abuse@Host.Invalid") + + def test_two_local_part_cases_stay_two_destinations(self): + """Only the receiving host knows whether its local parts fold. + + Folding them here would silently drop a desk that a host genuinely + distinguishes; not folding them costs a duplicate mail at worst. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["Abuse@host.invalid", "abuse@host.invalid"], + "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(sorted(d["target"] for d in destinations), + ["Abuse@host.invalid", "abuse@host.invalid"]) + + def test_a_destination_starts_with_no_body(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + self.assertIsNone(report.email_destinations(contacts)[0]["body"]) + + def test_the_same_contacts_produce_the_same_ids_twice(self): + """Ids must not depend on dict iteration luck or set ordering. + + Task 8 writes each body to bodies/.xarf and records its hash + against that id, so an id that moved between two runs over the + same input would compare one desk's body against another's. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["b@host.invalid", "a@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["c@host.invalid"], "source": "rdap"}, + ] + first = report.email_destinations(contacts) + second = report.email_destinations(contacts) + self.assertEqual([(d["id"], d["target"]) for d in first], + [(d["id"], d["target"]) for d in second]) + self.assertEqual([d["target"] for d in first], + ["b@host.invalid", "a@host.invalid", + "c@host.invalid"]) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3