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