path: root/tests/test_report.py

import copy
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"}])


if __name__ == "__main__":
    unittest.main()