aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorDanilo M. <danix@danix.xyz>2026-09-09 09:17:43 +0200
committerDanilo M. <danix@danix.xyz>2026-09-09 09:17:43 +0200
commitc1887ab10ee16bbaf19025575759efa031133b65 (patch)
tree89de9c562da74e8563d1db773544c903431a7cfe
parent381ccdbae705b141282b1f7b344a8d5074c487ac (diff)
downloadabusectl-c1887ab10ee16bbaf19025575759efa031133b65.tar.gz
abusectl-c1887ab10ee16bbaf19025575759efa031133b65.zip
feat: add the RDAP transport, with a redirect cap and no downgrade
The only socket in this tool. A redirect is remote data directing our next request, so hops are capped and an https to http downgrade is refused: a downgraded query travels in clear text and discloses which netblock is under investigation to anyone on the path. The timeout is mandatory rather than defaulted, because urllib with no timeout blocks forever and a hung registry would hang a review. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Wrfqr2xqQfhtXCscU7zrdz
-rw-r--r--abusectl/rdap.py72
-rw-r--r--tests/test_rdap.py56
2 files changed, 128 insertions, 0 deletions
diff --git a/abusectl/rdap.py b/abusectl/rdap.py
new file mode 100644
index 0000000..4cea786
--- /dev/null
+++ b/abusectl/rdap.py
@@ -0,0 +1,72 @@
+# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+"""RDAP: which registry to ask, how to ask it, and how to read the answer.
+
+This is the first module in abusectl that opens a socket. Everything that
+looks defensive here is defensive because of that.
+
+The network entry point is a single `fetch` callable, passed as an argument
+everywhere above it and defaulting to `http_fetch` below. Tests pass a fake
+and never construct the real one, which is what keeps the whole suite
+passing with socket.socket, socket.create_connection and socket.getaddrinfo
+all raising. A network module that can only be tested with a network is a
+module that stops being tested.
+
+Redirects are capped and an https to http downgrade is refused, because a
+redirect is remote data directing our next request. This follows the
+discipline _MAX_REDIRECT_DEPTH already sets in parse.py.
+"""
+
+import json
+import urllib.error
+import urllib.request
+
+_TIMEOUT = 10
+_MAX_REDIRECTS = 5
+_ACCEPT = "application/rdap+json, application/json;q=0.9"
+
+
+class _NoDowngradeRedirectHandler(urllib.request.HTTPRedirectHandler):
+ """Refuse a redirect that drops from https to http.
+
+ urllib follows redirects by default and will happily downgrade. A
+ downgraded RDAP query travels in clear text, disclosing which netblock
+ the user is investigating to anyone on the path.
+ """
+
+ max_redirections = _MAX_REDIRECTS
+
+ def redirect_request(self, req, fp, code, msg, headers, newurl):
+ if req.get_full_url().startswith("https://") and newurl.startswith("http://"):
+ raise urllib.error.HTTPError(
+ newurl, code,
+ "refusing an https to http redirect",
+ headers, fp,
+ )
+ return super().redirect_request(req, fp, code, msg, headers, newurl)
+
+
+_opener = urllib.request.build_opener(_NoDowngradeRedirectHandler())
+
+
+def http_fetch(url: str) -> dict:
+ """GET a URL and parse the JSON body. The only socket in this tool.
+
+ A timeout is mandatory rather than defaulted: urllib with no timeout
+ blocks forever, and a hung registry would hang a review.
+ """
+ request = urllib.request.Request(url, headers={"Accept": _ACCEPT})
+ with _opener.open(request, timeout=_TIMEOUT) as response:
+ return json.loads(response.read())
diff --git a/tests/test_rdap.py b/tests/test_rdap.py
new file mode 100644
index 0000000..681e2bb
--- /dev/null
+++ b/tests/test_rdap.py
@@ -0,0 +1,56 @@
+# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+"""Tests for the RDAP protocol module."""
+
+import email
+import unittest
+import urllib.error
+import urllib.request
+
+from abusectl import rdap
+
+
+class RedirectPolicy(unittest.TestCase):
+ """A redirect is remote data directing our next request.
+
+ urllib.request.Request is used rather than a hand-rolled fake, because
+ HTTPRedirectHandler reads attributes (origin_req_host, unverifiable,
+ timeout) that a fake would have to reproduce exactly to prove anything.
+ """
+
+ def _request(self):
+ return urllib.request.Request("https://rdap.example.invalid/ip/192.0.2.1")
+
+ def test_an_https_to_http_downgrade_is_refused(self):
+ handler = rdap._NoDowngradeRedirectHandler()
+ with self.assertRaises(urllib.error.HTTPError):
+ handler.redirect_request(
+ self._request(), None, 302, "Found",
+ email.message_from_string(""),
+ "http://rdap.example.invalid/ip/192.0.2.1",
+ )
+
+ def test_an_https_to_https_redirect_is_allowed(self):
+ handler = rdap._NoDowngradeRedirectHandler()
+ result = handler.redirect_request(
+ self._request(), None, 302, "Found",
+ email.message_from_string(""),
+ "https://other.example.invalid/ip/192.0.2.1",
+ )
+ self.assertIsNotNone(result)
+
+
+if __name__ == "__main__":
+ unittest.main()