path: root/tests/test_report.py

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()