docs: plan the contacts implementation

Twelve tasks, TDD throughout, building bottom up: transport, bootstrap
cache, server selection, jCard extraction, querying with the label walk,
the worklist, resolution, the subcommand, an offline proof, two sweeps
and the documentation.

The code in the plan was assembled and run before committing it, against
the plan's own tests, 44 of which pass. That found two defects in the
plan itself: a parameter that shadowed the cache_dir function and forced
a globals() lookup, now renamed to cache_root, and a hand-rolled fake
request in the redirect test that lacked the attributes
HTTPRedirectHandler reads, now a real urllib Request.

Also drops the em dashes from the Documents list in AGENTS.md, which the
no-em-dash writing rule covers and which the contacts entry had matched.

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

1 files changed, 1985 insertions, 0 deletions

diff --git a/docs/plans/2026-09-09-contacts.md b/docs/plans/2026-09-09-contacts.md
new file mode 100644
index 0000000..40392d8
--- /dev/null
+++ b/docs/plans/2026-09-09-contacts.md
@@ -0,0 +1,1985 @@
+# abusectl `contacts` Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Build `abusectl contacts <case>`, which resolves an abuse contact for every IP and domain indicator in a case manifest via RDAP and rewrites the manifest with a `contacts[]` array.
+
+**Architecture:** Two modules. `rdap.py` is the protocol (IANA bootstrap fetch and cache, server selection, query, jCard extraction). `contacts.py` is the policy (which indicators resolve, how hosts fold, what reaches the manifest). Both take a `fetch` callable as an argument, defaulting to a real urllib transport, so the entire test suite keeps passing with sockets raising.
+
+**Tech Stack:** Python 3.11+ standard library only. `urllib.request`, `ipaddress`, `json`, `email.utils`. No new dependencies; `requirements.txt` stays empty.
+
+**Read first:** `docs/specs/2026-09-09-contacts.md`. It states a FOURTH non-negotiable property, that a query carries a bare host or IP and never a URL, and every task below that touches a query exists to hold that property.
+
+---
+
+## Background for an engineer new to this codebase
+
+**RDAP** is the JSON successor to `whois`. You ask a registry about an IP or a
+domain and get JSON back. Which registry to ask is answered by three bootstrap
+files IANA publishes, mapping IP ranges and TLDs to server base URLs.
+
+**jCard** (RFC 7095) is how RDAP encodes contact details: vCard as nested JSON
+arrays rather than objects. An entity looks like this:
+
+```json
+{
+  "roles": ["abuse"],
+  "vcardArray": ["vcard", [
+    ["version", {}, "text", "4.0"],
+    ["fn", {}, "text", "Abuse Desk"],
+    ["email", {}, "text", "abuse@example.invalid"]
+  ]]
+}
+```
+
+Note the shape: `vcardArray[1]` is a list of property arrays, each
+`[name, params, type, value]`. The value is at index 3.
+
+**Existing modules you will use:**
+
+- `case.load(path) -> dict` reads a manifest, `case.save(path, manifest)`
+  writes it atomically. `case.py` is the only writer of a case directory.
+- `config.load(path) -> Config` with `.cases` and `.trusted_relays`.
+- `cli.py` dispatches subcommands and owns exit codes: `EXIT_OK = 0`,
+  `EXIT_ERROR = 1`, `EXIT_NOT_CONFIGURED = 3`.
+
+**Indicators in a manifest** look like `{"id": "ioc-1", "type": "ipv4",
+"value": "198.51.100.7", "origin": "received-chain"}`. Types in play:
+`ipv4`, `ipv6`, `domain`, `url`, `sha256`, `observation`.
+
+**Test conventions:** `python3 -m unittest discover tests`. Standard library
+`unittest`, no pytest, no fixtures directory beyond `tests/fixtures/*.eml`.
+Fixtures use `example.invalid`, `.invalid` and RFC 5737 documentation ranges
+(`192.0.2.0/24`, `198.51.100.0/24`, `203.0.113.0/24`) only. Never a real
+domain, address or netblock.
+
+**Commits are GPG-signed:** `git commit -S`. Never pass `--no-verify`.
+
+---
+
+## File structure
+
+| File | Responsibility |
+|---|---|
+| Create: `abusectl/rdap.py` | Bootstrap cache, server selection, query, jCard extraction |
+| Create: `abusectl/contacts.py` | Worklist from indicators, fold hosts, build `contacts[]` |
+| Create: `tests/test_rdap.py` | Protocol tests, fake transport |
+| Create: `tests/test_contacts.py` | Policy tests, including the fourth property |
+| Modify: `abusectl/cli.py` | Add the `contacts` subparser and `_cmd_contacts` |
+| Modify: `tests/test_cli.py` | Dispatch test for the new subcommand |
+
+Task order builds bottom-up: transport, then bootstrap, then selection, then
+extraction, then policy, then the command line. Every task ends green and
+committed.
+
+---
+
+### Task 1: The transport
+
+**Files:**
+- Create: `abusectl/rdap.py`
+- Test: `tests/test_rdap.py`
+
+The transport is the only code in this repository that opens a socket. It is
+a separate function taking no case state so that everything above it can be
+tested with a fake.
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `tests/test_rdap.py`:
+
+```python
+# 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()
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `python3 -m unittest tests.test_rdap -v`
+Expected: FAIL, `ModuleNotFoundError: No module named 'abusectl.rdap'`
+
+- [ ] **Step 3: Write the module and the transport**
+
+Create `abusectl/rdap.py`:
+
+```python
+# 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())
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `python3 -m unittest tests.test_rdap -v`
+Expected: PASS, 2 tests
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add abusectl/rdap.py tests/test_rdap.py
+git commit -S -m "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."
+```
+
+---
+
+### Task 2: Bootstrap cache
+
+**Files:**
+- Modify: `abusectl/rdap.py`
+- Test: `tests/test_rdap.py`
+
+IANA publishes `ipv4.json`, `ipv6.json` and `dns.json`. Cache them under
+`$XDG_CACHE_HOME/abusectl/rdap/`, TTL 7 days, and fall back to a stale copy
+when a refetch fails.
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `tests/test_rdap.py`, before the `if __name__` block:
+
+```python
+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)
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `python3 -m unittest tests.test_rdap.Bootstrap -v`
+Expected: FAIL, `AttributeError: module 'abusectl.rdap' has no attribute 'bootstrap'`
+
+- [ ] **Step 3: Implement the bootstrap cache**
+
+Add to `abusectl/rdap.py`, after `http_fetch`:
+
+```python
+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
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `python3 -m unittest tests.test_rdap -v`
+Expected: PASS, 7 tests
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add abusectl/rdap.py tests/test_rdap.py
+git commit -S -m "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."
+```
+
+---
+
+### Task 3: Server selection
+
+**Files:**
+- Modify: `abusectl/rdap.py`
+- Test: `tests/test_rdap.py`
+
+Bootstrap documents have the shape
+`{"services": [[["192.0.2.0/24", "198.51.100.0/24"], ["https://rdap.example.invalid/"]]]}`.
+For `dns.json` the first list holds TLDs rather than ranges.
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `tests/test_rdap.py`:
+
+```python
+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))
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `python3 -m unittest tests.test_rdap.ServerSelection -v`
+Expected: FAIL, `AttributeError: module 'abusectl.rdap' has no attribute 'server_for_ip'`
+
+- [ ] **Step 3: Implement selection**
+
+Add to `abusectl/rdap.py`:
+
+```python
+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
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `python3 -m unittest tests.test_rdap -v`
+Expected: PASS, 13 tests
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add abusectl/rdap.py tests/test_rdap.py
+git commit -S -m "feat: select an RDAP server by longest prefix and by TLD
+
+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 range would name the operator that gave it away.
+
+A TLD that publishes no RDAP service selects nothing, which is a normal
+outcome for many TLDs rather than a defect."
+```
+
+---
+
+### Task 4: Reading an abuse address out of a jCard
+
+**Files:**
+- Modify: `abusectl/rdap.py`
+- Test: `tests/test_rdap.py`
+
+This is the strict-role rule and its four sub-rules. Every test here has a
+victim named in the spec.
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `tests/test_rdap.py`:
+
+```python
+def _entity(roles, emails, entities=None):
+    """Build an RDAP entity in real jCard shape."""
+    properties = [["version", {}, "text", "4.0"]]
+    for address in emails:
+        properties.append(["email", {}, "text", address])
+    entity = {"roles": roles, "vcardArray": ["vcard", properties]}
+    if entities:
+        entity["entities"] = entities
+    return entity
+
+
+class AbuseExtraction(unittest.TestCase):
+    def test_an_abuse_entity_yields_its_address(self):
+        response = {"entities": [_entity(["abuse"], ["abuse@example.invalid"])]}
+        self.assertEqual(
+            rdap.abuse_addresses(response), ["abuse@example.invalid"]
+        )
+
+    def test_a_nested_abuse_entity_is_found(self):
+        """The abuse entity is usually a child of the organisation entity."""
+        response = {
+            "entities": [
+                _entity(
+                    ["registrant"], [],
+                    entities=[_entity(["abuse"], ["abuse@example.invalid"])],
+                )
+            ]
+        }
+        self.assertEqual(
+            rdap.abuse_addresses(response), ["abuse@example.invalid"]
+        )
+
+    def test_a_technical_only_response_yields_nothing(self):
+        """A technical contact is a named human who never volunteered to
+        receive abuse mail. Mailing them is useless and is a small privacy
+        harm to an uninvolved third party."""
+        response = {"entities": [_entity(["technical"], ["someone@example.invalid"])]}
+        self.assertEqual(rdap.abuse_addresses(response), [])
+
+    def test_every_abuse_address_is_kept(self):
+        """Some netblocks publish two desks, and picking one arbitrarily
+        can drop the one that would have answered."""
+        response = {
+            "entities": [
+                _entity(["abuse"], ["one@example.invalid", "two@example.invalid"])
+            ]
+        }
+        self.assertEqual(
+            rdap.abuse_addresses(response),
+            ["one@example.invalid", "two@example.invalid"],
+        )
+
+    def test_a_newline_in_an_address_is_rejected(self):
+        """The address becomes a mail recipient in report and submit, so a
+        CRLF here is header injection into mail this tool sends."""
+        response = {
+            "entities": [
+                _entity(["abuse"], ["abuse@example.invalid\r\nBcc: victim@example.org"])
+            ]
+        }
+        self.assertEqual(rdap.abuse_addresses(response), [])
+
+    def test_a_non_address_is_rejected(self):
+        response = {"entities": [_entity(["abuse"], ["not an address"])]}
+        self.assertEqual(rdap.abuse_addresses(response), [])
+
+    def test_recursion_is_depth_capped(self):
+        """Remote JSON must not be able to hang the tool."""
+        deep = _entity(["abuse"], ["deep@example.invalid"])
+        for _ in range(10):
+            deep = _entity(["registrant"], [], entities=[deep])
+        self.assertEqual(rdap.abuse_addresses({"entities": [deep]}), [])
+
+    def test_duplicate_addresses_collapse(self):
+        response = {
+            "entities": [
+                _entity(["abuse"], ["abuse@example.invalid"]),
+                _entity(["abuse"], ["abuse@example.invalid"]),
+            ]
+        }
+        self.assertEqual(
+            rdap.abuse_addresses(response), ["abuse@example.invalid"]
+        )
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `python3 -m unittest tests.test_rdap.AbuseExtraction -v`
+Expected: FAIL, `AttributeError: module 'abusectl.rdap' has no attribute 'abuse_addresses'`
+
+- [ ] **Step 3: Implement extraction**
+
+Add to `abusectl/rdap.py`:
+
+```python
+import email.utils
+
+_MAX_ENTITY_DEPTH = 4
+
+
+def _valid_address(raw: str) -> str | None:
+    """Return a usable address, or None.
+
+    This value becomes a mail recipient in report and submit, so it is
+    validated where it enters rather than where it is used: a control
+    character here is header injection into mail this tool sends.
+    """
+    if not isinstance(raw, str) or not raw.strip():
+        return None
+    if any(character in raw for character in "\r\n\t"):
+        return None
+    if any(ord(character) < 32 for character in raw):
+        return None
+
+    name, address = email.utils.parseaddr(raw)
+    if not address or address.count("@") != 1:
+        return None
+    local, _, domain = address.partition("@")
+    if not local or not domain or "." not in domain:
+        return None
+    return address
+
+
+def _emails_from_vcard(entity: dict) -> list[str]:
+    """Pull every email property value out of a jCard.
+
+    jCard encodes vCard as nested arrays: vcardArray[1] is a list of
+    [name, params, type, value] properties, so the value is at index 3.
+    """
+    found = []
+    vcard = entity.get("vcardArray")
+    if not isinstance(vcard, list) or len(vcard) < 2:
+        return found
+    for prop in vcard[1]:
+        if not isinstance(prop, list) or len(prop) < 4:
+            continue
+        if prop[0] != "email":
+            continue
+        address = _valid_address(prop[3])
+        if address:
+            found.append(address)
+    return found
+
+
+def abuse_addresses(response: dict) -> list[str]:
+    """Return every published abuse address in an RDAP response.
+
+    STRICT: only an entity whose roles contain "abuse" counts. There is no
+    fallback to a technical or registrant contact, because that is a named
+    human who never volunteered for abuse mail, and no fallback to
+    abuse@<domain> by convention, because for a phishing domain that
+    mailbox belongs to the ATTACKER and mailing it would confirm both the
+    catch and that the user's address is live.
+
+    A links referral is never followed. If the address is not in this
+    response, there is no address: following a URL the response chose for
+    us is an outbound fetch under remote control.
+    """
+    found: list[str] = []
+
+    def walk(entities, depth):
+        if depth > _MAX_ENTITY_DEPTH or not isinstance(entities, list):
+            return
+        for entity in entities:
+            if not isinstance(entity, dict):
+                continue
+            roles = entity.get("roles") or []
+            if isinstance(roles, list) and "abuse" in roles:
+                found.extend(_emails_from_vcard(entity))
+            walk(entity.get("entities"), depth + 1)
+
+    walk(response.get("entities"), 1)
+
+    seen = set()
+    unique = []
+    for address in found:
+        if address not in seen:
+            seen.add(address)
+            unique.append(address)
+    return unique
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `python3 -m unittest tests.test_rdap -v`
+Expected: PASS, 21 tests
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add abusectl/rdap.py tests/test_rdap.py
+git commit -S -m "feat: read abuse addresses from a jCard, strictly
+
+Only an entity whose roles contain abuse counts. No fallback to a
+technical or registrant contact, who is a named human that never
+volunteered for abuse mail, and no fallback to abuse@<domain> by
+convention: for a phishing domain that mailbox belongs to the attacker,
+so constructing it would confirm both the catch and that the reporter's
+address is live.
+
+Addresses are validated where they enter rather than where they are
+used, because the value becomes a mail recipient later and a control
+character in it is header injection into mail this tool sends.
+
+Entity recursion is depth capped so remote JSON cannot hang the tool."
+```
+
+---
+
+### Task 5: Querying, and the label walk
+
+**Files:**
+- Modify: `abusectl/rdap.py`
+- Test: `tests/test_rdap.py`
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `tests/test_rdap.py`:
+
+```python
+class Query(unittest.TestCase):
+    IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]}
+    DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]}
+
+    def test_an_ip_query_hits_the_selected_server(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"handle": "NET-1", "entities": []}
+
+        result = rdap.query_ip("198.51.100.7", self.IPV4, fetch=fetch)
+
+        self.assertEqual(calls, ["https://rir.example.invalid/ip/198.51.100.7"])
+        self.assertEqual(result["handle"], "NET-1")
+
+    def test_an_unlisted_ip_is_not_queried(self):
+        def fetch(url):
+            raise AssertionError(f"should not have fetched {url}")
+
+        self.assertIsNone(rdap.query_ip("203.0.113.9", self.IPV4, fetch=fetch))
+
+    def test_the_label_walk_stops_at_the_first_answer(self):
+        """mail.deep.example.invalid is not registrable; example.invalid is.
+
+        The registry is the authority on what is registrable, which is why
+        this walks rather than carrying a Public Suffix List that would go
+        stale weekly.
+        """
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            if url.endswith("/domain/example.invalid"):
+                return {"handle": "DOM-1", "entities": []}
+            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)
+
+        result, queried = rdap.query_domain(
+            "mail.deep.example.invalid", self.DNS, fetch=fetch
+        )
+
+        self.assertEqual(queried, "example.invalid")
+        self.assertEqual(result["handle"], "DOM-1")
+        self.assertEqual(len(calls), 3)
+
+    def test_the_walk_never_queries_a_bare_tld(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)
+
+        result, queried = rdap.query_domain(
+            "deep.example.invalid", self.DNS, fetch=fetch
+        )
+
+        self.assertIsNone(result)
+        self.assertNotIn("https://registry.example.invalid/domain/invalid", calls)
+
+    def test_a_tld_with_no_server_is_not_queried(self):
+        def fetch(url):
+            raise AssertionError(f"should not have fetched {url}")
+
+        result, queried = rdap.query_domain(
+            "example.test", self.DNS, fetch=fetch
+        )
+        self.assertIsNone(result)
+
+    def test_the_walk_is_capped(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)
+
+        host = "a.b.c.d.e.f.g.example.invalid"
+        rdap.query_domain(host, self.DNS, fetch=fetch)
+        self.assertLessEqual(len(calls), rdap._MAX_LABEL_WALK)
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `python3 -m unittest tests.test_rdap.Query -v`
+Expected: FAIL, `AttributeError: module 'abusectl.rdap' has no attribute 'query_ip'`
+
+- [ ] **Step 3: Implement querying**
+
+Add to `abusectl/rdap.py`:
+
+```python
+_MAX_LABEL_WALK = 5
+
+
+def query_ip(address: str, bootstrap_data: dict, fetch=http_fetch) -> dict | None:
+    """Query the registry responsible for an address.
+
+    Returns None when no registry is listed for it, which is a normal
+    outcome rather than an error.
+    """
+    base = server_for_ip(address, bootstrap_data)
+    if base is None:
+        return None
+    return fetch(f"{base.rstrip('/')}/ip/{address}")
+
+
+def query_domain(
+    host: str, bootstrap_data: dict, fetch=http_fetch
+) -> tuple[dict | None, str | None]:
+    """Query for a host, walking up the labels to find the registrable name.
+
+    RDAP wants the registrable domain, and mail.deep.example.invalid is not
+    one. Rather than carrying a Public Suffix List, which is a transcribed
+    table that goes stale weekly, this asks the registry: it is the
+    authority on what is registrable.
+
+    Returns (response, queried_name). A bare TLD is never queried.
+    """
+    labels = host.lower().strip(".").split(".")
+    if len(labels) < 2:
+        return None, None
+
+    tld = labels[-1]
+    base = server_for_tld(tld, bootstrap_data)
+    if base is None:
+        return None, None
+    base = base.rstrip("/")
+
+    attempts = 0
+    # Stop before the bare TLD: range end is len(labels) - 1, so the last
+    # candidate is the two-label name.
+    for start in range(0, len(labels) - 1):
+        if attempts >= _MAX_LABEL_WALK:
+            break
+        candidate = ".".join(labels[start:])
+        attempts += 1
+        try:
+            return fetch(f"{base}/domain/{candidate}"), candidate
+        except Exception:
+            continue
+
+    return None, None
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `python3 -m unittest tests.test_rdap -v`
+Expected: PASS, 27 tests
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add abusectl/rdap.py tests/test_rdap.py
+git commit -S -m "feat: query RDAP, walking up the labels for a registrable domain
+
+RDAP wants the registrable domain and a deep host is not one. Rather
+than bundling a Public Suffix List, which is a transcribed table that
+goes stale weekly and is the failure init.PROVIDERS already documents,
+this asks the registry, which is the authority on what is registrable.
+
+The walk is capped and never queries a bare TLD."
+```
+
+---
+
+### Task 6: The worklist, and the fourth property
+
+**Files:**
+- Create: `abusectl/contacts.py`
+- Test: `tests/test_contacts.py`
+
+This task holds the fourth non-negotiable property. Write these tests
+carefully; they are the ones that matter.
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `tests/test_contacts.py`:
+
+```python
+# 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 turning indicators into abuse contacts."""
+
+import unittest
+
+from abusectl import contacts
+
+
+class Worklist(unittest.TestCase):
+    def test_ips_and_domains_are_resolvable(self):
+        iocs = [
+            {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"},
+            {"id": "ioc-2", "type": "domain", "value": "example.invalid"},
+        ]
+        work = contacts.worklist(iocs)
+        self.assertEqual(
+            {(item.kind, item.query) for item in work},
+            {("ip", "198.51.100.7"), ("domain", "example.invalid")},
+        )
+
+    def test_hashes_and_observations_are_not_resolvable(self):
+        iocs = [
+            {"id": "ioc-1", "type": "sha256", "value": "e3b0c442"},
+            {"id": "ioc-2", "type": "observation",
+             "value": "display-name-carries-address"},
+        ]
+        self.assertEqual(contacts.worklist(iocs), [])
+
+    def test_a_url_contributes_only_its_host(self):
+        """THE FOURTH PROPERTY. A query discloses what the user is looking
+        at, and a URL path can carry recipient identity that
+        suspect_path_segments deliberately flags rather than redacts."""
+        iocs = [{
+            "id": "ioc-1", "type": "url",
+            "value": "https://login.example.invalid/verify/victim%40example.org?e=REDACTED",
+        }]
+        work = contacts.worklist(iocs)
+        self.assertEqual(len(work), 1)
+        self.assertEqual(work[0].kind, "domain")
+        self.assertEqual(work[0].query, "login.example.invalid")
+
+    def test_url_userinfo_never_reaches_the_query(self):
+        iocs = [{
+            "id": "ioc-1", "type": "url",
+            "value": "https://victim%40example.org:secret@login.example.invalid/x",
+        }]
+        work = contacts.worklist(iocs)
+        self.assertEqual(work[0].query, "login.example.invalid")
+
+    def test_a_url_port_is_stripped(self):
+        iocs = [{"id": "ioc-1", "type": "url",
+                 "value": "https://login.example.invalid:8443/x"}]
+        self.assertEqual(contacts.worklist(iocs)[0].query, "login.example.invalid")
+
+    def test_a_url_host_that_is_an_ip_resolves_as_an_ip(self):
+        iocs = [{"id": "ioc-1", "type": "url",
+                 "value": "http://198.51.100.7/login"}]
+        work = contacts.worklist(iocs)
+        self.assertEqual(work[0].kind, "ip")
+        self.assertEqual(work[0].query, "198.51.100.7")
+
+    def test_a_bracketed_ipv6_url_host_resolves_as_an_ip(self):
+        iocs = [{"id": "ioc-1", "type": "url",
+                 "value": "http://[2001:db8::1]/login"}]
+        work = contacts.worklist(iocs)
+        self.assertEqual(work[0].kind, "ip")
+        self.assertEqual(work[0].query, "2001:db8::1")
+
+    def test_hosts_fold_and_keep_every_contributing_ioc(self):
+        """Twenty URLs on one host must produce one query."""
+        iocs = [
+            {"id": "ioc-1", "type": "url", "value": "https://a.example.invalid/one"},
+            {"id": "ioc-2", "type": "url", "value": "https://a.example.invalid/two"},
+            {"id": "ioc-3", "type": "domain", "value": "a.example.invalid"},
+        ]
+        work = contacts.worklist(iocs)
+        self.assertEqual(len(work), 1)
+        self.assertEqual(work[0].iocs, ["ioc-1", "ioc-2", "ioc-3"])
+
+    def test_a_trailing_dot_folds_with_the_bare_host(self):
+        iocs = [
+            {"id": "ioc-1", "type": "domain", "value": "example.invalid."},
+            {"id": "ioc-2", "type": "domain", "value": "example.invalid"},
+        ]
+        self.assertEqual(len(contacts.worklist(iocs)), 1)
+
+    def test_an_untrusted_hop_is_still_resolved(self):
+        """A forged chain's IP may still be the real sender's; the
+        confidence marker stays in the manifest for review."""
+        iocs = [{"id": "ioc-1", "type": "ipv4", "value": "203.0.113.99",
+                 "confidence": "untrusted-hop"}]
+        self.assertEqual(len(contacts.worklist(iocs)), 1)
+
+    def test_a_malformed_url_contributes_nothing(self):
+        iocs = [{"id": "ioc-1", "type": "url", "value": "not a url"}]
+        self.assertEqual(contacts.worklist(iocs), [])
+
+
+if __name__ == "__main__":
+    unittest.main()
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `python3 -m unittest tests.test_contacts -v`
+Expected: FAIL, `ModuleNotFoundError: No module named 'abusectl.contacts'`
+
+- [ ] **Step 3: Implement the worklist**
+
+Create `abusectl/contacts.py`:
+
+```python
+# 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.
+"""Indicators to abuse contacts: which ones resolve, and to what.
+
+THE FOURTH NON-NEGOTIABLE PROPERTY lives here. An RDAP query tells a third
+party what the user is looking at, so a query carries a BARE HOST OR IP
+ADDRESS and never a URL.
+
+A URL path can carry recipient identity. parse.suspect_path_segments()
+FLAGS those rather than redacting them, deliberately, because a path
+segment may be the thing being reported, and that decision is safe only
+while the URL stays local. Property 1 governs what is PUBLISHED; a query
+is a disclosure that appears in no report, so property 1 does not cover
+it and this one does.
+
+Concretely: a url indicator contributes its HOST to the worklist and
+nothing else. Path, query and fragment never leave the machine.
+
+This is a trap rather than a theoretical concern. The obvious
+implementation resolves "a contact for each indicator" by reading each
+indicator's value, and for a url indicator that value is an entire URL.
+"""
+
+import ipaddress
+import urllib.parse
+from dataclasses import dataclass, field
+
+from . import rdap
+
+
+@dataclass
+class WorkItem:
+    """One thing to ask a registry about, and every indicator behind it."""
+
+    kind: str  # "ip" or "domain"
+    query: str
+    iocs: list[str] = field(default_factory=list)
+
+
+def _host_of(url: str) -> str | None:
+    """Return the bare host of a URL: no userinfo, no port, no path.
+
+    urlsplit().hostname does all three, which is why it is used rather
+    than netloc: netloc still carries userinfo and a port.
+    """
+    try:
+        parts = urllib.parse.urlsplit(url)
+    except ValueError:
+        return None
+    host = parts.hostname
+    if not host:
+        return None
+    return host.strip(".").lower() or None
+
+
+def _is_ip(value: str) -> bool:
+    try:
+        ipaddress.ip_address(value)
+        return True
+    except ValueError:
+        return False
+
+
+def worklist(iocs: list[dict]) -> list[WorkItem]:
+    """Build the deduplicated list of queries for a set of indicators.
+
+    Hosts fold: twenty URLs on one host produce one query, and the item
+    keeps every indicator id that contributed so nothing is lost.
+    """
+    items: dict[tuple[str, str], WorkItem] = {}
+
+    def add(kind, query, ioc_id):
+        key = (kind, query)
+        if key not in items:
+            items[key] = WorkItem(kind=kind, query=query)
+        if ioc_id not in items[key].iocs:
+            items[key].iocs.append(ioc_id)
+
+    for ioc in iocs:
+        ioc_type = ioc.get("type")
+        value = ioc.get("value") or ""
+        ioc_id = ioc.get("id")
+
+        if ioc_type in ("ipv4", "ipv6"):
+            if _is_ip(value):
+                add("ip", value, ioc_id)
+        elif ioc_type == "domain":
+            host = value.strip(".").lower()
+            if host:
+                add("ip" if _is_ip(host) else "domain", host, ioc_id)
+        elif ioc_type == "url":
+            host = _host_of(value)
+            if host:
+                add("ip" if _is_ip(host) else "domain", host, ioc_id)
+
+    return list(items.values())
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `python3 -m unittest tests.test_contacts -v`
+Expected: PASS, 11 tests
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add abusectl/contacts.py tests/test_contacts.py
+git commit -S -m "feat: build the contacts worklist, host only
+
+Adds the fourth non-negotiable property: a query carries a bare host or
+IP and never a URL. An RDAP query discloses what the user is looking at,
+and a URL path can carry recipient identity that parse deliberately
+flags rather than redacts, because a path segment may be the thing being
+reported. That decision is safe only while the URL stays local.
+
+Property 1 governs what is published and a query appears in no report,
+so property 1 does not cover this and this property does.
+
+Hosts fold, so twenty URLs on one host make one query while the item
+keeps every indicator id behind it."
+```
+
+---
+
+### Task 7: Resolving the worklist
+
+**Files:**
+- Modify: `abusectl/contacts.py`
+- Test: `tests/test_contacts.py`
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `tests/test_contacts.py`, before the `if __name__` block:
+
+```python
+class Resolve(unittest.TestCase):
+    IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]}
+    IPV6 = {"services": []}
+    DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]}
+
+    def _bootstraps(self):
+        return {"ipv4": self.IPV4, "ipv6": self.IPV6, "dns": self.DNS}
+
+    def test_an_ip_resolves_to_its_abuse_desk(self):
+        def fetch(url):
+            return {
+                "handle": "NET-1",
+                "entities": [{
+                    "roles": ["abuse"],
+                    "vcardArray": ["vcard", [
+                        ["version", {}, "text", "4.0"],
+                        ["email", {}, "text", "abuse@example.invalid"],
+                    ]],
+                }],
+            }
+
+        iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["iocs"], ["ioc-1"])
+        self.assertEqual(result[0]["query"], "198.51.100.7")
+        self.assertEqual(result[0]["abuse"], ["abuse@example.invalid"])
+        self.assertEqual(result[0]["handle"], "NET-1")
+        self.assertNotIn("error", result[0])
+
+    def test_no_abuse_role_records_a_reason_not_an_error(self):
+        def fetch(url):
+            return {"handle": "NET-2", "entities": []}
+
+        iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(result[0]["abuse"], [])
+        self.assertEqual(result[0]["error"], "no abuse role published")
+
+    def test_no_rdap_server_records_a_reason(self):
+        def fetch(url):
+            raise AssertionError(f"should not have fetched {url}")
+
+        iocs = [{"id": "ioc-1", "type": "domain", "value": "example.test"}]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(result[0]["abuse"], [])
+        self.assertIn("no rdap server", result[0]["error"])
+
+    def test_a_network_failure_is_per_query_and_does_not_stop_the_run(self):
+        def fetch(url):
+            if "198.51.100.7" in url:
+                raise OSError("connection timed out")
+            return {
+                "handle": "DOM-1",
+                "entities": [{
+                    "roles": ["abuse"],
+                    "vcardArray": ["vcard", [
+                        ["version", {}, "text", "4.0"],
+                        ["email", {}, "text", "abuse@example.invalid"],
+                    ]],
+                }],
+            }
+
+        iocs = [
+            {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"},
+            {"id": "ioc-2", "type": "domain", "value": "example.invalid"},
+        ]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(len(result), 2)
+        failed = [r for r in result if r["query"] == "198.51.100.7"][0]
+        worked = [r for r in result if r["query"] == "example.invalid"][0]
+        self.assertIn("connection timed out", failed["error"])
+        self.assertEqual(worked["abuse"], ["abuse@example.invalid"])
+
+    def test_one_query_per_host_however_many_iocs(self):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"handle": "DOM-1", "entities": []}
+
+        iocs = [
+            {"id": f"ioc-{n}", "type": "url",
+             "value": f"https://a.example.invalid/page{n}"}
+            for n in range(20)
+        ]
+        result = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(len(result[0]["iocs"]), 20)
+
+    def test_no_query_ever_carries_a_path(self):
+        """THE FOURTH PROPERTY, asserted at the transport."""
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"handle": "DOM-1", "entities": []}
+
+        iocs = [{
+            "id": "ioc-1", "type": "url",
+            "value": "https://a.example.invalid/verify/victim%40example.org?e=x",
+        }]
+        contacts.resolve(iocs, bootstraps=self._bootstraps(), fetch=fetch)
+
+        for url in calls:
+            self.assertNotIn("victim", url)
+            self.assertNotIn("verify", url)
+            self.assertNotIn("%40", url)
+            self.assertNotIn("?", url)
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `python3 -m unittest tests.test_contacts.Resolve -v`
+Expected: FAIL, `AttributeError: module 'abusectl.contacts' has no attribute 'resolve'`
+
+- [ ] **Step 3: Implement resolve**
+
+Add to `abusectl/contacts.py`:
+
+```python
+def resolve(iocs: list[dict], bootstraps: dict, fetch=rdap.http_fetch) -> list[dict]:
+    """Resolve every resolvable indicator to an abuse contact.
+
+    bootstraps is {"ipv4": ..., "ipv6": ..., "dns": ...}, passed in rather
+    than fetched here so this function stays testable with no network and
+    so the caller owns the cache policy.
+
+    A failure is recorded per query and never stops the run: a timeout on
+    one indicator must not cost the contacts that did resolve. A missing
+    contact is a normal outcome, not an error.
+    """
+    results = []
+
+    for item in worklist(iocs):
+        entry = {
+            "iocs": item.iocs,
+            "query": item.query,
+            "abuse": [],
+            "source": "rdap",
+        }
+
+        try:
+            if item.kind == "ip":
+                family = "ipv6" if ":" in item.query else "ipv4"
+                response = rdap.query_ip(
+                    item.query, bootstraps.get(family, {}), fetch=fetch
+                )
+                if response is None:
+                    entry["error"] = "no rdap server for this range"
+                    results.append(entry)
+                    continue
+            else:
+                response, queried = rdap.query_domain(
+                    item.query, bootstraps.get("dns", {}), fetch=fetch
+                )
+                if response is None:
+                    entry["error"] = "no rdap server for this tld, or no answer"
+                    results.append(entry)
+                    continue
+                if queried and queried != item.query:
+                    entry["queried"] = queried
+        except Exception as exc:
+            entry["error"] = f"{type(exc).__name__}: {exc}"
+            results.append(entry)
+            continue
+
+        handle = response.get("handle")
+        if handle:
+            entry["handle"] = handle
+
+        addresses = rdap.abuse_addresses(response)
+        entry["abuse"] = addresses
+        if not addresses:
+            entry["error"] = "no abuse role published"
+
+        results.append(entry)
+
+    return results
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `python3 -m unittest tests.test_contacts -v`
+Expected: PASS, 17 tests
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add abusectl/contacts.py tests/test_contacts.py
+git commit -S -m "feat: resolve the worklist to abuse contacts
+
+Failure is per query and never stops the run: a timeout on one indicator
+must not cost the contacts that did resolve, and a missing contact is a
+normal outcome rather than an error.
+
+Bootstraps are passed in rather than fetched here, so this stays
+testable with no network and the caller owns the cache policy."
+```
+
+---
+
+### Task 8: The `contacts` subcommand
+
+**Files:**
+- Modify: `abusectl/cli.py`
+- Modify: `tests/test_cli.py`
+
+- [ ] **Step 1: Write the failing test**
+
+Append to `tests/test_cli.py`, inside the existing test module and before any
+`if __name__` block. Match the surrounding style; if existing dispatch tests
+use a helper, reuse it.
+
+```python
+class ContactsCommand(unittest.TestCase):
+    def test_contacts_rewrites_the_manifest(self):
+        import json
+        import tempfile
+        from pathlib import Path
+        from unittest import mock
+
+        from abusectl import case
+
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            created = case.create(root, b"From: sender@example.invalid\r\n\r\nbody\r\n")
+            manifest = case.load(created.path)
+            manifest["iocs"] = [
+                {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}
+            ]
+            case.save(created.path, manifest)
+
+            fake_contacts = [{
+                "iocs": ["ioc-1"], "query": "198.51.100.7",
+                "abuse": ["abuse@example.invalid"], "source": "rdap",
+            }]
+
+            with mock.patch("abusectl.cli.contacts_module.resolve",
+                            return_value=fake_contacts) as resolve, \
+                 mock.patch("abusectl.cli.rdap_module.bootstrap",
+                            return_value={"services": []}):
+                code = cli.main(["contacts", str(created.path)])
+
+            self.assertEqual(code, cli.EXIT_OK)
+            self.assertTrue(resolve.called)
+            written = case.load(created.path)
+            self.assertEqual(written["contacts"], fake_contacts)
+
+    def test_a_missing_case_is_an_error_not_a_traceback(self):
+        code = cli.main(["contacts", "/nonexistent/case/path"])
+        self.assertEqual(code, cli.EXIT_ERROR)
+```
+
+- [ ] **Step 2: Run the test to verify it fails**
+
+Run: `python3 -m unittest tests.test_cli.ContactsCommand -v`
+Expected: FAIL, `SystemExit: 2` from argparse, since the subcommand does not exist
+
+- [ ] **Step 3: Wire the subcommand**
+
+In `abusectl/cli.py`, add to the imports at the top, matching the existing
+import style (`from . import parse as parse_module`):
+
+```python
+from . import contacts as contacts_module
+from . import rdap as rdap_module
+```
+
+In `_build_parser()`, after the `parse_parser` block and before `return parser`:
+
+```python
+    contacts_parser = subparsers.add_parser(
+        "contacts", help="resolve abuse contacts for a case"
+    )
+    contacts_parser.add_argument("case", type=Path)
+```
+
+Add the handler, after `_cmd_parse`:
+
+```python
+def _cmd_contacts(args) -> int:
+    """Resolve abuse contacts and rewrite the manifest.
+
+    A re-run overwrites contacts[] wholesale rather than merging. A merge
+    would let a contact resolved a week ago survive into a report filed
+    today, which is the stale-address hazard the response caching policy
+    already refuses. Overwriting makes a re-run always safe and always
+    current, which matters because a partial network failure makes
+    re-running the natural next step.
+    """
+    try:
+        manifest = case.load(args.case)
+    except FileNotFoundError:
+        print(f"abusectl contacts: no case at {args.case}", file=sys.stderr)
+        return EXIT_ERROR
+    except (ValueError, OSError) as exc:
+        print(f"abusectl contacts: {args.case}: {exc}", file=sys.stderr)
+        return EXIT_ERROR
+
+    try:
+        bootstraps = {
+            name: rdap_module.bootstrap(name) for name in ("ipv4", "ipv6", "dns")
+        }
+    except rdap_module.BootstrapUnavailable as exc:
+        print(f"abusectl contacts: {exc}", file=sys.stderr)
+        return EXIT_ERROR
+
+    resolved = contacts_module.resolve(manifest.get("iocs", []), bootstraps)
+
+    manifest["contacts"] = resolved
+    case.save(args.case, manifest)
+
+    unresolved = sum(1 for entry in resolved if not entry["abuse"])
+    print(f"{len(resolved)} contacts, {unresolved} without an abuse address")
+    return EXIT_OK
+```
+
+In `main()`, add the dispatch beside the existing ones:
+
+```python
+        if args.command == "contacts":
+            return _cmd_contacts(args)
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `python3 -m unittest tests.test_cli -v`
+Expected: PASS, including the two new tests
+
+- [ ] **Step 5: Run the whole suite**
+
+Run: `python3 -m unittest discover tests`
+Expected: OK, 112 existing plus the new tests
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add abusectl/cli.py tests/test_cli.py
+git commit -S -m "feat: add the contacts subcommand
+
+A re-run overwrites contacts[] wholesale rather than merging. A merge
+would let a contact resolved a week ago survive into a report filed
+today, which is the stale-address hazard the response caching policy
+already refuses, and overwriting makes a re-run always safe, which
+matters because a partial network failure makes re-running the natural
+next step."
+```
+
+---
+
+### Task 9: Prove the suite is still offline
+
+**Files:**
+- Test: `tests/test_offline.py` (create only if no equivalent exists)
+
+The umbrella design's property 2 is verified by running the suite with sockets
+raising. Confirm that still holds now that a network module exists.
+
+- [ ] **Step 1: Check whether the guard already exists**
+
+Run: `grep -rn "getaddrinfo\|create_connection" tests/`
+
+If a test already blocks sockets suite-wide, skip to Step 3 and just run it.
+
+- [ ] **Step 2: If none exists, create `tests/test_offline.py`**
+
+```python
+# 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.
+"""The suite must pass with no network, including the network modules.
+
+A network module that can only be tested with a network is a module that
+stops being tested. rdap and contacts both take an injected fetch, and
+this asserts that the default is never reached by accident during a test
+run: parse stays pure, and nothing above it opens a socket unasked.
+"""
+
+import socket
+import unittest
+from unittest import mock
+
+from abusectl import contacts, parse, rdap
+
+
+class NothingOpensASocket(unittest.TestCase):
+    def setUp(self):
+        for name in ("socket", "create_connection", "getaddrinfo"):
+            patcher = mock.patch.object(
+                socket, name,
+                side_effect=AssertionError(f"socket.{name} was called"),
+            )
+            patcher.start()
+            self.addCleanup(patcher.stop)
+
+    def test_parsing_opens_no_socket(self):
+        raw = (b"Received: from relay.example.invalid ([192.0.2.10])\r\n"
+               b"From: sender@example.invalid\r\n"
+               b"Subject: test\r\n\r\nbody\r\n")
+        parse.iocs(raw, trusted=["192.0.2.0/24"])
+
+    def test_resolving_with_an_injected_fetch_opens_no_socket(self):
+        iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}]
+        bootstraps = {
+            "ipv4": {"services": [[["198.51.100.0/24"],
+                                   ["https://rir.example.invalid/"]]]},
+            "ipv6": {"services": []},
+            "dns": {"services": []},
+        }
+        result = contacts.resolve(
+            iocs, bootstraps=bootstraps,
+            fetch=lambda url: {"handle": "NET-1", "entities": []},
+        )
+        self.assertEqual(len(result), 1)
+
+    def test_the_real_transport_is_never_the_default_in_a_test(self):
+        """Sanity: http_fetch exists and is the documented default."""
+        self.assertIs(contacts.resolve.__defaults__[-1], rdap.http_fetch)
+
+
+if __name__ == "__main__":
+    unittest.main()
+```
+
+- [ ] **Step 3: Run the whole suite**
+
+Run: `python3 -m unittest discover tests`
+Expected: OK, all tests pass
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add tests/test_offline.py
+git commit -S -m "test: prove the suite still opens no socket
+
+The umbrella design verifies property 2 by running with sockets raising.
+Now that a network module exists, that has to stay true for the WHOLE
+suite rather than for everything except contacts: a network module that
+can only be tested with a network is a module that stops being tested."
+```
+
+---
+
+### Task 10: Sweep A, offline, over the user's real spam
+
+**Files:**
+- Scratchpad only. Nothing in this task is committed to the repository.
+
+**Ask the user before reading their mail.** This is required by AGENTS.md.
+
+- [ ] **Step 1: Ask permission**
+
+Ask: "May I run the offline sweep over your spam corpus? It reads
+`tag:spam` from your notmuch index, builds the contacts worklist with a fake
+transport, and sends no packets."
+
+- [ ] **Step 2: Write the sweep script in the scratchpad**
+
+Write to the session scratchpad directory, NEVER into the repository:
+
+```python
+"""Sweep A: build the contacts worklist over real spam. No packets."""
+
+import re
+import subprocess
+import sys
+
+# Run this from the abusectl checkout, or point PYTHONPATH at it.
+sys.path.insert(0, "/home/you/path/to/abusectl")
+
+from abusectl import contacts, parse
+
+TRUSTED = ["192.0.2.0/24"]  # replace with the user's real config values
+
+mids = subprocess.run(
+    ["notmuch", "search", "--output=messages", "tag:spam"],
+    capture_output=True, text=True, check=True,
+).stdout.split()
+
+hosts = set()
+walk_depths = []
+malformed = []
+crashes = 0
+folded = 0
+
+for mid in mids:
+    try:
+        raw = subprocess.run(
+            ["notmuch", "show", "--format=raw", mid],
+            capture_output=True, check=True,
+        ).stdout
+        iocs = parse.iocs(raw, trusted=TRUSTED)
+    except Exception:
+        crashes += 1
+        continue
+
+    work = contacts.worklist(iocs)
+    folded += sum(len(item.iocs) for item in work) - len(work)
+
+    for item in work:
+        hosts.add((item.kind, item.query))
+        # THE ASSERTION: a query is a bare host or IP, nothing else.
+        if any(c in item.query for c in "/?#@:") and item.kind != "ip":
+            malformed.append(item.query)
+        if item.kind == "domain":
+            walk_depths.append(item.query.count(".") + 1)
+
+print(f"messages:        {len(mids)}")
+print(f"crashes:         {crashes}")
+print(f"unique targets:  {len(hosts)}")
+print(f"iocs folded away:{folded}")
+print(f"malformed:       {len(malformed)}")
+if walk_depths:
+    print(f"labels min/max:  {min(walk_depths)}/{max(walk_depths)}")
+assert not malformed, malformed[:5]
+print("OK: every query was a bare host or IP")
+```
+
+- [ ] **Step 3: Run it and record the counts**
+
+Run the script. Report to the user: message count, crash count, unique
+targets, how many indicators folded, label depth range, and whether the
+assertion held.
+
+**What may leave this script:** counts, tallies, TLDs, error reasons, walk
+depths, whether any query was malformed.
+**What may not:** an address, a real domain, a real abuse contact, or a URL
+from a real message.
+
+- [ ] **Step 4: If the sweep finds a defect, reproduce it synthetically**
+
+Write a new test in `tests/test_contacts.py` using `example.invalid` and RFC
+5737 values that reproduces the shape, watch it fail, fix, watch it pass,
+commit. The real message stays in the scratchpad.
+
+- [ ] **Step 5: Nothing to commit if the sweep was clean**
+
+Do not commit the script. Record the outcome in the handoff instead.
+
+---
+
+### Task 11: Sweep B, online, a deliberate handful
+
+**Files:**
+- Scratchpad only.
+
+**This task is OUTWARD-FACING and needs an explicit go-ahead**, separate from
+Task 10's. It discloses to registries, and possibly to the attacker's own
+registrar, which netblocks and domains the user is investigating, from their
+address, at a known time. That cannot be undone.
+
+- [ ] **Step 1: Ask for explicit permission**
+
+Ask: "Sweep B queries real registries for ten to twenty hand-picked
+indicators. Unlike every test so far this discloses what you are
+investigating to third parties and cannot be undone. Shall I run it?"
+
+If the user declines, stop here. The module is still fully tested; sweep B
+only confirms real response shapes.
+
+- [ ] **Step 2: Pick the sample by hand**
+
+Ten to twenty targets across distinct netblocks and TLDs, chosen with the
+user. Include at least one of each: a well-known netblock, a deep host that
+will exercise the label walk, and a TLD likely to publish no RDAP.
+
+- [ ] **Step 3: Run against the real transport**
+
+```python
+from abusectl import contacts, rdap
+
+bootstraps = {name: rdap.bootstrap(name) for name in ("ipv4", "ipv6", "dns")}
+
+iocs = [
+    {"id": "ioc-1", "type": "ipv4", "value": "..."},  # filled in with the user
+]
+
+for entry in contacts.resolve(iocs, bootstraps=bootstraps):
+    print(entry["query"], "->", len(entry["abuse"]), "addresses",
+          entry.get("error", ""))
+```
+
+Print the COUNT of addresses, never the addresses themselves.
+
+- [ ] **Step 4: Record shapes, fix any parsing defect**
+
+If a real response shape does not parse, reproduce it as a synthetic fixture
+with every address, handle and range replaced, add the failing test, fix,
+commit.
+
+- [ ] **Step 5: Report outcome**
+
+Counts and shapes only. No real contact reaches the repository or the
+conversation.
+
+---
+
+### Task 12: Documentation
+
+**Files:**
+- Modify: `AGENTS.md`
+- Modify: `README.md`
+- Modify: `docs/BACKLOG.md` if anything was deferred
+
+- [ ] **Step 1: Add the fourth property to AGENTS.md**
+
+The section is titled "THREE PROPERTIES THAT ARE NOT NEGOTIABLE". Rename it
+to "FOUR PROPERTIES THAT ARE NOT NEGOTIABLE" and add, after property 3:
+
+```markdown
+### 4. A query carries a bare host or IP, never a URL
+
+`contacts` is the first part that talks to anyone. An RDAP query tells a
+third party what the user is looking at, so it carries a BARE HOST OR IP
+ADDRESS and nothing else.
+
+Property 1 governs what is PUBLISHED. A query appears in no report, so
+property 1 does not cover it. A URL path can carry recipient identity, and
+`suspect_path_segments` deliberately FLAGS those rather than redacting them,
+which is safe only while the URL stays local.
+
+`contacts.worklist()` reduces a `url` indicator to `urlsplit().hostname`,
+which drops userinfo, port and path together. `tests/test_contacts.py`
+asserts at the transport that no query ever carried a path, and the offline
+sweep asserts the same thing against real mail.
+```
+
+- [ ] **Step 2: Update the architecture block in AGENTS.md**
+
+In the `## Architecture` section, move `contacts` out of "Planned" and into
+the module list:
+
+```
+  contacts.py   IOCs -> abuse contacts            network, read-only
+  rdap.py       bootstrap, query, jCard           network, read-only
+```
+
+Update the Planned line to name only `report`, `submit` and `retry`.
+
+- [ ] **Step 3: Document the command in README.md**
+
+Add `contacts` to the usage section beside `parse`, matching the existing
+style, including that it needs network and that a re-run is safe.
+
+- [ ] **Step 4: Run the suite one final time**
+
+Run: `python3 -m unittest discover tests`
+Expected: OK
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add AGENTS.md README.md docs/BACKLOG.md
+git commit -S -m "docs: record the fourth property and the contacts command
+
+An RDAP query discloses what the user is looking at, and property 1
+covers only what is published, so the query rule needed stating in its
+own right beside the other three."
+```
+
+---
+
+## Self-review against the spec
+
+| Spec section | Task |
+|---|---|
+| Modules, injected fetch | 1, 6, 9 |
+| Fourth property | 6, 7, 10, 12 |
+| What gets resolved (table) | 6 |
+| Bootstrap, TTL, stale fallback | 2 |
+| Server selection, longest prefix | 3 |
+| Label walk replacing a PSL | 5 |
+| Transport, timeout, redirect cap, no downgrade | 1 |
+| Caching policy, in-run dedup | 2, 6 |
+| Strict abuse role, four rules | 4 |
+| Manifest shape, `iocs` and `abuse` as lists | 7 |
+| Re-run overwrites wholesale | 8 |
+| Failure is per query | 7 |
+| Test list | 3, 4, 5, 6, 7 |
+| Sweep A offline | 10 |
+| Sweep B online, ask first | 11 |
+| Deliberately absent (no PSL, no whois, no disk response cache) | 2, 5 |
+
+**Note on test counts:** the running totals in each task assume the tests
+above are added in order and nothing else changes. If a count is off by a
+few, that is bookkeeping rather than a failure; what matters is that the
+named tests pass and `python3 -m unittest discover tests` is green.
+
+**Note on `no rdap server` wording:** Task 7's test asserts
+`"no rdap server" in entry["error"]`, and the implementation writes two
+different suffixes for the IP and domain cases. Keep the substring stable if
+you reword either message.