diff options
| -rw-r--r-- | AGENTS.md | 10 | ||||
| -rw-r--r-- | abusectl/case.py | 4 | ||||
| -rw-r--r-- | abusectl/cli.py | 3 | ||||
| -rw-r--r-- | abusectl/parse.py | 128 | ||||
| -rw-r--r-- | abusectl/report.py | 884 | ||||
| -rw-r--r-- | docs/BACKLOG.md | 67 | ||||
| -rw-r--r-- | docs/plans/2026-09-09-report.md | 1536 | ||||
| -rw-r--r-- | docs/specs/2026-09-09-contacts.md | 25 | ||||
| -rw-r--r-- | docs/specs/2026-09-09-report.md | 443 | ||||
| -rw-r--r-- | tests/fixtures/reportable.eml | 22 | ||||
| -rw-r--r-- | tests/test_case.py | 10 | ||||
| -rw-r--r-- | tests/test_cli.py | 25 | ||||
| -rw-r--r-- | tests/test_offline.py | 8 | ||||
| -rw-r--r-- | tests/test_parse.py | 136 | ||||
| -rw-r--r-- | tests/test_report.py | 1600 |
15 files changed, 4896 insertions, 5 deletions
@@ -367,3 +367,13 @@ settles only what they share. - `docs/BACKLOG.md`, open items, with the cause verified in the code rather than assumed. Read it before starting work; add to it rather than fixing something unasked. + +The author also keeps an idea note OUTSIDE this repository, in his Obsidian +vault at `~/Documents/Obsidian/note/notes on abusectl.md`. It is an inbox, +written as things occur to him while developing or using the tool, split into +done and not-done and in no particular order. **The note is the inbox, the +backlog is the tracked list.** Reconcile them in both directions when starting +substantial work: an idea in the note that is real work earns a backlog entry +with its cause verified in the code, and a backlog item he has marked done in +the note can be closed. Do not edit the note unasked; it is his, not the +repository's. diff --git a/abusectl/case.py b/abusectl/case.py index 4c8c2db..233bc5f 100644 --- a/abusectl/case.py +++ b/abusectl/case.py @@ -87,6 +87,10 @@ def create(root: Path, raw: bytes) -> Case: "auth": [], "contacts": [], "destinations": [], + # Seeded like the rest so a created-but-unparsed case has the same + # shape as a parsed one: report indexes this block rather than + # guarding every access. Additive, so FORMAT_VERSION is unchanged. + "headers": [], } save(path, manifest) diff --git a/abusectl/cli.py b/abusectl/cli.py index 2200e14..ed31a4b 100644 --- a/abusectl/cli.py +++ b/abusectl/cli.py @@ -263,6 +263,9 @@ def _cmd_parse(args) -> int: manifest = case.load(created.path) manifest["iocs"] = parse_module.iocs(raw, trusted=settings.trusted_relays) manifest["auth"] = parse_module.auth_results(raw) + manifest["headers"] = parse_module.report_headers( + raw, trusted=settings.trusted_relays + ) case.save(created.path, manifest) except parse_module.NoTrustBoundary as exc: print(f"abusectl parse: {exc}", file=sys.stderr) diff --git a/abusectl/parse.py b/abusectl/parse.py index 230ca2d..2217583 100644 --- a/abusectl/parse.py +++ b/abusectl/parse.py @@ -60,6 +60,55 @@ _URL = re.compile(r"https?://[^\s<>\"')]+", re.IGNORECASE) # stack or spin forever, independent of the `seen` cycle guard. _MAX_REDIRECT_DEPTH = 5 +# The optional "for <addr>" clause of a Received header (RFC 5321 4.4). Our +# own boundary relay writes the ENVELOPE RECIPIENT there, so the one Received +# line a report publishes carries the victim's address verbatim unless this +# is removed. +# +# Anchored on the ADDRESS, not on the clause terminator. Requiring "for" to +# be immediately followed by ";" or end of line is strictly stronger than +# the grammar: RFC 5321 4.4 puts For inside Opt-info, so With, ID, Via or a +# CFWS comment may legitimately follow it, and "for <a@b> (envelope-from +# <c@d>);" is routine Exim and Sendmail output. That terminator anchor +# stripped only the neatest shape and published the address in four +# ordinary ones. +# +# Angle brackets are optional because For is 1*( Path / Mailbox ) and +# Mailbox carries none. The address run stops at "<>;" and whitespace so the +# match cannot swallow the rest of the header, and a trailing ")" is trimmed +# so the clause inside a comment does not leave an orphan bracket. +_RECEIVED_FOR = re.compile(r"(?is)\bfor\s+<?[^\s<>;]+@[^\s<>;]+>?\)?") + +# Left behind once a clause is cut from the middle of a line: a doubled +# space, or a space now sitting against the ";" that ends the Opt-info. The +# line is published verbatim to a third party, so it should not read as the +# output of a broken tool. +_LEFTOVER_SPACE = re.compile(r"\s{2,}") +_ORPHAN_SEPARATOR = re.compile(r"\s+([;)])") + +# A comment that held nothing but the clause, e.g. "(for <a@b>)", is now an +# empty pair of brackets or a lone opening one. Cut rather than left as +# debris in a published line. +_EMPTY_COMMENT = re.compile(r"\s*\(\s*\)?\s*$") + +# The headers that may appear in a published report. A WHITELIST, never a +# blacklist: a blacklist means every header this parser learns to read later +# is a leak waiting for someone to remember. To, Cc, Delivered-To and +# X-Original-To are absent by construction, which is the same reason iocs() +# does not read them either. +_REPORT_HEADERS = ( + "From", + "Subject", + "Date", + "Message-ID", + "Reply-To", + "Return-Path", + "Authentication-Results", + "Received-SPF", + "MIME-Version", + "Content-Type", +) + @dataclass(frozen=True) class Hop: @@ -156,6 +205,85 @@ def sending_ip(raw: bytes, trusted: list[str]) -> str | None: return None +def _strip_envelope_recipient(received_value: str) -> str: + """Remove the "for <addr>" clause from a Received header. + + The boundary hop is written by OUR OWN relay, and that clause is where it + records the envelope recipient: the victim's address, in the one header a + report reproduces verbatim. Truncating the chain at the boundary does not + help here, because the leak is INSIDE the line being kept, which is why + this is a separate step rather than part of the walk. + + The clause is optional (RFC 5321 4.4) and carries nothing a desk needs: + the report is about who SENT the message. Only the clause goes, so the + hop's own evidence, the address and the receiving server, survives. + + Every shape of the clause is cut, not merely the tidy one: see + _RECEIVED_FOR, where anchoring on the terminator rather than on the + address let four ordinary shapes publish the address. + """ + cut = _RECEIVED_FOR.sub("", received_value) + cut = _LEFTOVER_SPACE.sub(" ", cut) + cut = _ORPHAN_SEPARATOR.sub(r"\1", cut) + return _EMPTY_COMMENT.sub("", cut).strip() + + +def report_headers(raw: bytes, trusted: list[str]) -> list[tuple[str, str]]: + """Return the headers that may be published, outermost Received first. + + `report` never opens source.eml: the decision about what may be disclosed + is made once, here, beside every other one. A second module filtering the + original at report time would put that decision in two places, and two + places to remember is how the fourth property leaked three times. + + Received is published as EXACTLY ONE line, the boundary hop, and the + chain is cut in both directions. Above it are our own relays: publishing + them tells a third party about the user's mail path. Below it is the + attacker's own writing, and that half is the dangerous one. A forged + chain names an innocent third party (the forged-chain fixture plants + 198.51.100.7 for this), so publishing a hop below the boundary puts + someone else's address into a report an abuse desk will act on. This is + the third property, applied to what gets DISCLOSED rather than to what + sending_ip() concludes, and the answer is the same for the same reason: + the boundary hop is the last line we can stand behind. + + That one surviving line still goes through _strip_envelope_recipient(): + it was written by our own relay and names the victim in its "for" clause, + so the whitelist alone does not make it safe. + + Everything else comes from the _REPORT_HEADERS whitelist, so a recipient + header is absent because it was never named rather than because it was + stripped. + + A Received line naming no parseable IP cannot be placed against the + boundary, so it is dropped rather than guessed at. Publishing an + unplaceable line risks disclosing exactly the two things the cut exists + to prevent; iocs() still records every hop it can read, so nothing is + lost to review, only to the report. + + Returned as a list of pairs rather than a dict because the whitelist may + later keep a header that repeats, and order carries meaning. + """ + message = _message(raw) + result: list[tuple[str, str]] = [] + + for value in message.get_all("received") or []: + ip = _extract_ip(str(value)) + if ip is None: + continue + if _in_any(ip, trusted): + continue + result.append(("Received", _strip_envelope_recipient(str(value)))) + break + + for name in _REPORT_HEADERS: + value = message.get(name) + if value is not None: + result.append((name, str(value))) + + return result + + def _address_of(header_value: str | None) -> str | None: """Return the addr-spec of a sender header, ignoring its display name. diff --git a/abusectl/report.py b/abusectl/report.py new file mode 100644 index 0000000..36bbbff --- /dev/null +++ b/abusectl/report.py @@ -0,0 +1,884 @@ +# 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. +"""IOCs and abuse contacts to report bodies: the last step before anything +irreversible happens. + +This module is PURE and OFFLINE. It opens no socket, sends no mail and reads +no file outside the case directory. What it produces is a document the user +reads, edits and approves, so the output is written for a human first and a +parser second. + +It takes the reporting identity as an ARGUMENT rather than reading the +config, the way parse.py takes the trust boundary. The identity is the one +thing in a report disclosed deliberately, and a module that reaches for it +itself is a module that can disclose it in a code path nobody reviewed. +""" + +import hashlib + +from email.header import Header +from email.headerregistry import Address +from email.message import EmailMessage +from email.policy import SMTP + + +VERSION = "0.1.0" + + +# Hard wrap column, from the spec: "Plain text, hard-wrapped at 72 columns." +# Abuse desks run ticketing systems that reflow or clip long lines, and a +# clipped line is the defect this whole wrapping scheme exists to prevent. +_WIDTH = 72 + +# The marker that says "this value continues on the next line". A trailing +# backslash is the convention shells, C and Makefiles all use, so it reads +# without a legend, and it is what makes a broken line UNAMBIGUOUS: a reader +# seeing no backslash knows the value ended there. +# +# The alternative, breaking silently, is the same defect as truncating. A +# desk that copies one line of a wrapped URL and acts on it has acted on a +# resource that was never reported. Something must mark the seam, and this +# marks it in the only direction that is safe: a fragment ANNOUNCES that it +# is a fragment, rather than a whole value having to prove it is whole. +# +# Because the marker is a character a VALUE may also contain, every literal +# backslash in a value is DOUBLED before wrapping and halved on the way +# back. Without that, a value ending in "\" is indistinguishable from a wrap +# marker, and the values here are attacker-supplied: a trailing backslash is +# legal in a URL path, and an attacker reading this source could append one +# to make their indicator garble itself in the report a desk reads. +# +# Doubling EVERY backslash rather than only a trailing one is what keeps the +# encoding unambiguous. Escaping just the last character leaves "x\\" (two +# literal backslashes) encoding to the same text as "x\" followed by a wrap, +# which is the same bug one character further along. +_CONTINUATION = "\\" +_ESCAPE = "\\" + + +def _escape(value: str) -> str: + """Double every backslash so none can be read as a continuation marker. + + The inverse is _unescape(). Applied to the value only, never to the + indent or the surrounding prose, so what a reader sees differs from the + literal value in exactly one way and unwrap() undoes exactly that. + """ + return value.replace(_ESCAPE, _ESCAPE + _ESCAPE) + + +def _unescape(value: str) -> str: + """Halve the doubled backslashes _escape() produced. + + Scans left to right, consuming a doubled pair as one character. On text + _escape() actually produced, every run is even and str.replace gives + the same answer, which a mutation test confirmed over every backslash + pattern up to length 11: this is NOT protecting the round trip, and + saying otherwise would overstate what the tests hold down. + + It is kept because unwrap() is public and may be handed a line a user + edited, where a run can be odd. The scan then consumes pairs strictly + left to right and leaves the odd one alone, which is the reading that + matches how the text was written; str.replace rescans its own output + and would fold a stray backslash into the pair beside it. + """ + out = [] + index = 0 + while index < len(value): + if value.startswith(_ESCAPE + _ESCAPE, index): + out.append(_ESCAPE) + index += 2 + else: + out.append(value[index]) + index += 1 + return "".join(out) + + +def _wrap_value(value: str, indent: str) -> list[str]: + """Break one long value across lines so no line exceeds _WIDTH. + + Breaks at an arbitrary column rather than at a word or a punctuation + boundary, DELIBERATELY. A URL has no whitespace, so a word-wrapper + leaves it over-long and the column guarantee fails on exactly the value + that matters most. Breaking after a "/" or a "&" instead would be + prettier and is wrong: those characters are meaningful inside the value, + so a break at one is a place a reader cannot tell a seam from content. + An arbitrary break plus an explicit marker is legible precisely because + the marker, not the position, carries the meaning. + + The value is ESCAPED first, so its own backslashes cannot be mistaken + for the marker, and the wrap arithmetic then runs over the escaped text: + the column limit governs what is PRINTED, and the escaped form is what + is printed. Measuring the original instead would let a value full of + backslashes overflow the line. + + A break must NEVER land between the two halves of an escaped pair, and + the loop below backs off by one character to guarantee it. This is not + tidiness: unwrap() decides whether a trailing backslash is a marker or + content by the PARITY of the run it ends, and a break inside a pair + splits that run across two lines, so both halves are counted wrongly. + A genuine marker then reads as content, the continuation line is + orphaned, and the tail of the value is silently dropped. + + That defect survived a first fix and a passing test suite, because it + needs a backslash to land exactly on the break column: it appears in + random adversarial values roughly one time in eight and in none of the + hand-written cases. Backing off one character costs a column on a line + that has a backslash at its edge, and buys an invariant that holds for + every input rather than for the inputs someone thought of. + + The value is never altered, only divided and escaped; unwrap() is the + exact inverse, asserted over adversarial values including trailing and + doubled backslashes. + """ + value = _escape(value) + room = _WIDTH - len(indent) - len(_CONTINUATION) + if len(indent) + len(value) <= _WIDTH: + return [indent + value] + lines = [] + while len(value) > room: + cut = room + # An odd run of backslashes ending at the cut means the last one is + # the first half of a pair; move the break before it. + if (len(value[:cut]) - len(value[:cut].rstrip(_ESCAPE))) % 2 == 1: + cut -= 1 + lines.append(indent + value[:cut] + _CONTINUATION) + value = value[cut:] + lines.append(indent + value) + return lines + + +def unwrap(text: str) -> str: + """Rejoin lines a continuation marker broke, giving back the values. + + The inverse of the wrapping above, and the reason the wrapping is + honest rather than merely tidy: a value that can be mechanically + reassembled is a value that was not damaged by being displayed. A desk + that scripts against the text part gets its indicators back exactly as + reported, and the tests assert reassembly rather than mere presence, + which is what rules out a truncation passing as a wrap. + + The leading whitespace of a continuation line is display indent, not + content: no value this module emits begins with a space, because every + one of them is an indicator, a header value or an identity, all of + which are stripped before they arrive. + + A trailing backslash is a marker only when the run of backslashes it + ends is ODD, because _escape() doubled every literal one. An even run is + entirely escaped content and the line ends there. Testing endswith("\\") + alone was the defect this parity check replaces: it read a value's own + trailing backslash as a marker and swallowed the following line, which + for a Subject meant absorbing the Date header beneath it. + + Rejoining happens BEFORE unescaping, so a doubled pair split across a + break is whole again before it is decoded. + """ + joined = [] + for line in text.splitlines(): + if joined and _ends_with_marker(joined[-1]): + joined[-1] = joined[-1][:-len(_CONTINUATION)] + line.lstrip() + else: + joined.append(line) + return "\n".join(_unescape(line) for line in joined) + + +def _ends_with_marker(line: str) -> bool: + """Whether this line ends in a continuation marker rather than content. + + The marker is one unescaped backslash, and _escape() doubled every + literal one, so the question is purely the PARITY of the trailing run: + odd means the last backslash has no partner and is the marker, even + means every one is half of an escaped pair and the line ends here. + """ + run = len(line) - len(line.rstrip(_ESCAPE)) + return run % 2 == 1 + + +# What each parse.py origin means in a sentence a desk can act on. +# +# The raw tokens are a PARSER's vocabulary: "header-list_unsubscribe" has an +# underscore in it and reads as debug output, which makes a careful report +# look machine-dumped and invites a desk to discount it. The mapping is +# deliberately small and flat, one line each, because the alternative, a +# sentence generated per indicator, is a second body of prose to keep true. +# +# An origin absent from this table is shown AS-IS rather than dropped: a +# newer parse.py may invent one, and losing the only line that says where an +# indicator was seen is worse than showing an ugly token. Ugly is also +# self-correcting, since it is visible to whoever reads the next report. +# +# Each phrase completes the sentence "seen ...", so every entry must read +# grammatically after that word. The first draft mixed "from the Received +# chain" with "the From header" and rendered "seen the From header", which +# reads as a typo and undercuts exactly the care the report is meant to +# show. Keep new entries in the same voice. +_ORIGINS = { + "received-chain": "in the Received chain", + "body": "in a link in the message body", + "redirect-target": "as a redirect target declared by another link", + "attachment": "as an attachment", + "header-from": "in the From header", + "header-reply_to": "in the Reply-To header", + "header-return_path": "in the Return-Path header", + "header-list_unsubscribe": "in the List-Unsubscribe header", +} + + +_REDACTION_NOTE = ( + "Recipient identifiers have been removed from this report by policy.", + "Parameter names are preserved, parameter values are not. Full", + "evidence is retained locally and is available on request.", +) + + +# The role mailboxes RFC 2142 mandates, which it also requires be matched +# case-insensitively. Only the ones an RDAP abuse entity plausibly +# publishes; this is not the full list and does not need to be. +_ROLE_MAILBOXES = frozenset({ + "abuse", "postmaster", "security", "noc", "hostmaster", +}) + + +def _is_mailable(address: str) -> bool: + """Whether this value can be the target of a mail at all. + + RDAP jCard data is third-party and occasionally malformed, so a + published "abuse address" is not guaranteed to be one. A value with no + "@", or with either half empty, cannot be delivered to anyone. + + The check exists because the alternative is silent and worse than a + missing contact. A malformed value used to become a destination with + status "pending" and an unsendable target, so the indicator appeared + on its way to a desk, no desk would ever receive it, and it was + excluded from unreportable() precisely because its contact HAD an + abuse entry. That is the failure the unreportable array exists to + prevent, arriving through the one door that array does not watch. + + Deliberately shallow: this is not address validation and must not + become it. Whether a syntactically fine address reaches a live desk is + the mail transport's answer, not a parser's, and rejecting an address + a desk actually reads would drop a report. It rejects only what cannot + be a mailbox under any reading. + """ + local, at, domain = address.rpartition("@") + return bool(at and local.strip() and domain.strip()) + + +def _group_key(address: str) -> str: + """The key two spellings of one desk must share, and no more than that. + + The DOMAIN is case-insensitive by every standard that touches it, so + "abuse@Host.Invalid" and "abuse@host.invalid" are one desk and must + not be mailed twice about one incident. + + The LOCAL PART folds only for the RFC 2142 ROLE MAILBOXES. Those are + standardised names that the same RFC requires be recognised regardless + of case, so no host runs "Abuse@" and "abuse@" as two different desks, + and treating them as two is a duplicate mail with nothing on the other + side of the trade. This is where the duplicate actually happens, since + an abuse entity publishes a role mailbox nearly every time. + + Any other local part is left exactly as published. RFC 5321 leaves its + interpretation to the receiving host, and only that host knows whether + it folds case. For a NAMED mailbox the asymmetry that governs the role + names reverses: folding two desks a host genuinely distinguishes would + silently drop one of them, and a dropped desk is worse than a duplicate + mail. So each half folds on the strength of its own standard, and + neither borrows the other's. + + The first version of this folded the domain alone and shipped with a + test that used a lowercase local part throughout, so the test passed + while "Abuse@Host.Invalid" and "abuse@host.invalid" produced two + destinations. A test that varies one half of its input proves nothing + about the other. + """ + local, at, domain = address.rpartition("@") + if not at: + # Not an address shape we can split. _is_mailable keeps these out + # of the destinations, but the key stays defined for anything that + # asks for an id directly; group it by its literal text rather + # than inventing a domain for it. + return address + if local.lower() in _ROLE_MAILBOXES: + local = local.lower() + return f"{local}@{domain.lower()}" + + +def email_destination_id(address: str) -> str: + """The id of the destination that reports to this desk. + + Derived from the ADDRESS, so an id names a desk rather than a position + in whatever list this run happened to build. + + That distinction is the whole point. `report` may run again on a case, + and `contacts` may have resolved a new indicator since; a positional id + then renumbers every desk after the newcomer. Bodies are written to + bodies/<id>.xarf and each body's SHA-256 is recorded against its id, so + after a renumber the file on disk belongs to a DIFFERENT desk than the + manifest entry sharing its id. The edit check would compare one desk's + body against another's, reporting an edit nobody made, or, if the two + happened to match, missing one that was. + + It hashes the same normalised form the grouping uses, so two spellings + of one desk get one id. Deriving it from the raw text instead would let + the spelling RDAP published first decide a body's filename. + """ + digest = hashlib.sha256(_group_key(address).encode("utf-8")).hexdigest() + # ponytail: 8 hex chars, a case has a handful of desks; widen if a + # collision is ever observed. A short id keeps a case directory + # readable to the human reviewing it, which is what it is for. + return f"email-{digest[:8]}" + + +def email_destinations(contacts: list[dict]) -> list[dict]: + """Group contacts into one destination per abuse ADDRESS. + + Contacts already fold by host, but two different contacts can still + resolve to the same address, an IP and a domain at one hoster being the + common case. One mail per address rather than per contact is what stops + a desk receiving two mails about one incident. + + Each id is derived from its own target address, so it survives both a + reordering and a change in the contacts: a desk keeps its id when a new + indicator resolves to a new desk ahead of it. See email_destination_id + for why that matters more than it first appears. + + The returned ORDER is still first-seen, because the destinations are a + list a human reads during review and the order the indicators were + found in is the most explicable one available. Nothing downstream may + key off that order; the id is what identifies a destination. + """ + by_address: dict[str, dict] = {} + + for contact in contacts: + for address in contact.get("abuse", []): + if not _is_mailable(address): + # unreportable() applies the same test, so the indicator + # is listed there rather than vanishing between the two. + continue + key = _group_key(address) + # First spelling seen wins the target. Any spelling reaches the + # desk, and picking one keeps the report stable across a re-run. + destination = by_address.setdefault(key, {"target": address, + "iocs": []}) + for ioc in contact.get("iocs", []): + if ioc not in destination["iocs"]: + destination["iocs"].append(ioc) + + return [ + { + "id": email_destination_id(destination["target"]), + "kind": "email", + "target": destination["target"], + "iocs": destination["iocs"], + "body": None, + "status": "pending", + } + for destination in by_address.values() + ] + + +def unreportable(contacts: list[dict]) -> list[dict]: + """List every IOC that reached no email destination, with the reason. + + A missing contact is a normal outcome, not an error: RDAP publishes no + abuse role for many netblocks. Making it visible is what keeps review + honest, since finding it any other way means diffing the IOC list + against every destination's IOC list. Same instinct as + suspect_path_segments flagging rather than redacting. + + The membership test is "reached no destination", not "sits in a + contact that resolved nothing", and those differ. Contacts fold by + HOST, so one indicator can appear in two contacts, a domain that + resolved and an IP that did not. Listing it because one of its + contacts failed would put it in the destinations AND in the list of + things no desk was found for, in one manifest. A reviewer reads the + second and hand-reports an indicator already on its way to a desk, + which costs the exact diffing this array exists to spare them. So an + indicator is unreportable only when NONE of its contacts produced a + mailable address, and the mailability test is the one + email_destinations applies, so no indicator can fall between them. + + Each indicator appears ONCE. Two failed contacts for one host are two + rows about one indicator otherwise, possibly with different reasons. + First reason seen wins, matching the first-seen ordering of the + destinations: both lists then read in the order the indicators were + found, which is the only ordering a human can explain. + """ + reachable = set() + for contact in contacts: + if any(_is_mailable(address) + for address in contact.get("abuse", [])): + reachable.update(contact.get("iocs", [])) + + result: list[dict] = [] + listed = set() + for contact in contacts: + if any(_is_mailable(address) + for address in contact.get("abuse", [])): + continue + # A contact's own error says more than the fallback, which is why + # it wins even when the contact published an unusable address. + # An empty string is not a reason: a manifest is a file the user + # edits, and a blank reason renders as a blank cell that tells + # them nothing. + reason = contact.get("error") or ( + "no usable abuse address published" if contact.get("abuse") + else "no abuse address resolved") + for ioc in contact.get("iocs", []): + if ioc in reachable or ioc in listed: + continue + listed.add(ioc) + result.append({"ioc": ioc, "reason": reason}) + return result + + +def _describe(entry: dict) -> str: + """The one-line "why you are seeing this" under an indicator. + + A boundary hop wins over its origin because it is the strongest claim + the tool makes: it is the hop sending_ip() resolved to, the last one we + can stand behind, and a desk needs to know it is being told "your + address sent this" rather than "your address appeared somewhere in a + chain the attacker partly wrote". + + Every other hop in the chain is attacker-writable, so it gets the + ordinary origin line and no claim of authorship. That distinction is + the third property expressed to a reader. + """ + if entry.get("confidence") == "boundary-hop": + return "sending IP, first hop outside our trust boundary" + origin = entry.get("origin", "") + if not origin: + return "" + return "seen " + _ORIGINS.get(origin, origin) + + +def text_part(manifest: dict, destination: dict, identity: dict) -> str: + """Build the human-readable part: the one that decides whether a desk + acts on the report. + + The ask goes first, because a desk triaging a queue must know in one + line what happened and what is wanted. Only THIS destination's own + indicators appear: a desk shown three IPs that are not theirs stops + reading, and, worse, has been told about a third party's infrastructure + for no reason. The lookup is by id against the manifest, so a + destination naming an id the manifest does not carry contributes no row + rather than raising; a manifest is a file the user edits and the two can + disagree. + + NOTHING here is truncated. Every value that does not fit is wrapped with + an explicit continuation marker instead, because a cut value is a WRONG + value rather than a short one: a desk acting on the first 72 characters + of a URL acts on a resource nobody reported, and a cut header misstates + what the message declared. See _wrap_value for why the marker is what + makes that safe. + + Introduces nothing that did not come from the manifest or the identity. + Everything it formats has already been through redact.py, so this is not + a filter and must not become one, but it also must not add: the + destination's target address is deliberately absent from the body, since + a desk knows its own address and printing it only adds a string to a + document whose whole discipline is that fewer strings leak less. + """ + by_id = {entry["id"]: entry for entry in manifest.get("iocs", [])} + mine = [by_id[i] for i in destination.get("iocs", []) if i in by_id] + + lines = [ + "Phishing message reported: infrastructure on your network was", + "used to send or host it. Requesting takedown and customer", + "notification.", + ] + + if mine: + lines += ["", "Observed on your infrastructure:", ""] + for entry in mine: + lines += _wrap_value(str(entry.get("value", "")), " ") + description = _describe(entry) + if description: + lines += _wrap_value(description, " ") + + # Pairs, not a dict: JSON has no tuple, so case.load() hands these back + # as lists. Both shapes destructure identically, and there is a test + # driving the round-tripped one because that is what actually arrives. + shown = [(name, value) for name, value in (manifest.get("headers") or []) + if name in ("Date", "From", "Subject")] + if shown: + lines += ["", "Message as declared:", ""] + for name, value in shown: + lines += _wrap_value(f"{name}: {value}", " ") + + auth = manifest.get("auth") or {} + if auth: + lines += ["", "Authentication results:", ""] + lines += _wrap_value( + " ".join(f"{key.upper()}: {value}" + for key, value in sorted(auth.items())), " ") + + lines += ["", *_REDACTION_NOTE, ""] + lines += _wrap_value( + f"Reported by: {identity['name']}, {identity['org']} " + f"<{identity['email']}>", "") + lines.append("Generated by abusectl.") + + return "\n".join(lines) + "\n" + + +# Every character that any reasonable reader of this part might treat as the +# end of a field, plus "%" itself. +# +# CR and LF are the ones that matter: a field value carrying one forges a +# field, and the forged field is read as something THIS TOOL asserted, on a +# document that carries the reporter's identity. That is header injection +# into mail we send, which the contacts spec already names as a hazard. +# +# The rest are here because "what counts as a line break" is not one answer. +# Python's own email module raises on U+2028 and U+2029 as readily as on LF, +# because it reaches for str.splitlines(), which also breaks on VT, FF, the +# three information separators and NEL. A value carrying one of those does +# not merely render oddly in the assembled document: it aborts the document. +# So the set is taken from splitlines() rather than from RFC 5322, on the +# principle that the defence must cover what the CONSUMERS break on, not what +# one specification says a break is. +# +# "%" is in the set for a different reason, and leaving it out is a second +# injection one step later: see _field_value(). +_UNSAFE = "".join(chr(c) for c in ( + 0x0A, 0x0B, 0x0C, 0x0D, 0x1C, 0x1D, 0x1E, 0x85, 0x2028, 0x2029, +)) + "%" + + +def _field_value(value) -> str: + """Make one attacker-supplied value safe to emit as a field value. + + ENCODES rather than rejects or strips, and the choice is the whole point + of this function. + + The value is reachable, not theoretical. redact.url_valued_parameters() + URL-DECODES a redirector's destination parameter in order to recover it + as an indicator in its own right, exactly as the second property + intends. So a message body carrying + + http://r.invalid/go?next=http%3A%2F%2Fa.invalid%2Fx%0AFeedback-Type... + + produces, through parse.iocs() on a real message, an IOC whose value + contains a literal newline followed by text shaped like a field. There + is no upstream filter between that and here. + + REJECTING the indicator loses a genuine redirect target, which is one of + the more actionable things a desk receives, over an attacker's choice of + byte. STRIPPING the character silently rewrites the indicator into a + different URL, and a desk that acts on the stripped form has acted on a + resource nobody reported: that is the truncation defect from the text + part wearing a different coat, and it is worse here because nothing in + the output says it happened. + + Percent-encoding is the URL's own native encoding, so it reads as + intended by the audience that receives it; it is EXACTLY REVERSIBLE, so + a desk or a later submit path recovers what the message declared; and it + is visible, so an altered value announces the alteration rather than + passing as a clean one. Same instinct as the continuation marker in the + text part: a fragment announces that it is a fragment. + + "%" must be encoded too, and this is not tidiness. A redacted URL + legitimately contains percent signs, so an encoder that leaves them + alone emits text whose reversal produces the very control character the + encoding existed to remove: an attacker writes the literal five + characters "%0A" into a path and unquote() hands the next reader a + newline. An encoding that is not a true inverse is not a defence. + + The escape runs over UTF-8 BYTES, not over code points, and that is not + a detail. Percent-encoding is defined on octets, so a character above + U+007F has more than one byte to spell: encoding U+2028 as "%2028" from + its ordinal produces text that unquote() reads back as "%20" followed by + the literal "28", which is a SPACE and the digits, not the character + that was there. The reversibility this function claims would then be + false for exactly the two characters that are here because Python's + email module breaks on them. Encoding each of its three UTF-8 bytes + gives "%E2%80%A8", and unquote() returns U+2028. + + Non-strings are coerced rather than raising. A manifest is a file the + user edits by hand and JSON has numbers; this module is the last step + before a reviewed case becomes a sent mail, and a report that raises + produces nothing at all. + """ + text = str(value) + if not any(ch in text for ch in _UNSAFE): + return text + out = [] + for ch in text: + if ch in _UNSAFE: + out.append("".join(f"%{b:02X}" for b in ch.encode("utf-8"))) + else: + out.append(ch) + return "".join(out) + + +def feedback_fields(manifest: dict, destination: dict) -> list[tuple[str, str]]: + """Build the machine-readable part: an RFC 5965 envelope carrying x-arf + fields inside it. + + RFC 5965 is an IETF standard and universally understood, but it was + designed for feedback loops, where a report is ABOUT A MESSAGE. These + reports are about INDICATORS, and 5965 has no field for "this specific + host is the thing being reported". x-arf's Source does. The envelope is + the standard's own extension point: 5965 requires an implementation to + ignore fields it does not support, so a standards parser reads what it + knows and x-arf tooling finds what it wants. + + Returned as PAIRS, not a dict, because Reported-Uri and Reported-Domain + repeat. Everything else does not, and that is enforced rather than + assumed: RFC 5965 gives Source-IP and Arrival-Date "once maximum". A + strict parser meeting a repeated single-occurrence field either rejects + the part or keeps the last occurrence, so a second Source-IP does not + add an address, it DISPLACES the primary one. Every address still + travels: the text part lists all of this destination's indicators, and + that is the part a human acts on. + + ARRIVAL-DATE IS DELIBERATELY ABSENT. 5965 defines it as when the + generating ADMD's own MTA received the message; the manifest's Date + header is when the SENDER CLAIMED to have sent it, which on a phishing + message is attacker-controlled free text. Copying one into the other + asserts an attacker's timestamp as our own observation, and a desk + correlating it against their logs finds nothing and discounts the + report. The honest source is the boundary Received hop's timestamp, + which parse.report_headers() already publishes; extracting it needs a + date parser, and 5965 makes the field optional, so it is omitted until + something actually needs it. An absent optional field misstates nothing. + + SOURCE IS OMITTED WHEN THERE IS NOTHING TO PUT IN IT, rather than + emitted empty. "Source:" with nothing after it tells a parser that the + thing being reported is the empty string; no field tells it nothing, + which is the truth. This happens when a desk's only indicators are a + sha256 or an observation, and both of those belong to the human part. + + sha256 and observation are NOT forced into the nearest-looking field. + Neither is a Source, a URI or a domain, and telling a desk that a file + hash is a URI is a false statement in the part meant to be machine-read. + + Every field NAME here is a literal in this module and none is derived + from data. The tempting generalisation, a table from an IOC's own type + string to a field name, would make a hand-edited manifest able to name + fields; the set is closed on purpose. Every field VALUE goes through + _field_value(), at the single point where the pairs are built, for the + reason the fourth property was learned three times over: validation + applied per branch gets forgotten on the next branch. + """ + by_id = {entry["id"]: entry for entry in manifest.get("iocs", [])} + mine = [by_id[i] for i in destination.get("iocs", []) if i in by_id] + + ips = [e.get("value") for e in mine if e.get("type") in ("ipv4", "ipv6")] + urls = [e.get("value") for e in mine if e.get("type") == "url"] + domains = [e.get("value") for e in mine if e.get("type") == "domain"] + + # Source is singular, so the primary indicator fills it and the rest + # travel in the repeatable fields and in the text part. An IP is the + # most actionable thing a hosting desk can act on, so it wins when + # present; a domain beats a URL because a desk suspending a name + # covers every URL under it. + primary = (ips or domains or urls or [None])[0] + + fields = [ + ("Feedback-Type", "abuse"), + ("User-Agent", f"abusectl/{VERSION}"), + ("Version", "1"), + ("Report-Type", "phishing"), + ] + + if primary is not None: + fields.append(("Source", primary)) + if ips: + # Once maximum, per RFC 5965. See the docstring. + fields.append(("Source-IP", ips[0])) + for domain in domains: + fields.append(("Reported-Domain", domain)) + for url in urls: + fields.append(("Reported-Uri", url)) + + return [(name, _field_value(value)) for name, value in fields] + + +# The header names the third part may carry, and the reason it is a list here +# rather than a trust in the manifest. +# +# parse.report_headers() already produces a whitelist, so in the normal path +# every name arriving here is one of these and this filter changes nothing. +# The manifest is a FILE THE USER EDITS, though, and the first property's +# structural argument is that report cannot disclose what it was never given. +# A hand-edited manifest is precisely a way it CAN be given something: a user +# pasting a header block back in, or a case written by a future parse.py whose +# whitelist grew, hands this module a "To:" pair and the argument no longer +# holds by construction. +# +# So the same whitelist is applied a second time at the point of publication. +# That is a deliberate exception to "two places to remember is how the fourth +# property leaked", and the trade runs the other way here: this is not a +# second DECISION about what may be disclosed, it is the same decision +# enforced where the disclosure actually happens. parse.py still decides; this +# refuses to publish anything it did not decide for. A duplicated whitelist +# that drifts loses a header from a report, which review sees. A missing one +# publishes a recipient address to the attacker, which nobody sees. +# +# Transcribed from the spec's "Where the headers come from" whitelist, which +# is the same list parse._REPORT_HEADERS holds. Matched case-insensitively, +# because a header name is case-insensitive by RFC 5322 and a hand-edited +# manifest will not have preserved anyone's capitalisation. +_PUBLISHABLE_HEADERS = frozenset({ + "received", "from", "subject", "date", "message-id", "reply-to", + "return-path", "authentication-results", "received-spf", + "mime-version", "content-type", +}) + + +def _header_line(name: str, value: str) -> str: + """One line of the third part, safe to emit even when the value is not. + + A header VALUE here is attacker-supplied free text. The spec keeps + Subject and the From display name deliberately, because they are what + lets a desk recognise a campaign, and it says plainly that the whitelist + governs WHICH headers travel and never what is inside one. A Subject + carrying a newline therefore reaches this function, and emitting it + verbatim forges a header line inside a part whose entire content is read + as headers: "Subject: evil\\nFrom: forged@attacker.invalid" becomes two + headers, the second of which a desk reads as something the message + declared. That is the Task 5 injection one part further along, and it is + worse here, because the forged line is grammatical where a forged x-arf + field is merely present. + + The answer differs from _field_value()'s percent-encoding, and the + difference is the audience. This part's content IS rfc822 headers, so the + encoding a reader of it already knows is RFC 2047, not URL escaping. An + encoded word neutralises the break by turning it into an RFC 5322 FOLD: + the value continues on a continuation line, a parser unfolds it back to + one header whose value is the original text including the character that + was there, and nothing new appears in the header list. Percent-encoding + would also be safe and would read as a bug to a mail parser, which is the + one audience this part has. + + Encoded ONLY when a value actually carries something unsafe. RFC 2047 + encodes indiscriminately, so applying it to every header would render an + ordinary Subject as "=?utf-8?q?Your_account?=" and cost the desk the + legibility this part exists for. The condition is the same _UNSAFE set + the machine part uses, minus "%": "%" is in that set because + percent-encoding must be a true inverse, and nothing here percent-encodes, + so a literal "%" in a Subject is just a character. + + Non-strings are coerced for the reason _field_value() coerces them: a + manifest is hand-edited, JSON has numbers, and a report that raises + produces nothing at all. + """ + text = str(value) + if any(ch in text for ch in _UNSAFE if ch != "%"): + # maxlinelen leaves room for "Name: " on the first line; the exact + # number only affects where a fold lands, never what unfolds back. + return f"{name}: {Header(text, 'utf-8', maxlinelen=64).encode()}" + return f"{name}: {text}" + + +def _headers_part(manifest: dict) -> str | None: + """The third part's body, or None when there is nothing to publish. + + Returns None rather than an empty string so build() can OMIT the part. + An empty text/rfc822-headers is a positive claim that the message + declared no headers, which is never true of a real message; absent says + the report carries none, which is the truth for a case parsed before the + headers block existed or one a user emptied by hand. Same rule as + feedback_fields() omitting Source rather than emitting it empty, and the + same rule as the config's absent-not-empty-string. + + Pairs, not a dict, matching what case.load() hands back: JSON has no + tuple, so these arrive as lists and both shapes destructure identically. + """ + lines = [ + _header_line(name, value) + for name, value in (manifest.get("headers") or []) + if str(name).lower() in _PUBLISHABLE_HEADERS + ] + if not lines: + return None + return "\n".join(lines) + "\n" + + +def build(manifest: dict, destination: dict, identity: dict) -> str: + """Assemble one destination's report as an RFC 5965 MIME document. + + Three parts: what a human reads, what a parser reads, and the headers. + + THE ORIGINAL MESSAGE IS NOT ATTACHED, and there is no message/rfc822 + part. source.eml carries every identifier the first property exists to + keep out: To, Cc, Delivered-To, unredacted URLs whose query and path + segments encode the recipient, the user's own Message-IDs and maildir + paths. An abuse desk forwards a report to the abused customer, who for a + phishing domain may be the attacker, and URLhaus is a public feed. RFC + 5965 provides text/rfc822-headers for exactly the case where the full + message cannot be included, so this is the standard's own answer rather + than a deviation from it. + + The FROM is built with email.headerregistry.Address rather than by + formatting a string. A reporting identity legitimately contains a comma, + "Example Consulting, Ltd" being the obvious one, and a comma is the + address-list separator: f"{name} <{email}>" then parses back as TWO + addresses, the first of which is a bogus addr-spec with no domain. A desk + replying to the report replies to that, and the reply reaches nobody. + Address quotes the display name when it needs quoting and RFC 2047-encodes + it when it is not ASCII, which are the two cases a consultant's org name + actually hits. + + Each subpart's type is set AFTER its content, which is the opposite of + what looks right and was checked rather than assumed: set_content() + REPLACES the Content-Type it derived from the payload, so setting the type + first leaves all three parts as text/plain. Setting it afterwards keeps + the transfer encoding and charset set_content() chose, which is what makes + a non-ASCII header value in the third part survive as base64 rather than + as a malformed 7bit line. + + The multipart boundary is chosen by the generator at SERIALISATION time, + after it has seen every payload, so a body containing something shaped + like a delimiter cannot collide with the real one; that is verified + against a payload carrying a literal "--===============0==" line. + """ + message = EmailMessage(policy=SMTP) + message["From"] = Address(str(identity.get("name", "")), + addr_spec=str(identity["email"])) + message["To"] = destination["target"] + # The case id is generated by case.py from a date and random hex, so it is + # not attacker-supplied and needs no escaping; it is the string the user + # greps for when a desk replies. + message["Subject"] = ( + f"Abuse report: phishing infrastructure, case {manifest['case_id']}" + ) + message.make_mixed() + message.set_type("multipart/report") + message.set_param("report-type", "feedback-report") + + human = EmailMessage(policy=SMTP) + human.set_content(text_part(manifest, destination, identity)) + message.attach(human) + + machine = EmailMessage(policy=SMTP) + machine.set_content( + "\n".join(f"{name}: {value}" + for name, value in feedback_fields(manifest, destination)) + + "\n") + machine.set_type("message/feedback-report") + message.attach(machine) + + body = _headers_part(manifest) + if body is not None: + headers = EmailMessage(policy=SMTP) + headers.set_content(body) + headers.set_type("text/rfc822-headers") + message.attach(headers) + + return message.as_string() diff --git a/docs/BACKLOG.md b/docs/BACKLOG.md index 79c0ab1..780cbf5 100644 --- a/docs/BACKLOG.md +++ b/docs/BACKLOG.md @@ -7,6 +7,8 @@ number and gains a status rather than being renumbered. |---|------|------|--------| | 1 | Skip boilerplate namespace URLs | XS | open | | 2 | An IDN indicator resolves to no contact | S | open | +| 3 | Expose kept cases so qtmaildir can tag spam | ? | open, unsized | +| 4 | `Report-Type: phishing` is unverified against x-arf | XS | open | ## 1. Skip boilerplate namespace URLs @@ -67,3 +69,68 @@ where the attacker wants the tool to normalise on their behalf, and a consultant chasing one indicator by hand is a smaller cost than a query made about a name the user never saw. Wait for a real IDN indicator in a sweep before building it. + +## 3. Expose kept cases so qtmaildir can tag spam + +**Source.** The author's idea note, not a defect found in the code. Unlike +items 1 and 2 the cause here has NOT been verified against the code, because +there is nothing built yet to verify: this is a feature request, and it is +recorded unsized on purpose. + +**Observed.** Case directories are permanent by design, so over time they +become a local corpus of messages the user has already judged to be phishing. +Nothing reads them back. The idea is that qtmaildir could ask this tool +whether an incoming message resembles one, and tag it as spam when it does. + +**Approach.** Undecided, and the shape matters more than the code. The +umbrella design already fixes the coupling between the two repositories: the +manifest format and a command name in qtmaildir's config, with no submodule. +A read-only subcommand answering a question about one message fits that +contract; a daemon, a socket or a shared database does not, and the umbrella +design rules out a database of this tool's own. + +**Constraints, and the real tension.** Deciding a message is spam by +resemblance is a classifier, and this tool has so far been deliberately +mechanical: it reports what a message declared, and refuses rather than +guesses when the trust boundary is unset. A resemblance score is the first +thing here that would be an opinion rather than an observation, and a wrong +one either hides real mail or teaches the user to distrust the tag. + +There is also a quieter question about what a match is allowed to be based on. +The obvious signals are the ones already in a manifest, a sending IP, a +domain, a URL shape, an attachment hash. Those are safe. Matching on the +message body would mean holding attacker-supplied text against new mail, and +`source.eml` is unredacted, so anything built here must not become a route by +which a stored recipient identifier reaches a comparison that is later +reported or logged. Property 1 governs what may be published, and a tag is not +a report, but the path from one to the other is short. + +**Before building.** Ask the author what "fits certain requisites" means to +him concretely, since that phrase is doing all the work in the note, and +whether he wants a judgement or only the facts, for instance a subcommand that +answers "this IP appears in three kept cases" and leaves the tagging decision +to qtmaildir. The second is much more in keeping with the rest of the tool. + +## 4. `Report-Type: phishing` is unverified against x-arf + +**Observed.** `report.feedback_fields()` emits `Report-Type: phishing` in the +machine-readable part. Every other field there was verified against RFC 5965 +itself; this one was not, because no primary source for x-arf's own field +semantics could be reached while building it. The abusix README documents only +the v3 to v4 deprecation and does not define the field. + +**Cause.** Not a defect found in the code. The value follows the worked +example in `docs/specs/2026-09-09-report.md`, so it is internally consistent, +and the hybrid envelope means a strict RFC 5965 parser ignores the field +either way (the RFC requires implementors ignore fields they do not support). +The exposure is limited to x-arf tooling reading a field name or value that +does not exist in the version it implements. + +**Approach.** Find a primary source for x-arf v4 field names, confirm or +correct the value, and record what it was checked against. If x-arf turns out +to name the field differently, the fix is one string and one test. + +**Constraints.** Low urgency: nothing here is a leak, and the failure mode is +a field an x-arf parser skips rather than acts on wrongly. Worth doing before +the first real report is filed, so a desk running x-arf tooling gets what it +expects. diff --git a/docs/plans/2026-09-09-report.md b/docs/plans/2026-09-09-report.md new file mode 100644 index 0000000..7e30e6e --- /dev/null +++ b/docs/plans/2026-09-09-report.md @@ -0,0 +1,1536 @@ +# abusectl `report` Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Build `abusectl report <case>`, which turns a case's IOCs and abuse +contacts into per-destination report bodies the user reviews before anything +is sent. + +**Architecture:** One new pure module, `report.py`, plus a whitelisted +`headers` block added to `parse.py` so `report` never opens `source.eml`. The +report is a `multipart/report` MIME document built with `email.message` from +the standard library. Grouping is per abuse address; a case with any sent +destination is frozen and cannot be regenerated. + +**Tech Stack:** Python 3.11+, standard library only (`email`, `hashlib`, +`json`, `unittest`). No new dependency, and `requirements.txt` is untouched. + +**Read first:** `docs/specs/2026-09-09-report.md` is the spec this implements, +and `AGENTS.md` states the four properties that must not be weakened. Task 1 +touches `parse.py`, which is the module the first property lives in. + +--- + +## File structure + +| File | Responsibility | Task | +|---|---|---| +| `abusectl/parse.py` | gains `report_headers()`, a whitelist extractor | 1 | +| `abusectl/report.py` | new: grouping, bodies, destinations, freeze check | 2-8 | +| `abusectl/config.py` | gains the `[reporter]` section | 9 | +| `abusectl/init.py` | gains three reporter prompts | 10 | +| `abusectl/cli.py` | gains the `report` subcommand dispatch | 11 | +| `tests/test_parse.py` | header whitelist tests | 1 | +| `tests/test_report.py` | new: everything in `report.py` | 2-8 | +| `tests/test_config.py` | reporter section tests | 9 | +| `tests/test_cli.py` | dispatch and exit codes | 11 | +| `tests/fixtures/reportable.eml` | new fixture with recipient headers present | 1 | + +`report.py` is one module, not two. The spec says why: there is no protocol +layer to split off, because `email.message` already is that layer. + +--- + +### Task 1: `parse.report_headers()`, the whitelist + +The spec's second decision: `parse` stores the whitelisted headers in the +manifest so `report` never opens `source.eml`. The test that matters is the +one proving `To` is DROPPED. + +**Files:** +- Create: `tests/fixtures/reportable.eml` +- Modify: `abusectl/parse.py` (add `_REPORT_HEADERS` and `report_headers()`) +- Test: `tests/test_parse.py` + +- [ ] **Step 1: Create the fixture** + +It must carry the headers the whitelist keeps AND the ones it must drop. +Check the weekday before writing it: `date -d 2026-09-07 +%A` returns +`Monday`. An RFC2822 parser validates the day against the date and a wrong +one reads as a malformed header. + +Create `tests/fixtures/reportable.eml`: + +``` +Received: from relay.example.org (relay.example.org [192.0.2.10]) + by mx.example.org with ESMTP id abc123 + for <you@example.org>; Mon, 07 Sep 2026 09:12:44 +0000 +Received: from sender.invalid (sender.invalid [203.0.113.42]) + by relay.example.org with ESMTP id def456; + Mon, 07 Sep 2026 09:12:40 +0000 +Return-Path: <bounce@sender.invalid> +Authentication-Results: mx.example.org; spf=fail; dkim=none; dmarc=fail +Received-SPF: fail (mx.example.org: domain of sender.invalid does not designate 203.0.113.42) +From: "Example Bank" <phish@sender.invalid> +To: victim@example.org +Cc: colleague@example.org +Delivered-To: victim@example.org +X-Original-To: victim@example.org +Reply-To: "Support" <reply@sender.invalid> +Subject: Your account requires verification +Date: Mon, 07 Sep 2026 09:12:40 +0000 +Message-ID: <case-one@sender.invalid> +MIME-Version: 1.0 +Content-Type: text/plain; charset=utf-8 + +Please verify at http://login.sender.invalid/verify?id=abc123 +``` + +- [ ] **Step 2: Write the failing tests** + +Add to `tests/test_parse.py`: + +```python +class ReportHeaders(unittest.TestCase): + def setUp(self): + self.raw = (FIXTURES / "reportable.eml").read_bytes() + + def test_the_whitelist_keeps_what_a_desk_needs(self): + headers = parse.report_headers(self.raw, trusted=["192.0.2.0/24"]) + names = [name for name, _ in headers] + self.assertIn("From", names) + self.assertIn("Subject", names) + self.assertIn("Date", names) + self.assertIn("Message-ID", names) + self.assertIn("Reply-To", names) + self.assertIn("Return-Path", names) + self.assertIn("Authentication-Results", names) + self.assertIn("Received-SPF", names) + + def test_recipient_headers_never_survive_the_whitelist(self): + headers = parse.report_headers(self.raw, trusted=["192.0.2.0/24"]) + blob = repr(headers) + for name in ("To", "Cc", "Delivered-To", "X-Original-To"): + self.assertNotIn(name, [n for n, _ in headers]) + self.assertNotIn("victim@example.org", blob) + self.assertNotIn("colleague@example.org", blob) + + def test_received_stops_at_the_boundary_hop(self): + # 192.0.2.10 is ours, so its Received line is our own infrastructure + # and must not be published; the hop below it is the one being + # reported and is kept. + headers = parse.report_headers(self.raw, trusted=["192.0.2.0/24"]) + received = [value for name, value in headers if name == "Received"] + self.assertEqual(len(received), 1) + self.assertIn("203.0.113.42", received[0]) + self.assertNotIn("mx.example.org with ESMTP id abc123", received[0]) + + def test_a_forged_chain_publishes_no_hop_below_the_boundary(self): + raw = (FIXTURES / "forged-chain.eml").read_bytes() + headers = parse.report_headers(raw, trusted=["192.0.2.0/24"]) + received = [value for name, value in headers if name == "Received"] + self.assertTrue(all("198.51.100.7" not in value for value in received)) +``` + +The last test is the one that protects an innocent party, the same job +`test_a_forged_chain_stops_at_the_first_untrusted_hop` already does for +`sending_ip()`. Read that test first and match its trusted-relay argument to +the fixture it uses; if `forged-chain.eml` uses a different boundary network, +use that one here rather than `192.0.2.0/24`. + +- [ ] **Step 3: Run to verify they fail** + +Run: `python3 -m unittest tests.test_parse.ReportHeaders -v` +Expected: FAIL, `AttributeError: module 'abusectl.parse' has no attribute 'report_headers'` + +- [ ] **Step 4: Implement** + +Add to `abusectl/parse.py`, near the other module constants: + +```python +# The headers that may appear in a published report. A WHITELIST, never a +# blacklist: a blacklist means every header this parser learns to read later +# is a leak waiting for someone to remember. To, Cc, Delivered-To and +# X-Original-To are absent by construction, which is the same reason iocs() +# does not read them either. +_REPORT_HEADERS = ( + "From", + "Subject", + "Date", + "Message-ID", + "Reply-To", + "Return-Path", + "Authentication-Results", + "Received-SPF", + "MIME-Version", + "Content-Type", +) +``` + +And the function, next to `received_hops()`: + +```python +def report_headers(raw: bytes, trusted: list[str]) -> list[tuple[str, str]]: + """Return the headers that may be published, outermost Received first. + + Received is truncated at the trust boundary: our own relays are our + infrastructure and publishing them tells a third party about the user's + mail path, so only the boundary hop and below are kept. Everything else + comes from a fixed whitelist. + + Returned as a list of pairs rather than a dict because Received repeats + and order carries meaning. + """ + message = _message(raw) + result: list[tuple[str, str]] = [] + + for value in message.get_all("received") or []: + ip = _extract_ip(str(value)) + if ip is not None and _in_any(ip, trusted): + continue + result.append(("Received", str(value))) + + for name in _REPORT_HEADERS: + value = message.get(name) + if value is not None: + result.append((name, str(value))) + + return result +``` + +- [ ] **Step 5: Run to verify they pass** + +Run: `python3 -m unittest tests.test_parse -v` +Expected: PASS, and every pre-existing parse test still passes. + +- [ ] **Step 6: Wire it into the manifest** + +In `abusectl/cli.py`, in the parse command around line 264, add the third +line: + +```python + manifest["iocs"] = parse_module.iocs(raw, trusted=settings.trusted_relays) + manifest["auth"] = parse_module.auth_results(raw) + manifest["headers"] = parse_module.report_headers( + raw, trusted=settings.trusted_relays + ) +``` + +- [ ] **Step 7: Run the whole suite** + +Run: `python3 -m unittest discover tests` +Expected: PASS, no regressions. + +- [ ] **Step 8: Commit** + +```bash +git add abusectl/parse.py abusectl/cli.py tests/test_parse.py tests/fixtures/reportable.eml +git commit -S -m "feat: store a whitelist of publishable headers in the manifest + +report must never open source.eml, so parse decides once what may be +published and report formats only what it is given. Received is truncated at +the trust boundary; To, Cc, Delivered-To and X-Original-To are absent by +construction rather than stripped." +``` + +--- + +### Task 2: Group contacts into destinations, one per abuse address + +**Files:** +- Create: `abusectl/report.py` +- Test: `tests/test_report.py` + +- [ ] **Step 1: Write the failing test** + +Create `tests/test_report.py`: + +```python +import unittest + +from abusectl import report + + +class Grouping(unittest.TestCase): + def test_two_contacts_at_one_address_become_one_destination(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["target"], "abuse@host.invalid") + self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"]) + + def test_a_contact_with_two_addresses_reaches_both_desks(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["a@host.invalid", "b@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual( + sorted(d["target"] for d in destinations), + ["a@host.invalid", "b@host.invalid"], + ) + + def test_a_contact_with_no_address_creates_no_destination(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + ] + self.assertEqual(report.email_destinations(contacts), []) + + def test_destinations_carry_stable_ids_and_pending_status(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(contacts)[0] + self.assertEqual(destination["id"], "email-1") + self.assertEqual(destination["kind"], "email") + self.assertEqual(destination["status"], "pending") +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_report.Grouping -v` +Expected: FAIL, `ModuleNotFoundError: No module named 'abusectl.report'` + +- [ ] **Step 3: Implement** + +Create `abusectl/report.py` with the GPLv2 header used by every other module +in this package (copy the fourteen-line block from the top of +`abusectl/case.py`, changing nothing but what follows it), then: + +```python +"""IOCs and abuse contacts to report bodies: the last step before anything +irreversible happens. + +This module is PURE and OFFLINE. It opens no socket, sends no mail and reads +no file outside the case directory. What it produces is a document the user +reads, edits and approves, so the output is written for a human first and a +parser second. + +It takes the reporting identity as an ARGUMENT rather than reading the +config, the way parse.py takes the trust boundary. The identity is the one +thing in a report disclosed deliberately, and a module that reaches for it +itself is a module that can disclose it in a code path nobody reviewed. +""" + +import hashlib + + +def email_destinations(contacts: list[dict]) -> list[dict]: + """Group contacts into one destination per abuse ADDRESS. + + Contacts already fold by host, but two different contacts can still + resolve to the same address, an IP and a domain at one hoster being the + common case. One mail per address rather than per contact is what stops + a desk receiving two mails about one incident. + """ + by_address: dict[str, list[str]] = {} + + for contact in contacts: + for address in contact.get("abuse", []): + iocs = by_address.setdefault(address, []) + for ioc in contact.get("iocs", []): + if ioc not in iocs: + iocs.append(ioc) + + return [ + { + "id": f"email-{index}", + "kind": "email", + "target": address, + "iocs": iocs, + "body": None, + "status": "pending", + } + for index, (address, iocs) in enumerate(by_address.items(), start=1) + ] +``` + +- [ ] **Step 4: Run to verify it passes** + +Run: `python3 -m unittest tests.test_report.Grouping -v` +Expected: PASS, 4 tests. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/report.py tests/test_report.py +git commit -S -m "feat: group abuse contacts into one destination per address + +Two contacts can resolve to the same desk, an IP and a domain at one hoster +being the common case, and grouping per contact would send that desk two +mails about one incident." +``` + +--- + +### Task 3: The unreportable list + +**Files:** +- Modify: `abusectl/report.py` +- Test: `tests/test_report.py` + +- [ ] **Step 1: Write the failing test** + +```python +class Unreportable(unittest.TestCase): + def test_an_ioc_with_no_desk_is_listed_with_its_reason(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2", "ioc-3"], "query": "example.invalid", + "abuse": [], "source": "rdap", + "error": "no abuse role published"}, + ] + self.assertEqual( + report.unreportable(contacts), + [ + {"ioc": "ioc-2", "reason": "no abuse role published"}, + {"ioc": "ioc-3", "reason": "no abuse role published"}, + ], + ) + + def test_a_missing_reason_still_produces_an_entry(self): + contacts = [{"iocs": ["ioc-9"], "query": "x.invalid", "abuse": [], + "source": "rdap"}] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-9", "reason": "no abuse address resolved"}], + ) + + def test_nothing_unreportable_is_an_empty_list_not_an_error(self): + contacts = [{"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}] + self.assertEqual(report.unreportable(contacts), []) +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_report.Unreportable -v` +Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'unreportable'` + +- [ ] **Step 3: Implement** + +Add to `abusectl/report.py`: + +```python +def unreportable(contacts: list[dict]) -> list[dict]: + """List every IOC that reached no email destination, with the reason. + + A missing contact is a normal outcome, not an error: RDAP publishes no + abuse role for many netblocks. Making it visible is what keeps review + honest, since finding it any other way means diffing the IOC list + against every destination's IOC list. Same instinct as + suspect_path_segments flagging rather than redacting. + """ + result = [] + for contact in contacts: + if contact.get("abuse"): + continue + reason = contact.get("error", "no abuse address resolved") + for ioc in contact.get("iocs", []): + result.append({"ioc": ioc, "reason": reason}) + return result +``` + +- [ ] **Step 4: Run to verify it passes** + +Run: `python3 -m unittest tests.test_report.Unreportable -v` +Expected: PASS, 3 tests. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/report.py tests/test_report.py +git commit -S -m "feat: list the indicators no abuse desk was found for + +Not an error and not an exit code: a case where nothing resolved still +reaches MISP and the vendors. Visible beats absent, so review can see it +without diffing IOC lists." +``` + +--- + +### Task 4: The plain-text part + +**Files:** +- Modify: `abusectl/report.py` +- Test: `tests/test_report.py` + +The wording is NOT asserted; the spec says whether it reads well to a desk has +no assertion and is hand-tested. What is asserted is what must and must not +appear. + +- [ ] **Step 1: Write the failing test** + +```python +IDENTITY = {"name": "A Reporter", "org": "Example Consulting", + "email": "reporter@example.org"} + +MANIFEST = { + "format": 1, + "case_id": "2026-09-07-aaaa", + "iocs": [ + {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", + "origin": "received-chain", "confidence": "boundary-hop"}, + {"id": "ioc-2", "type": "url", + "value": "http://login.sender.invalid/verify?id=REDACTED", + "origin": "body"}, + ], + "auth": {"spf": "fail", "dkim": "none", "dmarc": "fail"}, + "headers": [ + ("From", '"Example Bank" <phish@sender.invalid>'), + ("Subject", "Your account requires verification"), + ("Date", "Mon, 07 Sep 2026 09:12:40 +0000"), + ], + "contacts": [ + {"iocs": ["ioc-1", "ioc-2"], "query": "203.0.113.42", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ], +} + + +class TextPart(unittest.TestCase): + def setUp(self): + destination = report.email_destinations(MANIFEST["contacts"])[0] + self.text = report.text_part(MANIFEST, destination, IDENTITY) + + def test_the_redaction_note_is_always_present(self): + self.assertIn("Recipient identifiers", self.text) + + def test_the_reporter_identity_appears(self): + self.assertIn("A Reporter", self.text) + self.assertIn("Example Consulting", self.text) + self.assertIn("reporter@example.org", self.text) + + def test_the_destinations_own_indicators_appear(self): + self.assertIn("203.0.113.42", self.text) + self.assertIn("http://login.sender.invalid/verify?id=REDACTED", self.text) + + def test_an_indicator_belonging_to_another_desk_does_not_appear(self): + manifest = dict(MANIFEST) + manifest["iocs"] = MANIFEST["iocs"] + [ + {"id": "ioc-9", "type": "ipv4", "value": "192.0.2.99", + "origin": "received-chain"}, + ] + destination = report.email_destinations(MANIFEST["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + self.assertNotIn("192.0.2.99", text) + + def test_no_line_exceeds_seventy_two_columns(self): + for line in self.text.splitlines(): + self.assertLessEqual(len(line), 72, line) +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_report.TextPart -v` +Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'text_part'` + +- [ ] **Step 3: Implement** + +Add to `abusectl/report.py`: + +```python +_REDACTION_NOTE = ( + "Recipient identifiers have been removed from this report by policy.\n" + "Parameter names are preserved, parameter values are not. Full evidence\n" + "is retained locally and is available on request." +) + + +def _iocs_by_id(manifest: dict) -> dict: + return {entry["id"]: entry for entry in manifest.get("iocs", [])} + + +def text_part(manifest: dict, destination: dict, identity: dict) -> str: + """Build the human-readable part: the one that decides whether a desk + acts on the report. + + The ask goes first, because a desk triaging a queue must know in one + line what happened and what is wanted. Only this destination's own + indicators appear: a desk shown three IPs that are not theirs stops + reading. + """ + by_id = _iocs_by_id(manifest) + mine = [by_id[i] for i in destination["iocs"] if i in by_id] + + lines = [ + "Phishing message reported: infrastructure on your network was", + "used to send or host it. Requesting takedown and customer", + "notification.", + "", + "Observed on your infrastructure:", + "", + ] + + for entry in mine: + lines.append(f" {entry['value']}") + origin = entry.get("origin", "") + if entry.get("confidence") == "boundary-hop": + lines.append(" sending IP, first hop outside our boundary") + elif origin: + lines.append(f" seen in: {origin}") + + headers = manifest.get("headers") or [] + shown = [(name, value) for name, value in headers + if name in ("Date", "From", "Subject")] + if shown: + lines += ["", "Message as declared:", ""] + for name, value in shown: + lines.append(f" {name}: {value}"[:72]) + + auth = manifest.get("auth") or {} + if auth: + lines += ["", "Authentication results:", ""] + lines.append(" " + " ".join( + f"{key.upper()}: {value}" for key, value in sorted(auth.items()) + )[:70]) + + lines += ["", _REDACTION_NOTE, ""] + lines.append( + f"Reported by: {identity['name']}, {identity['org']} " + f"<{identity['email']}>"[:72] + ) + lines.append("Generated by abusectl.") + + return "\n".join(lines) + "\n" +``` + +- [ ] **Step 4: Run to verify it passes** + +Run: `python3 -m unittest tests.test_report.TextPart -v` +Expected: PASS, 5 tests. If the 72-column test fails on a long URL, that is a +real finding rather than a test to relax: wrap the URL onto its own line +rather than truncating it, because a truncated URL is a wrong indicator. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/report.py tests/test_report.py +git commit -S -m "feat: build the human-readable part of a report + +The ask goes first, only this desk's own indicators appear, and the redaction +note is unconditional: a desk seeing REDACTED with no explanation may read +the report as doctored." +``` + +--- + +### Task 5: The machine-readable part + +**Files:** +- Modify: `abusectl/report.py` +- Test: `tests/test_report.py` + +- [ ] **Step 1: Write the failing test** + +```python +class FeedbackPart(unittest.TestCase): + def setUp(self): + destination = report.email_destinations(MANIFEST["contacts"])[0] + self.fields = report.feedback_fields(MANIFEST, destination) + self.lookup = dict(self.fields) + + def test_the_three_rfc5965_required_fields_are_present(self): + self.assertEqual(self.lookup["Feedback-Type"], "abuse") + self.assertEqual(self.lookup["Version"], "1") + self.assertTrue(self.lookup["User-Agent"].startswith("abusectl/")) + + def test_the_xarf_report_type_is_phishing(self): + self.assertEqual(self.lookup["Report-Type"], "phishing") + + def test_source_is_the_primary_indicator(self): + self.assertEqual(self.lookup["Source"], "203.0.113.42") + self.assertEqual(self.lookup["Source-IP"], "203.0.113.42") + + def test_every_url_appears_as_a_reported_uri(self): + uris = [value for name, value in self.fields if name == "Reported-Uri"] + self.assertEqual( + uris, ["http://login.sender.invalid/verify?id=REDACTED"] + ) + + def test_a_destination_with_no_ip_omits_source_ip(self): + contacts = [{"iocs": ["ioc-2"], "query": "sender.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}] + destination = report.email_destinations(contacts)[0] + lookup = dict(report.feedback_fields(MANIFEST, destination)) + self.assertNotIn("Source-IP", lookup) + self.assertEqual( + lookup["Source"], "http://login.sender.invalid/verify?id=REDACTED" + ) +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_report.FeedbackPart -v` +Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'feedback_fields'` + +- [ ] **Step 3: Implement** + +Add to `abusectl/report.py`, with the version constant near the top of the +module: + +```python +VERSION = "0.1.0" + + +def feedback_fields(manifest: dict, destination: dict) -> list[tuple[str, str]]: + """Build the machine-readable part: an RFC 5965 envelope carrying x-arf + fields inside it. + + RFC 5965 is the standard and is universally understood, but it was + designed for feedback loops, where a report is about A MESSAGE. These + reports are about INDICATORS, and 5965 has no field for "this specific + host is the thing being reported". x-arf's Source does. The part is + key/value, so a 5965 parser reads what it knows and ignores the rest. + + Returned as pairs, not a dict: Reported-Uri repeats. + """ + by_id = _iocs_by_id(manifest) + mine = [by_id[i] for i in destination["iocs"] if i in by_id] + + ips = [e["value"] for e in mine if e.get("type") in ("ipv4", "ipv6")] + urls = [e["value"] for e in mine if e.get("type") == "url"] + domains = [e["value"] for e in mine if e.get("type") == "domain"] + + # Source is singular, so the primary indicator fills it and the rest + # travel in repeated fields and in the text part. An IP is the most + # actionable thing a hosting desk can act on, so it wins when present. + primary = (ips or domains or urls or [""])[0] + + fields = [ + ("Feedback-Type", "abuse"), + ("User-Agent", f"abusectl/{VERSION}"), + ("Version", "1"), + ("Report-Type", "phishing"), + ("Source", primary), + ] + + for ip in ips: + fields.append(("Source-IP", ip)) + for domain in domains: + fields.append(("Reported-Domain", domain)) + for url in urls: + fields.append(("Reported-Uri", url)) + + for name, value in manifest.get("headers") or []: + if name == "Date": + fields.append(("Arrival-Date", value)) + break + + return fields +``` + +- [ ] **Step 4: Run to verify it passes** + +Run: `python3 -m unittest tests.test_report.FeedbackPart -v` +Expected: PASS, 5 tests. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/report.py tests/test_report.py +git commit -S -m "feat: build the machine-readable feedback report part + +An RFC 5965 envelope carrying x-arf fields. 5965 reports are about a message +and these are about indicators, so x-arf's Source fills the gap while the +envelope keeps a standards parser working." +``` + +--- + +### Task 6: Assemble the MIME document + +**Files:** +- Modify: `abusectl/report.py` +- Test: `tests/test_report.py` + +- [ ] **Step 1: Write the failing test** + +```python +import email +import email.policy + + +class Document(unittest.TestCase): + def setUp(self): + self.destination = report.email_destinations(MANIFEST["contacts"])[0] + self.raw = report.build(MANIFEST, self.destination, IDENTITY) + self.parsed = email.message_from_string( + self.raw, policy=email.policy.default + ) + + def test_it_is_a_feedback_report_with_three_parts(self): + self.assertEqual(self.parsed.get_content_type(), "multipart/report") + self.assertEqual(self.parsed.get_param("report-type"), "feedback-report") + parts = list(self.parsed.iter_parts()) + self.assertEqual( + [part.get_content_type() for part in parts], + ["text/plain", "message/feedback-report", "text/rfc822-headers"], + ) + + def test_the_envelope_is_addressed_and_identified(self): + self.assertEqual(self.parsed["To"], "abuse@host.invalid") + self.assertIn("reporter@example.org", self.parsed["From"]) + self.assertTrue(self.parsed["Subject"]) + + def test_the_headers_part_carries_no_recipient_header(self): + headers_part = list(self.parsed.iter_parts())[2] + body = headers_part.get_content() + for name in ("To:", "Cc:", "Delivered-To:", "X-Original-To:"): + self.assertNotIn(name, body) + + def test_the_source_message_is_never_attached(self): + self.assertNotIn("message/rfc822", self.raw) +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_report.Document -v` +Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'build'` + +- [ ] **Step 3: Implement** + +Add the imports at the top of `abusectl/report.py`: + +```python +from email.message import EmailMessage +from email.policy import SMTP +``` + +And the function: + +```python +def build(manifest: dict, destination: dict, identity: dict) -> str: + """Assemble one destination's report as an RFC 5965 MIME document. + + Three parts: what a human reads, what a parser reads, and the headers. + The original message is NOT attached, and there is no message/rfc822 + part: source.eml carries every identifier the first property exists to + keep out, and an abuse desk forwards a report to the abused customer, + who for a phishing domain may be the attacker. RFC 5965 provides + text/rfc822-headers for exactly this case, so this is the standard's own + answer rather than a deviation from it. + """ + message = EmailMessage(policy=SMTP) + message["From"] = f"{identity['name']} <{identity['email']}>" + message["To"] = destination["target"] + message["Subject"] = ( + f"Abuse report: phishing infrastructure, case {manifest['case_id']}" + ) + message.make_mixed() + message.set_type("multipart/report") + message.set_param("report-type", "feedback-report") + + human = EmailMessage(policy=SMTP) + human.set_content(text_part(manifest, destination, identity)) + message.attach(human) + + machine = EmailMessage(policy=SMTP) + machine.set_content( + "\n".join(f"{name}: {value}" + for name, value in feedback_fields(manifest, destination)) + + "\n" + ) + machine.set_type("message/feedback-report") + message.attach(machine) + + headers = EmailMessage(policy=SMTP) + headers.set_content( + "\n".join(f"{name}: {value}" + for name, value in manifest.get("headers") or []) + + "\n" + ) + headers.set_type("text/rfc822-headers") + message.attach(headers) + + return message.as_string() +``` + +- [ ] **Step 4: Run to verify it passes** + +Run: `python3 -m unittest tests.test_report.Document -v` +Expected: PASS, 4 tests. If `set_type` on a subpart raises, set the type +BEFORE `set_content` on that part and re-run; the ordering matters in +`email.message`. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/report.py tests/test_report.py +git commit -S -m "feat: assemble the RFC 5965 report document + +Three parts and no message/rfc822: the original carries every identifier the +first property keeps out, and text/rfc822-headers is the standard's own +answer for a report that cannot include the message." +``` + +--- + +### Task 7: Body hashes and the frozen case + +**Files:** +- Modify: `abusectl/report.py` +- Test: `tests/test_report.py` + +- [ ] **Step 1: Write the failing test** + +```python +class Freeze(unittest.TestCase): + def test_a_frozen_case_refuses(self): + manifest = dict(MANIFEST) + manifest["frozen"] = {"at": "2026-09-07T10:00:00Z", "by": "abusedb"} + with self.assertRaises(report.Frozen) as caught: + report.check_regenerable(manifest, modified=[]) + self.assertIn("abusedb", str(caught.exception)) + + def test_a_frozen_case_refuses_even_when_forced(self): + manifest = dict(MANIFEST) + manifest["frozen"] = {"at": "2026-09-07T10:00:00Z", "by": "abusedb"} + with self.assertRaises(report.Frozen): + report.check_regenerable(manifest, modified=[], force=True) + + def test_a_modified_body_refuses_without_force(self): + with self.assertRaises(report.Modified) as caught: + report.check_regenerable(MANIFEST, modified=["email-1"]) + self.assertIn("email-1", str(caught.exception)) + + def test_a_modified_body_is_allowed_with_force(self): + report.check_regenerable(MANIFEST, modified=["email-1"], force=True) + + def test_an_untouched_case_regenerates(self): + report.check_regenerable(MANIFEST, modified=[]) + + +class Hashes(unittest.TestCase): + def test_the_hash_detects_a_changed_body(self): + first = report.body_hash("one") + self.assertNotEqual(first, report.body_hash("two")) + self.assertEqual(first, report.body_hash("one")) +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_report.Freeze tests.test_report.Hashes -v` +Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'Frozen'` + +- [ ] **Step 3: Implement** + +Add to `abusectl/report.py`: + +```python +class Frozen(Exception): + """Raised when a case has already been reported to at least one desk. + + There is no force override. The destinations are not independent + artifacts, they are one incident reported in parallel: regenerating one + body after another desk holds the report leaves two desks with + contradictory accounts of the same case. + """ + + +class Modified(Exception): + """Raised when a body was edited after generation and --force was not + given. Silently discarding a review that took twenty minutes is what + makes a tool untrustworthy. + """ + + +def body_hash(text: str) -> str: + """Hash a body's CONTENT, which is what the question actually is. + + An mtime is a poor witness in both directions: a git checkout, an rsync + or a backup restore all move it with no human having edited anything, + and an editor that preserves mtime hides a real edit. For a destination + that has been sent, this hash is also the record of what was disclosed. + """ + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def check_regenerable(manifest: dict, modified: list[str], + force: bool = False) -> None: + """Raise unless this case may be regenerated. Returns None when it may. + + Frozen wins over force: once a desk holds the report, the case is an + evidence record rather than a draft. + """ + frozen = manifest.get("frozen") + if frozen: + raise Frozen( + f"case reported to {frozen.get('by', 'a destination')} at " + f"{frozen.get('at', 'an unknown time')}; bodies cannot be " + f"regenerated. Edit a body by hand if it must change." + ) + + if modified and not force: + raise Modified( + "bodies edited since generation: " + ", ".join(modified) + + ". Re-run with --force to discard those edits; a timestamped " + "backup is kept." + ) +``` + +- [ ] **Step 4: Run to verify it passes** + +Run: `python3 -m unittest tests.test_report.Freeze tests.test_report.Hashes -v` +Expected: PASS, 6 tests. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/report.py tests/test_report.py +git commit -S -m "feat: freeze a reported case and detect edited bodies + +Content hash rather than mtime, because mtime is wrong in both directions. +Any sent destination freezes the whole case with no override: two desks +holding contradictory accounts of one incident is worse than a stale body." +``` + +--- + +### Task 8: Write the bodies to the case directory + +**Files:** +- Modify: `abusectl/report.py` +- Test: `tests/test_report.py` + +- [ ] **Step 1: Write the failing test** + +```python +import tempfile +from pathlib import Path + + +class Writing(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = Path(self.tmp.name) + self.addCleanup(self.tmp.cleanup) + + def test_it_writes_one_body_per_destination_and_records_the_hash(self): + manifest = report.generate(dict(MANIFEST), self.path, IDENTITY) + destination = manifest["destinations"][0] + body = self.path / destination["body"] + self.assertTrue(body.exists()) + self.assertEqual( + destination["body_sha256"], report.body_hash(body.read_text()) + ) + + def test_it_records_the_unreportable_indicators(self): + manifest = dict(MANIFEST) + manifest["contacts"] = MANIFEST["contacts"] + [ + {"iocs": ["ioc-4"], "query": "x.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + ] + result = report.generate(manifest, self.path, IDENTITY) + self.assertEqual( + result["unreportable"], + [{"ioc": "ioc-4", "reason": "no abuse role published"}], + ) + + def test_a_modified_body_is_backed_up_before_being_overwritten(self): + manifest = report.generate(dict(MANIFEST), self.path, IDENTITY) + body = self.path / manifest["destinations"][0]["body"] + body.write_text("hand edited during review\n") + + with self.assertRaises(report.Modified): + report.generate(manifest, self.path, IDENTITY) + + report.generate(manifest, self.path, IDENTITY, force=True) + backups = list((self.path / "bodies").glob("*.orig")) + self.assertEqual(len(backups), 1) + self.assertEqual(backups[0].read_text(), "hand edited during review\n") + + def test_a_deleted_body_regenerates_without_complaint(self): + manifest = report.generate(dict(MANIFEST), self.path, IDENTITY) + (self.path / manifest["destinations"][0]["body"]).unlink() + again = report.generate(manifest, self.path, IDENTITY) + self.assertTrue((self.path / again["destinations"][0]["body"]).exists()) +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_report.Writing -v` +Expected: FAIL, `AttributeError: module 'abusectl.report' has no attribute 'generate'` + +- [ ] **Step 3: Implement** + +Add the imports: + +```python +from datetime import datetime, timezone +from pathlib import Path +``` + +And the function: + +```python +def generate(manifest: dict, case_path: Path, identity: dict, + force: bool = False) -> dict: + """Write every body and return the manifest with destinations[] set. + + The manifest is RETURNED rather than saved: case.py is the only writer + of a case directory, so the caller saves. The bodies are this module's + to write because they are not the manifest. + """ + case_path = Path(case_path) + bodies = case_path / "bodies" + bodies.mkdir(exist_ok=True) + + modified = [] + for existing in manifest.get("destinations") or []: + recorded = existing.get("body_sha256") + if not recorded or not existing.get("body"): + continue + path = case_path / existing["body"] + if path.exists() and body_hash(path.read_text()) != recorded: + modified.append(existing["id"]) + + check_regenerable(manifest, modified, force=force) + + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + for destination_id in modified: + existing = next(d for d in manifest["destinations"] + if d["id"] == destination_id) + path = case_path / existing["body"] + path.rename(path.with_suffix(path.suffix + f".{stamp}.orig")) + + destinations = email_destinations(manifest.get("contacts", [])) + for destination in destinations: + text = build(manifest, destination, identity) + relative = f"bodies/{destination['id']}.xarf" + (case_path / relative).write_text(text) + destination["body"] = relative + destination["body_sha256"] = body_hash(text) + + manifest["destinations"] = destinations + manifest["unreportable"] = unreportable(manifest.get("contacts", [])) + return manifest +``` + +- [ ] **Step 4: Run to verify it passes** + +Run: `python3 -m unittest tests.test_report -v` +Expected: PASS, every class in the module. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/report.py tests/test_report.py +git commit -S -m "feat: write report bodies into the case directory + +case.py stays the only writer of the manifest, so generate returns it and +the caller saves. A modified body is backed up with a timestamp before +--force overwrites it." +``` + +--- + +### Task 9: The `[reporter]` config section + +**Files:** +- Modify: `abusectl/config.py` +- Test: `tests/test_config.py` + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_config.py`, following the file's existing pattern for +writing a temporary config (copy the helper the neighbouring tests use rather +than inventing one): + +```python +class Reporter(unittest.TestCase): + def test_the_reporter_identity_is_read(self): + settings = self._load(""" + [general] + trusted_relays = ["192.0.2.0/24"] + + [reporter] + name = "A Reporter" + org = "Example Consulting" + email = "reporter@example.org" + """) + self.assertEqual(settings.reporter["name"], "A Reporter") + self.assertEqual(settings.reporter["email"], "reporter@example.org") + + def test_an_absent_reporter_section_is_an_empty_dict_not_a_crash(self): + settings = self._load(""" + [general] + trusted_relays = ["192.0.2.0/24"] + """) + self.assertEqual(settings.reporter, {}) + + def test_an_empty_value_is_treated_as_absent(self): + settings = self._load(""" + [general] + trusted_relays = ["192.0.2.0/24"] + + [reporter] + name = "A Reporter" + org = "" + email = "reporter@example.org" + """) + self.assertNotIn("org", settings.reporter) +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_config.Reporter -v` +Expected: FAIL, `AttributeError: 'Config' object has no attribute 'reporter'` + +- [ ] **Step 3: Implement** + +In `abusectl/config.py`, add the field to the dataclass: + +```python +@dataclass(frozen=True) +class Config: + trusted_relays: list[str] + cases: pathlib.Path + reporter: dict +``` + +And in `load()`, before the `Config(...)` construction: + +```python + # A skipped answer is ABSENT, never an empty string: "" reads as + # configured-and-broken and produces a confusing failure much later, + # while absent reads as not-configured and the part that wants it can + # say so plainly. Same rule the rest of this file follows. + raw_reporter = data.get("reporter", {}) + reporter = { + key: value + for key, value in raw_reporter.items() + if isinstance(value, str) and value.strip() + } +``` + +Then pass `reporter=reporter` into the returned `Config`. Every other +construction of `Config` in the codebase and in the tests needs the new +field; run the full suite in step 4 to find them. + +- [ ] **Step 4: Run the whole suite** + +Run: `python3 -m unittest discover tests` +Expected: PASS. Fix any `Config()` construction the new field broke. + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/config.py tests/test_config.py +git commit -S -m "feat: read the reporter identity from config + +The reporter's identity is the one identifier this tool discloses +deliberately, so it comes from config only and parse never supplies it. An +empty value is absent, the same rule the rest of the config follows." +``` + +--- + +### Task 10: Three `init` prompts + +**Files:** +- Modify: `abusectl/init.py` + +The prompts are hand-tested, not unit-tested: the spec and `AGENTS.md` both +say whether a question reads clearly has no assertion. What IS tested is the +builder, if `init.py` has one that produces TOML. + +- [ ] **Step 1: Read the existing prompt flow** + +Run: `grep -n "def \|input(" abusectl/init.py` + +Follow the shape already there. Validate each answer AT the prompt that asked +for it and re-ask on a bad one, rather than erroring after the next question: +`AGENTS.md` records that a hand test found four defects of exactly that shape. + +- [ ] **Step 2: Add the three prompts** + +Ask for name, organisation and email, each skippable. A skipped answer must +be ABSENT from the generated TOML, never `""`. Emit the section only if at +least one answer was given. + +- [ ] **Step 3: Extend the builder test if one exists** + +If `tests/test_init.py` asserts on generated TOML, add a case that a skipped +reporter answer produces no key, matching the existing skipped-answer tests. + +Run: `python3 -m unittest tests.test_init -v` +Expected: PASS. + +- [ ] **Step 4: Hand test** + +Run: `python3 -m abusectl init --force` in a scratch `XDG_CONFIG_HOME` and +read the questions. This is the test that matters for prompts. + +```bash +XDG_CONFIG_HOME=$(mktemp -d) python3 -m abusectl init +``` + +- [ ] **Step 5: Commit** + +```bash +git add abusectl/init.py tests/test_init.py +git commit -S -m "feat: ask for the reporter identity during init + +Each answer is validated at the prompt that asked for it, and a skipped +answer is absent from the file rather than an empty string." +``` + +--- + +### Task 11: The `report` subcommand + +**Files:** +- Modify: `abusectl/cli.py` +- Test: `tests/test_cli.py` + +- [ ] **Step 1: Write the failing test** + +Follow the pattern `tests/test_cli.py` already uses for the contacts command. + +```python +class ReportCommand(unittest.TestCase): + def test_a_missing_case_is_an_error_not_a_traceback(self): + code = cli.main(["report", "/nonexistent/case"]) + self.assertEqual(code, cli.EXIT_ERROR) + + def test_a_case_with_no_reporter_configured_says_so(self): + # An unconfigured identity is not-configured, not a crash: the + # report would otherwise be filed with no reply address. + ... +``` + +Fill the second test in following the neighbouring tests' fixture setup; if +those tests build a case directory with a helper, reuse it rather than +writing a new one. + +- [ ] **Step 2: Run to verify it fails** + +Run: `python3 -m unittest tests.test_cli.ReportCommand -v` +Expected: FAIL, argparse rejects the unknown command `report`. + +- [ ] **Step 3: Implement the parser entry** + +In `abusectl/cli.py`, beside the contacts parser around line 65: + +```python + report_parser = subparsers.add_parser( + "report", help="build report bodies for a case" + ) + report_parser.add_argument("case", type=Path) + report_parser.add_argument( + "--force", + action="store_true", + help="discard hand edits to bodies, keeping a timestamped backup", + ) +``` + +- [ ] **Step 4: Implement the command** + +Add beside `cmd_contacts`, matching its error handling exactly: + +```python +def cmd_report(args) -> int: + """Build report bodies and rewrite the manifest. + + Offline and irreversible-free: nothing here sends anything. The output + is what the user reviews before submit does something that cannot be + recalled. + """ + try: + settings = config.load() + except config.NotConfigured as exc: + print(f"abusectl report: {exc}", file=sys.stderr) + return EXIT_NOT_CONFIGURED + + identity = settings.reporter + missing = [k for k in ("name", "org", "email") if k not in identity] + if missing: + print( + "abusectl report: no reporter identity configured " + f"(missing {', '.join(missing)}). Run `abusectl init`.", + file=sys.stderr, + ) + return EXIT_NOT_CONFIGURED + + try: + manifest = case.load(args.case) + except FileNotFoundError: + print(f"abusectl report: no case at {args.case}", file=sys.stderr) + return EXIT_ERROR + except ValueError as exc: + print(f"abusectl report: {args.case}: {exc}", file=sys.stderr) + return EXIT_ERROR + + try: + manifest = report_module.generate( + manifest, args.case, identity, force=args.force + ) + except (report_module.Frozen, report_module.Modified) as exc: + print(f"abusectl report: {exc}", file=sys.stderr) + return EXIT_ERROR + + case.save(args.case, manifest) + + count = len(manifest["destinations"]) + orphans = len(manifest["unreportable"]) + print(f"{count} destinations, {orphans} indicators with no abuse desk") + return EXIT_OK +``` + +Import the module at the top as `from abusectl import report as report_module`, +matching how `contacts` and `parse` are imported there, and add the dispatch +entry beside the others in `main()`. + +- [ ] **Step 5: Run the whole suite** + +Run: `python3 -m unittest discover tests` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add abusectl/cli.py tests/test_cli.py +git commit -S -m "feat: add the report subcommand + +Refuses without a configured reporter identity rather than filing a report +with no reply address, and turns a frozen or edited case into an error +message rather than a traceback." +``` + +--- + +### Task 12: Prove the suite still opens no socket + +The second property. `AGENTS.md` says it is verified, not asserted, and this +plan adds a module that must not break it. + +- [ ] **Step 1: Run the suite with the network unavailable** + +There is an existing test that does this; find it and confirm it covers the +new module. + +Run: `grep -rn "getaddrinfo\|create_connection" tests/` + +- [ ] **Step 2: Run the whole suite under that harness** + +Run: `python3 -m unittest discover tests` +Expected: PASS, including the offline-proof test. + +- [ ] **Step 3: Commit only if a change was needed** + +If the existing offline test already imports and exercises `report.py`, +nothing to commit. If it enumerates modules by name, add `report` to it and +commit: + +```bash +git add tests/test_offline.py +git commit -S -m "test: cover report.py in the no-socket proof" +``` + +--- + +### Task 13: Re-run both sweeps + +`AGENTS.md` requires this after any change to `parse.py`, and Task 1 changed +it. **Ask the user before reading their mail.** The script lives in the +scratchpad, never in the repository. + +- [ ] **Step 1: Ask permission** + +The corpus is the user's own spam in notmuch. Do not read it unasked. + +- [ ] **Step 2: Extend sweep A with the third assertion** + +The existing sweep asserts no address from the raw source appears in the IOC +output. Add the same assertion against every generated body: + +```python +raw = subprocess.run(["notmuch", "show", "--format=raw", mid], + capture_output=True, check=True).stdout +manifest = { + "format": 1, + "case_id": "sweep", + "iocs": parse.iocs(raw, trusted=TRUSTED), + "auth": parse.auth_results(raw), + "headers": parse.report_headers(raw, trusted=TRUSTED), + # contacts.worklist() is OFFLINE and issues no query; a fake abuse + # address per item is enough to force a body to be generated, which is + # what this assertion needs. contacts.resolve() must NOT be called here: + # sweep A sends nothing. + "contacts": [ + {"iocs": item.iocs, "query": item.query, + "abuse": ["desk@sweep.invalid"], "source": "rdap"} + for item in contacts.worklist(parse.iocs(raw, trusted=TRUSTED)) + if item.kind != "unusable" + ], +} +bodies = "".join( + report.build(manifest, destination, IDENTITY) + for destination in report.email_destinations(manifest["contacts"]) +) +for addr in set(re.findall(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+", + raw.decode("utf-8", "replace"))): + assert addr not in bodies, (mid, addr) +``` + +The assertion must stay that broad: every address in the raw source against +every byte of every body. Checking only the recipient misses an address the +parser invented from a display name, which is how the `_domain_of` defect +reached a report. + +Note that the reporter identity in the sweep must be a placeholder, not the +user's real address, or the assertion will fire on it. + +- [ ] **Step 3: Record the counts** + +Report messages swept, bodies generated, crashes, and any assertion failure. +Counts are evidence and may leave the script; addresses, subjects, +Message-IDs and real URLs may not. + +- [ ] **Step 4: Reproduce any finding as a synthetic fixture** + +A defect found in real mail becomes a fixture using `example.org`, `.invalid` +and RFC 5737 ranges, committed with its failing test. The real message stays +in the scratchpad. + +- [ ] **Step 5: Update the docs** + +Add the `report` spec to the Documents list in `AGENTS.md`, and record the +sweep result the way the contacts sweep is recorded there. + +```bash +git add AGENTS.md +git commit -S -m "docs: record the report spec and its sweep" +``` + +--- + +## Self-review notes + +Checked against `docs/specs/2026-09-09-report.md`: + +- Third part `text/rfc822-headers`, message never attached: Tasks 1, 6 +- Whitelist in `parse`, `report` never opens `source.eml`: Task 1 +- 5965 envelope with x-arf fields: Task 5 +- Identity from config, three keys, absent-not-empty: Tasks 9, 10 +- One destination per abuse address: Task 2 +- `unreportable[]`, not an error: Tasks 3, 8 +- SHA-256 not mtime, freeze with no override, explicit marker: Tasks 7, 8 +- Sweeps re-run with the third assertion: Task 13 + +Two spec items deliberately have no task, and both are correct as gaps: + +- **`api` destinations get rows but null bodies.** The spec narrowed this to + `submit`, so `email_destinations()` builds only email rows. When the submit + spec lands, the vendor rows join here. +- **`submit` writing `frozen` atomically with the first `sent`.** That is + `submit`'s work; Task 7 only reads the field. diff --git a/docs/specs/2026-09-09-contacts.md b/docs/specs/2026-09-09-contacts.md index c751bcb..d45cf83 100644 --- a/docs/specs/2026-09-09-contacts.md +++ b/docs/specs/2026-09-09-contacts.md @@ -234,19 +234,36 @@ resolved. The indicator still reaches MISP and the vendor feeds. "contacts": [ { "iocs": ["ioc-1"], "query": "198.51.100.7", "abuse": ["abuse@example.invalid"], - "source": "rdap", "handle": "AS64496", - "server": "rdap.example.invalid" }, - { "iocs": ["ioc-3", "ioc-7"], "query": "example.invalid", + "source": "rdap", "handle": "AS64496" }, + { "iocs": ["ioc-3", "ioc-7"], "query": "a.b.c.example.invalid", + "queried": "example.invalid", "abuse": [], "source": "rdap", "error": "no abuse role published" } ] ``` -Two departures from the sketch in the umbrella design, both deliberate: +Three departures from the sketch in the umbrella design, all deliberate: - **`iocs` is a list**, because hosts fold and one contact can serve several indicators. - **`abuse` is a list**, because multiple desks are real. +- **`queried` is present only when the label walk shortened the name.** It + records WHAT was asked about, not what the message contained: a contact + found for `a.b.c.example.invalid` at `example.invalid` belongs to the + registered domain rather than the exact host. Absent when the query and + the answer are the same name, and never present on an IP, which is always + asked as itself. + +**`server` is specified but NOT built.** An earlier draft of this example +carried it and nothing ever wrote it. It records WHO was asked, the RDAP +endpoint the bootstrap selected, which is a different fact from `queried` +and independent of it: one server answers thousands of names, and the same +name would move to another server if the bootstrap changed. Unlike `queried` +it is meaningful on the IP branch too, where longest-prefix selection picks +an endpoint. It is worth building when a desk disputes a report and the +answer is "this is the registry that published the address"; until `report` +needs that, the bootstrap cache on disk makes the mapping reproducible and +the field is dead weight. `query` records what was actually asked. Review can then see that a URL indicator was resolved by its host, which is the fourth property made visible diff --git a/docs/specs/2026-09-09-report.md b/docs/specs/2026-09-09-report.md new file mode 100644 index 0000000..12d2157 --- /dev/null +++ b/docs/specs/2026-09-09-report.md @@ -0,0 +1,443 @@ +# abusectl `report`: IOCs and contacts to report bodies + +Status: **agreed 2026-09-09**, in one brainstorming session with the user. +This spec settles the part the umbrella design left open, named there as +"the X-ARF (RFC 5965) schema version, which fields the user's reporting +identity fills, and the plain-text alternative for desks that do not parse +X-ARF". + +Read `docs/specs/2026-09-08-abusectl-design.md` first. This document assumes +its manifest format, its case directory and its ordering between the parts. +It changes one thing there and says so where it does: the re-run guard is a +content hash rather than a timestamp. + +## What it does + +``` +abusectl report <case> +``` + +Reads a case manifest, groups the resolved contacts into destinations, writes +a report body per destination under `bodies/`, and rewrites the manifest with +a `destinations[]` array. It is OFFLINE and PURE: it opens no socket, sends +no mail, and reads no file outside the case directory except the config. + +It is the last step before anything irreversible happens. What it produces is +a document the user reads, edits and approves, so its output is written for a +human first and a parser second. + +## Modules + +``` +abusectl/ + report.py IOCs + contacts -> bodies + destinations[] pure +``` + +One module. There is no protocol/policy split here of the kind that puts +`rdap.py` beside `contacts.py`, because there is no protocol: X-ARF is a MIME +document and `email.message` in the standard library already is that layer. + +`report.py` takes the reporting identity as an ARGUMENT, the way `parse.py` +takes the trust boundary. `cli.py` reads the config and passes it in. That is +what keeps the module testable with no files on disk, and it matters more here +than it did for `parse`: the identity is the one thing in a report that is +disclosed deliberately, and a module that reaches for it itself is a module +that can disclose it in a code path nobody reviewed. + +## The report + +Each email destination gets one MIME document, `multipart/report` with +`report-type=feedback-report`, per RFC 5965. Three parts, in order. + +### Part 1, `text/plain`: what a human reads + +This is the part that decides whether the report is acted on. Desks triage a +queue; a report whose ask is buried is a report that waits. + +``` +Phishing message received 2026-09-08, reporting infrastructure on your +network. Requesting takedown and customer notification. + +Observed on your infrastructure: + + 198.51.100.7 sending IP, first hop outside our trust boundary + example.invalid domain in message links, via a.b.c.example.invalid + +Message as declared: + + Date: Mon, 08 Sep 2026 09:12:44 +0000 + From: "Example Bank" <phish@example.invalid> + Subject: Your account requires verification + +Authentication results: + + SPF: fail DKIM: none DMARC: fail + +URLs, redacted: + + http://login-example.invalid/verify?id=REDACTED&src=REDACTED + +Recipient identifiers have been removed from this report by policy. +Parameter names are preserved, parameter values are not. Full evidence is +retained locally and is available on request. + +Reported by: Danilo M., Example Consulting <reporter@example.org> +Generated by abusectl/<version>. +``` + +Four decisions in that shape. + +**The ask is the first sentence.** Not the evidence, not the identity. A desk +reading one line must know what happened and what is wanted. + +**Only the recipient's own indicators appear.** Destinations are grouped per +abuse address (below), and a desk shown three IPs that are not theirs stops +reading. The `queried` field from `contacts` earns its keep here: "via +`a.b.c.example.invalid`" tells the desk why they are being mailed about a name +that is not literally in the message. + +**The redaction note is ALWAYS present, never conditional on whether anything +was redacted.** A desk that sees `?id=REDACTED` with no explanation may read +the report as malformed or doctored. One sentence turns that into a report +that looks careful, and it opens the door for a desk that genuinely needs more +to ask for it, which is the hand-paste route during review. + +**Plain text, hard-wrapped at 72 columns, with no HTML alternative.** Abuse +desks run ticketing systems and many strip HTML. An HTML part would be a +second body to keep in sync with the first for no reader. + +### Part 2, `message/feedback-report`: what a parser reads + +An RFC 5965 envelope carrying x-arf fields inside it. + +``` +Feedback-Type: abuse +User-Agent: abusectl/0.1.0 +Version: 1 +Report-Type: phishing +Source: 198.51.100.7 +Source-IP: 198.51.100.7 +Reported-Domain: example.invalid +Arrival-Date: Mon, 08 Sep 2026 09:12:44 +0000 +Reported-Uri: http://login-example.invalid/verify?id=REDACTED +``` + +`Feedback-Type`, `User-Agent` and `Version` are the three fields RFC 5965 +requires. The rest are optional there or come from x-arf. + +**Why an RFC 5965 envelope with x-arf fields inside, rather than either +alone.** RFC 5965 is an IETF standard and universally understood, but it was +designed for feedback loops, where a report is ABOUT A MESSAGE. These reports +are about INDICATORS, and 5965 has no natural field for "this specific host is +the thing being reported". x-arf's `Source` does. The envelope is the +standard's own extension point: the part is key/value, so a 5965 parser reads +the fields it knows and ignores the rest, and x-arf tooling finds what it +wants. + +**This choice deliberately does not depend on which of the two is more widely +deployed**, which is a number nobody publishes and which this document does +not claim to know. It was chosen so that the answer does not matter: a +standards parser works, x-arf tooling works, and the human part works +regardless of both. + +**`Source` is singular and a destination may carry several indicators.** The +primary indicator fills it; the full list appears in the text part and in +repeated `Reported-Uri` and `Source-IP` fields. RFC 5965 permits one report +part per indicator instead, and that was rejected as heavier for no reader: +the desk acts on the incident, not on each row. + +### Part 3, `text/rfc822-headers`: the message itself, almost + +**The original message is NOT attached.** `source.eml` carries every +identifier the first property exists to keep out: `To`, `Cc`, `Delivered-To`, +unredacted URLs whose query and path segments encode the recipient, the user's +own Message-IDs, maildir paths and account keys. An abuse desk forwards a +report to the abused customer, and for a phishing domain that customer may be +the attacker; URLhaus is a public feed. Attaching it would deanonymise the +reporter to the attacker, and for a consultant the tracking parameter may +carry a CLIENT's identifier rather than the user's own. + +RFC 5965 provides `text/rfc822-headers` for exactly the case where the full +message cannot be included, so this is the standard's own answer and not a +deviation from it. + +The headers included are a WHITELIST: + +``` +Received (down to the untrusted hop only, never below) +From, Subject, Date, Message-ID, Reply-To, Return-Path +Authentication-Results, Received-SPF +MIME-Version, Content-Type +``` + +**A whitelist, never a blacklist.** A blacklist means every header the parser +learns to read later is a leak waiting for someone to remember. This is the +same reasoning that has `parse.py` not reading `To` at all rather than +stripping it afterwards. + +`Subject` and the `From` display name are attacker-controlled free text, and a +sweep has already found a spoofed `Reply-To` display name. They are kept: they +are the message's own content rather than the recipient's identity, and they +are what lets a desk recognise a campaign they have seen before. + +**Considered and not built: a redacted body text part.** A desk analysing a +campaign wants the lure, the impersonated brand and the pretext, and none of +that survives headers-only. The body's INDICATORS already survive as IOCs +regardless, so what is lost is the prose. It is not built because the prose is +an unbounded attacker-supplied string, and deciding what is safe inside free +text is a judgement rather than a whitelist, which is the shape of every leak +this project has had. Build it when a desk actually asks for the lure, and +build it as a redaction rule with its own tests, not as a passthrough. + +## Where the headers come from + +**`parse` stores the whitelisted headers in the manifest, and `report` never +opens `source.eml`.** + +This is the same structural argument as the first property, applied one level +down: `report` cannot disclose a header it was never given. The alternative, +re-reading `source.eml` at report time and filtering there, would put a second +"what may be disclosed" decision in a second module, away from `parse.py` +where that decision currently lives, and two places to remember is how the +fourth property leaked three times. + +The cost is real and is accepted: this is a change to `parse.py`, a new +`headers` block in the manifest, and **both sweeps must be re-run**, per +`AGENTS.md`. The sweep assertion is what proves the whitelist does not carry +an address, and a whitelist written by hand is exactly the kind of thing a +sweep catches being wrong. + +### Two accepted disclosures, named so they are not mistaken for leaks + +The first property reads as an unqualified "recipient identifiers must never +reach a report". These are the deliberate exceptions the whitelist creates, +recorded here rather than left to be rediscovered in a test comment. + +**Our own receiving relay's hostname is published.** The boundary `Received` +line names it in its `by` clause and `Authentication-Results` names it as the +authserv-id, so `mx.example.org` travels with every report. That is the +user's mail host, not the user's identity, and an abuse desk learns it from +the report's own `From` regardless. It is accepted because removing it would +mean rewriting the inside of two headers whose value to a desk is precisely +that they are the receiving server's own verbatim words. The consequence is +that the manifest-wide "no bare `example.org`" assertion cannot hold over the +`headers` block; `tests/test_cli.py` narrows it there and asserts the +ADDRESS is still absent, which is the part that matters. + +**Attacker-controlled free text is published unfiltered.** `Subject` and the +`From` display name are kept deliberately, because they are what lets a desk +recognise a campaign. An attacker who writes the recipient's address into +one, plainly or obfuscated as `you%40example.org`, gets it published: the +whitelist governs WHICH headers travel, never what is inside one. This is +not fixed by filtering free text, which is the judgement-shaped problem that +`AGENTS.md` names as the origin of every leak this project has had. The sweep +over real mail is what covers this class, which is one more reason it is not +optional here. + +The envelope recipient is NOT in this list. Our own relay writes it into the +boundary `Received` line's optional `for` clause, and that clause is cut +before the line is stored, in every shape the grammar allows. + +## The reporting identity + +Three config keys, all under a `[reporter]` section: + +```toml +[reporter] +name = "Danilo M." +org = "Example Consulting" +email = "reporter@example.org" +``` + +They fill the report's `From`, the `Reported by:` line in the text part, and +nothing else. `User-Agent` is `abusectl/<version>` and is not configurable. + +**The reporter's identity is disclosed DELIBERATELY, and that is what makes it +different from every other identifier this tool refuses to publish.** The same +address recovered from a `To` header is a leak; supplied in a config file it +is the user choosing to be identified, and a report with no reply address is +one a desk deprioritises. The distinction is provenance, so it is enforced by +provenance: the identity comes from config ONLY, and `parse` must never supply +it. If the two ever became one path, the distinction would be a comment rather +than a guarantee. + +A skipped answer is ABSENT from the config, never an empty string, the same +rule the rest of the config follows. `init` grows prompts for these three, and +they are hand-tested like the rest of the prompts. + +**Noted, not built: a per-case reporting identity.** Reporting a campaign that +targeted a client, under the user's own name, tells the abuse desk which +consultant is working that incident, which is a disclosure about the +engagement rather than about the mail. The escape hatch already exists without +new machinery: review edits the bodies, and a `From` in a body is a line the +user can change. A config-level override belongs to the first real engagement +where it matters, not to this spec. + +## Destinations + +**One destination per abuse ADDRESS.** Every IOC whose contact resolved to +`abuse@example.invalid` is grouped into one report to that desk. + +Contacts already fold by host, but two different contacts can still resolve to +the same address, an IP and a domain both at one hoster being the common case. +Grouping per contact would send that desk two mails about one incident, which +is the duplicate-mail behaviour desks complain about. Grouping per IOC would +be a storm. + +`report` writes three kinds of destination: + +| kind | body | who builds the payload | +|---|---|---| +| `email` | `bodies/<id>.xarf` | `report`, the MIME document above | +| `api` | `null` for now, see below | `submit`, once its spec settles the shape | +| `misp` | `null` | `submit`, via PyMISP | + +MISP is a destination like the others with a null body, per the umbrella +design, so `submit` stays one loop with one ordering rule. + +**Only `email` bodies are built by THIS spec.** The `api` kinds are created as +destination entries with their IOC lists and a `pending` status, and their +bodies stay null until the `submit` spec settles each vendor's payload shape. +Writing a vendor's JSON now would mean guessing an endpoint's contract from +memory, which is the mistake the provider table already records: the first +draft was written from memory and every range was wrong. The destination +entries exist from the start so that `submit` fills bodies rather than +inventing rows, and so review can already see which vendors a case will reach. + +### An indicator with no abuse contact + +A missing contact is a normal outcome, not an error: the umbrella design +already settles that, and RDAP publishes no abuse role for many netblocks. + +`report` writes an `unreportable[]` array into the manifest, one entry per IOC +that reached no email destination, carrying the reason from its contact entry: + +```json +"unreportable": [ + { "ioc": "ioc-4", "reason": "no abuse role published" }, + { "ioc": "ioc-9", "reason": "not ASCII, and we do not guess at an IDN encoding" } +] +``` + +Two reasons for making it explicit. Review becomes honest: the user sees that +four indicators are going to MISP and the vendors but no desk was found for +them, which is a fact they may want to act on by hand, and finding it any other +way means diffing the IOC list against every destination's IOC list. And it is +the same instinct as `suspect_path_segments` flagging rather than redacting: a +failure that is visible beats a failure that is merely absent. + +**It is not an error and does not affect the exit code.** A case where nothing +resolved still produces MISP and vendor destinations and is a perfectly good +report. `report` exits non-zero only when it could not write. + +## Re-running, and the frozen case + +The umbrella design says `report` refuses to run again on a case whose bodies +were modified after generation, by a timestamp check. **The intent stands and +the mechanism changes**: it is a content hash. + +An mtime is a poor witness in both directions. A `git checkout`, an `rsync`, a +backup restore or an editor that writes-and-renames all move mtime with no +human having edited anything, and an editor that preserves mtime hides a real +edit. So `report` records the SHA-256 of each body it writes, in that body's +destination entry, and compares content rather than a rumour about content. +`hashlib` is standard library, so this costs a field and no dependency. + +The hash has a SECOND job, and the spec states it so a later change does not +drop it as redundant: for a destination that has been sent, the hash is the +record of what was actually disclosed to a third party. + +### The rule + +**A case where ANY destination has been sent is FROZEN.** `report` refuses, +and there is no `--force` override. + +The destinations are not independent artifacts, they are one incident reported +in parallel. If one desk holds the report and a body for another desk is then +regenerated with different content, two desks hold contradictory accounts of +the same case, and a desk that forwards to the other finds the reporter +unreliable. Regeneration is also not as isolated as it looks: the shared parts, +the identity, the header block, the IOC list, come from the manifest, so +regenerating one body after the manifest has changed produces a case whose +bodies were built from two different states. + +Freezing is recorded EXPLICITLY, written by `submit` at its first success: + +```json +"frozen": { "at": "2026-09-08T12:40:11Z", "by": "abusedb" } +``` + +Absent means not frozen, the same convention the config follows for a skipped +answer. It is an optional field, so existing manifests stay loadable and the +`format` version does not change. + +**Explicit rather than derived from the statuses**, because the marker is +write-once and monotonic. A status corrected by hand, or a status added by a +later schema that nobody remembered to add to a frozen set, would quietly +unfreeze a derived check. For a rule protecting an evidence record, a field +that can only be turned on is the right shape. + +`submit` must write the marker in the SAME atomic manifest write as the first +`sent` status. A separate write leaves a window where a crash produces a case +that has been disclosed and does not know it. + +### The whole rule, in order + +| case state | `report` does | +|---|---| +| `frozen` present | refuses, no override, names the destination that landed | +| bodies modified, not frozen | refuses without `--force` | +| bodies modified, `--force` | backs each up to `<name>.<timestamp>.orig`, regenerates | +| bodies missing or unmodified | regenerates | + +A deleted body regenerates silently. The only reasons to delete one are a +mistake or a deliberate start-over, and regeneration is what both want; the +case that looked like it needed protecting, a body deleted after it was sent, +is caught by the freeze rather than by the file check. + +Because `--force` can now only ever touch a case that nothing has left, it is +a far safer flag than it first appears. The backup is kept anyway: the +umbrella design's point about a review that took twenty minutes applies, and +`init` already backs up a config it is about to replace. + +A user who needs to change a body on a frozen or deferred case still can, by +editing it during review. That route is unaffected and is the right one: it is +a deliberate act with the user looking at the text. + +## Testing + +TDD, and the same rule as the rest of the repository: test what has a right +answer. + +Tested, because there is one: + +- the MIME structure: three parts, the right types, `report-type=feedback-report` +- the header whitelist keeps what it should and, more importantly, DROPS + `To`, `Cc`, `Delivered-To` and `X-Original-To` when a fixture carries them +- `Received` is truncated at the untrusted hop and never includes the ones + below it, against `forged-chain.eml` +- destinations group per address, including the two-contacts-one-address case +- an IOC with no contact lands in `unreportable[]` and creates no destination +- the body hash detects a modified body, and does not fire on an untouched one +- a frozen case refuses even with `--force` +- `--force` writes the timestamped backup before overwriting + +Not unit-tested: whether the text part READS well to an abuse desk. That has +no assertion, and it is the same category as the interactive prompts. The user +hand-tests it by reading a generated report. + +**The leak sweep must be re-run**, both A and B, because this spec changes +`parse.py`. Sweep A gains a third assertion: no address from the raw source +appears in any generated report body. That is the assertion that actually +proves the whitelist, and it should be as broad as the existing one, every +address in the source against every byte of every body. + +## What this leaves for `submit` + +Named here because this spec creates them, not to settle them: + +- writing the `frozen` marker atomically with the first `sent` status +- the vendor JSON shapes, and which side writes them. `report` creates the + `api` destination rows; whether it also learns to write their bodies is a + decision for that spec. If a vendor's payload turns out to need a value the + manifest does not hold, that is a change here, not a workaround there. diff --git a/tests/fixtures/reportable.eml b/tests/fixtures/reportable.eml new file mode 100644 index 0000000..e939059 --- /dev/null +++ b/tests/fixtures/reportable.eml @@ -0,0 +1,22 @@ +Received: from relay.example.org (relay.example.org [192.0.2.10]) + by mx.example.org with ESMTP id abc123 + for <you@example.org>; Mon, 07 Sep 2026 09:12:44 +0000 +Received: from sender.invalid (sender.invalid [203.0.113.42]) + by relay.example.org with ESMTP id def456 + for <you@example.org>; Mon, 07 Sep 2026 09:12:40 +0000 +Return-Path: <bounce@sender.invalid> +Authentication-Results: mx.example.org; spf=fail; dkim=none; dmarc=fail +Received-SPF: fail (mx.example.org: domain of sender.invalid does not designate 203.0.113.42) +From: "Example Bank" <phish@sender.invalid> +To: victim@example.org +Cc: colleague@example.org +Delivered-To: victim@example.org +X-Original-To: victim@example.org +Reply-To: "Support" <reply@sender.invalid> +Subject: Your account requires verification +Date: Mon, 07 Sep 2026 09:12:40 +0000 +Message-ID: <case-one@sender.invalid> +MIME-Version: 1.0 +Content-Type: text/plain; charset=utf-8 + +Please verify at http://login.sender.invalid/verify?id=abc123 diff --git a/tests/test_case.py b/tests/test_case.py index c6964a9..ab7eeeb 100644 --- a/tests/test_case.py +++ b/tests/test_case.py @@ -45,6 +45,16 @@ class TestCaseCreation(unittest.TestCase): manifest = json.loads((created.path / "manifest.json").read_text()) self.assertEqual(manifest["format"], case.FORMAT_VERSION) + def test_every_block_is_seeded_present_and_empty(self): + # A created-but-unparsed case must have the same SHAPE as a parsed + # one, so a later reader indexes a block rather than guarding every + # access. report will read headers and would hit a KeyError. + created = case.create(self.root, b"x") + manifest = json.loads((created.path / "manifest.json").read_text()) + for block in ("iocs", "auth", "contacts", "destinations", "headers"): + self.assertIn(block, manifest) + self.assertEqual(manifest[block], []) + def test_two_cases_do_not_collide(self): a = case.create(self.root, b"one") b = case.create(self.root, b"two") diff --git a/tests/test_cli.py b/tests/test_cli.py index 1c023c3..b8fffea 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -145,7 +145,30 @@ class TestParse(unittest.TestCase): str(FIXTURES / "simple.eml"), ) text = (pathlib.Path(out.strip()) / "manifest.json").read_text() - self.assertNotIn("example.org", text) + self.assertNotIn("you@example.org", text) + # The bare domain is still barred everywhere the IOCs live. The + # headers block is the one exception and it is a NARROW one: the + # whitelist publishes the boundary Received line and + # Authentication-Results, and both name our own receiving relay in a + # "by"/authserv-id clause. That is the user's mail host, not the + # user's identity, and a desk learns it from the report's own From + # regardless. The address itself must still be absent, which the + # assertion above and report_headers' own tests cover. + import json + + manifest = json.loads(text) + headers = manifest.pop("headers") + self.assertNotIn("example.org", json.dumps(manifest)) + # And nothing shaped like an address survives in the exception. + # Both spellings: you%40example.org is not a hypothetical, it is why + # leaky.eml exists, and docs/plans/2026-09-09-contacts.md records a + # From of phish@victim%40example.org.invalid. + blob = json.dumps(headers) + self.assertNotIn("@example.org", blob) + self.assertNotIn("you%40example.org", blob) + names = [name for name, _ in headers] + for name in ("To", "Cc", "Delivered-To", "X-Original-To"): + self.assertNotIn(name, names) def test_a_missing_config_points_at_init(self): code, _, err = self._run( diff --git a/tests/test_offline.py b/tests/test_offline.py index 19d9cee..547bd96 100644 --- a/tests/test_offline.py +++ b/tests/test_offline.py @@ -53,6 +53,14 @@ class NothingOpensASocket(unittest.TestCase): b"Subject: test\r\n\r\nbody\r\n") parse.iocs(raw, trusted=["192.0.2.0/24"]) + def test_selecting_report_headers_opens_no_socket(self): + # A new entry point into the parse path, so it is held to the same + # guarantee: choosing what to publish resolves nothing. + raw = (b"Received: from relay.example.invalid ([192.0.2.10])\r\n" + b"From: sender@example.invalid\r\n" + b"Subject: test\r\n\r\nbody\r\n") + parse.report_headers(raw, trusted=["192.0.2.0/24"]) + def test_resolving_with_an_injected_fetch_opens_no_socket(self): iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}] bootstraps = { diff --git a/tests/test_parse.py b/tests/test_parse.py index 11345c7..6e3b7a4 100644 --- a/tests/test_parse.py +++ b/tests/test_parse.py @@ -290,5 +290,141 @@ class TestIocAssembly(unittest.TestCase): self.assertNotIn("you%40example.org", blob) +class ReportHeaders(unittest.TestCase): + def test_the_whitelist_keeps_what_a_desk_needs(self): + headers = parse.report_headers(load("reportable.eml"), + trusted=["192.0.2.0/24"]) + names = [name for name, _ in headers] + for wanted in ("From", "Subject", "Date", "Message-ID", "Reply-To", + "Return-Path", "Authentication-Results", "Received-SPF"): + self.assertIn(wanted, names) + + def test_recipient_headers_never_survive_the_whitelist(self): + # The first property, at the one place a report reproduces header + # text verbatim. A blacklist would have to remember each of these; + # the whitelist never names them at all. + headers = parse.report_headers(load("reportable.eml"), + trusted=["192.0.2.0/24"]) + names = [name for name, _ in headers] + blob = repr(headers) + for name in ("To", "Cc", "Delivered-To", "X-Original-To"): + self.assertNotIn(name, names) + self.assertNotIn("victim@example.org", blob) + self.assertNotIn("colleague@example.org", blob) + + def test_received_stops_at_the_boundary_hop(self): + # 192.0.2.10 is ours, so its Received line is our own infrastructure + # and must not be published; the hop below it is the one being + # reported and is kept. + headers = parse.report_headers(load("reportable.eml"), + trusted=["192.0.2.0/24"]) + received = [value for name, value in headers if name == "Received"] + self.assertEqual(len(received), 1) + self.assertIn("203.0.113.42", received[0]) + self.assertNotIn("mx.example.org with ESMTP id abc123", received[0]) + + def test_the_published_hop_carries_no_envelope_recipient(self): + # The boundary Received line is written by OUR OWN relay, and its + # optional "for <addr>" clause is the envelope recipient: the + # victim's address, verbatim, in the one header a report reproduces + # in full. Truncating the chain is not enough on its own. + headers = parse.report_headers(load("reportable.eml"), + trusted=["192.0.2.0/24"]) + received = [value for name, value in headers if name == "Received"] + self.assertNotIn("you@example.org", received[0]) + self.assertNotIn("for <", received[0]) + # The rest of the hop survives; this is a cut, not a blanking. + self.assertIn("203.0.113.42", received[0]) + + def test_every_for_clause_shape_loses_the_address(self): + # RFC 5321 4.4 puts For inside Opt-info, so With, ID, Via or a CFWS + # comment may legitimately follow it, and its ABNF is + # 1*( Path / Mailbox ) where Mailbox carries no angle brackets. + # Anchoring on "for" being immediately followed by the clause + # terminator matched only the neatest shape and let four routine + # ones through, each publishing the victim's address. + hop = "from a.invalid (a.invalid [203.0.113.5]) by mx.example.org " + shapes = ( + "for <you@example.org> (envelope-from <b@c.invalid>); Mon, 07 Sep 2026 09:12:40 +0000", + "for you@example.org; Mon, 07 Sep 2026 09:12:40 +0000", + "for <you@example.org> with ESMTP; Mon, 07 Sep 2026 09:12:40 +0000", + "id qq; Mon, 07 Sep 2026 09:12:40 +0000 (for <you@example.org>)", + "for <you@example.org>; Mon, 07 Sep 2026 09:12:40 +0000", + ) + for tail in shapes: + with self.subTest(tail=tail): + stripped = parse._strip_envelope_recipient(hop + tail) + self.assertNotIn("you@example.org", stripped) + # The hop's own evidence survives: this is a cut, not a + # blanking, and a rule that ate the line would pass the + # assertion above while destroying the report. + self.assertIn("203.0.113.5", stripped) + self.assertIn("mx.example.org", stripped) + + def test_stripping_leaves_no_doubled_space_or_stray_separator(self): + # Cosmetic in isolation, but the result is published verbatim to a + # third party, so a mangled line reads as a broken tool. + hop = ("from a.invalid (a.invalid [203.0.113.5]) by mx.example.org" + " for <you@example.org>; Mon, 07 Sep 2026 09:12:40 +0000") + stripped = parse._strip_envelope_recipient(hop) + self.assertNotIn(" ", stripped) + self.assertNotIn(" ;", stripped) + self.assertIn("mx.example.org; Mon", stripped) + + def test_a_comment_holding_only_the_clause_leaves_no_debris(self): + hop = ("from a.invalid (a.invalid [203.0.113.5]) by mx.example.org" + " id qq; Mon, 07 Sep 2026 09:12:40 +0000 (for <you@example.org>)") + stripped = parse._strip_envelope_recipient(hop) + self.assertNotIn("you@example.org", stripped) + self.assertFalse(stripped.endswith("(")) + self.assertTrue(stripped.endswith("+0000")) + + def test_the_envelope_sender_comment_survives_the_cut(self): + # envelope-from is the SENDER, which is what the report is about, so + # cutting the recipient must not take it along. + hop = ("from a.invalid (a.invalid [203.0.113.5]) by mx.example.org" + " for <you@example.org> (envelope-from <bounce@sender.invalid>);" + " Mon, 07 Sep 2026 09:12:40 +0000") + stripped = parse._strip_envelope_recipient(hop) + self.assertNotIn("you@example.org", stripped) + self.assertIn("bounce@sender.invalid", stripped) + + def test_the_whitelist_does_not_filter_attacker_free_text(self): + # A DOCUMENTED LIMIT, not a guarantee. The spec keeps Subject and the + # From display name knowing both are attacker-controlled free text, + # because they are what lets a desk recognise a campaign. An attacker + # who writes the recipient's own address into one, obfuscated or not, + # gets it published: the whitelist governs WHICH headers travel, never + # what is inside one. + # + # This is asserted so the limit is visible and deliberate. Do not + # "fix" it by filtering free text, which is the judgement-shaped + # problem AGENTS.md names as the source of every leak here. The + # sweep over real mail is what covers this class, per AGENTS.md. + raw = (b"Received: from a.invalid (a.invalid [203.0.113.5])" + b" by mx.example.org with ESMTP id X;" + b" Mon, 07 Sep 2026 09:12:40 +0000\r\n" + b"From: <phish@sender.invalid>\r\n" + b"Subject: Verify you%40example.org\r\n\r\nbody\r\n") + headers = parse.report_headers(raw, trusted=["192.0.2.0/24"]) + subject = dict(headers)["Subject"] + self.assertIn("you%40example.org", subject) + + def test_a_forged_chain_publishes_no_hop_below_the_boundary(self): + # The same job test_a_forged_chain_stops_at_the_first_untrusted_hop + # does for sending_ip(), asserted over what actually gets published: + # 198.51.100.7 is an innocent party the attacker named. + headers = parse.report_headers(load("forged-chain.eml"), + trusted=["192.0.2.0/24"]) + received = [value for name, value in headers if name == "Received"] + # Asserted in BOTH directions: dropping Received altogether would + # satisfy the "not published" half on its own, and a test that + # passes when the feature is missing protects nothing. + self.assertEqual(len(received), 1) + self.assertIn("203.0.113.99", received[0]) + self.assertTrue(all("198.51.100.7" not in value for value in received)) + self.assertTrue(all("198.51.100.8" not in value for value in received)) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_report.py b/tests/test_report.py new file mode 100644 index 0000000..6a340c9 --- /dev/null +++ b/tests/test_report.py @@ -0,0 +1,1600 @@ +import copy +import email +import email.policy +import json +import unittest + +from abusectl import report + + +class Grouping(unittest.TestCase): + def test_two_contacts_at_one_address_become_one_destination(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["target"], "abuse@host.invalid") + self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"]) + + def test_a_contact_with_two_addresses_reaches_both_desks(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["a@host.invalid", "b@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual([d["target"] for d in destinations], + ["a@host.invalid", "b@host.invalid"]) + # Both desks carry the indicator, and each gets its own id: a + # destination that reached only one desk, or two rows sharing an + # id, would pass an assertion on the sorted targets alone. + self.assertEqual([d["iocs"] for d in destinations], + [["ioc-1"], ["ioc-1"]]) + self.assertEqual(len({d["id"] for d in destinations}), 2) + for destination in destinations: + self.assertEqual(destination["id"], + report.email_destination_id( + destination["target"])) + + def test_a_contact_with_no_address_creates_no_destination(self): + """The contact that resolved must still produce its destination. + + Asserted alongside one that DOES resolve, because "no destination + for this contact" is also what returning nothing at all looks + like, and that is not the behaviour being described. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + {"iocs": ["ioc-2"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual([d["target"] for d in destinations], + ["abuse@host.invalid"]) + self.assertEqual(destinations[0]["iocs"], ["ioc-2"]) + + def test_destinations_carry_stable_ids_and_pending_status(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(contacts)[0] + self.assertEqual(destination["id"], + report.email_destination_id("abuse@host.invalid")) + self.assertEqual(destination["kind"], "email") + self.assertEqual(destination["status"], "pending") + + def test_an_id_is_the_literal_shape_a_reviewer_will_read(self): + """Pin the shape, since it becomes a filename in bodies/. + + Computed by hand rather than by calling the code under test, so + this fails if the derivation changes rather than following it. + """ + self.assertEqual(report.email_destination_id("abuse@host.invalid"), + "email-bc50e369") + + def test_ids_are_derived_per_destination_not_per_contact(self): + """A contact that resolved to no desk must not shift another's id. + + The obvious implementation numbers destinations by position, and a + skipped contact then either burns an id or renumbers the rest. + Both are wrong for the same reason: an id names a desk. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + {"iocs": ["ioc-2"], "query": "198.51.100.7", + "abuse": ["a@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-3"], "query": "198.51.100.8", + "abuse": ["b@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual([d["id"] for d in destinations], + [report.email_destination_id("a@host.invalid"), + report.email_destination_id("b@host.invalid")]) + self.assertEqual([d["target"] for d in destinations], + ["a@host.invalid", "b@host.invalid"]) + + def test_one_desk_listed_twice_by_one_contact_is_one_destination(self): + """A duplicate in a contact's own abuse list must not duplicate a desk. + + RDAP jCards are attacker-adjacent data: an entity can publish the + same address in two vcard rows, and one destination per ADDRESS is + the rule regardless of how many rows produced it. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid", "abuse@host.invalid"], + "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["iocs"], ["ioc-1"]) + + def test_one_desk_spelled_with_two_domain_cases_is_one_destination(self): + """A domain is case-insensitive, so two spellings are one desk.""" + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@Host.Invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"]) + self.assertEqual(destinations[0]["target"], "abuse@Host.Invalid") + + def test_the_domain_folds_under_a_local_part_that_does_not(self): + """Isolate the domain fold from the role fold. + + The version of this test that first shipped used a lowercase + local part throughout, so it exercised only the domain and passed + while a capitalised role name produced two destinations. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["J.Smith@Host.Invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["J.Smith@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"]) + + def test_two_spellings_of_one_desk_share_an_id(self): + """The id derives from the same normalised form the grouping uses. + + Otherwise the spelling RDAP happened to publish first would decide + a body's filename, and a re-run that saw the other spelling first + would look like a different desk. + """ + self.assertEqual(report.email_destination_id("abuse@Host.Invalid"), + report.email_destination_id("abuse@host.invalid")) + + def test_a_role_mailbox_folds_in_both_halves(self): + """"Abuse@Host.Invalid" and "abuse@host.invalid" are one desk. + + The case that first shipped folded the domain only, so a jCard + publishing the role name capitalised produced two destinations and + two mails to one desk. RFC 2142 mandates the role mailboxes and + requires them case-insensitive, so no host runs "Abuse@" and + "abuse@" as different desks. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["Abuse@Host.Invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"]) + self.assertEqual(destinations[0]["target"], "Abuse@Host.Invalid") + + def test_one_contact_publishing_a_role_mailbox_twice_folds_it(self): + """The same fold applies within one contact's own abuse list. + + rdap.abuse_addresses dedupes case-sensitively, so a jCard with two + vcard rows spelling the role differently delivers both here. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["Abuse@host.invalid", "abuse@host.invalid"], + "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(len(destinations), 1) + self.assertEqual(destinations[0]["iocs"], ["ioc-1"]) + + def test_every_rfc2142_role_this_tool_can_meet_folds(self): + for role in ("abuse", "postmaster", "security", "noc", "hostmaster"): + with self.subTest(role=role): + self.assertEqual( + report.email_destination_id(f"{role.title()}@host.invalid"), + report.email_destination_id(f"{role}@host.invalid")) + + def test_a_personal_local_part_is_left_alone(self): + """Only the receiving host knows whether ITS local parts fold. + + A named mailbox is not a standardised role, so folding it could + silently merge two desks a host genuinely distinguishes and drop + one of them. Two mails to one desk is the lesser failure, and the + role names above are where the duplicate actually happens. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["J.Smith@host.invalid", "j.smith@host.invalid"], + "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual(sorted(d["target"] for d in destinations), + ["J.Smith@host.invalid", "j.smith@host.invalid"]) + self.assertNotEqual(destinations[0]["id"], destinations[1]["id"]) + + def test_a_destination_starts_with_no_body(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + self.assertIsNone(report.email_destinations(contacts)[0]["body"]) + + def test_the_contacts_passed_in_are_not_modified(self): + """The caller's contacts are the manifest's own array. + + case.py is the only writer of a manifest, so a grouping pass that + edited what it was handed would write through it from outside. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + before = copy.deepcopy(contacts) + report.email_destinations(contacts) + self.assertEqual(contacts, before) + + def test_a_destinations_ioc_list_is_its_own(self): + """Not aliased to the contact's list it was built from. + + Holds today because the grouping starts a fresh list, but nothing + else pins it: an implementation that reused contact["iocs"] for a + single-contact destination would pass every other test here and + leave a destination and a contact sharing one list in a manifest + about to be written. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(contacts)[0] + destination["iocs"].append("ioc-2") + self.assertEqual(contacts[0]["iocs"], ["ioc-1"]) + + def test_the_same_contacts_produce_the_same_ids_twice(self): + """Ids must not depend on dict iteration luck or set ordering.""" + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["b@host.invalid", "a@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["c@host.invalid"], "source": "rdap"}, + ] + first = report.email_destinations(contacts) + second = report.email_destinations(contacts) + self.assertEqual([(d["id"], d["target"]) for d in first], + [(d["id"], d["target"]) for d in second]) + self.assertEqual([d["target"] for d in first], + ["b@host.invalid", "a@host.invalid", + "c@host.invalid"]) + + def test_a_desks_id_survives_another_desk_appearing(self): + """An id names a DESK, not a position in this run's list. + + Task 8 writes each body to bodies/<id>.xarf and records its hash + against that id. With a positional id, re-running contacts on a + case that gained an indicator renumbers every desk after the new + one, so bodies/<id>.xarf on disk belongs to a different desk than + the manifest's entry of that id, and the edit check compares one + desk's body against another's. + """ + established = {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["b@host.invalid"], "source": "rdap"} + first = report.email_destinations([established]) + + # A later contacts run finds an indicator whose desk sorts ahead. + newcomer = {"iocs": ["ioc-2"], "query": "example.invalid", + "abuse": ["a@new.invalid"], "source": "rdap"} + second = report.email_destinations([newcomer, established]) + + by_target = {d["target"]: d["id"] for d in second} + self.assertEqual(by_target["b@host.invalid"], first[0]["id"]) + self.assertNotEqual(by_target["a@new.invalid"], first[0]["id"]) + + def test_an_ids_position_does_not_leak_into_it(self): + """The same desk alone and third in a list gets one id.""" + alone = report.email_destinations([ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["desk@host.invalid"], "source": "rdap"}, + ]) + crowded = report.email_destinations([ + {"iocs": ["ioc-2"], "query": "198.51.100.8", + "abuse": ["one@host.invalid", "two@host.invalid"], + "source": "rdap"}, + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["desk@host.invalid"], "source": "rdap"}, + ]) + self.assertEqual(crowded[2]["target"], "desk@host.invalid") + self.assertEqual(crowded[2]["id"], alone[0]["id"]) + + +class Unreportable(unittest.TestCase): + def test_an_ioc_with_no_desk_is_listed_with_its_reason(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2", "ioc-3"], "query": "example.invalid", + "abuse": [], "source": "rdap", + "error": "no abuse role published"}, + ] + self.assertEqual( + report.unreportable(contacts), + [ + {"ioc": "ioc-2", "reason": "no abuse role published"}, + {"ioc": "ioc-3", "reason": "no abuse role published"}, + ], + ) + + def test_a_missing_reason_still_produces_an_entry(self): + contacts = [{"iocs": ["ioc-9"], "query": "x.invalid", "abuse": [], + "source": "rdap"}] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-9", "reason": "no abuse address resolved"}], + ) + + def test_an_empty_reason_does_not_read_as_no_reason(self): + """`error: ""` must not be reported as the literal empty string. + + A contact entry is written by contacts.resolve, but a manifest is + a file on disk that a user edits during review. An empty reason + renders as a blank cell in the report the user reads, which says + nothing at all; the default at least says what happened. + """ + contacts = [{"iocs": ["ioc-9"], "query": "x.invalid", "abuse": [], + "source": "rdap", "error": ""}] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-9", "reason": "no abuse address resolved"}], + ) + + def test_nothing_unreportable_is_an_empty_list_not_an_error(self): + contacts = [{"iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}] + self.assertEqual(report.unreportable(contacts), []) + + def test_an_ioc_that_reached_a_desk_elsewhere_is_not_unreportable(self): + """Hosts fold, so one IOC can sit in a resolved and an unresolved + contact at once. It IS reportable, and listing it says otherwise. + + The plan's implementation listed it regardless, which puts an + indicator in both the destination list and the "no desk found" + list of one manifest. A reviewer reading the second acts on an + indicator that is already on its way to a desk, and the whole + point of the array is that it can be trusted without diffing. + """ + contacts = [ + {"iocs": ["ioc-1", "ioc-2"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + {"iocs": ["ioc-2", "ioc-3"], "query": "example.invalid", + "abuse": [], "source": "rdap", + "error": "no abuse role published"}, + ] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-3", "reason": "no abuse role published"}], + ) + + def test_one_ioc_unresolved_twice_is_listed_once(self): + """Two contacts, both unresolved, one shared indicator. + + A duplicate row is a second line in the report about one + indicator, and the reasons may differ, so which one wins has to + be decided rather than left to whichever contact came last. + First reason seen wins, matching the first-seen ordering the + destinations use. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + {"iocs": ["ioc-1"], "query": "198.51.100.7", "abuse": [], + "source": "rdap", "error": "no rdap server for this tld, " + "or no answer"}, + ] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-1", "reason": "no abuse role published"}], + ) + + def test_the_contacts_passed_in_are_not_modified(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + {"iocs": ["ioc-2"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + before = copy.deepcopy(contacts) + report.unreportable(contacts) + self.assertEqual(contacts, before) + + +class MalformedAddresses(unittest.TestCase): + """An abuse "address" with no @ cannot be mailed. + + RDAP jCard data is third-party and occasionally malformed, and a + destination built from such a value carries an unsendable target with + status "pending". That is the failure mode the unreportable array + exists to prevent: the indicator appears reportable, no desk ever + receives it, and nothing in the manifest says so. + """ + + def test_a_target_with_no_at_creates_no_destination(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["not-an-address"], "source": "rdap"}, + {"iocs": ["ioc-2"], "query": "198.51.100.7", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual([d["target"] for d in destinations], + ["abuse@host.invalid"]) + self.assertEqual(destinations[0]["iocs"], ["ioc-2"]) + + def test_an_ioc_whose_only_address_is_malformed_is_unreportable(self): + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["not-an-address"], "source": "rdap"}, + ] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-1", + "reason": "no usable abuse address published"}], + ) + + def test_a_usable_address_beside_a_malformed_one_still_reports(self): + """The good half of a jCard must survive the bad half. + + Discarding the contact wholesale would lose a real desk over a + neighbouring malformed row. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["not-an-address", "abuse@host.invalid"], + "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + self.assertEqual([d["target"] for d in destinations], + ["abuse@host.invalid"]) + self.assertEqual(report.unreportable(contacts), []) + + def test_an_addresss_own_error_is_not_overwritten_by_the_default(self): + """A contact that has both a reason and a malformed address. + + The contact's own error says more than "no usable address", so it + wins; the default is only for a contact that offered no reason. + """ + contacts = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["not-an-address"], "source": "rdap", + "error": "no abuse role published"}, + ] + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-1", "reason": "no abuse role published"}], + ) + + def test_the_two_lists_partition_every_indicator(self): + """The invariant the pair is for: each IOC is in exactly one. + + Every other test here pins one side. This pins the relationship, + which is what a reviewer actually relies on: an indicator missing + from both is silently unreported, and one in both is reported and + also flagged as unreported. Both failures come from the two + functions disagreeing about what counts as a desk, so they are + asserted against one input that exercises every branch. + """ + contacts = [ + {"iocs": ["ioc-1", "ioc-2"], "query": "198.51.100.7", + "abuse": ["Abuse@Host.Invalid"], "source": "rdap"}, + {"iocs": ["ioc-2", "ioc-3"], "query": "example.invalid", + "abuse": [], "source": "rdap", + "error": "no abuse role published"}, + {"iocs": ["ioc-4"], "query": "other.invalid", + "abuse": ["not-an-address"], "source": "rdap"}, + {"iocs": ["ioc-5"], "query": "mixed.invalid", + "abuse": ["broken", "abuse@host.invalid"], "source": "rdap"}, + ] + destinations = report.email_destinations(contacts) + reported = {i for d in destinations for i in d["iocs"]} + flagged = {e["ioc"] for e in report.unreportable(contacts)} + every = {i for c in contacts for i in c["iocs"]} + + self.assertEqual(reported & flagged, set()) + self.assertEqual(reported | flagged, every) + self.assertEqual(reported, {"ioc-1", "ioc-2", "ioc-5"}) + self.assertEqual(flagged, {"ioc-3", "ioc-4"}) + + def test_an_empty_or_whitespace_target_is_not_a_desk(self): + for value in ("", " ", "@host.invalid", "abuse@"): + with self.subTest(value=value): + contacts = [{"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": [value], "source": "rdap"}] + self.assertEqual(report.email_destinations(contacts), []) + self.assertEqual( + report.unreportable(contacts), + [{"ioc": "ioc-1", + "reason": "no usable abuse address published"}]) + + +IDENTITY = {"name": "A Reporter", "org": "Example Consulting", + "email": "reporter@example.org"} + +MANIFEST = { + "format": 1, + "case_id": "2026-09-07-aaaa", + "iocs": [ + {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", + "origin": "received-chain", "confidence": "boundary-hop"}, + {"id": "ioc-2", "type": "url", + "value": "http://login.sender.invalid/verify?id=REDACTED", + "origin": "body"}, + ], + "auth": {"spf": "fail", "dkim": "none", "dmarc": "fail"}, + "headers": [ + ("From", '"Example Bank" <phish@sender.invalid>'), + ("Subject", "Your account requires verification"), + ("Date", "Mon, 07 Sep 2026 09:12:40 +0000"), + ], + "contacts": [ + {"iocs": ["ioc-1", "ioc-2"], "query": "203.0.113.42", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ], +} + +# 120 characters, well past the 72-column wrap, built from .invalid only. +LONG_URL = ("http://very-long-host-name.example.invalid/a/rather/deep/path/" + "segment/tree/verify?campaign=REDACTED&id=REDACTED") + + +class TextPart(unittest.TestCase): + def setUp(self): + destination = report.email_destinations(MANIFEST["contacts"])[0] + self.text = report.text_part(MANIFEST, destination, IDENTITY) + + def test_the_redaction_note_is_always_present(self): + self.assertIn("Recipient identifiers", self.text) + + def test_the_reporter_identity_appears(self): + self.assertIn("A Reporter", self.text) + self.assertIn("Example Consulting", self.text) + self.assertIn("reporter@example.org", self.text) + + def test_the_destinations_own_indicators_appear(self): + self.assertIn("203.0.113.42", self.text) + self.assertIn("http://login.sender.invalid/verify?id=REDACTED", + self.text) + + def test_an_indicator_belonging_to_another_desk_does_not_appear(self): + manifest = dict(MANIFEST) + manifest["iocs"] = MANIFEST["iocs"] + [ + {"id": "ioc-9", "type": "ipv4", "value": "192.0.2.99", + "origin": "received-chain"}, + ] + destination = report.email_destinations(MANIFEST["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + self.assertNotIn("192.0.2.99", text) + + def test_no_line_exceeds_seventy_two_columns(self): + for line in self.text.splitlines(): + self.assertLessEqual(len(line), 72, line) + + # --- the parts the plan got wrong ------------------------------------- + + def _with(self, **fields) -> str: + manifest = copy.deepcopy(MANIFEST) + manifest.update(fields) + destination = report.email_destinations(manifest["contacts"])[0] + return report.text_part(manifest, destination, IDENTITY) + + def test_a_long_url_is_whole_and_still_within_seventy_two_columns(self): + """A truncated URL is a WRONG indicator, not a shortened one. + + The plan wrapped three lines with a `[:72]` slice and left the + indicator list unwrapped. Both halves are the same defect: a desk + acting on a prefix acts on a resource that is not the one reported, + and a prefix reads as complete because nothing says otherwise. + + So the value must survive intact, reassemblable by a reader, and + every line must still fit. Asserting only "the URL is in the text" + would pass on a long unwrapped line, and asserting only the column + limit would pass on a truncation; the two together admit neither. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "url", "value": LONG_URL, + "origin": "body"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + + for line in text.splitlines(): + self.assertLessEqual(len(line), 72, line) + # Rejoining the continuation lines must give back the exact value. + self.assertIn(LONG_URL, report.unwrap(text)) + + def test_a_long_header_value_is_not_truncated(self): + """A header is what the message DECLARED, and a cut one misstates it. + + The subject here is attacker-controlled free text of a length no + column limit accommodates. Truncating it publishes something the + message did not say. + """ + long_subject = ("Your account requires verification before " + "the end of the working day or it will be " + "suspended permanently") + manifest = copy.deepcopy(MANIFEST) + manifest["headers"] = [("Subject", long_subject)] + destination = report.email_destinations(manifest["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + + for line in text.splitlines(): + self.assertLessEqual(len(line), 72, line) + self.assertIn(long_subject, report.unwrap(text)) + + def test_a_long_reporter_identity_is_not_truncated(self): + """The identity is the one thing disclosed deliberately. + + A cut address is an address nobody can reply to, which defeats the + line's only purpose. The plan sliced it at 72. + """ + identity = {"name": "A Reporter With Rather A Long Name", + "org": "Example Consulting And Partners Limited", + "email": "a.reporter@consulting.example.org"} + destination = report.email_destinations(MANIFEST["contacts"])[0] + text = report.text_part(MANIFEST, destination, identity) + + for line in text.splitlines(): + self.assertLessEqual(len(line), 72, line) + joined = report.unwrap(text) + self.assertIn("A Reporter With Rather A Long Name", joined) + self.assertIn("a.reporter@consulting.example.org", joined) + + def test_headers_survive_a_json_round_trip(self): + """case.load() returns lists, not tuples: JSON has no tuple. + + The header block is the one place a pair is destructured, so it is + the one place the round-trip shape can break the report. + """ + manifest = json.loads(json.dumps(MANIFEST)) + self.assertIsInstance(manifest["headers"][0], list) + destination = report.email_destinations(manifest["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + self.assertIn("Your account requires verification", text) + + def test_an_origin_reads_as_english_not_as_an_internal_token(self): + """`header-list_unsubscribe` is a parser's word, not a desk's. + + A desk deciding whether to act needs to know where an indicator was + seen. An internal token with an underscore in it reads as debug + output and makes the whole report look machine-dumped. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "url", "value": "http://a.invalid/x", + "origin": "header-list_unsubscribe"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + self.assertNotIn("header-list_unsubscribe", text) + self.assertIn("List-Unsubscribe", text) + + def test_an_unknown_origin_is_shown_rather_than_dropped(self): + """A newer parse.py may invent an origin this table does not know. + + Dropping it would silently lose the one line saying where an + indicator came from, so an unknown token is shown as-is: ugly beats + absent, and it is visible enough to get the table updated. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "url", "value": "http://a.invalid/x", + "origin": "some-future-origin"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + self.assertIn("some-future-origin", text) + + def test_a_boundary_hop_is_described_as_the_sending_ip(self): + self.assertIn("sending IP", self.text) + + def test_an_ioc_id_a_destination_names_but_the_manifest_lacks(self): + """A manifest is a file the user edits, so the two can disagree. + + The report must still be produced for the indicators that do exist + rather than raising, and must not invent a row for the missing one. + """ + destination = {"id": "email-x", "kind": "email", + "target": "abuse@host.invalid", + "iocs": ["ioc-1", "ioc-404"], "body": None, + "status": "pending"} + text = report.text_part(MANIFEST, destination, IDENTITY) + self.assertIn("203.0.113.42", text) + self.assertNotIn("ioc-404", text) + + def test_a_manifest_with_no_auth_or_headers_still_reports(self): + """Both blocks are optional and an empty one must not print a + heading with nothing under it.""" + text = self._with(auth={}, headers=[]) + self.assertIn("203.0.113.42", text) + self.assertNotIn("Message as declared", text) + self.assertNotIn("Authentication results", text) + + +class BackslashRoundTrip(unittest.TestCase): + """A value's own backslash must never be read as a wrap marker. + + The continuation marker is a trailing "\\", and a URL path may legally + end in one. Until this was fixed the two were indistinguishable, so an + attacker who read this source could append a backslash and make their + own indicator garble itself in the report an abuse desk reads. That is + an adversarial trigger on attacker-supplied text, not an edge case. + + The property asserted throughout is the only one that closes it: + unwrap(text_part(...)) contains the value EXACTLY, for every value, + wrapped or not. Asserting "the value appears" without unwrap, or + asserting only on long values, both leave the short case open, and the + short case is the one that needs no wrapping to corrupt. + """ + + def _render_ioc(self, value: str) -> str: + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [{"id": "ioc-1", "type": "url", "value": value, + "origin": "body"}] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "example.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + return report.text_part(manifest, destination, IDENTITY) + + def _assert_round_trips(self, value: str) -> None: + text = self._render_ioc(value) + for line in text.splitlines(): + self.assertLessEqual(len(line), 72, line) + self.assertIn(value, report.unwrap(text)) + + def test_a_short_value_ending_in_a_backslash_survives(self): + """The case that needs no wrapping at all to corrupt. + + Nothing is wrapped here, yet unwrap() used to eat the following + line, merging the indicator with its own origin annotation and + rendering "http://a.invalid/xseen in a link in the message body". + One corrupt line where there were two, and a wrong indicator. + """ + self._assert_round_trips("http://a.invalid/x\\") + + def test_a_long_value_ending_in_a_backslash_survives(self): + self._assert_round_trips("http://a.invalid/" + "b" * 90 + "\\") + + def test_a_value_with_an_interior_backslash_survives(self): + self._assert_round_trips("http://a.invalid/x\\y/z") + + def test_a_value_ending_in_two_backslashes_survives(self): + """Whatever escaping is chosen must not have its own off-by-one. + + Doubling every backslash makes a trailing pair into four, and a + decoder that consumes them greedily or in the wrong order gives + back one backslash or three. This is the test that catches that. + """ + self._assert_round_trips("http://a.invalid/x\\\\") + + def test_adversarial_values_round_trip_exactly(self): + """A handful of shapes chosen to sit on the seams. + + The two boundary values matter most: a value that exactly fills a + line and one a single character over it are where an off-by-one in + the wrap arithmetic lives, and a backslash landing exactly on the + break column is where escaping and wrapping interact. + """ + indent = 2 + room = 72 - indent + values = [ + "http://a.invalid/x\\", + "http://a.invalid/x\\y/z", + "http://a.invalid/x\\\\", + "\\" + "a" * 40, + "a" * 40 + "\\", + "http://a.invalid/" + "b" * 90 + "\\", + "a" * room, # exactly fills the line + "a" * (room + 1), # one character over + "a" * (room - 1) + "\\", # backslash at the break + "a" * room + "\\", + "\\\\" + "c" * 80 + "\\\\", + ] + for value in values: + with self.subTest(value=value): + self._assert_round_trips(value) + + def test_a_backslash_landing_on_the_break_column_survives(self): + """The case that a passing suite still missed. + + Escaping doubles each backslash, and a break falling BETWEEN the + two halves of a pair splits the run unwrap() counts the parity of. + Both halves are then misread, a real marker reads as content, the + continuation line is orphaned and the tail of the value is silently + dropped. It needs a backslash at exactly the break column, so no + hand-written case found it; a randomised sweep failed 454 of 3538. + + Walking the backslash across every position around the boundary is + what makes this deterministic rather than luck. + """ + room = 72 - 2 # indent is two spaces for an indicator line + for offset in range(-4, 5): + position = room + offset + if position < 1: + continue + value = "a" * position + "\\" + "b" * 30 + with self.subTest(offset=offset): + self._assert_round_trips(value) + + def test_a_run_of_backslashes_across_the_break_survives(self): + """A run is where an off-by-one in the back-off hides. + + Backing off one character is correct only if the character it lands + on is the first half of a pair; a run of three or four exercises + whether the parity test looks at the run rather than at one + character. + """ + room = 72 - 2 + for length in range(1, 6): + for offset in range(-3, 4): + position = room + offset + if position < 1: + continue + value = "a" * position + "\\" * length + "b" * 20 + with self.subTest(length=length, offset=offset): + self._assert_round_trips(value) + + def test_a_value_that_is_entirely_backslashes_survives(self): + """Escaping doubles the length, so this is the worst case for both + the wrap arithmetic and the parity test at once.""" + for length in (1, 2, 3, 34, 35, 36, 70, 71): + with self.subTest(length=length): + self._assert_round_trips("\\" * length) + + def test_an_attacker_subject_cannot_corrupt_the_header_block(self): + """Subject is attacker-controlled and sits beside headers it can eat. + + This is worse than the URL case: a trailing backslash on Subject + used to swallow the following line, rendering + "Subject: Verify nowDate: Mon, 07 Sep 2026 09:12:40 +0000". The + attacker's own text destroys a DIFFERENT field's value, so the + block misstates what the message declared, which is the one thing + that block exists to report faithfully. + """ + subject = "Verify now\\" + manifest = copy.deepcopy(MANIFEST) + manifest["headers"] = [ + ("Subject", subject), + ("Date", "Mon, 07 Sep 2026 09:12:40 +0000"), + ] + destination = report.email_destinations(manifest["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + + for line in text.splitlines(): + self.assertLessEqual(len(line), 72, line) + joined = report.unwrap(text) + self.assertIn(f"Subject: {subject}", joined) + # The Date must survive intact rather than being absorbed. + self.assertIn("Date: Mon, 07 Sep 2026 09:12:40 +0000", joined) + + def test_a_display_name_ending_in_a_backslash_survives(self): + """The From display name is attacker-controlled too, and a sweep + has already found a spoofed one.""" + value = '"Example Bank\\" <phish@sender.invalid>' + manifest = copy.deepcopy(MANIFEST) + manifest["headers"] = [("From", value), + ("Subject", "Your account requires check")] + destination = report.email_destinations(manifest["contacts"])[0] + text = report.text_part(manifest, destination, IDENTITY) + + joined = report.unwrap(text) + self.assertIn(f"From: {value}", joined) + self.assertIn("Subject: Your account requires check", joined) + + def test_unwrap_reads_a_marker_by_parity_not_by_a_trailing_backslash(self): + """unwrap() is PUBLIC, so its input is not only our own output. + + A case manifest is a file the user edits and a desk may script + against the text part, so unwrap() must decide correctly on a line + it did not generate. Inside generated text the wrap back-off means + a marker always follows an even run, so parity and a plain + endswith() agree and neither is distinguishable by a round-trip + test. They disagree here, on a line ending in an escaped pair and + nothing else: that is content, and the following line must NOT be + absorbed into it. + """ + # "a\\" escaped is a value ending in one literal backslash, whole + # on its line. endswith() reads the second half as a marker. + self.assertEqual(report.unwrap(" a\\\\\n next"), " a\\\n next") + # An odd run IS a marker: two escaped halves plus the marker. + self.assertEqual(report.unwrap(" a\\\\\\\n next"), " a\\next") + + def test_an_identity_containing_a_backslash_survives(self): + identity = {"name": "A Reporter\\", "org": "Example Consulting", + "email": "reporter@example.org"} + destination = report.email_destinations(MANIFEST["contacts"])[0] + text = report.text_part(MANIFEST, destination, identity) + self.assertIn("A Reporter\\", report.unwrap(text)) + self.assertIn("Generated by abusectl.", report.unwrap(text)) + + +class FeedbackPart(unittest.TestCase): + def setUp(self): + destination = report.email_destinations(MANIFEST["contacts"])[0] + self.fields = report.feedback_fields(MANIFEST, destination) + self.lookup = dict(self.fields) + + def test_the_three_rfc5965_required_fields_are_present(self): + self.assertEqual(self.lookup["Feedback-Type"], "abuse") + self.assertEqual(self.lookup["Version"], "1") + self.assertTrue(self.lookup["User-Agent"].startswith("abusectl/")) + + def test_the_xarf_report_type_is_phishing(self): + self.assertEqual(self.lookup["Report-Type"], "phishing") + + def test_source_is_the_primary_indicator(self): + self.assertEqual(self.lookup["Source"], "203.0.113.42") + self.assertEqual(self.lookup["Source-IP"], "203.0.113.42") + + def test_every_url_appears_as_a_reported_uri(self): + uris = [value for name, value in self.fields if name == "Reported-Uri"] + self.assertEqual( + uris, ["http://login.sender.invalid/verify?id=REDACTED"] + ) + + def test_a_destination_with_no_ip_omits_source_ip(self): + contacts = [{"iocs": ["ioc-2"], "query": "sender.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}] + destination = report.email_destinations(contacts)[0] + lookup = dict(report.feedback_fields(MANIFEST, destination)) + self.assertNotIn("Source-IP", lookup) + self.assertEqual( + lookup["Source"], "http://login.sender.invalid/verify?id=REDACTED" + ) + + # --- the parts the plan got wrong ------------------------------------- + + def _fields_for(self, iocs: list[dict]) -> list[tuple[str, str]]: + """Render the machine part for a hand-built IOC list. + + Every IOC reaches one destination, so the field list is exactly what + those indicators produce and nothing is filtered out behind the test. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = iocs + manifest["contacts"] = [ + {"iocs": [entry["id"] for entry in iocs], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + return report.feedback_fields(manifest, destination) + + def test_source_ip_is_emitted_at_most_once(self): + """RFC 5965 says Source-IP appears "once maximum". + + The plan emitted one per IP. A strict parser meeting a repeated + single-occurrence field either rejects the part or keeps whichever + occurrence it saw last, so the field a repeat was meant to add is + the field that displaces the primary one. Every IP still travels, + in the text part and in Reported-Uri's sibling below. + """ + fields = self._fields_for([ + {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", + "origin": "received-chain", "confidence": "boundary-hop"}, + {"id": "ioc-2", "type": "ipv4", "value": "203.0.113.43", + "origin": "received-chain"}, + ]) + ips = [v for n, v in fields if n == "Source-IP"] + self.assertEqual(ips, ["203.0.113.42"]) + self.assertEqual(dict(fields)["Source"], "203.0.113.42") + + def test_no_field_appears_twice_unless_the_rfc_allows_it(self): + """The invariant behind the test above, stated once for every field. + + Reported-Uri and Reported-Domain are "any number of times"; every + other field this module emits is once-maximum. Asserting only on + Source-IP would let the next repeated field ship unnoticed. + """ + fields = self._fields_for([ + {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", + "origin": "received-chain"}, + {"id": "ioc-2", "type": "ipv6", "value": "2001:db8::1", + "origin": "received-chain"}, + {"id": "ioc-3", "type": "url", "value": "http://a.invalid/x", + "origin": "body"}, + {"id": "ioc-4", "type": "url", "value": "http://b.invalid/y", + "origin": "body"}, + {"id": "ioc-5", "type": "domain", "value": "a.invalid", + "origin": "header-from"}, + {"id": "ioc-6", "type": "domain", "value": "b.invalid", + "origin": "header-reply_to"}, + ]) + seen: dict[str, int] = {} + for name, _ in fields: + seen[name] = seen.get(name, 0) + 1 + repeatable = {"Reported-Uri", "Reported-Domain"} + for name, count in seen.items(): + if name not in repeatable: + self.assertEqual(count, 1, f"{name} appeared {count} times") + self.assertEqual(seen["Reported-Uri"], 2) + self.assertEqual(seen["Reported-Domain"], 2) + + def test_an_ipv6_indicator_fills_source_ip_too(self): + """"ipv6" is a distinct type string from parse.iocs(). + + A branch testing only for "ipv4" drops every IPv6 sender, and the + given tests use IPv4 throughout so none of them would notice. + """ + fields = dict(self._fields_for([ + {"id": "ioc-1", "type": "ipv6", "value": "2001:db8::1", + "origin": "received-chain"}, + ])) + self.assertEqual(fields["Source"], "2001:db8::1") + self.assertEqual(fields["Source-IP"], "2001:db8::1") + + def test_a_destination_with_no_typed_indicator_omits_source(self): + """An empty Source is worse than an absent one. + + "Source:" with nothing after it asserts that the thing being + reported is the empty string. A 5965 parser reading a present-but- + empty field has been told a value; reading no field it has been + told nothing, which is the truth. Only sha256 and observation + indicators reach a desk here, and both belong in the text part. + """ + fields = self._fields_for([ + {"id": "ioc-1", "type": "observation", + "value": "display-name-carries-address", + "origin": "display-name-from"}, + ]) + lookup = dict(fields) + self.assertNotIn("Source", lookup) + self.assertNotIn("Source-IP", lookup) + # The envelope is still well formed: a desk gets a valid part. + self.assertEqual(lookup["Feedback-Type"], "abuse") + self.assertEqual(lookup["Version"], "1") + + def test_a_type_this_module_does_not_place_is_not_invented_into_one(self): + """sha256 and observation have no 5965 or x-arf field. + + Neither is a Source, a Reported-Uri or a Reported-Domain, and + forcing one into the nearest-looking field would tell a desk that a + file hash is a URI. They travel in the human part, which is where a + desk reads what an attachment was. + """ + fields = self._fields_for([ + {"id": "ioc-1", "type": "ipv4", "value": "203.0.113.42", + "origin": "received-chain"}, + {"id": "ioc-2", "type": "sha256", "value": "a" * 64, + "origin": "attachment", "filename": "invoice.zip"}, + {"id": "ioc-3", "type": "observation", + "value": "display-name-carries-address", + "origin": "display-name-from"}, + ]) + blob = repr(fields) + self.assertNotIn("a" * 64, blob) + self.assertNotIn("display-name-carries-address", blob) + self.assertEqual(dict(fields)["Source"], "203.0.113.42") + + def test_an_ioc_id_the_manifest_lacks_is_skipped_not_raised(self): + """A manifest is a file the user edits, so the two can disagree. + + text_part() already tolerates this; the machine part indexed with + destination["iocs"] straight into a dict would raise instead, and + the two parts of one document must not disagree about whether the + case can be reported at all. + """ + destination = {"id": "email-x", "kind": "email", + "target": "abuse@host.invalid", + "iocs": ["ioc-1", "ioc-404"], "body": None, + "status": "pending"} + lookup = dict(report.feedback_fields(MANIFEST, destination)) + self.assertEqual(lookup["Source"], "203.0.113.42") + self.assertNotIn("ioc-404", repr(lookup)) + + def test_a_destination_with_no_iocs_key_still_renders(self): + """destination.get("iocs"), not destination["iocs"].""" + destination = {"id": "email-x", "kind": "email", + "target": "abuse@host.invalid", "body": None, + "status": "pending"} + lookup = dict(report.feedback_fields(MANIFEST, destination)) + self.assertEqual(lookup["Feedback-Type"], "abuse") + + def test_arrival_date_is_not_taken_from_the_senders_date_header(self): + """RFC 5965: Arrival-Date is when the generating ADMD's MTA received + the message. The Date header is when the SENDER CLAIMS it was sent. + + The plan copied Date into Arrival-Date. On a phishing message that + header is attacker-controlled free text, so the report would assert + as our own observation a timestamp the attacker chose, and a desk + correlating it against their own logs would look in the wrong place + or find nothing and discount the report. + + The honest source is the boundary Received hop's own timestamp, + which parse.report_headers() already publishes. Parsing one is a + date parser this task does not need, so the field is OMITTED: 5965 + makes it optional, and an absent optional field misstates nothing. + """ + lookup = dict(self.fields) + self.assertNotIn("Arrival-Date", lookup) + self.assertNotIn("Mon, 07 Sep 2026 09:12:40 +0000", repr(lookup)) + + +class FeedbackInjection(unittest.TestCase): + """A field value carrying a line break forges a field in the report. + + This is not hypothetical and it is not stopped upstream. redact.py + URL-DECODES a redirector's destination parameter to recover it as an + indicator, so a message body carrying + + http://r.invalid/go?next=http%3A%2F%2Fa.invalid%2Fx%0AFeedback-Type... + + produces, through parse.iocs() on a real .eml, an IOC whose value is + "http://a.invalid/x\\nFeedback-Type: not-abuse". Emitted verbatim, the + abuse desk's parser reads a Feedback-Type this tool never asserted, on a + report that carries the reporter's identity. That is an attacker writing + fields into mail sent under our name. + + The answer here is to PERCENT-ENCODE the control characters rather than + to drop the indicator or strip them. Dropping loses a real redirect + target; stripping silently rewrites an indicator into a different one a + desk would then act on. Percent-encoding is the URL's own native + encoding, is exactly reversible, and leaves the value visibly altered + rather than quietly wrong. + """ + + def _value_out(self, value: str) -> str | None: + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [{"id": "ioc-1", "type": "url", "value": value, + "origin": "redirect-target"}] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + fields = report.feedback_fields(manifest, destination) + for name, out in fields: + if name == "Reported-Uri": + return out + return None + + def _assert_no_break(self, fields: list[tuple[str, str]]) -> None: + for name, value in fields: + for bad in ("\r", "\n", "
", "
", "\v", "\f", + "\x1c", "\x1d", "\x1e", "\x85"): + self.assertNotIn(bad, name) + self.assertNotIn(bad, value) + + def test_a_newline_in_a_value_cannot_forge_a_field(self): + out = self._value_out("http://a.invalid/x\nFeedback-Type: not-abuse") + self.assertNotIn("\n", out) + self.assertIn("%0A", out) + # The forged field name must not survive as a line of its own, but + # the text of the indicator is still legible and reversible. + self.assertEqual(out, + "http://a.invalid/x%0AFeedback-Type: not-abuse") + + def test_the_real_parse_output_that_makes_this_reachable(self): + """End to end from an .eml, not from a hand-written IOC. + + A test that only feeds feedback_fields() a crafted string proves the + encoder works; it does not prove the encoder is needed. This runs + the actual redirector through parse.iocs() so the fixture and the + defence cannot drift apart. + """ + from abusectl import parse + raw = ( + "Received: from evil.invalid ([203.0.113.9]) by mx.example.org; " + "Mon, 07 Sep 2026 09:12:40 +0000\r\n" + "From: <phish@sender.invalid>\r\n" + "Subject: verify\r\n" + "Date: Mon, 07 Sep 2026 09:12:40 +0000\r\n" + "Content-Type: text/plain\r\n\r\n" + "http://r.invalid/go?next=http%3A%2F%2Fa.invalid%2Fx%0A" + "Feedback-Type%3A%20not-abuse\r\n" + ).encode() + iocs = parse.iocs(raw, trusted=["192.0.2.0/24"]) + injected = [e for e in iocs if "\n" in e["value"]] + self.assertTrue(injected, "the injection vector itself has changed") + + manifest = {"format": 1, "iocs": iocs, "headers": [], "auth": {}} + destination = {"id": "email-x", "kind": "email", + "target": "abuse@host.invalid", + "iocs": [e["id"] for e in iocs], "body": None, + "status": "pending"} + fields = report.feedback_fields(manifest, destination) + self._assert_no_break(fields) + + def test_every_line_breaking_shape_is_neutralised(self): + """The adversarial sweep, not a handful of cases. + + U+2028 and U+2029 are in here because Python's own email module + raises on them: str.splitlines() treats them as breaks, so a value + carrying one would make the whole document fail to assemble in + Task 6 rather than merely render oddly. + """ + breaks = ["\n", "\r", "\r\n", "\n\r", "
", "
", + "\v", "\f", "\x1c", "\x1d", "\x1e", "\x85"] + shapes = [] + for brk in breaks: + shapes += [ + brk, + "http://a.invalid/x" + brk, + brk + "http://a.invalid/x", + "http://a.invalid/x" + brk + "Feedback-Type: not-abuse", + "http://a.invalid/" + brk * 3 + "Source: 192.0.2.1", + ] + for value in shapes: + with self.subTest(value=repr(value)): + out = self._value_out(value) + self.assertIsNotNone(out) + for bad in breaks: + if len(bad) == 1: + self.assertNotIn(bad, out) + + def test_a_value_that_is_only_a_newline_still_yields_a_field(self): + """It must not become an empty value or vanish silently.""" + out = self._value_out("\n") + self.assertEqual(out, "%0A") + + def test_encoding_is_reversible_so_the_indicator_is_not_misstated(self): + """The property that makes encoding honest rather than a strip. + + A desk, or a later submit path, must be able to recover exactly what + the message declared. Stripping the character would pass every + assertion above and hand the desk a DIFFERENT URL. + """ + from urllib.parse import unquote + for value in ("http://a.invalid/x\nFeedback-Type: not-abuse", + "http://a.invalid/\r\n\r\n", + "http://a.invalid/x
y"): + with self.subTest(value=repr(value)): + self.assertEqual(unquote(self._value_out(value)), value) + + def test_a_literal_percent_is_encoded_so_the_reversal_is_unambiguous(self): + """Without this, "%0A" typed by the attacker decodes to a newline. + + A redacted URL legitimately contains percent signs, and an encoder + that leaves them alone produces text that unquote() turns into the + very control character the encoder existed to remove. The reversal + must be a true inverse or it is a second injection one step later. + """ + from urllib.parse import unquote + value = "http://a.invalid/x?a=%0AFeedback-Type: not-abuse" + out = self._value_out(value) + self.assertNotIn("\n", out) + self.assertEqual(unquote(out), value) + + def test_an_injected_field_name_in_a_domain_is_neutralised_too(self): + """Reported-Domain and Source take the same path as Reported-Uri. + + The defence must not live in one branch. That is the exact shape of + the fourth property's three leaks: validation applied per branch + gets forgotten on the next branch. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "domain", + "value": "a.invalid\nSource: 192.0.2.1", + "origin": "header-from"}, + {"id": "ioc-2", "type": "ipv4", + "value": "203.0.113.42\nVersion: 9", + "origin": "received-chain"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1", "ioc-2"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + fields = report.feedback_fields(manifest, destination) + self._assert_no_break(fields) + self.assertEqual(dict(fields)["Version"], "1") + lookup = dict(fields) + self.assertEqual(lookup["Source"], "203.0.113.42%0AVersion: 9") + self.assertEqual(lookup["Reported-Domain"], + "a.invalid%0ASource: 192.0.2.1") + + def test_the_rendered_part_survives_pythons_own_header_setter(self): + """The end the whole defence is for: Task 6 assembles with email. + + EmailMessage raises ValueError on a header value containing a break, + so an unencoded value does not merely render oddly, it aborts the + document. Asserting through the real setter is what makes this a + test of the outcome rather than of my own notion of a break. + """ + from email.message import EmailMessage + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "url", + "value": "http://a.invalid/x\r\nFeedback-Type: not-abuse
z", + "origin": "redirect-target"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + part = EmailMessage() + for name, value in report.feedback_fields(manifest, destination): + part[name] = value + rendered = part.as_string() + # The property is per LINE, not per substring: the attacker's text + # may legitimately appear INSIDE a value, and asserting it absent + # would forbid reporting a URL that merely contains the words. What + # must not exist is a line a parser reads as a field of its own. + names = [line.split(":", 1)[0] for line in rendered.splitlines() + if line and not line[0].isspace() and ":" in line] + self.assertEqual(names.count("Feedback-Type"), 1) + self.assertEqual( + [n for n in names if n == "Feedback-Type"], ["Feedback-Type"]) + for line in rendered.splitlines(): + self.assertNotEqual(line.strip(), "Feedback-Type: not-abuse") + + def test_a_field_name_is_never_taken_from_data(self): + """Names are literals in this module, so no input can invent one. + + Pinned because the obvious "generalise it" refactor is a table + mapping an IOC's own type string to a field name, and a manifest is + a file the user edits: a type of "x: y\\nFeedback-Type" would then + BE a field name. The set is closed on purpose. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "url\nFeedback-Type", "value": "x", + "origin": "body"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + fields = report.feedback_fields(manifest, destination) + self.assertEqual( + {name for name, _ in fields}, + {"Feedback-Type", "User-Agent", "Version", "Report-Type"}) + + def test_a_non_string_value_does_not_crash_the_report(self): + """A manifest is edited by hand and JSON has numbers. + + Not a security property, but a report that raises produces nothing + at all, and this is the one module standing between a reviewed case + and a sent mail. + """ + manifest = copy.deepcopy(MANIFEST) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "ipv4", "value": 42, + "origin": "received-chain"}, + ] + manifest["contacts"] = [ + {"iocs": ["ioc-1"], "query": "x.invalid", + "abuse": ["abuse@host.invalid"], "source": "rdap"}, + ] + destination = report.email_destinations(manifest["contacts"])[0] + lookup = dict(report.feedback_fields(manifest, destination)) + self.assertEqual(lookup["Source"], "42") + + +class Document(unittest.TestCase): + def setUp(self): + self.destination = report.email_destinations(MANIFEST["contacts"])[0] + self.raw = report.build(MANIFEST, self.destination, IDENTITY) + self.parsed = email.message_from_string( + self.raw, policy=email.policy.default + ) + + def test_it_is_a_feedback_report_with_three_parts(self): + self.assertEqual(self.parsed.get_content_type(), "multipart/report") + self.assertEqual(self.parsed.get_param("report-type"), + "feedback-report") + parts = list(self.parsed.iter_parts()) + self.assertEqual( + [part.get_content_type() for part in parts], + ["text/plain", "message/feedback-report", "text/rfc822-headers"], + ) + + def test_the_envelope_is_addressed_and_identified(self): + self.assertEqual(self.parsed["To"], "abuse@host.invalid") + self.assertIn("reporter@example.org", self.parsed["From"]) + self.assertTrue(self.parsed["Subject"]) + + def test_the_source_message_is_never_attached(self): + self.assertNotIn("message/rfc822", self.raw) + + def test_the_headers_part_carries_what_the_manifest_declared(self): + """The third part is the whitelisted headers, unmangled. + + Asserted by PARSING the part as headers rather than by looking for + substrings, because what a desk does with this part is parse it. + """ + headers = self._headers_of(self.parsed) + self.assertEqual(headers.keys(), ["From", "Subject", "Date"]) + self.assertEqual(headers["Subject"], + "Your account requires verification") + self.assertEqual(headers["Date"], "Mon, 07 Sep 2026 09:12:40 +0000") + # The From is asserted twice over: on the wire text, which is what + # is actually published, and on the address a desk would act on. + # A structured header re-renders the display name's quoting on + # read-back, so only the raw text pins what was written. + self.assertIn('From: "Example Bank" <phish@sender.invalid>', self.raw) + self.assertEqual([a.addr_spec for a in headers["From"].addresses], + ["phish@sender.invalid"]) + + @staticmethod + def _headers_of(parsed): + """Parse the third part's body the way a desk's parser would.""" + body = list(parsed.iter_parts())[2].get_content() + return email.message_from_string(body, policy=email.policy.default) + + +class DocumentHeaders(unittest.TestCase): + """The third part: what a hand-edited manifest can and cannot publish.""" + + def _build(self, headers): + manifest = copy.deepcopy(MANIFEST) + manifest["headers"] = headers + destination = report.email_destinations(manifest["contacts"])[0] + return report.build(manifest, destination, IDENTITY) + + def _part(self, raw, index=2): + parsed = email.message_from_string(raw, policy=email.policy.default) + return list(parsed.iter_parts())[index] + + def test_a_recipient_header_in_the_manifest_is_not_published(self): + """A manifest is a FILE THE USER EDITS, so it can carry a "To". + + parse.report_headers() would never produce one, which is exactly + why asserting on its output proves nothing: the property has to + survive a manifest nobody generated. The victim of getting this + wrong is the recipient, whose address reaches the abuse desk and + through it the attacker. + """ + raw = self._build([ + ("To", "victim@example.org"), + ("Cc", "other@example.org"), + ("Delivered-To", "victim@example.org"), + ("X-Original-To", "victim@example.org"), + ("Subject", "kept"), + ]) + body = self._part(raw).get_content() + self.assertNotIn("victim", raw) + self.assertNotIn("other@example.org", raw) + published = email.message_from_string(body, + policy=email.policy.default) + self.assertEqual(published.keys(), ["Subject"]) + + def test_a_newline_in_a_value_cannot_forge_a_header(self): + """Subject is attacker-controlled free text and is kept deliberately. + + Emitted verbatim it forges a header line in a part whose entire + content is read as headers. The value must survive intact and the + header list must not grow. + """ + raw = self._build([ + ("Subject", "lure\nFrom: forged@attacker.invalid"), + ]) + published = email.message_from_string( + self._part(raw).get_content(), policy=email.policy.default) + self.assertEqual(published.keys(), ["Subject"]) + self.assertEqual(published["Subject"], + "lure From: forged@attacker.invalid") + self.assertEqual(published["From"], None) + + def test_every_break_character_is_neutralised(self): + """Not only LF. Python's own parsers break on more than RFC 5322 does. + + One header in, one header out, for each character in turn. + """ + for ch in ("\r", "\n", "\r\n", "\v", "\f", "\x1c", "\x1d", "\x1e", + "\x85", "
", "
"): + with self.subTest(ch=repr(ch)): + raw = self._build([("Subject", f"a{ch}From: forged@x.invalid")]) + published = email.message_from_string( + self._part(raw).get_content(), + policy=email.policy.default) + self.assertEqual(published.keys(), ["Subject"]) + self.assertEqual(published["From"], None) + + def test_an_ordinary_value_is_left_legible(self): + """RFC 2047 is applied only when it is needed. + + Encoding every header would render an ordinary Subject as + "=?utf-8?q?..." and cost the desk the legibility this part is for. + """ + raw = self._build([("Subject", "Your account requires verification")]) + # Asserted on the THIRD PART's own text, not on the whole document. + # The text part prints the same header under "Message as declared", + # so a whole-document substring passes even when this part is + # entirely encoded, which a mutation confirmed. + body = self._part(raw).get_content() + self.assertEqual(body.splitlines(), + ["Subject: Your account requires verification"]) + self.assertNotIn("=?utf-8?", body) + + def test_a_case_with_no_headers_omits_the_part(self): + """An empty third part claims the message declared no headers. + + That is never true of a real message. Absent says the report + carries none, which is the truth for a case parsed before the + headers block existed. + """ + for headers in ([], None): + with self.subTest(headers=headers): + raw = self._build(headers) + parsed = email.message_from_string( + raw, policy=email.policy.default) + self.assertEqual( + [p.get_content_type() for p in parsed.iter_parts()], + ["text/plain", "message/feedback-report"]) + self.assertNotIn("rfc822-headers", raw) + + def test_a_manifest_carrying_only_unpublishable_headers_omits_the_part( + self): + """The filter must not leave an empty part behind either.""" + raw = self._build([("To", "victim@example.org")]) + parsed = email.message_from_string(raw, policy=email.policy.default) + self.assertEqual( + [p.get_content_type() for p in parsed.iter_parts()], + ["text/plain", "message/feedback-report"]) + self.assertNotIn("victim", raw) + + def test_a_header_name_is_matched_case_insensitively(self): + """A header name is case-insensitive, and a hand edit will not match. + + Both directions matter: a lowercased "subject" must still publish, + and an uppercased "TO" must still be refused. + """ + raw = self._build([("subject", "lower"), ("SUBJECT", "upper"), + ("Subject", "mixed"), ("TO", "victim@example.org")]) + published = email.message_from_string( + self._part(raw).get_content(), policy=email.policy.default) + # All three spellings publish. The whitelist is stored lowercase, so + # a case-SENSITIVE comparison would still admit "subject" and the + # assertion would pass while the other two vanished; a mutation + # found exactly that. The capitalised spellings are what pin it. + self.assertEqual(published.keys(), ["subject", "SUBJECT", "Subject"]) + self.assertNotIn("victim", raw) + + +class DocumentIdentity(unittest.TestCase): + """The From header: the one identifier disclosed deliberately.""" + + def _from(self, identity): + destination = report.email_destinations(MANIFEST["contacts"])[0] + raw = report.build(MANIFEST, destination, identity) + parsed = email.message_from_string(raw, policy=email.policy.default) + return parsed["From"].addresses + + def test_a_comma_in_the_org_name_does_not_split_the_address(self): + """"Example Consulting, Ltd" is a legitimate name, and a comma is + the address-list separator. + + Formatted into an f-string it parses back as TWO addresses, the + first a bogus addr-spec with no domain, and a desk replying to the + report replies to nobody. + """ + addresses = self._from({"name": "Example Consulting, Ltd", + "org": "Example Consulting", + "email": "reporter@example.org"}) + self.assertEqual(len(addresses), 1) + self.assertEqual(addresses[0].addr_spec, "reporter@example.org") + self.assertEqual(addresses[0].display_name, "Example Consulting, Ltd") + + def test_awkward_names_still_yield_one_reachable_address(self): + for name in ('A "Quoted" Reporter', "Angle <brackets>", "Dänilo Ü", + "Back\\slash", "semi;colon", "at@sign"): + with self.subTest(name=name): + addresses = self._from({"name": name, "org": "o", + "email": "reporter@example.org"}) + self.assertEqual(len(addresses), 1) + self.assertEqual(addresses[0].addr_spec, + "reporter@example.org") + self.assertEqual(addresses[0].display_name, name) + + +if __name__ == "__main__": + unittest.main() |