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
report-type keeps a standards parser working.

Field values are percent-encoded against header injection. This is reachable,
not theoretical: redact.url_valued_parameters() URL-decodes a redirector's
destination to recover it as an indicator, so a body carrying
"?next=http%3A%2F%2Fa.invalid%2Fx%0AFeedback-Type%3A%20not-abuse" produces,
through parse.iocs() on a real message, an IOC whose value holds a literal
newline followed by text shaped like a field. Emitted verbatim it forges a
field in a report carrying the reporter's identity. Encoding rather than
dropping keeps a genuine redirect target; encoding rather than stripping
keeps the indicator recoverable, since a stripped URL is a different URL a
desk would then act on. The escape covers everything str.splitlines() breaks
on, because Python's own email module raises on U+2028 as readily as on LF,
and it runs over UTF-8 bytes and escapes "%" so the reversal is a true
inverse rather than a second injection one step later.

Four corrections to the plan. Source-IP is emitted once, not once per IP:
RFC 5965 gives it "once maximum" and a repeat displaces the primary rather
than adding an address. Source is omitted when there is no typed indicator
rather than emitted empty, since an empty field asserts that the reported
thing is the empty string. Arrival-Date is dropped rather than copied from
the sender's Date header, which is attacker-controlled and means when the
sender claimed to send, not when our MTA received. sha256 and observation
are left to the text part rather than forced into the nearest field.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Xj1ayFRSUQ2u7cwb3S4axE

1 files changed, 455 insertions, 0 deletions

diff --git a/tests/test_report.py b/tests/test_report.py
index 3b13d11..f414841 100644
--- a/tests/test_report.py
+++ b/tests/test_report.py
@@ -922,5 +922,460 @@ class BackslashRoundTrip(unittest.TestCase):
         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")
+
+
 if __name__ == "__main__":
     unittest.main()