diff options
| -rw-r--r-- | abusectl/rdap.py | 69 |
1 files changed, 60 insertions, 9 deletions
diff --git a/abusectl/rdap.py b/abusectl/rdap.py index e1ca07f..66638c7 100644 --- a/abusectl/rdap.py +++ b/abusectl/rdap.py @@ -134,6 +134,45 @@ def bootstrap(registry: str, cache_root=None, fetch=http_fetch) -> dict: import ipaddress +def _service_entries(bootstrap_data): + """Yield (keys, urls) for every well-formed entry, skipping the rest. + + The bootstrap document arrives over the same network as an RDAP + response, and abuse_addresses already treats a response as hostile. + A malformed entry here used to raise IndexError, TypeError or + KeyError out of server_for_ip, and bootstrap() caches whatever JSON it + is handed for seven days, so one bad response would break every + contacts run for a week with a traceback pointing at nothing the user + can act on. A malformed entry is skipped, never fatal. + """ + if not isinstance(bootstrap_data, dict): + return + services = bootstrap_data.get("services") + if not isinstance(services, list): + return + for entry in services: + if not isinstance(entry, list) or len(entry) < 2: + continue + keys, urls = entry[0], entry[1] + if not isinstance(keys, list) or not isinstance(urls, list) or not urls: + continue + yield keys, urls + + +def _secure_base(urls) -> str | None: + """Return the entry's first URL, but only when it is an https string. + + _NoDowngradeRedirectHandler guards REDIRECTS only, so a bootstrap + naming an http:// base would send the query in clear text and + disclose which netblock the user is investigating to anyone on the + path. Refusing it here makes the guarantee hold end to end. + """ + url = urls[0] + if not isinstance(url, str) or not url.startswith("https://"): + return None + return url + + def server_for_ip(address: str, bootstrap_data: dict) -> str | None: """Return the RDAP base URL for an address, by longest prefix. @@ -149,11 +188,21 @@ def server_for_ip(address: str, bootstrap_data: dict) -> str | None: best_length = -1 best_url = None - for entry in bootstrap_data.get("services", []): - ranges, urls = entry[0], entry[1] - if not urls: + for ranges, urls in _service_entries(bootstrap_data): + url = _secure_base(urls) + if url is None: continue for cidr in ranges: + # ipaddress.ip_network(16909060) does NOT raise, it returns + # 1.2.3.4/32, and a /32 is the longest possible prefix so it + # wins every contest. An attacker who can influence the + # bootstrap would steer the query for their own address to a + # server they control, which answers with an abuse address of + # their choosing, and the user files the phishing report to + # the phisher. Only a string is a CIDR. config.py rejects the + # same trap for the trusted relays. + if not isinstance(cidr, str): + continue try: network = ipaddress.ip_network(cidr, strict=False) except ValueError: @@ -162,7 +211,7 @@ def server_for_ip(address: str, bootstrap_data: dict) -> str | None: continue if network.prefixlen > best_length: best_length = network.prefixlen - best_url = urls[0] + best_url = url return best_url @@ -174,12 +223,14 @@ def server_for_tld(tld: str, bootstrap_data: dict) -> str | None: 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: + for names, urls in _service_entries(bootstrap_data): + url = _secure_base(urls) + if url is None: continue - if any(name.lower() == wanted for name in names): - return urls[0] + # A non-string name raised AttributeError on .lower(); see the CIDR + # note in server_for_ip for why the document cannot be trusted. + if any(isinstance(name, str) and name.lower() == wanted for name in names): + return url return None |