diff options
Diffstat (limited to 'docs')
| -rw-r--r-- | docs/plans/2026-09-08-parse.md | 2322 | ||||
| -rw-r--r-- | docs/specs/2026-09-08-abusectl-design.md | 84 |
2 files changed, 2404 insertions, 2 deletions
diff --git a/docs/plans/2026-09-08-parse.md b/docs/plans/2026-09-08-parse.md new file mode 100644 index 0000000..579ff0e --- /dev/null +++ b/docs/plans/2026-09-08-parse.md @@ -0,0 +1,2322 @@ +# abusectl `parse` 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:** `abusectl parse msg.eml` writes a case directory holding the message and a manifest of its indicators, with no network access and no recipient identifiers. + +**Architecture:** Four pure modules over values (`redact`, `parse`, `case`, `init`) and one thin `cli` over them. Nothing opens a socket. The trusted-relay boundary is an argument, not a config read, so `parse` is testable with no files on disk. + +**Tech Stack:** Python 3.12, standard library only (`email`, `ipaddress`, `hashlib`, `tomllib`, `json`, `unittest`). No third-party dependencies in this part. + +**Spec:** `docs/specs/2026-09-08-abusectl-design.md` + +--- + +## File structure + +| File | Responsibility | +|---|---| +| `abusectl/__init__.py` | version string, nothing else | +| `abusectl/redact.py` | URL redaction; the safety rule, alone and testable | +| `abusectl/parse.py` | `.eml` bytes -> IOC list. Pure, no config, no network | +| `abusectl/case.py` | case directory: create, manifest read/write, atomic | +| `abusectl/config.py` | read TOML, locate config, typed access | +| `abusectl/init.py` | pure config builder + prompt shell + provider table | +| `abusectl/cli.py` | argparse dispatch, exit codes, wiring only | +| `tests/test_redact.py` | redaction rules | +| `tests/test_parse.py` | extraction, per IOC type | +| `tests/test_case.py` | directory layout, manifest round trip, atomicity | +| `tests/test_init.py` | config builder, not the prompts | +| `tests/fixtures/*.eml` | hand-written messages, `example.org` only | + +`redact.py` is separate from `parse.py` deliberately: it is the safety +property, and a module of its own gets tests that name it rather than tests +that reach it incidentally. + +--- + +## Task 1: Package skeleton and version + +**Files:** +- Create: `abusectl/__init__.py` +- Create: `tests/__init__.py` +- Test: `tests/test_version.py` + +- [ ] **Step 1: Write the failing test** + +```python +# tests/test_version.py +import unittest + +import abusectl + + +class TestVersion(unittest.TestCase): + def test_version_is_a_dotted_string(self): + self.assertRegex(abusectl.__version__, r"^\d+\.\d+\.\d+$") + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run it and watch it fail** + +Run: `python3 -m unittest tests.test_version -v` +Expected: FAIL, `ModuleNotFoundError: No module named 'abusectl'` + +- [ ] **Step 3: Create the package** + +```python +# abusectl/__init__.py +# 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. +"""Abuse reporting for phishing mail.""" + +__version__ = "0.1.0" +``` + +```python +# tests/__init__.py +``` + +(Empty file. It makes `tests` a package so `python3 -m unittest` discovers it.) + +- [ ] **Step 4: Run it and watch it pass** + +Run: `python3 -m unittest tests.test_version -v` +Expected: PASS, 1 test + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/__init__.py tests/__init__.py tests/test_version.py +git commit -S -m "feat: package skeleton" +``` + +--- + +## Task 2: URL redaction + +The safety rule, built before anything that produces URLs so nothing can +bypass it. Keep scheme, host, path and parameter NAMES; redact parameter +VALUES. Flag a path segment that looks like an encoded identifier rather than +redacting it, since a path may be meaningful. + +**Files:** +- Create: `abusectl/redact.py` +- Test: `tests/test_redact.py` + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_redact.py +import unittest + +from abusectl import redact + + +class TestRedactUrl(unittest.TestCase): + def test_query_values_are_redacted_and_names_kept(self): + # The names fingerprint the kit; the values identify the recipient. + self.assertEqual( + redact.url("http://login.example.invalid/verify?id=abc&src=mail"), + "http://login.example.invalid/verify?id=REDACTED&src=REDACTED", + ) + + def test_a_url_with_no_query_is_unchanged(self): + self.assertEqual( + redact.url("http://login.example.invalid/verify"), + "http://login.example.invalid/verify", + ) + + def test_scheme_host_and_path_survive(self): + self.assertEqual( + redact.url("https://a.example.invalid/one/two/three?x=1"), + "https://a.example.invalid/one/two/three?x=REDACTED", + ) + + def test_a_valueless_parameter_keeps_its_shape(self): + self.assertEqual( + redact.url("http://a.example.invalid/p?flag"), + "http://a.example.invalid/p?flag=REDACTED", + ) + + def test_repeated_parameter_names_are_all_redacted(self): + self.assertEqual( + redact.url("http://a.example.invalid/p?t=1&t=2"), + "http://a.example.invalid/p?t=REDACTED&t=REDACTED", + ) + + +class TestSuspectPathSegments(unittest.TestCase): + def test_a_base64_looking_segment_is_flagged(self): + # Flagged for review, NOT redacted: a path may be meaningful. + found = redact.suspect_path_segments( + "http://a.example.invalid/verify/dGVzdEBleGFtcGxlLm9yZw/" + ) + self.assertEqual(found, ["dGVzdEBleGFtcGxlLm9yZw"]) + + def test_a_long_hex_segment_is_flagged(self): + found = redact.suspect_path_segments( + "http://a.example.invalid/c/5f4dcc3b5aa765d61d8327deb882cf99" + ) + self.assertEqual(found, ["5f4dcc3b5aa765d61d8327deb882cf99"]) + + def test_ordinary_path_words_are_not_flagged(self): + found = redact.suspect_path_segments( + "http://a.example.invalid/account/verify/now" + ) + self.assertEqual(found, []) + + def test_a_short_segment_is_not_flagged(self): + # "news" is base64-shaped and four characters. Too short to carry an + # address, and flagging it would train the user to ignore the flag. + found = redact.suspect_path_segments("http://a.example.invalid/news") + self.assertEqual(found, []) + + +class TestUrlValuedParameters(unittest.TestCase): + def test_a_redirect_target_is_recovered(self): + found = redact.url_valued_parameters( + "http://t.example.invalid/c?url=http%3A%2F%2Fevil.example.invalid%2Fp" + ) + self.assertEqual(found, ["http://evil.example.invalid/p"]) + + def test_a_tracking_token_is_not_mistaken_for_one(self): + found = redact.url_valued_parameters( + "http://t.example.invalid/c?u=dGVzdEBleGFtcGxlLm9yZw" + ) + self.assertEqual(found, []) + + def test_the_original_is_still_fully_redacted(self): + # Recovery does not loosen the rule: the redirector itself keeps every + # value blanked, including the one the target was recovered from. + raw = "http://t.example.invalid/c?url=http%3A%2F%2Fe.example.invalid%2Fp&u=tok" + self.assertEqual( + redact.url(raw), + "http://t.example.invalid/c?url=REDACTED&u=REDACTED", + ) + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run them and watch them fail** + +Run: `python3 -m unittest tests.test_redact -v` +Expected: FAIL, `ImportError: cannot import name 'redact'` + +- [ ] **Step 3: Implement** + +```python +# abusectl/redact.py +# 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. +"""Redaction of recipient identifiers hidden inside URLs. + +A phishing URL commonly carries the recipient's identity in its query +string: `?e=<address>`, `?u=<base64 of it>`, `?id=<md5 of it>`. Publishing +that to a vendor or an abuse desk is the same leak as publishing the To +header, one level down, and it deanonymises the reporter to the attacker, +since abuse desks forward reports and URLhaus is a public feed. + +Parameter NAMES are kept because they fingerprint the kit; parameter VALUES +are redacted because they identify the recipient. The token is unique per +recipient by design, so keeping it would make correlation WORSE: two +messages from one campaign would look like different URLs. + +The full URL survives in the case directory's source.eml either way. This +module decides only what may be published. +""" + +import re +from urllib.parse import parse_qsl, unquote, urlencode, urlsplit, urlunsplit + +REDACTED = "REDACTED" + +# A segment long enough to carry an encoded address, made only of characters +# base64 or hex use. 16 is above ordinary path words ("subscribe" is 9) and +# below any encoding of an email address. +_MIN_SUSPECT_LENGTH = 16 +_SUSPECT = re.compile(r"^[A-Za-z0-9+/=_-]{%d,}$" % _MIN_SUSPECT_LENGTH) + + +def url(raw: str) -> str: + """Return `raw` with every query parameter value replaced.""" + parts = urlsplit(raw) + if not parts.query: + return raw + + # keep_blank_values so `?flag` survives as a name rather than vanishing: + # its presence is part of the fingerprint. + pairs = parse_qsl(parts.query, keep_blank_values=True) + redacted = [(name, REDACTED) for name, _ in pairs] + return urlunsplit(parts._replace(query=urlencode(redacted))) + + +def url_valued_parameters(raw: str) -> list[str]: + """Parameter values that are themselves http(s) URLs. + + A redirector carries its destination in a parameter, which is exactly + what `url()` blanks. The destination is an INDICATOR rather than a + recipient identifier, so it is recovered and reported in its own right; + every other value stays redacted. + """ + parts = urlsplit(raw) + if not parts.query: + return [] + + found = [] + for _, value in parse_qsl(parts.query, keep_blank_values=True): + candidate = unquote(value).strip() + if candidate.lower().startswith(("http://", "https://")): + found.append(candidate) + return found + + +def suspect_path_segments(raw: str) -> list[str]: + """Path segments that look like an encoded identifier. + + Flagged for review, never redacted: unlike a query value, a path segment + may be the thing being reported. The user decides. + """ + parts = urlsplit(raw) + return [seg for seg in parts.path.split("/") if seg and _SUSPECT.match(seg)] +``` + +- [ ] **Step 4: Run them and watch them pass** + +Run: `python3 -m unittest tests.test_redact -v` +Expected: PASS, 12 tests + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/redact.py tests/test_redact.py +git commit -S -m "feat: redact recipient identifiers inside URLs" +``` + +--- + +## Task 3: Fixtures + +Hand-written messages. **`example.org`, `example.invalid` and the RFC 5737 +documentation IP ranges only** (`192.0.2.0/24`, `198.51.100.0/24`, +`203.0.113.0/24`). No real phishing sample goes in this repository: it would +carry the recipient identifiers this tool exists to keep out of reports, and a +repository is potentially public. + +**Files:** +- Create: `tests/fixtures/simple.eml` +- Create: `tests/fixtures/forged-chain.eml` +- Create: `tests/fixtures/with-attachment.eml` + +- [ ] **Step 1: Write `simple.eml`** + +``` +Received: from mx.example.org (mx.example.org [192.0.2.11]) + by mail.example.org (Postfix) with ESMTP id AAA11 + for <you@example.org>; Tue, 8 Sep 2026 10:15:02 +0200 (CEST) +Received: from sender.example.invalid (sender.example.invalid [203.0.113.42]) + by mx.example.org (Postfix) with ESMTP id BBB22 + for <you@example.org>; Tue, 8 Sep 2026 10:15:01 +0200 (CEST) +Authentication-Results: mx.example.org; + spf=fail smtp.mailfrom=sender.example.invalid; + dkim=none; + dmarc=fail header.from=bank.example.invalid +Return-Path: <bounce@sender.example.invalid> +From: "Your Bank" <security@bank.example.invalid> +Reply-To: <collect@drop.example.invalid> +To: <you@example.org> +Subject: Verify your account +Message-ID: <aaa111@sender.example.invalid> +Date: Tue, 8 Sep 2026 10:15:00 +0200 +MIME-Version: 1.0 +Content-Type: text/html; charset=utf-8 + +<html><body> +<p>Please <a href="http://login.bank-verify.example.invalid/verify?id=dGVzdEBleGFtcGxlLm9yZw">confirm</a>.</p> +</body></html> +``` + +**Check the weekday before committing this file.** `Qt::RFC2822Date`-style +validators reject a date whose weekday disagrees with the date, and the same +trap has already cost qtmaildir two broken fixtures. Verify with +`date -d 2026-09-08 +%A`, which must print `Tuesday`. + +- [ ] **Step 2: Write `forged-chain.eml`** + +The attacker prepends two `Received` headers of their own. Only the two +outermost, added by our own MTAs, are trustworthy. + +``` +Received: from mx.example.org (mx.example.org [192.0.2.11]) + by mail.example.org (Postfix) with ESMTP id CCC33 + for <you@example.org>; Tue, 8 Sep 2026 11:00:02 +0200 (CEST) +Received: from evil.example.invalid (evil.example.invalid [203.0.113.99]) + by mx.example.org (Postfix) with ESMTP id DDD44 + for <you@example.org>; Tue, 8 Sep 2026 11:00:01 +0200 (CEST) +Received: from innocent.example.invalid (innocent.example.invalid [198.51.100.7]) + by evil.example.invalid (Postfix) with ESMTP id EEE55; Tue, 8 Sep 2026 10:59:00 +0200 (CEST) +Received: from also-forged.example.invalid (also-forged.example.invalid [198.51.100.8]) + by innocent.example.invalid (Postfix) with ESMTP id FFF66; Tue, 8 Sep 2026 10:58:00 +0200 (CEST) +Return-Path: <bounce@evil.example.invalid> +From: "Support" <help@evil.example.invalid> +To: <you@example.org> +Subject: Action required +Message-ID: <bbb222@evil.example.invalid> +Date: Tue, 8 Sep 2026 10:58:00 +0200 +MIME-Version: 1.0 +Content-Type: text/plain; charset=utf-8 + +Visit http://evil.example.invalid/go?u=dGVzdEBleGFtcGxlLm9yZw to continue. +``` + +This fixture is the one that matters. With `192.0.2.0/24` trusted, the sending +IP is `203.0.113.99`, and `198.51.100.7` must NOT be reported as the sender: +it is an innocent third party named in a header the attacker wrote. + +- [ ] **Step 3: Write `with-attachment.eml`** + +``` +Received: from sender.example.invalid (sender.example.invalid [203.0.113.42]) + by mail.example.org (Postfix) with ESMTP id GGG77 + for <you@example.org>; Tue, 8 Sep 2026 12:00:00 +0200 (CEST) +Return-Path: <bounce@sender.example.invalid> +From: "Accounts" <billing@sender.example.invalid> +To: <you@example.org> +Subject: Invoice attached +Message-ID: <ccc333@sender.example.invalid> +Date: Tue, 8 Sep 2026 12:00:00 +0200 +MIME-Version: 1.0 +Content-Type: multipart/mixed; boundary="BOUND1" + +--BOUND1 +Content-Type: text/plain; charset=utf-8 + +See the attached invoice. + +--BOUND1 +Content-Type: application/pdf; name="invoice.pdf" +Content-Disposition: attachment; filename="invoice.pdf" +Content-Transfer-Encoding: base64 + +SGVsbG8sIHdvcmxkIQ== + +--BOUND1-- +``` + +The attachment body decodes to `Hello, world!`, whose SHA-256 is +`315f5bdb76d078c43b8ac0064e4a0164612b1fce77c869345bfc94c75894edd3`. Confirm +with: + +```bash +printf 'Hello, world!' | sha256sum +``` + +- [ ] **Step 3b: Write `redirector.eml`** + +A tracking link that carries its destination in a parameter, beside a +recipient token that must NOT survive. + +``` +Received: from sender.example.invalid (sender.example.invalid [203.0.113.42]) + by mail.example.org (Postfix) with ESMTP id HHH88 + for <you@example.org>; Tue, 8 Sep 2026 14:00:00 +0200 (CEST) +Return-Path: <bounce@sender.example.invalid> +From: "Delivery" <notice@sender.example.invalid> +To: <you@example.org> +Subject: Your parcel +Message-ID: <ddd444@sender.example.invalid> +Date: Tue, 8 Sep 2026 14:00:00 +0200 +MIME-Version: 1.0 +Content-Type: text/plain; charset=utf-8 + +Track it: http://t.example.invalid/c?url=http%3A%2F%2Fevil.example.invalid%2Fpay%3Fref%3D99&u=dGVzdEBleGFtcGxlLm9yZw + +- [ ] **Step 4: Verify the fixtures parse as MIME at all** + +Run: + +```bash +python3 -c " +from email import policy +from email.parser import BytesParser +import pathlib +for p in sorted(pathlib.Path('tests/fixtures').glob('*.eml')): + m = BytesParser(policy=policy.default).parsebytes(p.read_bytes()) + print(p.name, '->', m['subject'], '|', len(m.get_all('received') or []), 'received') +" +``` + +Expected: + +``` +forged-chain.eml -> Action required | 4 received +redirector.eml -> Your parcel | 1 received +simple.eml -> Verify your account | 2 received +with-attachment.eml -> Invoice attached | 1 received +``` + +- [ ] **Step 5: Commit** + +```bash +git add tests/fixtures/ +git commit -S -m "test: fixtures for the parser, documentation ranges only" +``` + +--- + +## Task 4: Received-chain walking and the trust boundary + +The trap this whole part exists for. `Received` headers are prepended, so the +list runs newest first, and everything below our own infrastructure is +attacker-controlled. + +**Files:** +- Create: `abusectl/parse.py` +- Test: `tests/test_parse.py` + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_parse.py +import pathlib +import unittest + +from abusectl import parse + +FIXTURES = pathlib.Path(__file__).parent / "fixtures" + + +def load(name: str) -> bytes: + return (FIXTURES / name).read_bytes() + + +class TestReceivedChain(unittest.TestCase): + def test_hops_are_returned_outermost_first(self): + hops = parse.received_hops(load("simple.eml")) + self.assertEqual([h.ip for h in hops], ["192.0.2.11", "203.0.113.42"]) + + def test_the_first_untrusted_hop_is_the_sender(self): + ip = parse.sending_ip(load("simple.eml"), trusted=["192.0.2.0/24"]) + self.assertEqual(ip, "203.0.113.42") + + def test_a_forged_chain_stops_at_the_first_untrusted_hop(self): + # The attacker prepended two hops naming an innocent third party. + # Walking past the boundary would report 198.51.100.7, which is + # someone else's address in a header the attacker wrote. + ip = parse.sending_ip(load("forged-chain.eml"), trusted=["192.0.2.0/24"]) + self.assertEqual(ip, "203.0.113.99") + + def test_hops_below_the_boundary_are_still_recorded(self): + # Recorded, but as untrusted: they may be useful and must not be + # presented as fact. + hops = parse.received_hops(load("forged-chain.eml")) + self.assertEqual( + [h.ip for h in hops], + ["192.0.2.11", "203.0.113.99", "198.51.100.7", "198.51.100.8"], + ) + + def test_no_trusted_relays_is_an_error_not_a_guess(self): + # Guessing the outermost public IP is wrong in exactly the case that + # matters, and a confident wrong answer gets a third party reported. + with self.assertRaises(parse.NoTrustBoundary): + parse.sending_ip(load("simple.eml"), trusted=[]) + + def test_a_chain_entirely_inside_the_boundary_has_no_sender(self): + ip = parse.sending_ip(load("simple.eml"), trusted=["192.0.2.0/24", "203.0.113.0/24"]) + self.assertIsNone(ip) + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run them and watch them fail** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: FAIL, `ImportError: cannot import name 'parse'` + +- [ ] **Step 3: Implement** + +```python +# abusectl/parse.py +# 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. +"""Extract indicators from a message. + +Pure and offline. This module opens no socket and reads no config: the +trusted-relay boundary arrives as an ARGUMENT, which is what lets the whole +thing be tested against fixtures with no setup. + +**Nothing here resolves or fetches anything.** Not the URLs, not the redirect +chains, not remote images. Following a link confirms the address is live to +the sender and fires exactly the tracker the message wanted. That is a safety +property, not a performance choice. +""" + +import ipaddress +import re +from dataclasses import dataclass +from email import policy +from email.parser import BytesParser + + +class NoTrustBoundary(Exception): + """Raised when no trusted relays were supplied. + + Not a warning: the outermost public IP is the usual guess and it is wrong + in exactly the case that matters, an attacker who forges extra Received + headers. Reporting the wrong IP gets an innocent party abuse-reported. + """ + + +@dataclass(frozen=True) +class Hop: + """One Received header, reduced to what can be reported.""" + + ip: str + trusted: bool = False + + +# The bracketed literal is the only part of a Received header worth trusting +# structurally: the hostnames beside it are supplied by the connecting client. +_IP_IN_BRACKETS = re.compile(r"\[([0-9a-fA-F.:]+)\]") + + +def _message(raw: bytes): + return BytesParser(policy=policy.default).parsebytes(raw) + + +def _ip_of(header: str) -> str | None: + for candidate in _IP_IN_BRACKETS.findall(header): + try: + return str(ipaddress.ip_address(candidate)) + except ValueError: + continue + return None + + +def received_hops(raw: bytes) -> list[Hop]: + """Every Received hop that names an IP, OUTERMOST FIRST. + + Received headers are prepended by each MTA, so the header list is already + newest first: our own infrastructure is at the top and the sender at the + bottom. Anything below our own hops was written by whoever was speaking to + us and can be fabricated wholesale. + """ + message = _message(raw) + hops = [] + for header in message.get_all("received") or []: + ip = _ip_of(str(header)) + if ip is not None: + hops.append(Hop(ip=ip)) + return hops + + +def _in_any(ip: str, networks: list[str]) -> bool: + address = ipaddress.ip_address(ip) + for network in networks: + if address in ipaddress.ip_network(network, strict=False): + return True + return False + + +def sending_ip(raw: bytes, trusted: list[str]) -> str | None: + """The first hop outside the trusted boundary, walking outermost inward. + + Returns None when every hop is inside the boundary, which means the + message never crossed it and there is no external sender to report. + """ + if not trusted: + raise NoTrustBoundary( + "no trusted_relays configured: run `abusectl init`" + ) + + for hop in received_hops(raw): + if not _in_any(hop.ip, trusted): + return hop.ip + return None +``` + +- [ ] **Step 4: Run them and watch them pass** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: PASS, 6 tests + +- [ ] **Step 5: Mutation check, which is the point of this task** + +Break the boundary deliberately and confirm the fixture catches it. Change +`sending_ip` to walk to the LAST untrusted hop rather than the first: + +```python + last = None + for hop in received_hops(raw): + if not _in_any(hop.ip, trusted): + last = hop.ip + return last +``` + +Run: `python3 -m unittest tests.test_parse -v` +Expected: FAIL on `test_a_forged_chain_stops_at_the_first_untrusted_hop`, +reporting `198.51.100.8` where `203.0.113.99` was expected. + +**Then put the correct implementation back** and re-run to confirm PASS. A +test that cannot fail is not protecting anything, and this is the one test in +the plan whose failure means an innocent party gets reported. + +- [ ] **Step 6: Commit** + +```bash +git add abusectl/parse.py tests/test_parse.py +git commit -S -m "feat: walk the Received chain to the trust boundary" +``` + +--- + +## Task 5: Sender domains and auth results + +**Files:** +- Modify: `abusectl/parse.py` +- Modify: `tests/test_parse.py` + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_parse.py`, above the `if __name__` block: + +```python +class TestSenderDomains(unittest.TestCase): + def test_the_three_sender_headers_are_collected(self): + domains = parse.sender_domains(load("simple.eml")) + self.assertEqual( + domains, + { + "return_path": "sender.example.invalid", + "from": "bank.example.invalid", + "reply_to": "drop.example.invalid", + }, + ) + + def test_reply_to_is_absent_when_it_matches_from(self): + # Only a DIFFERING Reply-To is an indicator; repeating From adds noise. + domains = parse.sender_domains(load("with-attachment.eml")) + self.assertNotIn("reply_to", domains) + + def test_recipient_headers_are_never_returned(self): + # The safety property, asserted rather than assumed. + domains = parse.sender_domains(load("simple.eml")) + self.assertNotIn("example.org", domains.values()) + + +class TestAuthResults(unittest.TestCase): + def test_verdicts_are_read_as_the_server_recorded_them(self): + auth = parse.auth_results(load("simple.eml")) + self.assertEqual(auth, {"spf": "fail", "dkim": "none", "dmarc": "fail"}) + + def test_a_message_with_no_auth_header_reports_nothing(self): + self.assertEqual(parse.auth_results(load("with-attachment.eml")), {}) +``` + +- [ ] **Step 2: Run them and watch them fail** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: FAIL, `AttributeError: module 'abusectl.parse' has no attribute 'sender_domains'` + +- [ ] **Step 3: Implement** + +Append to `abusectl/parse.py`: + +```python +_ADDRESS = re.compile(r"[<\s]?([^<>@\s]+)@([^<>@\s]+?)[>\s]?$") +_AUTH_VERDICT = re.compile(r"\b(spf|dkim|dmarc)=([a-z]+)", re.IGNORECASE) + + +def _domain_of(header_value: str | None) -> str | None: + if not header_value: + return None + match = _ADDRESS.search(header_value.strip()) + return match.group(2).lower() if match else None + + +def sender_domains(raw: bytes) -> dict[str, str]: + """Domains from Return-Path, From, and a DIFFERING Reply-To. + + Recipient headers are never read. The guarantee is that this module + cannot disclose an identifier it was never given, so To, Cc, + Delivered-To and X-Original-To are not consulted at all. + """ + message = _message(raw) + + domains = {} + for key, header in (("return_path", "return-path"), ("from", "from")): + domain = _domain_of(message.get(header)) + if domain: + domains[key] = domain + + reply_to = _domain_of(message.get("reply-to")) + if reply_to and reply_to != domains.get("from"): + domains["reply_to"] = reply_to + + return domains + + +def auth_results(raw: bytes) -> dict[str, str]: + """SPF, DKIM and DMARC verdicts as the RECEIVING server recorded them. + + Read, never recomputed: recomputing needs DNS, and this module resolves + nothing. The receiving server's verdict is also the honest one, since it + is what actually happened at delivery time. + """ + message = _message(raw) + header = message.get("authentication-results") + if not header: + return {} + + verdicts = {} + for mechanism, verdict in _AUTH_VERDICT.findall(str(header)): + verdicts.setdefault(mechanism.lower(), verdict.lower()) + return verdicts +``` + +- [ ] **Step 4: Run them and watch them pass** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: PASS, 11 tests + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/parse.py tests/test_parse.py +git commit -S -m "feat: extract sender domains and auth verdicts" +``` + +--- + +## Task 6: URLs and attachments + +**Files:** +- Modify: `abusectl/parse.py` +- Modify: `tests/test_parse.py` + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_parse.py`: + +```python +class TestUrls(unittest.TestCase): + def test_an_href_is_found_and_redacted(self): + urls = parse.urls(load("simple.eml")) + self.assertEqual( + urls, + ["http://login.bank-verify.example.invalid/verify?id=REDACTED"], + ) + + def test_a_plain_text_url_is_found_and_redacted(self): + urls = parse.urls(load("forged-chain.eml")) + self.assertEqual(urls, ["http://evil.example.invalid/go?u=REDACTED"]) + + def test_urls_are_deduplicated_and_ordered(self): + raw = ( + b"From: <a@b.example.invalid>\r\n" + b"Subject: t\r\n" + b"Content-Type: text/plain\r\n\r\n" + b"http://z.example.invalid/ and http://a.example.invalid/ and " + b"http://z.example.invalid/ again\r\n" + ) + self.assertEqual( + parse.urls(raw), + ["http://a.example.invalid/", "http://z.example.invalid/"], + ) + + +class TestRedirectChains(unittest.TestCase): + def test_a_declared_target_is_recovered_as_a_hop(self): + chains = parse.redirect_chains(load("redirector.eml")) + self.assertEqual(len(chains), 1) + source, target = chains[0] + self.assertTrue(source.startswith("http://t.example.invalid/c")) + self.assertTrue(target.startswith("http://evil.example.invalid/pay")) + + def test_the_recovered_target_is_itself_redacted(self): + _, target = parse.redirect_chains(load("redirector.eml"))[0] + self.assertEqual(target, "http://evil.example.invalid/pay?ref=REDACTED") + + def test_the_recipient_token_does_not_survive(self): + # The whole point: the destination is an indicator, the token is not. + chains = parse.redirect_chains(load("redirector.eml")) + self.assertNotIn("dGVzdEBleGFtcGxlLm9yZw", repr(chains)) + + def test_a_message_with_no_redirector_reports_none(self): + self.assertEqual(parse.redirect_chains(load("simple.eml")), []) + + +class TestAttachments(unittest.TestCase): + def test_filename_and_sha256_are_recorded(self): + found = parse.attachments(load("with-attachment.eml")) + self.assertEqual(len(found), 1) + self.assertEqual(found[0].filename, "invoice.pdf") + self.assertEqual( + found[0].sha256, + "315f5bdb76d078c43b8ac0064e4a0164612b1fce77c869345bfc94c75894edd3", + ) + + def test_a_message_with_no_attachment_reports_none(self): + self.assertEqual(parse.attachments(load("simple.eml")), []) +``` + +- [ ] **Step 2: Run them and watch them fail** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: FAIL, `AttributeError: module 'abusectl.parse' has no attribute 'urls'` + +- [ ] **Step 3: Implement** + +Append to `abusectl/parse.py` (and add `import hashlib` and +`from abusectl import redact` to the imports at the top of the file): + +```python +# Deliberately permissive about the tail: a URL in mail is often broken across +# lines or followed by punctuation, and over-matching is corrected by the +# redaction step, while under-matching loses the indicator entirely. +_URL = re.compile(r"https?://[^\s<>\"')]+", re.IGNORECASE) + + +@dataclass(frozen=True) +class Attachment: + filename: str + sha256: str + + +def _text_parts(message) -> list[str]: + bodies = [] + for part in message.walk(): + if part.get_content_maintype() != "text": + continue + if part.get_content_disposition() == "attachment": + continue + try: + bodies.append(part.get_content()) + except (LookupError, UnicodeDecodeError): + # An unknown charset is not a reason to lose the whole message. + payload = part.get_payload(decode=True) or b"" + bodies.append(payload.decode("utf-8", "replace")) + return bodies + + +def urls(raw: bytes) -> list[str]: + """Every http(s) URL in the text parts, redacted, sorted, deduplicated. + + NOTHING IS FETCHED. Redirect chains are read from what the message + declares, never by following a link: a request would confirm the address + is live and fire the tracker. + """ + message = _message(raw) + + found = set() + for body in _text_parts(message): + for match in _URL.findall(body): + found.add(redact.url(match.rstrip(".,;:!?"))) + return sorted(found) + + +# A redirector may point at another redirector. Bounded because the chain is +# read from the message rather than followed, so a hostile URL cannot make the +# parser walk forever, but a nested value is still attacker-supplied. +_MAX_REDIRECT_DEPTH = 5 + + +def redirect_chains(raw: bytes) -> list[tuple[str, str]]: + """Declared redirect hops, as (from, to) pairs. + + DECLARED, never followed: the destination is read out of the redirector's + own parameters. A request would confirm the address is live and fire the + tracker, which is the thing this tool exists to avoid. + + The target is reported in its own right because it is an indicator rather + than a recipient identifier; it is still redacted itself, so a token in + the destination's own query string does not survive. + """ + chains: list[tuple[str, str]] = [] + seen: set[str] = set() + + def walk(current: str, depth: int) -> None: + if depth >= _MAX_REDIRECT_DEPTH or current in seen: + return + seen.add(current) + for target in redact.url_valued_parameters(current): + safe = redact.url(target) + chains.append((redact.url(current), safe)) + walk(target, depth + 1) + + message = _message(raw) + for body in _text_parts(message): + for match in _URL.findall(body): + walk(match.rstrip(".,;:!?"), 0) + + return chains + + +def attachments(raw: bytes) -> list[Attachment]: + """Attachment filenames and SHA-256 digests. + + A hash is reportable and a filename is an indicator, so both are kept. + The filename is untrusted text: it is recorded, never used as a path. + """ + message = _message(raw) + + found = [] + for part in message.walk(): + if part.get_content_disposition() != "attachment": + continue + payload = part.get_payload(decode=True) + if payload is None: + continue + found.append( + Attachment( + filename=part.get_filename() or "", + sha256=hashlib.sha256(payload).hexdigest(), + ) + ) + return found +``` + +- [ ] **Step 4: Run them and watch them pass** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: PASS, 20 tests + +- [ ] **Step 5: Prove nothing on the network was touched** + +The never-fetch rule deserves a check that is not a reading of the code. Run +the suite with sockets disabled: + +```bash +python3 - <<'EOF' +import socket +import sys +import unittest + + +class Blocked(Exception): + pass + + +def refuse(*args, **kwargs): + raise Blocked("the parser attempted a network connection") + + +socket.socket = refuse +socket.create_connection = refuse +socket.getaddrinfo = refuse + +loader = unittest.TestLoader() +suite = loader.discover("tests", pattern="test_parse.py") +result = unittest.TextTestRunner(verbosity=2).run(suite) +sys.exit(0 if result.wasSuccessful() else 1) +EOF +``` + +Expected: every test passes. A failure naming `Blocked` means something in +the parse path resolves or fetches, which is a safety defect rather than a +bug. + +- [ ] **Step 6: Commit** + +```bash +git add abusectl/parse.py tests/test_parse.py +git commit -S -m "feat: extract URLs and attachment hashes, fetching nothing" +``` + +--- + +## Task 7: The case directory + +**Files:** +- Create: `abusectl/case.py` +- Test: `tests/test_case.py` + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_case.py +import json +import pathlib +import tempfile +import unittest + +from abusectl import case + + +class TestCaseCreation(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = pathlib.Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_a_case_holds_the_source_and_a_manifest(self): + created = case.create(self.root, b"From: <a@b.example.invalid>\r\n\r\nhi") + self.assertTrue((created.path / "source.eml").is_file()) + self.assertTrue((created.path / "manifest.json").is_file()) + + def test_the_source_is_stored_byte_for_byte(self): + raw = b"From: <a@b.example.invalid>\r\n\r\nhi\r\n" + created = case.create(self.root, raw) + self.assertEqual((created.path / "source.eml").read_bytes(), raw) + + def test_the_manifest_carries_a_format_version(self): + created = case.create(self.root, b"x") + manifest = json.loads((created.path / "manifest.json").read_text()) + self.assertEqual(manifest["format"], case.FORMAT_VERSION) + + def test_two_cases_do_not_collide(self): + a = case.create(self.root, b"one") + b = case.create(self.root, b"two") + self.assertNotEqual(a.path, b.path) + + def test_a_case_id_is_filesystem_safe(self): + created = case.create(self.root, b"x") + self.assertRegex(created.path.name, r"^\d{4}-\d{2}-\d{2}-[0-9a-f]{4}$") + + def test_a_manifest_round_trips(self): + created = case.create(self.root, b"x") + manifest = case.load(created.path) + manifest["iocs"] = [{"id": "ioc-1", "type": "ipv4", "value": "203.0.113.1"}] + case.save(created.path, manifest) + self.assertEqual(case.load(created.path)["iocs"][0]["value"], "203.0.113.1") + + def test_saving_leaves_no_temporary_file_behind(self): + # The manifest is written atomically, temp file plus rename, because a + # half-written manifest during a review is a corrupted evidence record. + created = case.create(self.root, b"x") + case.save(created.path, case.load(created.path)) + leftovers = [p.name for p in created.path.iterdir() if p.suffix == ".tmp"] + self.assertEqual(leftovers, []) + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run them and watch them fail** + +Run: `python3 -m unittest tests.test_case -v` +Expected: FAIL, `ImportError: cannot import name 'case'` + +- [ ] **Step 3: Implement** + +```python +# abusectl/case.py +# 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 case directory: the state every subcommand reads and writes. + +State lives on disk rather than in memory so a review can take a week and +survive a reboot. This module is the ONLY writer of a case directory. + +The manifest is written atomically, temp file plus rename, because a +half-written manifest during a review is a corrupted evidence record. +Nothing here deletes a case: they are the user's evidence. + +source.eml holds the message UNREDACTED. The redaction rule is about what +may be published, not about what is kept locally, so a case directory is +sensitive at rest and the submit path must never attach source.eml wholesale. +""" + +import json +import os +import secrets +import tempfile +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path + +FORMAT_VERSION = 1 + +MANIFEST = "manifest.json" +SOURCE = "source.eml" +BODIES = "bodies" + + +@dataclass(frozen=True) +class Case: + path: Path + case_id: str + + +def _now() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _new_id() -> str: + # Date for browsing, four random hex for collision resistance. Not a hash + # of the message: two reports of the same campaign are separate cases. + return f"{datetime.now(timezone.utc):%Y-%m-%d}-{secrets.token_hex(2)}" + + +def create(root: Path, raw: bytes) -> Case: + """Create a case under `root` holding `raw`, and return it.""" + root = Path(root) + root.mkdir(parents=True, exist_ok=True) + + while True: + case_id = _new_id() + path = root / case_id + try: + path.mkdir() + break + except FileExistsError: + continue + + (path / BODIES).mkdir() + (path / SOURCE).write_bytes(raw) + + save( + path, + { + "format": FORMAT_VERSION, + "case_id": case_id, + "created": _now(), + "source": SOURCE, + "iocs": [], + "auth": {}, + "contacts": [], + "destinations": [], + }, + ) + return Case(path=path, case_id=case_id) + + +def load(path: Path) -> dict: + """Read a case's manifest, refusing a format this build does not know.""" + manifest = json.loads((Path(path) / MANIFEST).read_text(encoding="utf-8")) + + # Refusing an unknown format is correct rather than cautious: a newer + # writer may mean fields this build would silently drop on the next save. + found = manifest.get("format") + if found != FORMAT_VERSION: + raise ValueError( + f"manifest format {found} is not supported (expected {FORMAT_VERSION})" + ) + return manifest + + +def save(path: Path, manifest: dict) -> None: + """Write a case's manifest atomically.""" + path = Path(path) + target = path / MANIFEST + + handle, temporary = tempfile.mkstemp(dir=path, prefix=".manifest-", suffix=".tmp") + try: + with os.fdopen(handle, "w", encoding="utf-8") as stream: + json.dump(manifest, stream, indent=2, sort_keys=False) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, target) + except BaseException: + # A failed write must not leave a stray temp file in an evidence + # directory, and must not have touched the existing manifest. + try: + os.unlink(temporary) + except FileNotFoundError: + pass + raise +``` + +- [ ] **Step 4: Run them and watch them pass** + +Run: `python3 -m unittest tests.test_case -v` +Expected: PASS, 7 tests + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/case.py tests/test_case.py +git commit -S -m "feat: case directory with an atomically written manifest" +``` + +--- + +## Task 8: Assemble the IOC list + +Joins the extractors to the manifest schema, giving each IOC an `id`, an +`origin` and, for a hop, a `confidence`. + +**Files:** +- Modify: `abusectl/parse.py` +- Modify: `tests/test_parse.py` + +- [ ] **Step 1: Write the failing tests** + +Append to `tests/test_parse.py`: + +```python +class TestIocAssembly(unittest.TestCase): + def test_every_ioc_has_a_unique_id_and_an_origin(self): + iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"]) + ids = [i["id"] for i in iocs] + self.assertEqual(len(ids), len(set(ids))) + self.assertTrue(all(i["origin"] for i in iocs)) + + def test_the_sending_ip_is_present_and_marked_trusted_hop(self): + iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"]) + ips = [i for i in iocs if i["type"] == "ipv4"] + self.assertEqual(ips[0]["value"], "203.0.113.42") + self.assertEqual(ips[0]["confidence"], "boundary-hop") + + def test_hops_below_the_boundary_are_marked_untrusted(self): + iocs = parse.iocs(load("forged-chain.eml"), trusted=["192.0.2.0/24"]) + ips = {i["value"]: i for i in iocs if i["type"] == "ipv4"} + self.assertEqual(ips["203.0.113.99"]["confidence"], "boundary-hop") + self.assertEqual(ips["198.51.100.7"]["confidence"], "untrusted-hop") + + def test_urls_carry_their_redacted_form(self): + iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"]) + urls = [i for i in iocs if i["type"] == "url"] + self.assertEqual(len(urls), 1) + self.assertIn("REDACTED", urls[0]["value"]) + + def test_a_suspect_path_segment_is_flagged_on_the_ioc(self): + iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"]) + url = next(i for i in iocs if i["type"] == "url") + self.assertEqual(url["suspect_path_segments"], ["dGVzdEBleGFtcGxlLm9yZw"]) + + def test_no_ioc_holds_a_recipient_address(self): + # The safety property, asserted over the whole output. + for name in ("simple.eml", "forged-chain.eml", "with-attachment.eml", + "redirector.eml"): + iocs = parse.iocs(load(name), trusted=["192.0.2.0/24"]) + blob = repr(iocs) + self.assertNotIn("you@example.org", blob) + self.assertNotIn("example.org", blob) +``` + +- [ ] **Step 2: Run them and watch them fail** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: FAIL, `AttributeError: module 'abusectl.parse' has no attribute 'iocs'` + +- [ ] **Step 3: Implement** + +Append to `abusectl/parse.py`: + +```python +def iocs(raw: bytes, trusted: list[str]) -> list[dict]: + """Every indicator, in the manifest's own shape. + + Each carries an `id` that contacts and destinations reference, so a value + is corrected in one place, and an `origin` saying where it came from, + because during review the user needs to know whether an IP came from a + header to trust or one the attacker wrote. + """ + sender = sending_ip(raw, trusted) # raises NoTrustBoundary if unset + + found = [] + + def add(**fields): + fields["id"] = f"ioc-{len(found) + 1}" + found.append(fields) + + for hop in received_hops(raw): + if _in_any(hop.ip, trusted): + continue + add( + type="ipv6" if ":" in hop.ip else "ipv4", + value=hop.ip, + origin="received-chain", + # The boundary hop is the one we can stand behind. Everything + # below it was written by whoever was speaking to our MTA. + confidence="boundary-hop" if hop.ip == sender else "untrusted-hop", + ) + + for key, domain in sender_domains(raw).items(): + add(type="domain", value=domain, origin=f"header-{key}") + + for url_value in urls(raw): + entry = { + "type": "url", + "value": url_value, + "origin": "body", + } + suspects = redact.suspect_path_segments(url_value) + if suspects: + # Flagged, not redacted: a path segment may be the thing being + # reported, so the user decides during review. + entry["suspect_path_segments"] = suspects + add(**entry) + + for hop_from, hop_to in redirect_chains(raw): + add( + type="url", + value=hop_to, + origin="redirect-target", + redirect_from=hop_from, + ) + + for attachment in attachments(raw): + add( + type="sha256", + value=attachment.sha256, + origin="attachment", + filename=attachment.filename, + ) + + return found +``` + +- [ ] **Step 4: Run them and watch them pass** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: PASS, 26 tests + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/parse.py tests/test_parse.py +git commit -S -m "feat: assemble IOCs in the manifest's shape" +``` + +--- + +## Task 9: Config reading + +**Files:** +- Create: `abusectl/config.py` +- Test: `tests/test_config.py` + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_config.py +import pathlib +import tempfile +import unittest + +from abusectl import config + + +class TestConfig(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = pathlib.Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _write(self, text: str) -> pathlib.Path: + path = self.root / "config.toml" + path.write_text(text, encoding="utf-8") + return path + + def test_trusted_relays_and_cases_are_read(self): + path = self._write( + '[general]\n' + 'cases = "~/cases"\n' + 'trusted_relays = ["192.0.2.0/24"]\n' + ) + loaded = config.load(path) + self.assertEqual(loaded.trusted_relays, ["192.0.2.0/24"]) + self.assertEqual(loaded.cases, pathlib.Path.home() / "cases") + + def test_a_missing_file_is_reported_as_not_configured(self): + with self.assertRaises(config.NotConfigured): + config.load(self.root / "absent.toml") + + def test_an_empty_relay_list_is_not_configured(self): + # Present but empty is the same as absent: parse must refuse either + # way rather than guess, so they are one error. + path = self._write('[general]\ntrusted_relays = []\n') + with self.assertRaises(config.NotConfigured): + config.load(path) + + def test_a_malformed_cidr_is_rejected_at_load(self): + path = self._write('[general]\ntrusted_relays = ["not-a-network"]\n') + with self.assertRaises(ValueError): + config.load(path) + + def test_the_cases_path_has_a_default(self): + path = self._write('[general]\ntrusted_relays = ["192.0.2.0/24"]\n') + self.assertEqual(config.load(path).cases, config.DEFAULT_CASES) + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run them and watch them fail** + +Run: `python3 -m unittest tests.test_config -v` +Expected: FAIL, `ImportError: cannot import name 'config'` + +- [ ] **Step 3: Implement** + +```python +# abusectl/config.py +# 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. +"""Reading ~/.config/abusectl/config.toml. + +stdlib tomllib, no dependency. A key that was skipped at setup is ABSENT +rather than empty: `api_key = ""` reads as configured-and-broken and produces +a confusing auth error much later, where an absent key reads as +not-configured and the part that wants it can say so plainly. +""" + +import ipaddress +import os +import tomllib +from dataclasses import dataclass +from pathlib import Path + +DEFAULT_CASES = Path( + os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share") +) / "abusectl" + + +class NotConfigured(Exception): + """Raised when the config is missing or names no trusted relays.""" + + +def path() -> Path: + """Where the config lives.""" + root = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) + return root / "abusectl" / "config.toml" + + +@dataclass(frozen=True) +class Config: + trusted_relays: list[str] + cases: Path + + +def load(from_path: Path | None = None) -> Config: + """Read and validate the config.""" + source = Path(from_path) if from_path else path() + + try: + raw = source.read_bytes() + except FileNotFoundError as error: + raise NotConfigured( + f"no config at {source}: run `abusectl init`" + ) from error + + general = tomllib.loads(raw.decode("utf-8")).get("general", {}) + + relays = general.get("trusted_relays") or [] + if not relays: + raise NotConfigured( + f"no trusted_relays in {source}: run `abusectl init`" + ) + + # Validated here rather than at parse time so a typo is reported against + # the file that holds it. + for relay in relays: + ipaddress.ip_network(relay, strict=False) + + cases = general.get("cases") + return Config( + trusted_relays=list(relays), + cases=Path(cases).expanduser() if cases else DEFAULT_CASES, + ) +``` + +- [ ] **Step 4: Run them and watch them pass** + +Run: `python3 -m unittest tests.test_config -v` +Expected: PASS, 5 tests + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/config.py tests/test_config.py +git commit -S -m "feat: read and validate the config" +``` + +--- + +## Task 10: `init`, the pure builder + +The builder only. The prompts come in Task 11 and are hand-tested. + +**Files:** +- Create: `abusectl/init.py` +- Test: `tests/test_init.py` + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_init.py +import pathlib +import tempfile +import tomllib +import unittest + +from abusectl import init + + +class TestBuildConfig(unittest.TestCase): + def test_the_answers_become_readable_toml(self): + text = init.build({"trusted_relays": ["192.0.2.0/24"], "cases": "~/c"}) + parsed = tomllib.loads(text) + self.assertEqual(parsed["general"]["trusted_relays"], ["192.0.2.0/24"]) + self.assertEqual(parsed["general"]["cases"], "~/c") + + def test_a_skipped_answer_is_absent_not_empty(self): + # An empty string reads as configured-and-broken later on. + text = init.build({"trusted_relays": ["192.0.2.0/24"], "cases": ""}) + self.assertNotIn("cases", tomllib.loads(text)["general"]) + + def test_a_malformed_relay_is_rejected(self): + with self.assertRaises(ValueError): + init.build({"trusted_relays": ["nonsense"]}) + + def test_no_relays_at_all_is_rejected(self): + with self.assertRaises(ValueError): + init.build({"trusted_relays": []}) + + def test_the_result_loads_back_through_config(self): + from abusectl import config + + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text( + init.build({"trusted_relays": ["192.0.2.0/24"]}), encoding="utf-8" + ) + self.assertEqual(config.load(path).trusted_relays, ["192.0.2.0/24"]) + + +class TestProviderTable(unittest.TestCase): + def test_a_known_provider_resolves_to_ranges(self): + self.assertTrue(init.provider_relays("gmail")) + + def test_lookup_is_case_insensitive(self): + self.assertEqual(init.provider_relays("Gmail"), init.provider_relays("gmail")) + + def test_an_unknown_provider_returns_nothing(self): + self.assertEqual(init.provider_relays("nosuchprovider"), []) + + def test_every_shipped_range_is_a_valid_network(self): + import ipaddress + + for name, ranges in init.PROVIDERS.items(): + for entry in ranges: + ipaddress.ip_network(entry, strict=False) + + +class TestSampleChain(unittest.TestCase): + def test_hops_are_offered_for_picking(self): + raw = ( + pathlib.Path(__file__).parent / "fixtures" / "simple.eml" + ).read_bytes() + hops = init.hops_from_sample(raw) + self.assertEqual(hops, ["192.0.2.11", "203.0.113.42"]) + + +class TestWriteGuard(unittest.TestCase): + def test_writing_over_an_existing_config_refuses_without_force(self): + # The refusal is what makes the interactive confirm meaningful: the + # caller has to have decided something. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text("[general]\n", encoding="utf-8") + with self.assertRaises(FileExistsError): + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}) + + def test_force_overwrites_and_leaves_a_backup(self): + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text('[general]\ntrusted_relays = ["10.0.0.0/8"]\n', + encoding="utf-8") + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}, force=True) + + self.assertIn("192.0.2.0/24", path.read_text()) + backups = list(pathlib.Path(tmp).glob("config.toml.bak-*")) + self.assertEqual(len(backups), 1) + self.assertIn("10.0.0.0/8", backups[0].read_text()) + + def test_a_backup_is_not_world_readable_either(self): + # It holds the same secrets the config does. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text("[general]\n", encoding="utf-8") + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}, force=True) + backup = next(pathlib.Path(tmp).glob("config.toml.bak-*")) + self.assertEqual(backup.stat().st_mode & 0o077, 0) + + def test_an_unsupplied_section_survives_a_rewrite(self): + # Once the config holds a MISP key, an init that only sets the relays + # must not silently discard it. The backup makes that recoverable; + # not losing it is better. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text( + '[general]\ntrusted_relays = ["10.0.0.0/8"]\n\n' + '[misp]\nurl = "https://misp.example.invalid"\n' + 'api_key = "kept"\n', + encoding="utf-8", + ) + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}, force=True) + + rewritten = path.read_text() + self.assertIn("192.0.2.0/24", rewritten) + self.assertIn("[misp]", rewritten) + self.assertIn("kept", rewritten) + + def test_a_written_config_is_not_world_readable(self): + # It will hold API keys as later parts land. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}) + self.assertEqual(path.stat().st_mode & 0o077, 0) + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run them and watch them fail** + +Run: `python3 -m unittest tests.test_init -v` +Expected: FAIL, `ImportError: cannot import name 'init'` + +- [ ] **Step 3: Implement** + +```python +# abusectl/init.py +# 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. +"""First-run configuration. + +A PURE BUILDER plus a thin prompt shell. `build()` takes the answers as a +mapping and returns TOML text; the interactive prompts and the command-line +flags are two front ends over it, so the writing logic is testable with no +terminal and the two routes cannot drift. + +`--non-interactive` exists so an agent can run setup: every question is also +a flag, and there is no answer reachable only by typing. +""" + +import ipaddress +import os +import tomllib +from datetime import datetime +from pathlib import Path + +from abusectl import parse + +# Sending ranges the providers publish themselves. Static rather than read +# from SPF at runtime: SPF is a DNS lookup, and while the never-resolve rule +# is about parsing hostile mail rather than setup, a static table keeps the +# boundary unambiguous. Verify against the provider's own documentation +# before adding an entry. +PROVIDERS: dict[str, list[str]] = { + "gmail": ["35.190.247.0/24", "64.233.160.0/19", "66.102.0.0/20", + "66.249.80.0/20", "72.14.192.0/18", "74.125.0.0/16", + "108.177.8.0/21", "173.194.0.0/16", "209.85.128.0/17", + "216.58.192.0/19", "216.239.32.0/19"], + "fastmail": ["66.111.4.0/24", "103.168.172.0/22"], + "proton": ["185.70.40.0/22", "51.89.119.103/32"], +} + + +def provider_relays(name: str) -> list[str]: + """The published sending ranges for a known provider, or an empty list.""" + return list(PROVIDERS.get(name.strip().lower(), [])) + + +def hops_from_sample(raw: bytes) -> list[str]: + """The Received chain of a known-good message, outermost first. + + Setup shows this list and asks which hops are the user's own, which turns + an abstract question into picking from a real one. + """ + return [hop.ip for hop in parse.received_hops(raw)] + + +def build(answers: dict) -> str: + """Render the answers as config TOML. + + A skipped answer is OMITTED rather than written empty, so a later part + can tell "not configured" from "configured to nothing". + """ + relays = [r.strip() for r in answers.get("trusted_relays") or [] if r.strip()] + if not relays: + raise ValueError("at least one trusted relay is required") + for relay in relays: + ipaddress.ip_network(relay, strict=False) + + lines = [ + "# abusectl configuration.", + "# Written by `abusectl init`. Safe to edit by hand.", + "", + "[general]", + "", + "# The hops your own mail infrastructure adds. Everything below the", + "# outermost of these was written by whoever was speaking to your MTA", + "# and can be forged, so this boundary decides which IP gets reported.", + "trusted_relays = [", + ] + lines += [f' "{relay}",' for relay in relays] + lines.append("]") + + cases = (answers.get("cases") or "").strip() + if cases: + lines += ["", "# Where case directories are written. Nothing deletes them.", + f'cases = "{cases}"'] + + lines += [ + "", + "# Later parts add their own sections here as they are built:", + "# [misp] url and api_key, written by `abusectl init` once", + "# submit exists", + "# [vendors] abusedb, urlhaus, virustotal keys", + "# [reporting] the identity X-ARF reports are sent under", + "", + ] + return "\n".join(lines) + + +def existing_summary(path: Path) -> str: + """A short description of a config already in place, for the confirm. + + Values are NOT shown: the file holds API keys as later parts land, and + echoing a secret to the terminal to ask about overwriting it is a poor + trade. Section and key names are enough to recognise what would be lost. + """ + try: + parsed = tomllib.loads(Path(path).read_bytes().decode("utf-8")) + except (OSError, tomllib.TOMLDecodeError): + return "unreadable" + + parts = [] + for section, values in parsed.items(): + if isinstance(values, dict): + parts.append(f"[{section}]: {', '.join(sorted(values))}") + return "; ".join(parts) or "empty" + + +def _preserved_sections(path: Path) -> str: + """Sections of an existing config that `build()` does not write. + + An init run that sets only the relays must not silently drop a MISP key + set by an earlier one. Carried across verbatim rather than re-rendered, + so a comment or a field this build does not understand also survives. + """ + try: + text = Path(path).read_bytes().decode("utf-8") + except OSError: + return "" + + kept: list[str] = [] + keeping = False + for line in text.splitlines(): + stripped = line.strip() + if stripped.startswith("["): + # [general] is rewritten from the answers; everything else is + # somebody else's section and is preserved untouched. + keeping = stripped != "[general]" + if keeping: + kept.append(line) + continue + if keeping: + kept.append(line) + + return "\n".join(kept).strip() + + +def back_up(path: Path) -> Path | None: + """Copy an existing config aside, returning where it went. + + Timestamped rather than a single .bak, so a second mistake does not + overwrite the recovery from the first. + """ + path = Path(path) + if not path.exists(): + return None + + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + backup = path.with_name(f"{path.name}.bak-{stamp}") + + # Same mode as the config: the backup holds the same secrets. + handle = os.open(backup, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(handle, "wb") as stream: + stream.write(path.read_bytes()) + return backup + + +def write(path: Path, answers: dict, force: bool = False) -> Path: + """Write the config, backing up whatever was there. + + Refuses an existing file unless `force`, which is what makes the + interactive confirmation meaningful: the caller has to have decided. + The backup happens either way, so a wrong answer is recoverable rather + than needing to have been foreseen. + """ + path = Path(path) + if path.exists() and not force: + raise FileExistsError( + f"{path} already exists: pass --force to overwrite it" + ) + + preserved = _preserved_sections(path) if path.exists() else "" + text = build(answers) + if preserved: + text = f"{text}\n{preserved}\n" + + path.parent.mkdir(parents=True, exist_ok=True) + back_up(path) + + # 0600 before anything is written: the file holds API keys as later parts + # land, and a world-readable moment is a world-readable moment. + handle = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(handle, "w", encoding="utf-8") as stream: + stream.write(text) + os.chmod(path, 0o600) + return path +``` + +- [ ] **Step 4: Run them and watch them pass** + +Run: `python3 -m unittest tests.test_init -v` +Expected: PASS, 15 tests + +- [ ] **Step 5: Verify the shipped provider ranges before trusting them** + +The table is static data and can go stale. Confirm each provider's ranges +against its own published documentation, and record the date checked in a +comment above `PROVIDERS`. A wrong range here means a hop is treated as the +user's own and the real sender is never reported. + +- [ ] **Step 6: Commit** + +```bash +git add abusectl/init.py tests/test_init.py +git commit -S -m "feat: first-run config builder" +``` + +--- + +## Task 11: The CLI + +**Files:** +- Create: `abusectl/cli.py` +- Create: `abusectl/__main__.py` + +- [ ] **Step 1: Write the CLI** + +```python +# abusectl/cli.py +# 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. +"""Command-line entry point. Dispatch and exit codes, no logic of its own.""" + +import argparse +import sys +from pathlib import Path + +from abusectl import __version__, case, config, init, parse + +EXIT_OK = 0 +EXIT_ERROR = 1 +EXIT_NOT_CONFIGURED = 3 + + +def _prompt_answers(sample: Path | None) -> dict: + """The interactive front end. Hand-tested, not unit-tested.""" + print("abusectl setup\n") + + relays: list[str] = [] + + if sample: + print(f"Received chain of {sample}, outermost first:\n") + hops = init.hops_from_sample(sample.read_bytes()) + for number, ip in enumerate(hops, start=1): + print(f" {number} {ip}") + picked = input("\nWhich of these are yours? (e.g. 1,2) > ").strip() + for index in picked.replace(" ", "").split(","): + if index.isdigit() and 1 <= int(index) <= len(hops): + relays.append(f"{hops[int(index) - 1]}/32") + + while not relays: + typed = input( + "Trusted relays as CIDR, comma separated.\n" + " Leave empty to name a provider instead.\n> " + ).strip() + if typed: + relays = [r.strip() for r in typed.split(",") if r.strip()] + break + + provider = input(f"Provider ({', '.join(sorted(init.PROVIDERS))}) > ").strip() + relays = init.provider_relays(provider) + if not relays: + print(f" unknown provider {provider!r}\n") + + cases = input(f"\nCase directory [{config.DEFAULT_CASES}] > ").strip() + return {"trusted_relays": relays, "cases": cases} + + +def _cmd_init(args: argparse.Namespace) -> int: + target = args.config or config.path() + force = args.force + + if args.non_interactive: + if not args.trusted_relays: + print( + "--trusted-relays is required with --non-interactive", + file=sys.stderr, + ) + return EXIT_ERROR + answers = { + "trusted_relays": args.trusted_relays, + "cases": args.cases or "", + } + else: + # The soft route: show what is there and ask, rather than making the + # user re-run with --force just to change one answer. A backup is + # written either way, so saying yes here is recoverable. + if target.exists() and not force: + print(f"A configuration already exists at {target}") + print(f" it holds: {init.existing_summary(target)}") + print(" sections this run does not set are kept, and the current") + print(" file is backed up beside it before anything is written.") + if input("\nSet it up again? [y/N] > ").strip().lower() not in ("y", "yes"): + print("left unchanged") + return EXIT_OK + force = True + + answers = _prompt_answers(args.from_sample) + + try: + written = init.write(target, answers, force=force) + except (FileExistsError, ValueError) as error: + print(str(error), file=sys.stderr) + return EXIT_ERROR + + print(f"\nwrote {written}") + return EXIT_OK + + +def _cmd_parse(args: argparse.Namespace) -> int: + try: + settings = config.load(args.config) + except config.NotConfigured as error: + print(str(error), file=sys.stderr) + return EXIT_NOT_CONFIGURED + + raw = args.message.read_bytes() + created = case.create(settings.cases, raw) + + manifest = case.load(created.path) + manifest["iocs"] = parse.iocs(raw, trusted=settings.trusted_relays) + manifest["auth"] = parse.auth_results(raw) + case.save(created.path, manifest) + + print(created.path) + return EXIT_OK + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="abusectl") + parser.add_argument("--version", action="version", version=__version__) + parser.add_argument("--config", type=Path, help="use this config file") + sub = parser.add_subparsers(dest="command", required=True) + + setup = sub.add_parser("init", help="write the first-run configuration") + setup.add_argument("--trusted-relays", nargs="*", metavar="CIDR") + setup.add_argument("--cases", metavar="DIR") + setup.add_argument("--from-sample", type=Path, metavar="EML", + help="pick your hops from a known-good message") + setup.add_argument("--non-interactive", action="store_true", + help="take every answer from flags, never prompt") + setup.add_argument("--force", action="store_true", + help="overwrite an existing config") + setup.set_defaults(func=_cmd_init) + + reader = sub.add_parser("parse", help="extract indicators from a message") + reader.add_argument("message", type=Path) + reader.set_defaults(func=_cmd_parse) + + args = parser.parse_args(argv) + try: + return args.func(args) + except parse.NoTrustBoundary as error: + print(str(error), file=sys.stderr) + return EXIT_NOT_CONFIGURED + + +if __name__ == "__main__": + raise SystemExit(main()) +``` + +```python +# abusectl/__main__.py +# 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. +"""`python3 -m abusectl`.""" + +from abusectl.cli import main + +raise SystemExit(main()) +``` + +- [ ] **Step 2: Check the whole suite still passes** + +Run: `python3 -m unittest discover tests -v` +Expected: PASS, 66 tests + +- [ ] **Step 3: Drive it end to end, non-interactively** + +```bash +TMP=$(mktemp -d) +python3 -m abusectl --config "$TMP/config.toml" init \ + --non-interactive --trusted-relays 192.0.2.0/24 --cases "$TMP/cases" +CASE=$(python3 -m abusectl --config "$TMP/config.toml" parse tests/fixtures/forged-chain.eml) +cat "$CASE/manifest.json" +``` + +Expected: the manifest lists `203.0.113.99` with `"confidence": "boundary-hop"`, +`198.51.100.7` and `198.51.100.8` with `"untrusted-hop"`, one URL with +`u=REDACTED`, and **no occurrence of `you@example.org`**. Confirm the last +with: + +```bash +grep -c "example.org" "$CASE/manifest.json" || echo "clean: no recipient data" +``` + +- [ ] **Step 4: Check the not-configured path** + +```bash +python3 -m abusectl --config "$TMP/absent.toml" parse tests/fixtures/simple.eml +echo "exit: $?" +``` + +Expected: `no config at ...: run `abusectl init`` on stderr, exit 3. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/cli.py abusectl/__main__.py +git commit -S -m "feat: command line for init and parse" +``` + +--- + +## Task 12: Hand test the interactive setup + +**Not automated, deliberately.** Whether a question reads clearly has no +assertion, and a test driving stdin would assert the wording it was written +against and break on a rewording that improved it. The builder underneath is +already covered by Task 10. + +**Run each of these and report what happens.** The wrong answers matter more +than the right ones: that is where prompts actually fail. + +- [ ] **1. The ordinary path.** `python3 -m abusectl --config /tmp/t1.toml init`, + answer with real CIDRs. Does the question make it clear these are YOUR + relays rather than the sender's? + +- [ ] **2. A malformed CIDR.** Type `192.0.2.0/99`. Where does it fail, and does + the failure lose the answers already given? + +- [ ] **3. The provider route.** Press Enter at the CIDR prompt, then type + `gmail`. Then try again with `Gmail` and with `gmial`. + +- [ ] **4. The sample route.** `init --from-sample tests/fixtures/simple.eml`. + Is it clear which hops to pick? Try picking none, and picking `9`. + +- [ ] **5. Interrupt it.** Ctrl+C halfway. Is a half-written config left behind? + There must not be one. + +- [ ] **6. The default.** Press Enter at the case directory prompt. Is the path + that gets written the one that was shown? + +- [ ] **7. Re-run over an existing config.** It must show what is already + there and ask. Answer `n`: nothing changes, and no backup is written. + Does the summary make it clear what would be replaced? + +- [ ] **8. Re-run and answer `y`.** It goes through setup again. Check a + `config.toml.bak-<timestamp>` appeared beside it, holding the previous + contents, and that `ls -l` shows `-rw-------` on both. + +- [ ] **8b. Re-run over a config with a hand-added section.** Add + `[misp]\nurl = "https://misp.example.invalid"` to the file, run `init` + again, answer `y`, and confirm the `[misp]` section is still there + afterwards. This is the one that matters once real keys are in the file. + +- [ ] **8c. `--force` non-interactively.** + `init --non-interactive --trusted-relays 192.0.2.0/24 --force` over an + existing config: no prompt, backup written, other sections kept. + +- [ ] **9. Agent route.** `init --non-interactive` with no `--trusted-relays` + must fail with a usable message rather than prompting. + +Report back with anything that reads badly, and I will change the prompts. +Nothing in this task gets a test. + +--- + +## Task 13: README and a `retry`/`contacts` note + +**Files:** +- Modify: `README.md` + +- [ ] **Step 1: Mark what exists** + +Replace the "Status: early" line with: + +```markdown +**Status: `init` and `parse` are built.** The rest of the pipeline is +designed but not written; see `docs/specs/2026-09-08-abusectl-design.md`. +``` + +- [ ] **Step 2: Add a usage section after "What it does"** + +```markdown +## Getting started + +```bash +abusectl init # asks, writes ~/.config/abusectl/config.toml +abusectl parse message.eml # prints the case directory it created +``` + +`init` needs to know which `Received` hops your own mail infrastructure adds, +because everything below that boundary was written by whoever was talking to +your server and can be forged. It offers three ways to answer: type the CIDRs, +name a known provider, or point it at a message you know arrived legitimately +with `--from-sample` and pick your hops from the real chain. + +`parse` refuses to run until that boundary is set. Guessing it wrong means +reporting an innocent third party, so it does not guess. + +For scripted or agent-driven setup, every question is also a flag: + +```bash +abusectl init --non-interactive --trusted-relays 192.0.2.0/24 198.51.100.0/24 +``` +``` + +- [ ] **Step 3: Commit** + +```bash +git add README.md +git commit -S -m "docs: README covers init and parse" +``` + +--- + +## Done when + +- `python3 -m unittest discover tests` passes, 66 tests +- The socket-blocked run in Task 6 Step 5 passes, proving nothing resolves +- The forged-chain fixture reports `203.0.113.99`, never `198.51.100.7` +- No manifest produced from any fixture contains `example.org` +- The Task 4 mutation check was run and seen to fail before being reverted +- Task 12 was hand-tested and its findings reported + +## Not in this plan + +`contacts`, `report`, `submit`, `retry` and the qtmaildir dialog. Each gets +its own spec first, per the umbrella design: `contacts` has real unknowns +(RDAP referral chasing, caching, netblocks that publish no abuse contact) and +`submit` more so. diff --git a/docs/specs/2026-09-08-abusectl-design.md b/docs/specs/2026-09-08-abusectl-design.md index ed984dd..6b2e559 100644 --- a/docs/specs/2026-09-08-abusectl-design.md +++ b/docs/specs/2026-09-08-abusectl-design.md @@ -56,6 +56,7 @@ One tool, subcommands, one repository. Each subcommand reads and writes a therefore take a week and survive a reboot. ``` +abusectl init -> config.toml first run, then exits abusectl parse msg.eml -> case dir, IOCs offline, pure abusectl contacts <case> -> + abuse contacts network, read-only abusectl report <case> -> + report bodies offline, pure @@ -175,11 +176,24 @@ The full URL survives in `source.eml` either way, so the evidence exists locally; it is simply not what gets published by default, and the review gate allows pasting one in by hand when a particular desk genuinely needs it. +**One exception, and it is the rule's own logic rather than a hole in it.** A +redirector carries its DESTINATION in a parameter, which the rule above would +blank. That destination is an indicator rather than a recipient identifier, so +a parameter value that parses as an http(s) URL is recovered and reported in +its own right, redacted itself, and the hop is recorded as a chain. Every +value that is not a URL stays blanked, including a tracking token sitting in +the same query string. The recovery is bounded in depth, because a redirector +may point at another one and the nested value is attacker-supplied. + +Reading a destination out of a parameter is not fetching it. The chain is what +the message DECLARES; nothing is followed. + ## Components ``` abusectl/ cli.py argparse dispatch, exit codes. No logic. + init.py first-run config: pure builder + prompt shell case.py case dir: create, load, save manifest, atomic writes parse.py .eml -> IOCs stdlib only, pure contacts.py IOCs -> abuse contacts (RDAP) network, read-only @@ -210,6 +224,68 @@ an edit to `submit.py`. This is the one place a plugin shape earns itself, because there are four known members with genuinely different APIs. There is no discovery mechanism; it is a package with four members. +## First run: `abusectl init` + +**`parse` REFUSES to run with no `trusted_relays` configured**, rather than +guessing. The outermost public IP is the usual guess and it is wrong in +exactly the case that matters: an attacker who forges extra `Received` +headers. A confident wrong answer here gets an innocent third party reported. + +Refusing is only defensible with a route out, so `abusectl init` writes the +config and exits, and `parse`'s error names it rather than stating a bare +failure. + +**It asks only what the part being built needs.** Today that is the trusted +relays and the cases directory. The MISP URL and key, the vendor keys and the +X-ARF identity arrive as questions when the parts that use them are built, so +no question is written before its part is designed. + +**A skipped answer is ABSENT from the file, never an empty string.** +`api_key = ""` reads as configured-and-broken and produces a confusing auth +error much later; an absent key reads as not-configured, and the part that +wants it can say so plainly. + +### The trusted-relay question has three tiers + +It is the answer a user is most likely to get wrong, and it is the one that +decides whether the reported IP is the attacker's or an innocent relay's. + +1. **Ask for CIDRs.** Validated with stdlib `ipaddress`, so a malformed entry + is rejected at the prompt rather than at parse time. +2. **A known-provider table.** Gmail, Fastmail, Proton and similar publish + their sending ranges. The table ships with the tool as static data. + Deliberately NOT read from SPF at runtime: that is a DNS lookup, and while + the never-resolve rule is about parsing hostile mail rather than about + setup, a static table keeps the boundary unambiguous. +3. **`--from-sample <msg.eml>`.** For self-hosted mail and anything absent + from the table: show the `Received` chain of a message the user knows + arrived legitimately, and let them pick which hops are theirs. This turns + an abstract question into choosing from a real list. + +### Two front ends, one writer + +`init` is a **pure builder plus a thin prompt shell**. One function takes the +answers as a mapping and returns the TOML text; the interactive prompts and +the command-line flags are two front ends over it. So the config-writing logic +is testable with no terminal and no files, and the two routes cannot drift. + +**`--non-interactive` exists so an agent can run setup**, taking every answer +as a flag and failing on a missing one rather than prompting. Every question +is also a flag; there is no answer reachable only by typing. + +**`init` never overwrites silently.** Re-running over an existing config +refuses unless `--force`, and `--force` still never drops a key the new run +did not supply: a config holding a MISP key must not lose it to a later run +that only set the relays. + +### What is tested and what is hand-tested + +The builder has right answers and is tested: the TOML it produces, a skipped +key being absent, a malformed CIDR being rejected, an existing config not +being clobbered. **The prompts are hand-tested by the user**, because whether +a question reads clearly has no assertion; a test driving stdin would assert +the wording it was written against and break on a rewording that improved it. + **`config.py`** reads TOML through stdlib `tomllib`, no dependency. It holds the cases path, the MISP URL and key, vendor keys, the user's reporting identity for X-ARF, `max_attempts` for the retry cap, and the trusted-relay @@ -228,6 +304,7 @@ already runs `mailsync.sh` from, and not to a process of this tool's own. | Part | Needs | |---|---| +| `init` | stdlib only | | `parse` | stdlib only | | `contacts` | an HTTP client | | `report` | stdlib only | @@ -491,8 +568,11 @@ keys. infrastructure.** Headers can be forged wholesale, and only the hops the user's own MTA added are trustworthy. Without a configured trusted-relay boundary, the "sending IP" is whatever the attacker chose to write. The -boundary is config; IOCs below it are marked `untrusted-hop` rather than -presented as fact. +boundary is config, supplied to `parse` as an ARGUMENT so the module stays +pure and config-free; `cli.py` reads it and passes it in. IOCs below the +boundary are marked `untrusted-hop` rather than presented as fact, and with no +boundary configured at all `parse` refuses to run rather than guessing, per +the `init` section above. **Never resolve and never fetch.** Not the URLs, not the redirects, not remote images. Following a link confirms the address is live to the sender and fires |