diff options
| -rw-r--r-- | abusectl/rdap.py | 25 | ||||
| -rw-r--r-- | tests/test_rdap.py | 89 |
2 files changed, 112 insertions, 2 deletions
diff --git a/abusectl/rdap.py b/abusectl/rdap.py index 66638c7..14099e2 100644 --- a/abusectl/rdap.py +++ b/abusectl/rdap.py @@ -31,6 +31,7 @@ discipline _MAX_REDIRECT_DEPTH already sets in parse.py. import json import urllib.error +import urllib.parse import urllib.request _TIMEOUT = 10 @@ -324,6 +325,23 @@ def abuse_addresses(response: dict) -> list[str]: _MAX_LABEL_WALK = 5 +def _quoted(component: str) -> str: + """Percent-encode a component so it cannot escape its path segment. + + safe="" rather than urllib's default safe="/": leaving the separator + intact is exactly the traversal shape this guards against, and a + component ending in "/../.." would address a path the registry never + published. A bare "%", as an IPv6 scope id carries, is a truncated + escape rather than a literal percent, so it is encoded too. + + Defence in depth, not validation. contacts.is_queryable() is the + admission point and stays the real gate; this is the layer that holds + if a later branch reaches these functions without passing through it, + which is how the same property leaked twice already. + """ + return urllib.parse.quote(component, safe="") + + def query_ip(address: str, bootstrap_data: dict, fetch=http_fetch) -> dict | None: """Query the registry responsible for an address. @@ -333,7 +351,7 @@ def query_ip(address: str, bootstrap_data: dict, fetch=http_fetch) -> dict | Non base = server_for_ip(address, bootstrap_data) if base is None: return None - return fetch(f"{base.rstrip('/')}/ip/{address}") + return fetch(f"{base.rstrip('/')}/ip/{_quoted(address)}") def query_domain( @@ -367,7 +385,10 @@ def query_domain( candidate = ".".join(labels[start:]) attempts += 1 try: - return fetch(f"{base}/domain/{candidate}"), candidate + # The URL is quoted, the returned candidate is NOT: the manifest + # records and the review dialog shows the name the user has to + # recognise, and %2E%2E is not that name. + return fetch(f"{base}/domain/{_quoted(candidate)}"), candidate except Exception: continue diff --git a/tests/test_rdap.py b/tests/test_rdap.py index 7f2f0d5..815c311 100644 --- a/tests/test_rdap.py +++ b/tests/test_rdap.py @@ -445,5 +445,94 @@ class Query(unittest.TestCase): self.assertEqual(len(calls), 5) +class QuotedQueryComponent(unittest.TestCase): + """The interpolated component cannot escape its path segment. + + contacts.is_queryable() is the admission point and validates every + candidate, but that guarantee has already leaked three times in this + module's history, each time the same shape: a validator applied to one + branch and forgotten on its sibling. These call query_ip and + query_domain DIRECTLY, bypassing contacts entirely, because a future + branch that skips worklist() is exactly the failure this layer exists + to survive. + """ + + IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]} + IPV6 = {"services": [[["fe80::/10"], ["https://rir.example.invalid/"]]]} + DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]} + + def _recorder(self): + calls = [] + + def fetch(url): + calls.append(url) + return {"handle": "X", "entities": []} + + return calls, fetch + + def test_a_traversal_in_an_ip_stays_under_the_ip_segment(self): + """server_for_ip refuses this string today, so the base is forced. + + Forcing it is the point: this asserts what query_ip does with a + component it was handed, not what today's lookup happens to reject. + """ + calls, fetch = self._recorder() + original = rdap.server_for_ip + rdap.server_for_ip = lambda address, data: "https://rir.example.invalid/" + try: + rdap.query_ip("198.51.100.7/../../etc", self.IPV4, fetch=fetch) + finally: + rdap.server_for_ip = original + + self.assertEqual( + calls, + ["https://rir.example.invalid/ip/198.51.100.7%2F..%2F..%2Fetc"], + ) + + def test_a_scope_id_in_an_ip_is_encoded_not_left_malformed(self): + """A bare % in a URL is a truncated escape, not a literal percent. + + server_for_ip accepts fe80::1%eth0 today, so this one reaches the + wire through the normal path. + """ + calls, fetch = self._recorder() + rdap.query_ip("fe80::1%eth0", self.IPV6, fetch=fetch) + + self.assertEqual( + calls, ["https://rir.example.invalid/ip/fe80%3A%3A1%25eth0"] + ) + + def test_a_query_and_fragment_in_a_host_are_not_live(self): + calls, fetch = self._recorder() + rdap.query_domain("victim?e=x#frag.example.invalid", self.DNS, fetch=fetch) + + self.assertEqual(len(calls), 1) + url = calls[0] + self.assertNotIn("?", url) + self.assertNotIn("#", url) + self.assertIn("%3F", url) + self.assertIn("%23", url) + + def test_a_normal_ip_url_is_unchanged(self): + calls, fetch = self._recorder() + rdap.query_ip("198.51.100.7", self.IPV4, fetch=fetch) + self.assertEqual(calls, ["https://rir.example.invalid/ip/198.51.100.7"]) + + def test_a_normal_host_url_is_unchanged(self): + calls, fetch = self._recorder() + rdap.query_domain("example.invalid", self.DNS, fetch=fetch) + self.assertEqual( + calls, ["https://registry.example.invalid/domain/example.invalid"] + ) + + def test_the_returned_candidate_is_unquoted(self): + """The manifest and the review dialog show the name, not the URL.""" + calls, fetch = self._recorder() + _, queried = rdap.query_domain( + "victim?e=x#frag.example.invalid", self.DNS, fetch=fetch + ) + self.assertEqual(queried, "victim?e=x#frag.example.invalid") + + if __name__ == "__main__": unittest.main() |