feat: cache the IANA bootstrap, and prefer a stale copy to none

Seven day TTL under XDG_CACHE_HOME, deliberately not in a case
directory: this is a copy of a public map, not evidence.

A failed refetch falls back to the stale copy. Staleness is safe in this
direction, since a server that has moved simply misses and reads as no
contact, while losing IANA entirely would stop the user filing a report.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Wrfqr2xqQfhtXCscU7zrdz

2 files changed, 132 insertions, 0 deletions

diff --git a/abusectl/rdap.py b/abusectl/rdap.py
index 4cea786..92466b5 100644
--- a/abusectl/rdap.py
+++ b/abusectl/rdap.py
@@ -70,3 +70,62 @@ def http_fetch(url: str) -> dict:
     request = urllib.request.Request(url, headers={"Accept": _ACCEPT})
     with _opener.open(request, timeout=_TIMEOUT) as response:
         return json.loads(response.read())
+
+
+import os
+import pathlib
+import time
+
+_BOOTSTRAP_TTL = 7 * 24 * 60 * 60
+_BOOTSTRAP_URL = "https://data.iana.org/rdap/{}.json"
+_REGISTRIES = ("ipv4", "ipv6", "dns")
+
+
+class BootstrapUnavailable(Exception):
+    """No bootstrap data: the fetch failed and there is no cached copy."""
+
+
+def cache_dir() -> pathlib.Path:
+    """Return the bootstrap cache directory, reading XDG at call time.
+
+    Deliberately not inside a case directory: this is a copy of a public
+    map, not evidence.
+    """
+    base = os.environ.get(
+        "XDG_CACHE_HOME", str(pathlib.Path.home() / ".cache")
+    )
+    return pathlib.Path(base) / "abusectl" / "rdap"
+
+
+def bootstrap(registry: str, cache_root=None, fetch=http_fetch) -> dict:
+    """Return an IANA bootstrap document, from cache when it is fresh.
+
+    A failed refetch falls back to the stale copy rather than failing the
+    run. Staleness is safe here: a moved server misses and reads as no
+    contact, whereas losing IANA entirely would stop the user filing a
+    report at all.
+    """
+    if registry not in _REGISTRIES:
+        raise ValueError(f"unknown registry {registry!r}")
+
+    directory = pathlib.Path(cache_root) if cache_root is not None else cache_dir()
+    path = directory / f"{registry}.json"
+
+    cached = None
+    if path.exists():
+        cached = json.loads(path.read_text())
+        if time.time() - path.stat().st_mtime < _BOOTSTRAP_TTL:
+            return cached
+
+    try:
+        data = fetch(_BOOTSTRAP_URL.format(registry))
+    except Exception:
+        if cached is not None:
+            return cached
+        raise BootstrapUnavailable(
+            f"cannot fetch the {registry} bootstrap and no cached copy exists"
+        )
+
+    directory.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data))
+    return data
diff --git a/tests/test_rdap.py b/tests/test_rdap.py
index 681e2bb..3fc2c65 100644
--- a/tests/test_rdap.py
+++ b/tests/test_rdap.py
@@ -52,5 +52,78 @@ class RedirectPolicy(unittest.TestCase):
         self.assertIsNotNone(result)
 
 
+import json
+import tempfile
+import time
+from pathlib import Path
+
+
+class Bootstrap(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.cache = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+
+    def test_a_missing_file_is_fetched_and_cached(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"services": []}
+
+        data = rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)
+
+        self.assertEqual(data, {"services": []})
+        self.assertEqual(calls, ["https://data.iana.org/rdap/ipv4.json"])
+        self.assertTrue((self.cache / "ipv4.json").exists())
+
+    def test_a_fresh_cache_is_not_refetched(self):
+        (self.cache / "ipv4.json").write_text(json.dumps({"services": ["cached"]}))
+
+        def fetch(url):
+            raise AssertionError(f"should not have fetched {url}")
+
+        data = rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)
+        self.assertEqual(data, {"services": ["cached"]})
+
+    def test_a_stale_cache_is_refetched(self):
+        path = self.cache / "ipv4.json"
+        path.write_text(json.dumps({"services": ["old"]}))
+        old = time.time() - (rdap._BOOTSTRAP_TTL + 60)
+        import os
+        os.utime(path, (old, old))
+
+        data = rdap.bootstrap(
+            "ipv4", cache_root=self.cache, fetch=lambda url: {"services": ["new"]}
+        )
+        self.assertEqual(data, {"services": ["new"]})
+
+    def test_a_failed_refetch_falls_back_to_the_stale_copy(self):
+        """Losing IANA must not stop the user filing a report.
+
+        Last week's map is almost certainly still correct, and a stale
+        bootstrap fails safe: the worst case is querying a server that has
+        moved, which misses and reads as no contact.
+        """
+        path = self.cache / "ipv4.json"
+        path.write_text(json.dumps({"services": ["old"]}))
+        old = time.time() - (rdap._BOOTSTRAP_TTL + 60)
+        import os
+        os.utime(path, (old, old))
+
+        def fetch(url):
+            raise OSError("network is unreachable")
+
+        data = rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)
+        self.assertEqual(data, {"services": ["old"]})
+
+    def test_a_failed_fetch_with_no_cache_raises(self):
+        def fetch(url):
+            raise OSError("network is unreachable")
+
+        with self.assertRaises(rdap.BootstrapUnavailable):
+            rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)
+
+
 if __name__ == "__main__":
     unittest.main()