# abusectl `report` 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 report `, which turns a case's IOCs and abuse contacts into per-destination report bodies the user reviews before anything is sent. **Architecture:** One new pure module, `report.py`, plus a whitelisted `headers` block added to `parse.py` so `report` never opens `source.eml`. The report is a `multipart/report` MIME document built with `email.message` from the standard library. Grouping is per abuse address; a case with any sent destination is frozen and cannot be regenerated. **Tech Stack:** Python 3.11+, standard library only (`email`, `hashlib`, `json`, `unittest`). No new dependency, and `requirements.txt` is untouched. **Read first:** `docs/specs/2026-09-09-report.md` is the spec this implements, and `AGENTS.md` states the four properties that must not be weakened. Task 1 touches `parse.py`, which is the module the first property lives in. --- ## File structure | File | Responsibility | Task | |---|---|---| | `abusectl/parse.py` | gains `report_headers()`, a whitelist extractor | 1 | | `abusectl/report.py` | new: grouping, bodies, destinations, freeze check | 2-8 | | `abusectl/config.py` | gains the `[reporter]` section | 9 | | `abusectl/init.py` | gains three reporter prompts | 10 | | `abusectl/cli.py` | gains the `report` subcommand dispatch | 11 | | `tests/test_parse.py` | header whitelist tests | 1 | | `tests/test_report.py` | new: everything in `report.py` | 2-8 | | `tests/test_config.py` | reporter section tests | 9 | | `tests/test_cli.py` | dispatch and exit codes | 11 | | `tests/fixtures/reportable.eml` | new fixture with recipient headers present | 1 | `report.py` is one module, not two. The spec says why: there is no protocol layer to split off, because `email.message` already is that layer. --- ### Task 1: `parse.report_headers()`, the whitelist The spec's second decision: `parse` stores the whitelisted headers in the manifest so `report` never opens `source.eml`. The test that matters is the one proving `To` is DROPPED. **Files:** - Create: `tests/fixtures/reportable.eml` - Modify: `abusectl/parse.py` (add `_REPORT_HEADERS` and `report_headers()`) - Test: `tests/test_parse.py` - [ ] **Step 1: Create the fixture** It must carry the headers the whitelist keeps AND the ones it must drop. Check the weekday before writing it: `date -d 2026-09-07 +%A` returns `Monday`. An RFC2822 parser validates the day against the date and a wrong one reads as a malformed header. Create `tests/fixtures/reportable.eml`: ``` Received: from relay.example.org (relay.example.org [192.0.2.10]) by mx.example.org with ESMTP id abc123 for ; Mon, 07 Sep 2026 09:12:44 +0000 Received: from sender.invalid (sender.invalid [203.0.113.42]) by relay.example.org with ESMTP id def456; Mon, 07 Sep 2026 09:12:40 +0000 Return-Path: Authentication-Results: mx.example.org; spf=fail; dkim=none; dmarc=fail Received-SPF: fail (mx.example.org: domain of sender.invalid does not designate 203.0.113.42) From: "Example Bank" To: victim@example.org Cc: colleague@example.org Delivered-To: victim@example.org X-Original-To: victim@example.org Reply-To: "Support" Subject: Your account requires verification Date: Mon, 07 Sep 2026 09:12:40 +0000 Message-ID: MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Please verify at http://login.sender.invalid/verify?id=abc123 ``` - [ ] **Step 2: Write the failing tests** Add to `tests/test_parse.py`: ```python class ReportHeaders(unittest.TestCase): def setUp(self): self.raw = (FIXTURES / "reportable.eml").read_bytes() def test_the_whitelist_keeps_what_a_desk_needs(self): headers = parse.report_headers(self.raw, trusted=["192.0.2.0/24"]) names = [name for name, _ in headers] self.assertIn("From", names) self.assertIn("Subject", names) self.assertIn("Date", names) self.assertIn("Message-ID", names) self.assertIn("Reply-To", names) self.assertIn("Return-Path", names) self.assertIn("Authentication-Results", names) self.assertIn("Received-SPF", names) def test_recipient_headers_never_survive_the_whitelist(self): headers = parse.report_headers(self.raw, trusted=["192.0.2.0/24"]) blob = repr(headers) for name in ("To", "Cc", "Delivered-To", "X-Original-To"): self.assertNotIn(name, [n for n, _ in headers]) self.assertNotIn("victim@example.org", blob) self.assertNotIn("colleague@example.org", blob) def test_received_stops_at_the_boundary_hop(self): # 192.0.2.10 is ours, so its Received line is our own infrastructure # and must not be published; the hop below it is the one being # reported and is kept. headers = parse.report_headers(self.raw, trusted=["192.0.2.0/24"]) received = [value for name, value in headers if name == "Received"] self.assertEqual(len(received), 1) self.assertIn("203.0.113.42", received[0]) self.assertNotIn("mx.example.org with ESMTP id abc123", received[0]) def test_a_forged_chain_publishes_no_hop_below_the_boundary(self): raw = (FIXTURES / "forged-chain.eml").read_bytes() headers = parse.report_headers(raw, trusted=["192.0.2.0/24"]) received = [value for name, value in headers if name == "Received"] self.assertTrue(all("198.51.100.7" not in value for value in received)) ``` The last test is the one that protects an innocent party, the same job `test_a_forged_chain_stops_at_the_first_untrusted_hop` already does for `sending_ip()`. Read that test first and match its trusted-relay argument to the fixture it uses; if `forged-chain.eml` uses a different boundary network, use that one here rather than `192.0.2.0/24`. - [ ] **Step 3: Run to verify they fail** Run: `python3 -m unittest tests.test_parse.ReportHeaders -v` Expected: FAIL, `AttributeError: module 'abusectl.parse' has no attribute 'report_headers'` - [ ] **Step 4: Implement** Add to `abusectl/parse.py`, near the other module constants: ```python # The headers that may appear in a published report. A WHITELIST, never a # blacklist: a blacklist means every header this parser learns to read later # is a leak waiting for someone to remember. To, Cc, Delivered-To and # X-Original-To are absent by construction, which is the same reason iocs() # does not read them either. _REPORT_HEADERS = ( "From", "Subject", "Date", "Message-ID", "Reply-To", "Return-Path", "Authentication-Results", "Received-SPF", "MIME-Version", "Content-Type", ) ``` And the function, next to `received_hops()`: ```python def report_headers(raw: bytes, trusted: list[str]) -> list[tuple[str, str]]: """Return the headers that may be published, outermost Received first. Received is truncated at the trust boundary: our own relays are our infrastructure and publishing them tells a third party about the user's mail path, so only the boundary hop and below are kept. Everything else comes from a fixed whitelist. Returned as a list of pairs rather than a dict because Received repeats and order carries meaning. """ message = _message(raw) result: list[tuple[str, str]] = [] for value in message.get_all("received") or []: ip = _extract_ip(str(value)) if ip is not None and _in_any(ip, trusted): continue result.append(("Received", str(value))) for name in _REPORT_HEADERS: value = message.get(name) if value is not None: result.append((name, str(value))) return result ``` - [ ] **Step 5: Run to verify they pass** Run: `python3 -m unittest tests.test_parse -v` Expected: PASS, and every pre-existing parse test still passes. - [ ] **Step 6: Wire it into the manifest** In `abusectl/cli.py`, in the parse command around line 264, add the third line: ```python manifest["iocs"] = parse_module.iocs(raw, trusted=settings.trusted_relays) manifest["auth"] = parse_module.auth_results(raw) manifest["headers"] = parse_module.report_headers( raw, trusted=settings.trusted_relays ) ``` - [ ] **Step 7: Run the whole suite** Run: `python3 -m unittest discover tests` Expected: PASS, no regressions. - [ ] **Step 8: Commit** ```bash git add abusectl/parse.py abusectl/cli.py tests/test_parse.py tests/fixtures/reportable.eml git commit -S -m "feat: store a whitelist of publishable headers in the manifest report must never open source.eml, so parse decides once what may be published and report formats only what it is given. Received is truncated at the trust boundary; To, Cc, Delivered-To and X-Original-To are absent by construction rather than stripped." ``` --- ### Task 2: Group contacts into destinations, one per abuse address **Files:** - Create: `abusectl/report.py` - Test: `tests/test_report.py` - [ ] **Step 1: Write the failing test** Create `tests/test_report.py`: ```python import unittest from abusectl import report class Grouping(unittest.TestCase): def test_two_contacts_at_one_address_become_one_destination(self): contacts = [ {"iocs": ["ioc-1"], "query": "198.51.100.7", "abuse": ["abuse@host.invalid"], "source": "rdap"}, {"iocs": ["ioc-2"], "query": "example.invalid", "abuse": ["abuse@host.invalid"], "source": "rdap"}, ] destinations = report.email_destinations(contacts) self.assertEqual(len(destinations), 1) self.assertEqual(destinations[0]["target"], "abuse@host.invalid") self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"]) def test_a_contact_with_two_addresses_reaches_both_desks(self): contacts = [ {"iocs": ["ioc-1"], "query": "198.51.100.7", "abuse": ["a@host.invalid", "b@host.invalid"], "source": "rdap"}, ] destinations = report.email_destinations(contacts) self.assertEqual( sorted(d["target"] for d in destinations), ["a@host.invalid", "b@host.invalid"], ) def test_a_contact_with_no_address_creates_no_destination(self): contacts = [ {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], "source": "rdap", "error": "no abuse role published"}, ] self.assertEqual(report.email_destinations(contacts), []) def test_destinations_carry_stable_ids_and_pending_status(self): contacts = [ {"iocs": ["ioc-1"], "query": "198.51.100.7", "abuse": ["abuse@host.invalid"], "source": "rdap"}, ] destination = report.email_destinations(contacts)[0] self.assertEqual(destination["id"], "email-1") self.assertEqual(destination["kind"], "email") self.assertEqual(destination["status"], "pending") ``` - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_report.Grouping -v` Expected: FAIL, `ModuleNotFoundError: No module named 'abusectl.report'` - [ ] **Step 3: Implement** Create `abusectl/report.py` with the GPLv2 header used by every other module in this package (copy the fourteen-line block from the top of `abusectl/case.py`, changing nothing but what follows it), then: ```python """IOCs and abuse contacts to report bodies: the last step before anything irreversible happens. This module is PURE and OFFLINE. It opens no socket, sends no mail and reads no file outside the case directory. What it produces is a document the user reads, edits and approves, so the output is written for a human first and a parser second. It takes the reporting identity as an ARGUMENT rather than reading the config, the way parse.py takes the trust boundary. The identity is the one thing in a report disclosed deliberately, and a module that reaches for it itself is a module that can disclose it in a code path nobody reviewed. """ import hashlib def email_destinations(contacts: list[dict]) -> list[dict]: """Group contacts into one destination per abuse ADDRESS. Contacts already fold by host, but two different contacts can still resolve to the same address, an IP and a domain at one hoster being the common case. One mail per address rather than per contact is what stops a desk receiving two mails about one incident. """ by_address: dict[str, list[str]] = {} for contact in contacts: for address in contact.get("abuse", []): iocs = by_address.setdefault(address, []) for ioc in contact.get("iocs", []): if ioc not in iocs: iocs.append(ioc) return [ { "id": f"email-{index}", "kind": "email", "target": address, "iocs": iocs, "body": None, "status": "pending", } for index, (address, iocs) in enumerate(by_address.items(), start=1) ] ``` - [ ] **Step 4: Run to verify it passes** Run: `python3 -m unittest tests.test_report.Grouping -v` Expected: PASS, 4 tests. - [ ] **Step 5: Commit** ```bash git add abusectl/report.py tests/test_report.py git commit -S -m "feat: group abuse contacts into one destination per address Two contacts can resolve to the same desk, an IP and a domain at one hoster being the common case, and grouping per contact would send that desk two mails about one incident." ``` --- ### Task 3: The unreportable list **Files:** - Modify: `abusectl/report.py` - Test: `tests/test_report.py` - [ ] **Step 1: Write the failing test** ```python class Unreportable(unittest.TestCase): def test_an_ioc_with_no_desk_is_listed_with_its_reason(self): contacts = [ {"iocs": ["ioc-1"], "query": "198.51.100.7", "abuse": ["abuse@host.invalid"], "source": "rdap"}, {"iocs": ["ioc-2", "ioc-3"], "query": "example.invalid", "abuse": [], "source": "rdap", "error": "no abuse role published"}, ] self.assertEqual( report.unreportable(contacts), [ {"ioc": "ioc-2", "reason": "no abuse role published"}, {"ioc": "ioc-3", "reason": "no abuse role published"}, ], ) def test_a_missing_reason_still_produces_an_entry(self): contacts = [{"iocs": ["ioc-9"], "query": "x.invalid", "abuse": [], "source": "rdap"}] self.assertEqual( report.unreportable(contacts), [{"ioc": "ioc-9", "reason": "no abuse address resolved"}], ) def test_nothing_unreportable_is_an_empty_list_not_an_error(self): contacts = [{"iocs": ["ioc-1"], "query": "198.51.100.7", "abuse": ["abuse@host.invalid"], "source": "rdap"}] self.assertEqual(report.unreportable(contacts), []) ``` - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_report.Unreportable -v` Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'unreportable'` - [ ] **Step 3: Implement** Add to `abusectl/report.py`: ```python def unreportable(contacts: list[dict]) -> list[dict]: """List every IOC that reached no email destination, with the reason. A missing contact is a normal outcome, not an error: RDAP publishes no abuse role for many netblocks. Making it visible is what keeps review honest, since finding it any other way means diffing the IOC list against every destination's IOC list. Same instinct as suspect_path_segments flagging rather than redacting. """ result = [] for contact in contacts: if contact.get("abuse"): continue reason = contact.get("error", "no abuse address resolved") for ioc in contact.get("iocs", []): result.append({"ioc": ioc, "reason": reason}) return result ``` - [ ] **Step 4: Run to verify it passes** Run: `python3 -m unittest tests.test_report.Unreportable -v` Expected: PASS, 3 tests. - [ ] **Step 5: Commit** ```bash git add abusectl/report.py tests/test_report.py git commit -S -m "feat: list the indicators no abuse desk was found for Not an error and not an exit code: a case where nothing resolved still reaches MISP and the vendors. Visible beats absent, so review can see it without diffing IOC lists." ``` --- ### Task 4: The plain-text part **Files:** - Modify: `abusectl/report.py` - Test: `tests/test_report.py` The wording is NOT asserted; the spec says whether it reads well to a desk has no assertion and is hand-tested. What is asserted is what must and must not appear. - [ ] **Step 1: Write the failing test** ```python IDENTITY = {"name": "A Reporter", "org": "Example Consulting", "email": "reporter@example.org"} MANIFEST = { "format": 1, "case_id": "2026-09-07-aaaa", "iocs": [ {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", "origin": "received-chain", "confidence": "boundary-hop"}, {"id": "ioc-2", "type": "url", "value": "http://login.sender.invalid/verify?id=REDACTED", "origin": "body"}, ], "auth": {"spf": "fail", "dkim": "none", "dmarc": "fail"}, "headers": [ ("From", '"Example Bank" '), ("Subject", "Your account requires verification"), ("Date", "Mon, 07 Sep 2026 09:12:40 +0000"), ], "contacts": [ {"iocs": ["ioc-1", "ioc-2"], "query": "203.0.113.42", "abuse": ["abuse@host.invalid"], "source": "rdap"}, ], } class TextPart(unittest.TestCase): def setUp(self): destination = report.email_destinations(MANIFEST["contacts"])[0] self.text = report.text_part(MANIFEST, destination, IDENTITY) def test_the_redaction_note_is_always_present(self): self.assertIn("Recipient identifiers", self.text) def test_the_reporter_identity_appears(self): self.assertIn("A Reporter", self.text) self.assertIn("Example Consulting", self.text) self.assertIn("reporter@example.org", self.text) def test_the_destinations_own_indicators_appear(self): self.assertIn("203.0.113.42", self.text) self.assertIn("http://login.sender.invalid/verify?id=REDACTED", self.text) def test_an_indicator_belonging_to_another_desk_does_not_appear(self): manifest = dict(MANIFEST) manifest["iocs"] = MANIFEST["iocs"] + [ {"id": "ioc-9", "type": "ipv4", "value": "192.0.2.99", "origin": "received-chain"}, ] destination = report.email_destinations(MANIFEST["contacts"])[0] text = report.text_part(manifest, destination, IDENTITY) self.assertNotIn("192.0.2.99", text) def test_no_line_exceeds_seventy_two_columns(self): for line in self.text.splitlines(): self.assertLessEqual(len(line), 72, line) ``` - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_report.TextPart -v` Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'text_part'` - [ ] **Step 3: Implement** Add to `abusectl/report.py`: ```python _REDACTION_NOTE = ( "Recipient identifiers have been removed from this report by policy.\n" "Parameter names are preserved, parameter values are not. Full evidence\n" "is retained locally and is available on request." ) def _iocs_by_id(manifest: dict) -> dict: return {entry["id"]: entry for entry in manifest.get("iocs", [])} def text_part(manifest: dict, destination: dict, identity: dict) -> str: """Build the human-readable part: the one that decides whether a desk acts on the report. The ask goes first, because a desk triaging a queue must know in one line what happened and what is wanted. Only this destination's own indicators appear: a desk shown three IPs that are not theirs stops reading. """ by_id = _iocs_by_id(manifest) mine = [by_id[i] for i in destination["iocs"] if i in by_id] lines = [ "Phishing message reported: infrastructure on your network was", "used to send or host it. Requesting takedown and customer", "notification.", "", "Observed on your infrastructure:", "", ] for entry in mine: lines.append(f" {entry['value']}") origin = entry.get("origin", "") if entry.get("confidence") == "boundary-hop": lines.append(" sending IP, first hop outside our boundary") elif origin: lines.append(f" seen in: {origin}") headers = manifest.get("headers") or [] shown = [(name, value) for name, value in headers if name in ("Date", "From", "Subject")] if shown: lines += ["", "Message as declared:", ""] for name, value in shown: lines.append(f" {name}: {value}"[:72]) auth = manifest.get("auth") or {} if auth: lines += ["", "Authentication results:", ""] lines.append(" " + " ".join( f"{key.upper()}: {value}" for key, value in sorted(auth.items()) )[:70]) lines += ["", _REDACTION_NOTE, ""] lines.append( f"Reported by: {identity['name']}, {identity['org']} " f"<{identity['email']}>"[:72] ) lines.append("Generated by abusectl.") return "\n".join(lines) + "\n" ``` - [ ] **Step 4: Run to verify it passes** Run: `python3 -m unittest tests.test_report.TextPart -v` Expected: PASS, 5 tests. If the 72-column test fails on a long URL, that is a real finding rather than a test to relax: wrap the URL onto its own line rather than truncating it, because a truncated URL is a wrong indicator. - [ ] **Step 5: Commit** ```bash git add abusectl/report.py tests/test_report.py git commit -S -m "feat: build the human-readable part of a report The ask goes first, only this desk's own indicators appear, and the redaction note is unconditional: a desk seeing REDACTED with no explanation may read the report as doctored." ``` --- ### Task 5: The machine-readable part **Files:** - Modify: `abusectl/report.py` - Test: `tests/test_report.py` - [ ] **Step 1: Write the failing test** ```python class FeedbackPart(unittest.TestCase): def setUp(self): destination = report.email_destinations(MANIFEST["contacts"])[0] self.fields = report.feedback_fields(MANIFEST, destination) self.lookup = dict(self.fields) def test_the_three_rfc5965_required_fields_are_present(self): self.assertEqual(self.lookup["Feedback-Type"], "abuse") self.assertEqual(self.lookup["Version"], "1") self.assertTrue(self.lookup["User-Agent"].startswith("abusectl/")) def test_the_xarf_report_type_is_phishing(self): self.assertEqual(self.lookup["Report-Type"], "phishing") def test_source_is_the_primary_indicator(self): self.assertEqual(self.lookup["Source"], "203.0.113.42") self.assertEqual(self.lookup["Source-IP"], "203.0.113.42") def test_every_url_appears_as_a_reported_uri(self): uris = [value for name, value in self.fields if name == "Reported-Uri"] self.assertEqual( uris, ["http://login.sender.invalid/verify?id=REDACTED"] ) def test_a_destination_with_no_ip_omits_source_ip(self): contacts = [{"iocs": ["ioc-2"], "query": "sender.invalid", "abuse": ["abuse@host.invalid"], "source": "rdap"}] destination = report.email_destinations(contacts)[0] lookup = dict(report.feedback_fields(MANIFEST, destination)) self.assertNotIn("Source-IP", lookup) self.assertEqual( lookup["Source"], "http://login.sender.invalid/verify?id=REDACTED" ) ``` - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_report.FeedbackPart -v` Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'feedback_fields'` - [ ] **Step 3: Implement** Add to `abusectl/report.py`, with the version constant near the top of the module: ```python VERSION = "0.1.0" def feedback_fields(manifest: dict, destination: dict) -> list[tuple[str, str]]: """Build the machine-readable part: an RFC 5965 envelope carrying x-arf fields inside it. RFC 5965 is the standard and is universally understood, but it was designed for feedback loops, where a report is about A MESSAGE. These reports are about INDICATORS, and 5965 has no field for "this specific host is the thing being reported". x-arf's Source does. The part is key/value, so a 5965 parser reads what it knows and ignores the rest. Returned as pairs, not a dict: Reported-Uri repeats. """ by_id = _iocs_by_id(manifest) mine = [by_id[i] for i in destination["iocs"] if i in by_id] ips = [e["value"] for e in mine if e.get("type") in ("ipv4", "ipv6")] urls = [e["value"] for e in mine if e.get("type") == "url"] domains = [e["value"] for e in mine if e.get("type") == "domain"] # Source is singular, so the primary indicator fills it and the rest # travel in repeated fields and in the text part. An IP is the most # actionable thing a hosting desk can act on, so it wins when present. primary = (ips or domains or urls or [""])[0] fields = [ ("Feedback-Type", "abuse"), ("User-Agent", f"abusectl/{VERSION}"), ("Version", "1"), ("Report-Type", "phishing"), ("Source", primary), ] for ip in ips: fields.append(("Source-IP", ip)) for domain in domains: fields.append(("Reported-Domain", domain)) for url in urls: fields.append(("Reported-Uri", url)) for name, value in manifest.get("headers") or []: if name == "Date": fields.append(("Arrival-Date", value)) break return fields ``` - [ ] **Step 4: Run to verify it passes** Run: `python3 -m unittest tests.test_report.FeedbackPart -v` Expected: PASS, 5 tests. - [ ] **Step 5: Commit** ```bash git add abusectl/report.py tests/test_report.py git commit -S -m "feat: build the machine-readable feedback report part An RFC 5965 envelope carrying x-arf fields. 5965 reports are about a message and these are about indicators, so x-arf's Source fills the gap while the envelope keeps a standards parser working." ``` --- ### Task 6: Assemble the MIME document **Files:** - Modify: `abusectl/report.py` - Test: `tests/test_report.py` - [ ] **Step 1: Write the failing test** ```python import email import email.policy class Document(unittest.TestCase): def setUp(self): self.destination = report.email_destinations(MANIFEST["contacts"])[0] self.raw = report.build(MANIFEST, self.destination, IDENTITY) self.parsed = email.message_from_string( self.raw, policy=email.policy.default ) def test_it_is_a_feedback_report_with_three_parts(self): self.assertEqual(self.parsed.get_content_type(), "multipart/report") self.assertEqual(self.parsed.get_param("report-type"), "feedback-report") parts = list(self.parsed.iter_parts()) self.assertEqual( [part.get_content_type() for part in parts], ["text/plain", "message/feedback-report", "text/rfc822-headers"], ) def test_the_envelope_is_addressed_and_identified(self): self.assertEqual(self.parsed["To"], "abuse@host.invalid") self.assertIn("reporter@example.org", self.parsed["From"]) self.assertTrue(self.parsed["Subject"]) def test_the_headers_part_carries_no_recipient_header(self): headers_part = list(self.parsed.iter_parts())[2] body = headers_part.get_content() for name in ("To:", "Cc:", "Delivered-To:", "X-Original-To:"): self.assertNotIn(name, body) def test_the_source_message_is_never_attached(self): self.assertNotIn("message/rfc822", self.raw) ``` - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_report.Document -v` Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'build'` - [ ] **Step 3: Implement** Add the imports at the top of `abusectl/report.py`: ```python from email.message import EmailMessage from email.policy import SMTP ``` And the function: ```python def build(manifest: dict, destination: dict, identity: dict) -> str: """Assemble one destination's report as an RFC 5965 MIME document. Three parts: what a human reads, what a parser reads, and the headers. The original message is NOT attached, and there is no message/rfc822 part: source.eml carries every identifier the first property exists to keep out, and an abuse desk forwards a report to the abused customer, who for a phishing domain may be the attacker. RFC 5965 provides text/rfc822-headers for exactly this case, so this is the standard's own answer rather than a deviation from it. """ message = EmailMessage(policy=SMTP) message["From"] = f"{identity['name']} <{identity['email']}>" message["To"] = destination["target"] message["Subject"] = ( f"Abuse report: phishing infrastructure, case {manifest['case_id']}" ) message.make_mixed() message.set_type("multipart/report") message.set_param("report-type", "feedback-report") human = EmailMessage(policy=SMTP) human.set_content(text_part(manifest, destination, identity)) message.attach(human) machine = EmailMessage(policy=SMTP) machine.set_content( "\n".join(f"{name}: {value}" for name, value in feedback_fields(manifest, destination)) + "\n" ) machine.set_type("message/feedback-report") message.attach(machine) headers = EmailMessage(policy=SMTP) headers.set_content( "\n".join(f"{name}: {value}" for name, value in manifest.get("headers") or []) + "\n" ) headers.set_type("text/rfc822-headers") message.attach(headers) return message.as_string() ``` - [ ] **Step 4: Run to verify it passes** Run: `python3 -m unittest tests.test_report.Document -v` Expected: PASS, 4 tests. If `set_type` on a subpart raises, set the type BEFORE `set_content` on that part and re-run; the ordering matters in `email.message`. - [ ] **Step 5: Commit** ```bash git add abusectl/report.py tests/test_report.py git commit -S -m "feat: assemble the RFC 5965 report document Three parts and no message/rfc822: the original carries every identifier the first property keeps out, and text/rfc822-headers is the standard's own answer for a report that cannot include the message." ``` --- ### Task 7: Body hashes and the frozen case **Files:** - Modify: `abusectl/report.py` - Test: `tests/test_report.py` - [ ] **Step 1: Write the failing test** ```python class Freeze(unittest.TestCase): def test_a_frozen_case_refuses(self): manifest = dict(MANIFEST) manifest["frozen"] = {"at": "2026-09-07T10:00:00Z", "by": "abusedb"} with self.assertRaises(report.Frozen) as caught: report.check_regenerable(manifest, modified=[]) self.assertIn("abusedb", str(caught.exception)) def test_a_frozen_case_refuses_even_when_forced(self): manifest = dict(MANIFEST) manifest["frozen"] = {"at": "2026-09-07T10:00:00Z", "by": "abusedb"} with self.assertRaises(report.Frozen): report.check_regenerable(manifest, modified=[], force=True) def test_a_modified_body_refuses_without_force(self): with self.assertRaises(report.Modified) as caught: report.check_regenerable(MANIFEST, modified=["email-1"]) self.assertIn("email-1", str(caught.exception)) def test_a_modified_body_is_allowed_with_force(self): report.check_regenerable(MANIFEST, modified=["email-1"], force=True) def test_an_untouched_case_regenerates(self): report.check_regenerable(MANIFEST, modified=[]) class Hashes(unittest.TestCase): def test_the_hash_detects_a_changed_body(self): first = report.body_hash("one") self.assertNotEqual(first, report.body_hash("two")) self.assertEqual(first, report.body_hash("one")) ``` - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_report.Freeze tests.test_report.Hashes -v` Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'Frozen'` - [ ] **Step 3: Implement** Add to `abusectl/report.py`: ```python class Frozen(Exception): """Raised when a case has already been reported to at least one desk. There is no force override. The destinations are not independent artifacts, they are one incident reported in parallel: regenerating one body after another desk holds the report leaves two desks with contradictory accounts of the same case. """ class Modified(Exception): """Raised when a body was edited after generation and --force was not given. Silently discarding a review that took twenty minutes is what makes a tool untrustworthy. """ def body_hash(text: str) -> str: """Hash a body's CONTENT, which is what the question actually is. An mtime is a poor witness in both directions: a git checkout, an rsync or a backup restore all move it with no human having edited anything, and an editor that preserves mtime hides a real edit. For a destination that has been sent, this hash is also the record of what was disclosed. """ return hashlib.sha256(text.encode("utf-8")).hexdigest() def check_regenerable(manifest: dict, modified: list[str], force: bool = False) -> None: """Raise unless this case may be regenerated. Returns None when it may. Frozen wins over force: once a desk holds the report, the case is an evidence record rather than a draft. """ frozen = manifest.get("frozen") if frozen: raise Frozen( f"case reported to {frozen.get('by', 'a destination')} at " f"{frozen.get('at', 'an unknown time')}; bodies cannot be " f"regenerated. Edit a body by hand if it must change." ) if modified and not force: raise Modified( "bodies edited since generation: " + ", ".join(modified) + ". Re-run with --force to discard those edits; a timestamped " "backup is kept." ) ``` - [ ] **Step 4: Run to verify it passes** Run: `python3 -m unittest tests.test_report.Freeze tests.test_report.Hashes -v` Expected: PASS, 6 tests. - [ ] **Step 5: Commit** ```bash git add abusectl/report.py tests/test_report.py git commit -S -m "feat: freeze a reported case and detect edited bodies Content hash rather than mtime, because mtime is wrong in both directions. Any sent destination freezes the whole case with no override: two desks holding contradictory accounts of one incident is worse than a stale body." ``` --- ### Task 8: Write the bodies to the case directory **Files:** - Modify: `abusectl/report.py` - Test: `tests/test_report.py` - [ ] **Step 1: Write the failing test** ```python import tempfile from pathlib import Path class Writing(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = Path(self.tmp.name) self.addCleanup(self.tmp.cleanup) def test_it_writes_one_body_per_destination_and_records_the_hash(self): manifest = report.generate(dict(MANIFEST), self.path, IDENTITY) destination = manifest["destinations"][0] body = self.path / destination["body"] self.assertTrue(body.exists()) self.assertEqual( destination["body_sha256"], report.body_hash(body.read_text()) ) def test_it_records_the_unreportable_indicators(self): manifest = dict(MANIFEST) manifest["contacts"] = MANIFEST["contacts"] + [ {"iocs": ["ioc-4"], "query": "x.invalid", "abuse": [], "source": "rdap", "error": "no abuse role published"}, ] result = report.generate(manifest, self.path, IDENTITY) self.assertEqual( result["unreportable"], [{"ioc": "ioc-4", "reason": "no abuse role published"}], ) def test_a_modified_body_is_backed_up_before_being_overwritten(self): manifest = report.generate(dict(MANIFEST), self.path, IDENTITY) body = self.path / manifest["destinations"][0]["body"] body.write_text("hand edited during review\n") with self.assertRaises(report.Modified): report.generate(manifest, self.path, IDENTITY) report.generate(manifest, self.path, IDENTITY, force=True) backups = list((self.path / "bodies").glob("*.orig")) self.assertEqual(len(backups), 1) self.assertEqual(backups[0].read_text(), "hand edited during review\n") def test_a_deleted_body_regenerates_without_complaint(self): manifest = report.generate(dict(MANIFEST), self.path, IDENTITY) (self.path / manifest["destinations"][0]["body"]).unlink() again = report.generate(manifest, self.path, IDENTITY) self.assertTrue((self.path / again["destinations"][0]["body"]).exists()) ``` - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_report.Writing -v` Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'generate'` - [ ] **Step 3: Implement** Add the imports: ```python from datetime import datetime, timezone from pathlib import Path ``` And the function: ```python def generate(manifest: dict, case_path: Path, identity: dict, force: bool = False) -> dict: """Write every body and return the manifest with destinations[] set. The manifest is RETURNED rather than saved: case.py is the only writer of a case directory, so the caller saves. The bodies are this module's to write because they are not the manifest. """ case_path = Path(case_path) bodies = case_path / "bodies" bodies.mkdir(exist_ok=True) modified = [] for existing in manifest.get("destinations") or []: recorded = existing.get("body_sha256") if not recorded or not existing.get("body"): continue path = case_path / existing["body"] if path.exists() and body_hash(path.read_text()) != recorded: modified.append(existing["id"]) check_regenerable(manifest, modified, force=force) stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") for destination_id in modified: existing = next(d for d in manifest["destinations"] if d["id"] == destination_id) path = case_path / existing["body"] path.rename(path.with_suffix(path.suffix + f".{stamp}.orig")) destinations = email_destinations(manifest.get("contacts", [])) for destination in destinations: text = build(manifest, destination, identity) relative = f"bodies/{destination['id']}.xarf" (case_path / relative).write_text(text) destination["body"] = relative destination["body_sha256"] = body_hash(text) manifest["destinations"] = destinations manifest["unreportable"] = unreportable(manifest.get("contacts", [])) return manifest ``` - [ ] **Step 4: Run to verify it passes** Run: `python3 -m unittest tests.test_report -v` Expected: PASS, every class in the module. - [ ] **Step 5: Commit** ```bash git add abusectl/report.py tests/test_report.py git commit -S -m "feat: write report bodies into the case directory case.py stays the only writer of the manifest, so generate returns it and the caller saves. A modified body is backed up with a timestamp before --force overwrites it." ``` --- ### Task 9: The `[reporter]` config section **Files:** - Modify: `abusectl/config.py` - Test: `tests/test_config.py` - [ ] **Step 1: Write the failing test** Add to `tests/test_config.py`, following the file's existing pattern for writing a temporary config (copy the helper the neighbouring tests use rather than inventing one): ```python class Reporter(unittest.TestCase): def test_the_reporter_identity_is_read(self): settings = self._load(""" [general] trusted_relays = ["192.0.2.0/24"] [reporter] name = "A Reporter" org = "Example Consulting" email = "reporter@example.org" """) self.assertEqual(settings.reporter["name"], "A Reporter") self.assertEqual(settings.reporter["email"], "reporter@example.org") def test_an_absent_reporter_section_is_an_empty_dict_not_a_crash(self): settings = self._load(""" [general] trusted_relays = ["192.0.2.0/24"] """) self.assertEqual(settings.reporter, {}) def test_an_empty_value_is_treated_as_absent(self): settings = self._load(""" [general] trusted_relays = ["192.0.2.0/24"] [reporter] name = "A Reporter" org = "" email = "reporter@example.org" """) self.assertNotIn("org", settings.reporter) ``` - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_config.Reporter -v` Expected: FAIL, `AttributeError: 'Config' object has no attribute 'reporter'` - [ ] **Step 3: Implement** In `abusectl/config.py`, add the field to the dataclass: ```python @dataclass(frozen=True) class Config: trusted_relays: list[str] cases: pathlib.Path reporter: dict ``` And in `load()`, before the `Config(...)` construction: ```python # A skipped answer is ABSENT, never an empty string: "" reads as # configured-and-broken and produces a confusing failure much later, # while absent reads as not-configured and the part that wants it can # say so plainly. Same rule the rest of this file follows. raw_reporter = data.get("reporter", {}) reporter = { key: value for key, value in raw_reporter.items() if isinstance(value, str) and value.strip() } ``` Then pass `reporter=reporter` into the returned `Config`. Every other construction of `Config` in the codebase and in the tests needs the new field; run the full suite in step 4 to find them. - [ ] **Step 4: Run the whole suite** Run: `python3 -m unittest discover tests` Expected: PASS. Fix any `Config()` construction the new field broke. - [ ] **Step 5: Commit** ```bash git add abusectl/config.py tests/test_config.py git commit -S -m "feat: read the reporter identity from config The reporter's identity is the one identifier this tool discloses deliberately, so it comes from config only and parse never supplies it. An empty value is absent, the same rule the rest of the config follows." ``` --- ### Task 10: Three `init` prompts **Files:** - Modify: `abusectl/init.py` The prompts are hand-tested, not unit-tested: the spec and `AGENTS.md` both say whether a question reads clearly has no assertion. What IS tested is the builder, if `init.py` has one that produces TOML. - [ ] **Step 1: Read the existing prompt flow** Run: `grep -n "def \|input(" abusectl/init.py` Follow the shape already there. Validate each answer AT the prompt that asked for it and re-ask on a bad one, rather than erroring after the next question: `AGENTS.md` records that a hand test found four defects of exactly that shape. - [ ] **Step 2: Add the three prompts** Ask for name, organisation and email, each skippable. A skipped answer must be ABSENT from the generated TOML, never `""`. Emit the section only if at least one answer was given. - [ ] **Step 3: Extend the builder test if one exists** If `tests/test_init.py` asserts on generated TOML, add a case that a skipped reporter answer produces no key, matching the existing skipped-answer tests. Run: `python3 -m unittest tests.test_init -v` Expected: PASS. - [ ] **Step 4: Hand test** Run: `python3 -m abusectl init --force` in a scratch `XDG_CONFIG_HOME` and read the questions. This is the test that matters for prompts. ```bash XDG_CONFIG_HOME=$(mktemp -d) python3 -m abusectl init ``` - [ ] **Step 5: Commit** ```bash git add abusectl/init.py tests/test_init.py git commit -S -m "feat: ask for the reporter identity during init Each answer is validated at the prompt that asked for it, and a skipped answer is absent from the file rather than an empty string." ``` --- ### Task 11: The `report` subcommand **Files:** - Modify: `abusectl/cli.py` - Test: `tests/test_cli.py` - [ ] **Step 1: Write the failing test** Follow the pattern `tests/test_cli.py` already uses for the contacts command. ```python class ReportCommand(unittest.TestCase): def test_a_missing_case_is_an_error_not_a_traceback(self): code = cli.main(["report", "/nonexistent/case"]) self.assertEqual(code, cli.EXIT_ERROR) def test_a_case_with_no_reporter_configured_says_so(self): # An unconfigured identity is not-configured, not a crash: the # report would otherwise be filed with no reply address. ... ``` Fill the second test in following the neighbouring tests' fixture setup; if those tests build a case directory with a helper, reuse it rather than writing a new one. - [ ] **Step 2: Run to verify it fails** Run: `python3 -m unittest tests.test_cli.ReportCommand -v` Expected: FAIL, argparse rejects the unknown command `report`. - [ ] **Step 3: Implement the parser entry** In `abusectl/cli.py`, beside the contacts parser around line 65: ```python report_parser = subparsers.add_parser( "report", help="build report bodies for a case" ) report_parser.add_argument("case", type=Path) report_parser.add_argument( "--force", action="store_true", help="discard hand edits to bodies, keeping a timestamped backup", ) ``` - [ ] **Step 4: Implement the command** Add beside `cmd_contacts`, matching its error handling exactly: ```python def cmd_report(args) -> int: """Build report bodies and rewrite the manifest. Offline and irreversible-free: nothing here sends anything. The output is what the user reviews before submit does something that cannot be recalled. """ try: settings = config.load() except config.NotConfigured as exc: print(f"abusectl report: {exc}", file=sys.stderr) return EXIT_NOT_CONFIGURED identity = settings.reporter missing = [k for k in ("name", "org", "email") if k not in identity] if missing: print( "abusectl report: no reporter identity configured " f"(missing {', '.join(missing)}). Run `abusectl init`.", file=sys.stderr, ) return EXIT_NOT_CONFIGURED try: manifest = case.load(args.case) except FileNotFoundError: print(f"abusectl report: no case at {args.case}", file=sys.stderr) return EXIT_ERROR except ValueError as exc: print(f"abusectl report: {args.case}: {exc}", file=sys.stderr) return EXIT_ERROR try: manifest = report_module.generate( manifest, args.case, identity, force=args.force ) except (report_module.Frozen, report_module.Modified) as exc: print(f"abusectl report: {exc}", file=sys.stderr) return EXIT_ERROR case.save(args.case, manifest) count = len(manifest["destinations"]) orphans = len(manifest["unreportable"]) print(f"{count} destinations, {orphans} indicators with no abuse desk") return EXIT_OK ``` Import the module at the top as `from abusectl import report as report_module`, matching how `contacts` and `parse` are imported there, and add the dispatch entry beside the others in `main()`. - [ ] **Step 5: Run the whole suite** Run: `python3 -m unittest discover tests` Expected: PASS. - [ ] **Step 6: Commit** ```bash git add abusectl/cli.py tests/test_cli.py git commit -S -m "feat: add the report subcommand Refuses without a configured reporter identity rather than filing a report with no reply address, and turns a frozen or edited case into an error message rather than a traceback." ``` --- ### Task 12: Prove the suite still opens no socket The second property. `AGENTS.md` says it is verified, not asserted, and this plan adds a module that must not break it. - [ ] **Step 1: Run the suite with the network unavailable** There is an existing test that does this; find it and confirm it covers the new module. Run: `grep -rn "getaddrinfo\|create_connection" tests/` - [ ] **Step 2: Run the whole suite under that harness** Run: `python3 -m unittest discover tests` Expected: PASS, including the offline-proof test. - [ ] **Step 3: Commit only if a change was needed** If the existing offline test already imports and exercises `report.py`, nothing to commit. If it enumerates modules by name, add `report` to it and commit: ```bash git add tests/test_offline.py git commit -S -m "test: cover report.py in the no-socket proof" ``` --- ### Task 13: Re-run both sweeps `AGENTS.md` requires this after any change to `parse.py`, and Task 1 changed it. **Ask the user before reading their mail.** The script lives in the scratchpad, never in the repository. - [ ] **Step 1: Ask permission** The corpus is the user's own spam in notmuch. Do not read it unasked. - [ ] **Step 2: Extend sweep A with the third assertion** The existing sweep asserts no address from the raw source appears in the IOC output. Add the same assertion against every generated body: ```python raw = subprocess.run(["notmuch", "show", "--format=raw", mid], capture_output=True, check=True).stdout manifest = { "format": 1, "case_id": "sweep", "iocs": parse.iocs(raw, trusted=TRUSTED), "auth": parse.auth_results(raw), "headers": parse.report_headers(raw, trusted=TRUSTED), # contacts.worklist() is OFFLINE and issues no query; a fake abuse # address per item is enough to force a body to be generated, which is # what this assertion needs. contacts.resolve() must NOT be called here: # sweep A sends nothing. "contacts": [ {"iocs": item.iocs, "query": item.query, "abuse": ["desk@sweep.invalid"], "source": "rdap"} for item in contacts.worklist(parse.iocs(raw, trusted=TRUSTED)) if item.kind != "unusable" ], } bodies = "".join( report.build(manifest, destination, IDENTITY) for destination in report.email_destinations(manifest["contacts"]) ) for addr in set(re.findall(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+", raw.decode("utf-8", "replace"))): assert addr not in bodies, (mid, addr) ``` The assertion must stay that broad: every address in the raw source against every byte of every body. Checking only the recipient misses an address the parser invented from a display name, which is how the `_domain_of` defect reached a report. Note that the reporter identity in the sweep must be a placeholder, not the user's real address, or the assertion will fire on it. - [ ] **Step 3: Record the counts** Report messages swept, bodies generated, crashes, and any assertion failure. Counts are evidence and may leave the script; addresses, subjects, Message-IDs and real URLs may not. - [ ] **Step 4: Reproduce any finding as a synthetic fixture** A defect found in real mail becomes a fixture using `example.org`, `.invalid` and RFC 5737 ranges, committed with its failing test. The real message stays in the scratchpad. - [ ] **Step 5: Update the docs** Add the `report` spec to the Documents list in `AGENTS.md`, and record the sweep result the way the contacts sweep is recorded there. ```bash git add AGENTS.md git commit -S -m "docs: record the report spec and its sweep" ``` --- ## Self-review notes Checked against `docs/specs/2026-09-09-report.md`: - Third part `text/rfc822-headers`, message never attached: Tasks 1, 6 - Whitelist in `parse`, `report` never opens `source.eml`: Task 1 - 5965 envelope with x-arf fields: Task 5 - Identity from config, three keys, absent-not-empty: Tasks 9, 10 - One destination per abuse address: Task 2 - `unreportable[]`, not an error: Tasks 3, 8 - SHA-256 not mtime, freeze with no override, explicit marker: Tasks 7, 8 - Sweeps re-run with the third assertion: Task 13 Two spec items deliberately have no task, and both are correct as gaps: - **`api` destinations get rows but null bodies.** The spec narrowed this to `submit`, so `email_destinations()` builds only email rows. When the submit spec lands, the vendor rows join here. - **`submit` writing `frozen` atomically with the first `sent`.** That is `submit`'s work; Task 7 only reads the field.