aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorDanilo M. <danix@danix.xyz>2026-09-09 19:25:12 +0200
committerDanilo M. <danix@danix.xyz>2026-09-09 19:25:12 +0200
commitb0031f905eb63fc76aa6cad0421c4623ebad2b88 (patch)
treed1fe969903fc72a398d151ccc07665f4f5d138f9
parent725c8ee4a3dff16c54bae6723a12ca2ef17d7e2f (diff)
downloadabusectl-b0031f905eb63fc76aa6cad0421c4623ebad2b88.tar.gz
abusectl-b0031f905eb63fc76aa6cad0421c4623ebad2b88.zip
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Xj1ayFRSUQ2u7cwb3S4axE
-rw-r--r--abusectl/report.py196
-rw-r--r--tests/test_report.py209
2 files changed, 405 insertions, 0 deletions
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" <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)
+
+
if __name__ == "__main__":
unittest.main()