diff options
| author | Danilo M. <danix@danix.xyz> | 2026-09-08 13:20:51 +0200 |
|---|---|---|
| committer | Danilo M. <danix@danix.xyz> | 2026-09-08 13:20:51 +0200 |
| commit | e65e598f4fbdcdfdd0b837abe3c794862f66a3e6 (patch) | |
| tree | 837966360441b70927206514d09550262211d0ce /docs | |
| parent | 9ac5e7485e5f53f089c8932a8b13bc725b1ba029 (diff) | |
| download | abusectl-e65e598f4fbdcdfdd0b837abe3c794862f66a3e6.tar.gz abusectl-e65e598f4fbdcdfdd0b837abe3c794862f66a3e6.zip | |
plan: implementation plan for init and parse
Thirteen tasks, TDD throughout, stdlib only. parse is pure and offline:
the trusted-relay boundary arrives as an argument rather than a config
read, so the whole extractor is testable against fixtures with no setup.
The plan carries three checks that are not ordinary unit tests. The
Received-chain task has a mutation step, because walking one hop too far
reports an innocent third party named in a header the attacker wrote, and
a test that cannot fail would not protect against it. The URL task runs
the suite with sockets refused, so the never-fetch rule is verified
rather than read. And every fixture is asserted to leave no recipient
address anywhere in the manifest.
init exists because parse refuses to guess the trust boundary. It asks
for CIDRs, offers a static table of known provider ranges, or reads the
chain of a known-good sample and lets the user pick their own hops. A
pure builder with the prompts and the flags as two front ends over it, so
--non-interactive covers agent-driven setup and the config writing is
tested without a terminal. Re-running shows what is already configured
and asks; either route backs the old file up first and preserves sections
this run does not set, so a later init cannot silently drop a MISP key.
The prompts themselves are hand-tested rather than driven from stdin: a
test there would assert the wording it was written against and break on a
rewording that improved it. Task 12 is the checklist, weighted towards
wrong answers.
Redirect chains were missing from the first draft of the plan and are now
specified: a parameter whose value is itself a URL is recovered as an
indicator while every other value stays redacted, which resolves the
conflict between reporting the destination and never publishing a
tracking token.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KphFXTc2QajxXsHWyvGJ4R
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 |