feat: query RDAP, walking up the labels for a registrable domain

RDAP wants the registrable domain and a deep host is not one. Rather
than bundling a Public Suffix List, which is a transcribed table that
goes stale weekly and is the failure init.PROVIDERS already documents,
this asks the registry, which is the authority on what is registrable.

The walk is capped and never queries a bare TLD.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Wrfqr2xqQfhtXCscU7zrdz

2 files changed, 133 insertions, 0 deletions

diff --git a/abusectl/rdap.py b/abusectl/rdap.py
index 94a5d02..e1ca07f 100644
--- a/abusectl/rdap.py
+++ b/abusectl/rdap.py
@@ -268,3 +268,56 @@ def abuse_addresses(response: dict) -> list[str]:
             seen.add(address)
             unique.append(address)
     return unique
+
+
+_MAX_LABEL_WALK = 5
+
+
+def query_ip(address: str, bootstrap_data: dict, fetch=http_fetch) -> dict | None:
+    """Query the registry responsible for an address.
+
+    Returns None when no registry is listed for it, which is a normal
+    outcome rather than an error.
+    """
+    base = server_for_ip(address, bootstrap_data)
+    if base is None:
+        return None
+    return fetch(f"{base.rstrip('/')}/ip/{address}")
+
+
+def query_domain(
+    host: str, bootstrap_data: dict, fetch=http_fetch
+) -> tuple[dict | None, str | None]:
+    """Query for a host, walking up the labels to find the registrable name.
+
+    RDAP wants the registrable domain, and mail.deep.example.invalid is not
+    one. Rather than carrying a Public Suffix List, which is a transcribed
+    table that goes stale weekly, this asks the registry: it is the
+    authority on what is registrable.
+
+    Returns (response, queried_name). A bare TLD is never queried.
+    """
+    labels = host.lower().strip(".").split(".")
+    if len(labels) < 2:
+        return None, None
+
+    tld = labels[-1]
+    base = server_for_tld(tld, bootstrap_data)
+    if base is None:
+        return None, None
+    base = base.rstrip("/")
+
+    attempts = 0
+    # Stop before the bare TLD: range end is len(labels) - 1, so the last
+    # candidate is the two-label name.
+    for start in range(0, len(labels) - 1):
+        if attempts >= _MAX_LABEL_WALK:
+            break
+        candidate = ".".join(labels[start:])
+        attempts += 1
+        try:
+            return fetch(f"{base}/domain/{candidate}"), candidate
+        except Exception:
+            continue
+
+    return None, None
diff --git a/tests/test_rdap.py b/tests/test_rdap.py
index b7baa21..9be5bcd 100644
--- a/tests/test_rdap.py
+++ b/tests/test_rdap.py
@@ -261,5 +261,85 @@ class AbuseExtraction(unittest.TestCase):
         )
 
 
+class Query(unittest.TestCase):
+    IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]}
+    DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]}
+
+    def test_an_ip_query_hits_the_selected_server(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"handle": "NET-1", "entities": []}
+
+        result = rdap.query_ip("198.51.100.7", self.IPV4, fetch=fetch)
+
+        self.assertEqual(calls, ["https://rir.example.invalid/ip/198.51.100.7"])
+        self.assertEqual(result["handle"], "NET-1")
+
+    def test_an_unlisted_ip_is_not_queried(self):
+        def fetch(url):
+            raise AssertionError(f"should not have fetched {url}")
+
+        self.assertIsNone(rdap.query_ip("203.0.113.9", self.IPV4, fetch=fetch))
+
+    def test_the_label_walk_stops_at_the_first_answer(self):
+        """mail.deep.example.invalid is not registrable; example.invalid is.
+
+        The registry is the authority on what is registrable, which is why
+        this walks rather than carrying a Public Suffix List that would go
+        stale weekly.
+        """
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            if url.endswith("/domain/example.invalid"):
+                return {"handle": "DOM-1", "entities": []}
+            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)
+
+        result, queried = rdap.query_domain(
+            "mail.deep.example.invalid", self.DNS, fetch=fetch
+        )
+
+        self.assertEqual(queried, "example.invalid")
+        self.assertEqual(result["handle"], "DOM-1")
+        self.assertEqual(len(calls), 3)
+
+    def test_the_walk_never_queries_a_bare_tld(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)
+
+        result, queried = rdap.query_domain(
+            "deep.example.invalid", self.DNS, fetch=fetch
+        )
+
+        self.assertIsNone(result)
+        self.assertNotIn("https://registry.example.invalid/domain/invalid", calls)
+
+    def test_a_tld_with_no_server_is_not_queried(self):
+        def fetch(url):
+            raise AssertionError(f"should not have fetched {url}")
+
+        result, queried = rdap.query_domain(
+            "example.test", self.DNS, fetch=fetch
+        )
+        self.assertIsNone(result)
+
+    def test_the_walk_is_capped(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)
+
+        host = "a.b.c.d.e.f.g.example.invalid"
+        rdap.query_domain(host, self.DNS, fetch=fetch)
+        self.assertLessEqual(len(calls), rdap._MAX_LABEL_WALK)
+
+
 if __name__ == "__main__":
     unittest.main()