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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Wrfqr2xqQfhtXCscU7zrdz

2 files changed, 186 insertions, 0 deletions

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()