diff options
Diffstat (limited to 'docs')
| -rw-r--r-- | docs/plans/2026-09-09-contacts.md | 1985 |
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. |