From 96beac190b8444fc5b7629daebbf4a67f5ab11e2 Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Wed, 9 Sep 2026 09:25:36 +0200 Subject: feat: resolve the worklist to abuse contacts Failure is per query and never stops the run: a timeout on one indicator must not cost the contacts that did resolve, and a missing contact is a normal outcome rather than an error. Bootstraps are passed in rather than fetched here, so this stays testable with no network and the caller owns the cache policy. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Wrfqr2xqQfhtXCscU7zrdz --- abusectl/contacts.py | 60 +++++++++++++++++++++++ tests/test_contacts.py | 126 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 186 insertions(+) diff --git a/abusectl/contacts.py b/abusectl/contacts.py index c4c263d..a0857c4 100644 --- a/abusectl/contacts.py +++ b/abusectl/contacts.py @@ -106,3 +106,63 @@ def worklist(iocs: list[dict]) -> list[WorkItem]: add("ip" if _is_ip(host) else "domain", host, ioc_id) return list(items.values()) + + +def resolve(iocs: list[dict], bootstraps: dict, fetch=rdap.http_fetch) -> list[dict]: + """Resolve every resolvable indicator to an abuse contact. + + bootstraps is {"ipv4": ..., "ipv6": ..., "dns": ...}, passed in rather + than fetched here so this function stays testable with no network and + so the caller owns the cache policy. + + A failure is recorded per query and never stops the run: a timeout on + one indicator must not cost the contacts that did resolve. A missing + contact is a normal outcome, not an error. + """ + results = [] + + for item in worklist(iocs): + entry = { + "iocs": item.iocs, + "query": item.query, + "abuse": [], + "source": "rdap", + } + + try: + if item.kind == "ip": + family = "ipv6" if ":" in item.query else "ipv4" + response = rdap.query_ip( + item.query, bootstraps.get(family, {}), fetch=fetch + ) + if response is None: + entry["error"] = "no rdap server for this range" + results.append(entry) + continue + else: + response, queried = rdap.query_domain( + item.query, bootstraps.get("dns", {}), fetch=fetch + ) + if response is None: + entry["error"] = "no rdap server for this tld, or no answer" + results.append(entry) + continue + if queried and queried != item.query: + entry["queried"] = queried + except Exception as exc: + entry["error"] = f"{type(exc).__name__}: {exc}" + results.append(entry) + continue + + handle = response.get("handle") + if handle: + entry["handle"] = handle + + addresses = rdap.abuse_addresses(response) + entry["abuse"] = addresses + if not addresses: + entry["error"] = "no abuse role published" + + results.append(entry) + + return results diff --git a/tests/test_contacts.py b/tests/test_contacts.py index a6ddabe..0ac197d 100644 --- a/tests/test_contacts.py +++ b/tests/test_contacts.py @@ -109,5 +109,131 @@ class Worklist(unittest.TestCase): self.assertEqual(contacts.worklist(iocs), []) +class Resolve(unittest.TestCase): + IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]} + IPV6 = {"services": []} + DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]} + + def _bootstraps(self): + return {"ipv4": self.IPV4, "ipv6": self.IPV6, "dns": self.DNS} + + def test_an_ip_resolves_to_its_abuse_desk(self): + def fetch(url): + return { + "handle": "NET-1", + "entities": [{ + "roles": ["abuse"], + "vcardArray": ["vcard", [ + ["version", {}, "text", "4.0"], + ["email", {}, "text", "abuse@example.invalid"], + ]], + }], + } + + iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}] + result = contacts.resolve( + iocs, bootstraps=self._bootstraps(), fetch=fetch + ) + + self.assertEqual(len(result), 1) + self.assertEqual(result[0]["iocs"], ["ioc-1"]) + self.assertEqual(result[0]["query"], "198.51.100.7") + self.assertEqual(result[0]["abuse"], ["abuse@example.invalid"]) + self.assertEqual(result[0]["handle"], "NET-1") + self.assertNotIn("error", result[0]) + + def test_no_abuse_role_records_a_reason_not_an_error(self): + def fetch(url): + return {"handle": "NET-2", "entities": []} + + iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}] + result = contacts.resolve( + iocs, bootstraps=self._bootstraps(), fetch=fetch + ) + + self.assertEqual(result[0]["abuse"], []) + self.assertEqual(result[0]["error"], "no abuse role published") + + def test_no_rdap_server_records_a_reason(self): + def fetch(url): + raise AssertionError(f"should not have fetched {url}") + + iocs = [{"id": "ioc-1", "type": "domain", "value": "example.test"}] + result = contacts.resolve( + iocs, bootstraps=self._bootstraps(), fetch=fetch + ) + + self.assertEqual(result[0]["abuse"], []) + self.assertIn("no rdap server", result[0]["error"]) + + def test_a_network_failure_is_per_query_and_does_not_stop_the_run(self): + def fetch(url): + if "198.51.100.7" in url: + raise OSError("connection timed out") + return { + "handle": "DOM-1", + "entities": [{ + "roles": ["abuse"], + "vcardArray": ["vcard", [ + ["version", {}, "text", "4.0"], + ["email", {}, "text", "abuse@example.invalid"], + ]], + }], + } + + iocs = [ + {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}, + {"id": "ioc-2", "type": "domain", "value": "example.invalid"}, + ] + result = contacts.resolve( + iocs, bootstraps=self._bootstraps(), fetch=fetch + ) + + self.assertEqual(len(result), 2) + failed = [r for r in result if r["query"] == "198.51.100.7"][0] + worked = [r for r in result if r["query"] == "example.invalid"][0] + self.assertIn("connection timed out", failed["error"]) + self.assertEqual(worked["abuse"], ["abuse@example.invalid"]) + + def test_one_query_per_host_however_many_iocs(self): + calls = [] + + def fetch(url): + calls.append(url) + return {"handle": "DOM-1", "entities": []} + + iocs = [ + {"id": f"ioc-{n}", "type": "url", + "value": f"https://a.example.invalid/page{n}"} + for n in range(20) + ] + result = contacts.resolve( + iocs, bootstraps=self._bootstraps(), fetch=fetch + ) + + self.assertEqual(len(calls), 1) + self.assertEqual(len(result[0]["iocs"]), 20) + + def test_no_query_ever_carries_a_path(self): + """THE FOURTH PROPERTY, asserted at the transport.""" + calls = [] + + def fetch(url): + calls.append(url) + return {"handle": "DOM-1", "entities": []} + + iocs = [{ + "id": "ioc-1", "type": "url", + "value": "https://a.example.invalid/verify/victim%40example.org?e=x", + }] + contacts.resolve(iocs, bootstraps=self._bootstraps(), fetch=fetch) + + for url in calls: + self.assertNotIn("victim", url) + self.assertNotIn("verify", url) + self.assertNotIn("%40", url) + self.assertNotIn("?", url) + + if __name__ == "__main__": unittest.main() -- cgit v1.2.3