From a3ff8f3b056c708872c5aedb4750d5cd32c5833d Mon Sep 17 00:00:00 2001
From: "Danilo M." <danix@danix.xyz>
Date: Wed, 9 Sep 2026 09:18:49 +0200
Subject: feat: select an RDAP server by longest prefix and by TLD

Longest prefix rather than first match: a block delegated to a new
operator appears as a more specific range inside its parent, and the
wider range would name the operator that gave it away.

A TLD that publishes no RDAP service selects nothing, which is a normal
outcome for many TLDs rather than a defect.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Wrfqr2xqQfhtXCscU7zrdz
---
 abusectl/rdap.py   | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 tests/test_rdap.py | 51 +++++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 103 insertions(+)

diff --git a/abusectl/rdap.py b/abusectl/rdap.py
index 92466b5..fc9cb12 100644
--- a/abusectl/rdap.py
+++ b/abusectl/rdap.py
@@ -129,3 +129,55 @@ def bootstrap(registry: str, cache_root=None, fetch=http_fetch) -> dict:
     directory.mkdir(parents=True, exist_ok=True)
     path.write_text(json.dumps(data))
     return data
+
+
+import ipaddress
+
+
+def server_for_ip(address: str, bootstrap_data: dict) -> str | None:
+    """Return the RDAP base URL for an address, by longest prefix.
+
+    Longest prefix rather than first match: a block delegated to a new
+    operator appears as a more specific range inside its parent, and the
+    wider one would name the operator that gave it away.
+    """
+    try:
+        ip = ipaddress.ip_address(address)
+    except ValueError:
+        return None
+
+    best_length = -1
+    best_url = None
+
+    for entry in bootstrap_data.get("services", []):
+        ranges, urls = entry[0], entry[1]
+        if not urls:
+            continue
+        for cidr in ranges:
+            try:
+                network = ipaddress.ip_network(cidr, strict=False)
+            except ValueError:
+                continue
+            if ip.version != network.version or ip not in network:
+                continue
+            if network.prefixlen > best_length:
+                best_length = network.prefixlen
+                best_url = urls[0]
+
+    return best_url
+
+
+def server_for_tld(tld: str, bootstrap_data: dict) -> str | None:
+    """Return the RDAP base URL for a TLD, or None when none is published.
+
+    Many TLDs publish no RDAP service at all, and that is a normal
+    outcome rather than a defect.
+    """
+    wanted = tld.lower().strip(".")
+    for entry in bootstrap_data.get("services", []):
+        names, urls = entry[0], entry[1]
+        if not urls:
+            continue
+        if any(name.lower() == wanted for name in names):
+            return urls[0]
+    return None
diff --git a/tests/test_rdap.py b/tests/test_rdap.py
index 3fc2c65..14a1fd4 100644
--- a/tests/test_rdap.py
+++ b/tests/test_rdap.py
@@ -125,5 +125,56 @@ class Bootstrap(unittest.TestCase):
             rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)
 
 
+class ServerSelection(unittest.TestCase):
+    IPV4 = {
+        "services": [
+            [["192.0.2.0/24"], ["https://wide.example.invalid/"]],
+            [["192.0.2.128/25"], ["https://narrow.example.invalid/"]],
+            [["198.51.100.0/24"], ["https://other.example.invalid/"]],
+        ]
+    }
+    DNS = {
+        "services": [
+            [["invalid"], ["https://registry.example.invalid/"]],
+            [["test"], ["https://test.example.invalid/"]],
+        ]
+    }
+
+    def test_an_address_selects_its_range(self):
+        self.assertEqual(
+            rdap.server_for_ip("198.51.100.7", self.IPV4),
+            "https://other.example.invalid/",
+        )
+
+    def test_the_longest_prefix_wins(self):
+        """192.0.2.200 is in both /24 and /25; the /25 is more specific.
+
+        Choosing the wider range would ask a registry that has delegated
+        the block away, and its answer would name the wrong operator.
+        """
+        self.assertEqual(
+            rdap.server_for_ip("192.0.2.200", self.IPV4),
+            "https://narrow.example.invalid/",
+        )
+
+    def test_an_unlisted_address_selects_nothing(self):
+        self.assertIsNone(rdap.server_for_ip("203.0.113.9", self.IPV4))
+
+    def test_a_tld_selects_its_registry(self):
+        self.assertEqual(
+            rdap.server_for_tld("invalid", self.DNS),
+            "https://registry.example.invalid/",
+        )
+
+    def test_tld_matching_ignores_case(self):
+        self.assertEqual(
+            rdap.server_for_tld("INVALID", self.DNS),
+            "https://registry.example.invalid/",
+        )
+
+    def test_an_unlisted_tld_selects_nothing(self):
+        self.assertIsNone(rdap.server_for_tld("example", self.DNS))
+
+
 if __name__ == "__main__":
     unittest.main()
-- 
cgit v1.2.3