diff options
| -rw-r--r-- | abusectl/report.py | 99 | ||||
| -rw-r--r-- | tests/test_report.py | 143 |
2 files changed, 242 insertions, 0 deletions
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. <danix@danix.xyz> +# +# 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/<id>.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/<id>.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() |