diff options
| -rw-r--r-- | abusectl/report.py | 37 | ||||
| -rw-r--r-- | tests/test_report.py | 57 |
2 files changed, 94 insertions, 0 deletions
diff --git a/abusectl/report.py b/abusectl/report.py index def7f3a..9897cc8 100644 --- a/abusectl/report.py +++ b/abusectl/report.py @@ -423,6 +423,43 @@ def email_destinations(contacts: list[dict]) -> list[dict]: ] +def vendor_destinations(iocs: list[dict], configured: set) -> list[dict]: + """Build one row per configured destination that has something to send. + + BOTH conditions have to hold. A row for an unconfigured destination is a + promise that can only fail; a row for a configured one with nothing it + accepts is a promise with no content, an AbuseIPDB submission with + nothing to put in its `ip` parameter. + + Each row carries ONLY the IOCs its own destination accepts. A row is + what submit iterates, so a urlhaus row listing an IP is a submission + that gets built wrong or dropped at send time, whichever the + implementer notices first. + + Order comes from DESTINATIONS rather than from `configured`, which is a + set and therefore has no order worth writing into a manifest twice. + """ + rows = [] + for name, spec in DESTINATIONS.items(): + if name not in configured: + continue + accepted = [ioc["id"] for ioc in iocs if ioc.get("type") in spec["accepts"]] + if not accepted: + continue + rows.append({ + "id": name, + "kind": spec["kind"], + "iocs": accepted, + # Null until submit settles each payload shape. Writing a + # vendor's JSON now would mean guessing an endpoint's contract. + # No body_sha256 either: that hash records what was disclosed, + # and nothing has been. + "body": None, + "status": "pending", + }) + return rows + + def unreportable(contacts: list[dict]) -> list[dict]: """List every IOC that reached no email destination, with the reason. diff --git a/tests/test_report.py b/tests/test_report.py index 6d1f845..d0e8239 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -2049,3 +2049,60 @@ class DestinationTable(unittest.TestCase): # MISP only, so this test names them rather than deriving them. self.assertEqual(report.ALL_TYPES, ("ipv4", "ipv6", "domain", "url", "sha256")) + + +class VendorDestinations(unittest.TestCase): + IOCS = [ + {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}, + {"id": "ioc-2", "type": "domain", "value": "example.invalid"}, + {"id": "ioc-3", "type": "url", "value": "http://example.invalid/a"}, + ] + + def test_a_row_carries_only_the_types_its_destination_accepts(self): + rows = report.vendor_destinations(self.IOCS, {"abusedb", "urlhaus"}) + by_id = {row["id"]: row for row in rows} + self.assertEqual(by_id["abusedb"]["iocs"], ["ioc-1"]) + self.assertEqual(by_id["urlhaus"]["iocs"], ["ioc-3"]) + + def test_misp_carries_every_ioc_including_what_no_vendor_takes(self): + rows = report.vendor_destinations(self.IOCS, {"misp"}) + self.assertEqual(rows[0]["iocs"], ["ioc-1", "ioc-2", "ioc-3"]) + + def test_a_destination_with_no_acceptable_ioc_gets_no_row(self): + # The rule's whole point: a case with no URL must not promise a + # urlhaus submission that would have nothing to submit. + no_urls = [ioc for ioc in self.IOCS if ioc["type"] != "url"] + rows = report.vendor_destinations(no_urls, {"abusedb", "urlhaus"}) + self.assertEqual([row["id"] for row in rows], ["abusedb"]) + + def test_an_unconfigured_destination_gets_no_row(self): + rows = report.vendor_destinations(self.IOCS, set()) + self.assertEqual(rows, []) + + def test_the_row_shape_is_the_shape_submit_iterates(self): + rows = report.vendor_destinations(self.IOCS, {"abusedb"}) + self.assertEqual(rows[0], { + "id": "abusedb", + "kind": "api", + "iocs": ["ioc-1"], + "body": None, + "status": "pending", + }) + + def test_no_row_carries_a_target(self): + # An endpoint is a property of the vendor, not of the case. Writing + # one into the evidence record would let a hand-edit redirect a + # submission somewhere the user never named. + rows = report.vendor_destinations(self.IOCS, {"misp", "abusedb"}) + for row in rows: + self.assertNotIn("target", row) + + def test_rows_come_out_in_table_order_not_set_order(self): + # The configured set is a set, so a stable order has to come from + # somewhere else or the manifest churns between runs. + first = report.vendor_destinations(self.IOCS, {"urlhaus", "misp", "abusedb"}) + second = report.vendor_destinations(self.IOCS, {"abusedb", "misp", "urlhaus"}) + self.assertEqual([row["id"] for row in first], + [row["id"] for row in second]) + self.assertEqual([row["id"] for row in first], + ["misp", "abusedb", "urlhaus"]) |