From b0031f905eb63fc76aa6cad0421c4623ebad2b88 Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Wed, 9 Sep 2026 19:25:12 +0200 Subject: 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. Nothing is truncated. The plan cut three lines with a [:72] slice and left the indicator list unwrapped, which is the same defect twice: a cut URL is a WRONG indicator rather than a short one, and a desk acting on the first 72 characters acts on a resource nobody reported. Long values are divided with an explicit trailing-backslash continuation instead, and unwrap() is the exact inverse, so the tests assert reassembly rather than mere presence. That is what rules out a truncation passing as a wrap. Origins are mapped to English. "header-list_unsubscribe" is a parser's vocabulary and reads as debug output; an origin the table does not know is shown as-is, because losing the only line saying where an indicator came from is worse than showing an ugly token. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xj1ayFRSUQ2u7cwb3S4axE --- abusectl/report.py | 196 +++++++++++++++++++++++++++++++++++++++++++++++ tests/test_report.py | 209 +++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 405 insertions(+) diff --git a/abusectl/report.py b/abusectl/report.py index 76b195b..7410ae2 100644 --- a/abusectl/report.py +++ b/abusectl/report.py @@ -29,6 +29,111 @@ itself is a module that can disclose it in a code path nobody reviewed. import hashlib +# 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. +_CONTINUATION = "\\" + + +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 never altered, only divided; unwrap() is the exact inverse + and there is a test asserting the round trip on a 120-character URL. + """ + room = _WIDTH - len(indent) - len(_CONTINUATION) + if len(indent) + len(value) <= _WIDTH: + return [indent + value] + lines = [] + while len(value) > room: + lines.append(indent + value[:room] + _CONTINUATION) + value = value[room:] + 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. + """ + out = [] + for line in text.splitlines(): + if out and out[-1].endswith(_CONTINUATION): + out[-1] = out[-1][:-len(_CONTINUATION)] + line.lstrip() + else: + out.append(line) + return "\n".join(out) + + +# 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. @@ -229,3 +334,94 @@ def unreportable(contacts: list[dict]) -> list[dict]: 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" diff --git a/tests/test_report.py b/tests/test_report.py index b572b54..35c68ba 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -1,4 +1,5 @@ import copy +import json import unittest from abusectl import report @@ -514,5 +515,213 @@ class MalformedAddresses(unittest.TestCase): "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" '), + ("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) + + if __name__ == "__main__": unittest.main() -- cgit v1.2.3