diff options
| -rw-r--r-- | abusectl/report.py | 91 | ||||
| -rw-r--r-- | tests/test_report.py | 208 |
2 files changed, 296 insertions, 3 deletions
diff --git a/abusectl/report.py b/abusectl/report.py index a624142..76b195b 100644 --- a/abusectl/report.py +++ b/abusectl/report.py @@ -37,6 +37,31 @@ _ROLE_MAILBOXES = frozenset({ }) +def _is_mailable(address: str) -> bool: + """Whether this value can be the target of a mail at all. + + RDAP jCard data is third-party and occasionally malformed, so a + published "abuse address" is not guaranteed to be one. A value with no + "@", or with either half empty, cannot be delivered to anyone. + + The check exists because the alternative is silent and worse than a + missing contact. A malformed value used to become a destination with + status "pending" and an unsendable target, so the indicator appeared + on its way to a desk, no desk would ever receive it, and it was + excluded from unreportable() precisely because its contact HAD an + abuse entry. That is the failure the unreportable array exists to + prevent, arriving through the one door that array does not watch. + + Deliberately shallow: this is not address validation and must not + become it. Whether a syntactically fine address reaches a live desk is + the mail transport's answer, not a parser's, and rejecting an address + a desk actually reads would drop a report. It rejects only what cannot + be a mailbox under any reading. + """ + local, at, domain = address.rpartition("@") + return bool(at and local.strip() and domain.strip()) + + def _group_key(address: str) -> str: """The key two spellings of one desk must share, and no more than that. @@ -67,9 +92,10 @@ def _group_key(address: str) -> str: """ 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. + # Not an address shape we can split. _is_mailable keeps these out + # of the destinations, but the key stays defined for anything that + # asks for an id directly; group it by its literal text rather + # than inventing a domain for it. return address if local.lower() in _ROLE_MAILBOXES: local = local.lower() @@ -124,6 +150,10 @@ def email_destinations(contacts: list[dict]) -> list[dict]: for contact in contacts: for address in contact.get("abuse", []): + if not _is_mailable(address): + # unreportable() applies the same test, so the indicator + # is listed there rather than vanishing between the two. + continue 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. @@ -144,3 +174,58 @@ def email_destinations(contacts: list[dict]) -> list[dict]: } for destination in by_address.values() ] + + +def unreportable(contacts: list[dict]) -> list[dict]: + """List every IOC that reached no email destination, with the reason. + + A missing contact is a normal outcome, not an error: RDAP publishes no + abuse role for many netblocks. Making it visible is what keeps review + honest, since finding it any other way means diffing the IOC list + against every destination's IOC list. Same instinct as + suspect_path_segments flagging rather than redacting. + + The membership test is "reached no destination", not "sits in a + contact that resolved nothing", and those differ. Contacts fold by + HOST, so one indicator can appear in two contacts, a domain that + resolved and an IP that did not. Listing it because one of its + contacts failed would put it in the destinations AND in the list of + things no desk was found for, in one manifest. A reviewer reads the + second and hand-reports an indicator already on its way to a desk, + which costs the exact diffing this array exists to spare them. So an + indicator is unreportable only when NONE of its contacts produced a + mailable address, and the mailability test is the one + email_destinations applies, so no indicator can fall between them. + + Each indicator appears ONCE. Two failed contacts for one host are two + rows about one indicator otherwise, possibly with different reasons. + First reason seen wins, matching the first-seen ordering of the + destinations: both lists then read in the order the indicators were + found, which is the only ordering a human can explain. + """ + reachable = set() + for contact in contacts: + if any(_is_mailable(address) + for address in contact.get("abuse", [])): + reachable.update(contact.get("iocs", [])) + + result: list[dict] = [] + listed = set() + for contact in contacts: + if any(_is_mailable(address) + for address in contact.get("abuse", [])): + continue + # A contact's own error says more than the fallback, which is why + # it wins even when the contact published an unusable address. + # An empty string is not a reason: a manifest is a file the user + # edits, and a blank reason renders as a blank cell that tells + # them nothing. + reason = contact.get("error") or ( + "no usable abuse address published" if contact.get("abuse") + else "no abuse address resolved") + for ioc in contact.get("iocs", []): + if ioc in reachable or ioc in listed: + continue + listed.add(ioc) + result.append({"ioc": ioc, "reason": reason}) + return result diff --git a/tests/test_report.py b/tests/test_report.py index 5649b0b..b572b54 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -306,5 +306,213 @@ class Grouping(unittest.TestCase): self.assertEqual(crowded[2]["id"], alone[0]["id"]) +class Unreportable(unittest.TestCase): + def test_an_ioc_with_no_desk_is_listed_with_its_reason(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2", "ioc-3"], "query": "example.invalid", + "abuse": [], "source": "rdap", + "error": "no abuse role published"}, + ] + self.assertEqual( + report.unreportable(contacts), + [ + {"ioc": "ioc-2", "reason": "no abuse role published"}, + {"ioc": "ioc-3", "reason": "no abuse role published"}, + ], + ) + + def test_a_missing_reason_still_produces_an_entry(self): + contacts = [{"iocs": ["ioc-9"], "query": "x.invalid", "abuse": [], + "source": "rdap"}] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-9", "reason": "no abuse address resolved"}], + ) + + def test_an_empty_reason_does_not_read_as_no_reason(self): + """`error: ""` must not be reported as the literal empty string. + + A contact entry is written by contacts.resolve, but a manifest is + a file on disk that a user edits during review. An empty reason + renders as a blank cell in the report the user reads, which says + nothing at all; the default at least says what happened. + """ + contacts = [{"iocs": ["ioc-9"], "query": "x.invalid", "abuse": [], + "source": "rdap", "error": ""}] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-9", "reason": "no abuse address resolved"}], + ) + + def test_nothing_unreportable_is_an_empty_list_not_an_error(self): + contacts = [{"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}] + self.assertEqual(report.unreportable(contacts), []) + + def test_an_ioc_that_reached_a_desk_elsewhere_is_not_unreportable(self): + """Hosts fold, so one IOC can sit in a resolved and an unresolved + contact at once. It IS reportable, and listing it says otherwise. + + The plan's implementation listed it regardless, which puts an + indicator in both the destination list and the "no desk found" + list of one manifest. A reviewer reading the second acts on an + indicator that is already on its way to a desk, and the whole + point of the array is that it can be trusted without diffing. + """ + contacts = [ + {"iocs": ["ioc-1", "ioc-2"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2", "ioc-3"], "query": "example.invalid", + "abuse": [], "source": "rdap", + "error": "no abuse role published"}, + ] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-3", "reason": "no abuse role published"}], + ) + + def test_one_ioc_unresolved_twice_is_listed_once(self): + """Two contacts, both unresolved, one shared indicator. + + A duplicate row is a second line in the report about one + indicator, and the reasons may differ, so which one wins has to + be decided rather than left to whichever contact came last. + First reason seen wins, matching the first-seen ordering the + destinations use. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + {"iocs": ["ioc-1"], "query": "198.51.100.7", "abuse": [], + "source": "rdap", "error": "no rdap server for this tld, " + "or no answer"}, + ] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-1", "reason": "no abuse role published"}], + ) + + def test_the_contacts_passed_in_are_not_modified(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + {"iocs": ["ioc-2"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + before = copy.deepcopy(contacts) + report.unreportable(contacts) + self.assertEqual(contacts, before) + + +class MalformedAddresses(unittest.TestCase): + """An abuse "address" with no @ cannot be mailed. + + RDAP jCard data is third-party and occasionally malformed, and a + destination built from such a value carries an unsendable target with + status "pending". That is the failure mode the unreportable array + exists to prevent: the indicator appears reportable, no desk ever + receives it, and nothing in the manifest says so. + """ + + def test_a_target_with_no_at_creates_no_destination(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["not-an-address"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual([d["target"] for d in destinations], + ["abuse@host.invalid"]) + self.assertEqual(destinations[0]["iocs"], ["ioc-2"]) + + def test_an_ioc_whose_only_address_is_malformed_is_unreportable(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["not-an-address"], "source": "rdap"}, + ] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-1", + "reason": "no usable abuse address published"}], + ) + + def test_a_usable_address_beside_a_malformed_one_still_reports(self): + """The good half of a jCard must survive the bad half. + + Discarding the contact wholesale would lose a real desk over a + neighbouring malformed row. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["not-an-address", "abuse@host.invalid"], + "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual([d["target"] for d in destinations], + ["abuse@host.invalid"]) + self.assertEqual(report.unreportable(contacts), []) + + def test_an_addresss_own_error_is_not_overwritten_by_the_default(self): + """A contact that has both a reason and a malformed address. + + The contact's own error says more than "no usable address", so it + wins; the default is only for a contact that offered no reason. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["not-an-address"], "source": "rdap", + "error": "no abuse role published"}, + ] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-1", "reason": "no abuse role published"}], + ) + + def test_the_two_lists_partition_every_indicator(self): + """The invariant the pair is for: each IOC is in exactly one. + + Every other test here pins one side. This pins the relationship, + which is what a reviewer actually relies on: an indicator missing + from both is silently unreported, and one in both is reported and + also flagged as unreported. Both failures come from the two + functions disagreeing about what counts as a desk, so they are + asserted against one input that exercises every branch. + """ + contacts = [ + {"iocs": ["ioc-1", "ioc-2"], "query": "198.51.100.7", + "abuse": ["Abuse@Host.Invalid"], "source": "rdap"}, + {"iocs": ["ioc-2", "ioc-3"], "query": "example.invalid", + "abuse": [], "source": "rdap", + "error": "no abuse role published"}, + {"iocs": ["ioc-4"], "query": "other.invalid", + "abuse": ["not-an-address"], "source": "rdap"}, + {"iocs": ["ioc-5"], "query": "mixed.invalid", + "abuse": ["broken", "abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + reported = {i for d in destinations for i in d["iocs"]} + flagged = {e["ioc"] for e in report.unreportable(contacts)} + every = {i for c in contacts for i in c["iocs"]} + + self.assertEqual(reported & flagged, set()) + self.assertEqual(reported | flagged, every) + self.assertEqual(reported, {"ioc-1", "ioc-2", "ioc-5"}) + self.assertEqual(flagged, {"ioc-3", "ioc-4"}) + + def test_an_empty_or_whitespace_target_is_not_a_desk(self): + for value in ("", " ", "@host.invalid", "abuse@"): + with self.subTest(value=value): + contacts = [{"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": [value], "source": "rdap"}] + self.assertEqual(report.email_destinations(contacts), []) + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-1", + "reason": "no usable abuse address published"}]) + + if __name__ == "__main__": unittest.main() |