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()