path: root/docs/plans/2026-09-09-contacts.md

# 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, and no dependency file in the repository.

**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):
        # socket.socket.connect, NOT socket.socket: replacing the class
        # itself breaks the ssl module at import time and produces false
        # failures that have nothing to do with network use.
        patcher = mock.patch.object(
            socket.socket, "connect",
            side_effect=AssertionError("socket.socket.connect was called"),
        )
        patcher.start()
        self.addCleanup(patcher.stop)

        for name in ("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: Tidy the imports in `abusectl/rdap.py`**

The module was built one task at a time, so `os`, `pathlib`, `time`,
`ipaddress` and `email.utils` are imported mid-file, after function
definitions, rather than grouped at the top. Move every import to the top of
the file in one block, standard library alphabetical, matching the other
modules in `abusectl/`. Change nothing else: this is a move, not a rewrite.

Run: `python3 -m unittest discover tests`
Expected: OK, the same count as before the move. If the count changes, the
move broke something; revert and redo it.

- [ ] **Step 5: Run the suite one final time**

Run: `python3 -m unittest discover tests`
Expected: OK

- [ ] **Step 6: Commit**

```bash
git add AGENTS.md README.md docs/BACKLOG.md abusectl/rdap.py
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.