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 --- abusectl/report.py | 99 +++++++++++++++++++++++++++++++++++ tests/test_report.py | 143 +++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 242 insertions(+) create mode 100644 abusectl/report.py create mode 100644 tests/test_report.py diff --git a/abusectl/report.py b/abusectl/report.py new file mode 100644 index 0000000..cf839f3 --- /dev/null +++ b/abusectl/report.py @@ -0,0 +1,99 @@ +# Copyright (C) 2026 Danilo M. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +"""IOCs and abuse contacts to report bodies: the last step before anything +irreversible happens. + +This module is PURE and OFFLINE. It opens no socket, sends no mail and reads +no file outside the case directory. What it produces is a document the user +reads, edits and approves, so the output is written for a human first and a +parser second. + +It takes the reporting identity as an ARGUMENT rather than reading the +config, the way parse.py takes the trust boundary. The identity is the one +thing in a report disclosed deliberately, and a module that reaches for it +itself is a module that can disclose it in a code path nobody reviewed. +""" + + +def _group_key(address: str) -> str: + """The key two spellings of one desk must share, and no more than that. + + The DOMAIN is case-insensitive by every standard that touches it, so + "abuse@Host.Invalid" and "abuse@host.invalid" are one desk and must + not be mailed twice about one incident. + + The LOCAL PART is left exactly as published. RFC 5321 leaves its + interpretation to the receiving host, and only that host knows whether + it folds case. In practice it almost always does, but "almost always" + is the wrong standard for the one field that decides whether a report + arrives: folding two desks that a host genuinely distinguishes would + silently drop one of them, and the cost of being wrong the other way + is a duplicate mail. A dropped desk is the worse failure, so the + conservative direction is to fold only what is defined to fold. + """ + local, at, domain = address.rpartition("@") + if not at: + # Not an address shape we can split. RDAP data is third-party and + # occasionally malformed; group it by its literal text rather than + # inventing a domain for it. + return address + return f"{local}@{domain.lower()}" + + +def email_destinations(contacts: list[dict]) -> list[dict]: + """Group contacts into one destination per abuse ADDRESS. + + Contacts already fold by host, but two different contacts can still + resolve to the same address, an IP and a domain at one hoster being the + common case. One mail per address rather than per contact is what stops + a desk receiving two mails about one incident. + + Ids are numbered over the DESTINATIONS produced, not over the contacts + read, so a contact that resolved to no desk leaves no hole: a reviewer + who sees "email-2" and "email-4" reasonably reads two reports as + missing. + + Insertion order carries the numbering, so the same contacts in the same + order always produce the same ids. That matters downstream: bodies are + written to bodies/.xarf and hashed against that id, so an id that + wandered between two runs over one input would compare one desk's body + against another's. It is NOT a promise that an id survives a change in + the contacts themselves; adding a desk earlier in the list renumbers + every desk after it. + """ + by_address: dict[str, dict] = {} + + for contact in contacts: + for address in contact.get("abuse", []): + key = _group_key(address) + # First spelling seen wins the target. Any spelling reaches the + # desk, and picking one keeps the report stable across a re-run. + destination = by_address.setdefault(key, {"target": address, + "iocs": []}) + for ioc in contact.get("iocs", []): + if ioc not in destination["iocs"]: + destination["iocs"].append(ioc) + + return [ + { + "id": f"email-{index}", + "kind": "email", + "target": destination["target"], + "iocs": destination["iocs"], + "body": None, + "status": "pending", + } + for index, destination in enumerate(by_address.values(), start=1) + ] 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