diff options
| -rw-r--r-- | abusectl/rdap.py | 52 | ||||
| -rw-r--r-- | tests/test_rdap.py | 51 |
2 files changed, 103 insertions, 0 deletions
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() |