path: root/tests/test_parse.py

# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""Tests for the Received-chain trust boundary walk."""

import pathlib
import unittest

from abusectl import parse

FIXTURES = pathlib.Path(__file__).parent / "fixtures"


def load(name: str) -> bytes:
    return (FIXTURES / name).read_bytes()


class TestReceivedChain(unittest.TestCase):
    def test_hops_are_returned_outermost_first(self):
        hops = parse.received_hops(load("simple.eml"))
        self.assertEqual([h.ip for h in hops], ["192.0.2.11", "203.0.113.42"])

    def test_the_first_untrusted_hop_is_the_sender(self):
        ip = parse.sending_ip(load("simple.eml"), trusted=["192.0.2.0/24"])
        self.assertEqual(ip, "203.0.113.42")

    def test_a_forged_chain_stops_at_the_first_untrusted_hop(self):
        # The attacker prepended two hops naming an innocent third party.
        # Walking past the boundary would report 198.51.100.7, which is
        # someone else's address in a header the attacker wrote.
        ip = parse.sending_ip(load("forged-chain.eml"), trusted=["192.0.2.0/24"])
        self.assertEqual(ip, "203.0.113.99")

    def test_hops_below_the_boundary_are_still_recorded(self):
        # Recorded, but as untrusted: they may be useful and must not be
        # presented as fact.
        hops = parse.received_hops(load("forged-chain.eml"))
        self.assertEqual(
            [h.ip for h in hops],
            ["192.0.2.11", "203.0.113.99", "198.51.100.7", "198.51.100.8"],
        )

    def test_no_trusted_relays_is_an_error_not_a_guess(self):
        # Guessing the outermost public IP is wrong in exactly the case that
        # matters, and a confident wrong answer gets a third party reported.
        with self.assertRaises(parse.NoTrustBoundary):
            parse.sending_ip(load("simple.eml"), trusted=[])

    def test_a_chain_entirely_inside_the_boundary_has_no_sender(self):
        ip = parse.sending_ip(
            load("simple.eml"), trusted=["192.0.2.0/24", "203.0.113.0/24"]
        )
        self.assertIsNone(ip)

    def test_a_helo_literal_does_not_beat_the_observed_address(self):
        # Postfix writes the client's own HELO string first and the address
        # it actually observed second. The first is attacker-chosen.
        raw = (
            b"Received: from [198.51.100.7] (unknown [203.0.113.99])"
            b" by mx.example.org with ESMTP id X;"
            b" Tue, 8 Sep 2026 10:00:00 +0200\r\n"
            b"From: <a@evil.invalid>\r\nSubject: t\r\n\r\nbody\r\n"
        )
        self.assertEqual(parse.sending_ip(raw, trusted=["192.0.2.0/24"]),
                         "203.0.113.99")

    def test_a_single_bracketed_address_still_works(self):
        raw = (
            b"Received: from x.invalid (x.invalid [203.0.113.5])"
            b" by mx.example.org with ESMTP id Y;"
            b" Tue, 8 Sep 2026 10:00:00 +0200\r\n"
            b"From: <a@evil.invalid>\r\nSubject: t\r\n\r\nbody\r\n"
        )
        self.assertEqual(parse.sending_ip(raw, trusted=["192.0.2.0/24"]),
                         "203.0.113.5")


class TestSenderDomains(unittest.TestCase):
    def test_the_three_sender_headers_are_collected(self):
        domains = parse.sender_domains(load("simple.eml"))
        self.assertEqual(
            domains,
            {
                "return_path": "sender.example.invalid",
                "from": "bank.example.invalid",
                "reply_to": "drop.example.invalid",
            },
        )

    def test_sender_is_collected_when_it_differs_from_from(self):
        # Sender names the party who actually injected the message, which on
        # a spam run is often a relay distinct from the forged From.
        domains = parse.sender_domains(load("leaky.eml"))
        self.assertEqual(domains["sender"], "relay.example.invalid")

    def test_reply_to_is_absent_when_it_matches_from(self):
        # Only a DIFFERING Reply-To is an indicator; repeating From adds noise.
        domains = parse.sender_domains(load("with-attachment.eml"))
        self.assertNotIn("reply_to", domains)

    def test_recipient_headers_are_never_returned(self):
        # The safety property, asserted rather than assumed.
        domains = parse.sender_domains(load("simple.eml"))
        self.assertNotIn("example.org", domains.values())

    def test_the_domain_comes_from_the_address_not_the_display_name(self):
        # A display name is attacker-controlled and sits BEFORE the angle
        # brackets, so a regex scanning the raw header finds it first. Doing
        # that files the report against whoever the attacker named.
        domains = parse.sender_domains(load("leaky.eml"))
        self.assertEqual(domains["from"], "sender.example.invalid")

    def test_a_display_name_address_is_kept_as_its_own_indicator(self):
        # Spoofing a recognisable address in the display name is a real
        # signal, so it is reported, but as a spoof rather than as a sender.
        spoofed = parse.display_name_addresses(load("leaky.eml"))
        self.assertEqual(spoofed, {"from": "you@example.org"})

    def test_the_spoof_reaches_the_iocs_as_a_flag_carrying_no_value(self):
        # The impersonated identity can be the recipient's own, so the IOC
        # records only THAT it happened. A domain here would leak in the
        # exact case the flag exists to report.
        iocs = parse.iocs(load("leaky.eml"), trusted=["192.0.2.0/24"])
        flags = [i for i in iocs if i["type"] == "observation"]
        self.assertEqual(len(flags), 1)
        self.assertEqual(flags[0]["origin"], "display-name-from")
        self.assertNotIn("example.org", flags[0]["value"])


class TestAuthResults(unittest.TestCase):
    def test_verdicts_are_read_as_the_server_recorded_them(self):
        auth = parse.auth_results(load("simple.eml"))
        self.assertEqual(auth, {"spf": "fail", "dkim": "none", "dmarc": "fail"})

    def test_a_message_with_no_auth_header_reports_nothing(self):
        self.assertEqual(parse.auth_results(load("with-attachment.eml")), {})


class TestUrls(unittest.TestCase):
    def test_an_href_is_found_and_redacted(self):
        urls = parse.urls(load("simple.eml"))
        self.assertEqual(
            urls,
            ["http://login.bank-verify.example.invalid/verify?id=REDACTED"],
        )

    def test_a_plain_text_url_is_found_and_redacted(self):
        urls = parse.urls(load("forged-chain.eml"))
        self.assertEqual(urls, ["http://evil.example.invalid/go?u=REDACTED"])

    def test_urls_are_deduplicated_and_ordered(self):
        raw = (
            b"From: <a@b.example.invalid>\r\n"
            b"Subject: t\r\n"
            b"Content-Type: text/plain\r\n\r\n"
            b"http://z.example.invalid/ and http://a.example.invalid/ and "
            b"http://z.example.invalid/ again\r\n"
        )
        self.assertEqual(
            parse.urls(raw),
            ["http://a.example.invalid/", "http://z.example.invalid/"],
        )


class TestRedirectChains(unittest.TestCase):
    def test_a_declared_target_is_recovered_as_a_hop(self):
        chains = parse.redirect_chains(load("redirector.eml"))
        self.assertEqual(len(chains), 1)
        source, target = chains[0]
        self.assertTrue(source.startswith("http://t.example.invalid/c"))
        self.assertTrue(target.startswith("http://evil.example.invalid/pay"))

    def test_the_recovered_target_is_itself_redacted(self):
        _, target = parse.redirect_chains(load("redirector.eml"))[0]
        self.assertEqual(target, "http://evil.example.invalid/pay?ref=REDACTED")

    def test_the_recipient_token_does_not_survive(self):
        # The whole point: the destination is an indicator, the token is not.
        chains = parse.redirect_chains(load("redirector.eml"))
        self.assertNotIn("dGVzdEBleGFtcGxlLm9yZw", repr(chains))

    def test_a_message_with_no_redirector_reports_none(self):
        self.assertEqual(parse.redirect_chains(load("simple.eml")), [])


class TestAttachments(unittest.TestCase):
    def test_filename_and_sha256_are_recorded(self):
        found = parse.attachments(load("with-attachment.eml"))
        self.assertEqual(len(found), 1)
        self.assertEqual(found[0].filename, "invoice.pdf")
        self.assertEqual(
            found[0].sha256,
            "315f5bdb76d078c43b8ac0064e4a0164612b1fce77c869345bfc94c75894edd3",
        )

    def test_a_message_with_no_attachment_reports_none(self):
        self.assertEqual(parse.attachments(load("simple.eml")), [])


class TestIocAssembly(unittest.TestCase):
    def test_every_ioc_has_a_unique_id_and_an_origin(self):
        iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"])
        ids = [i["id"] for i in iocs]
        self.assertEqual(len(ids), len(set(ids)))
        self.assertTrue(all(i["origin"] for i in iocs))

    def test_the_sending_ip_is_present_and_marked_boundary_hop(self):
        iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"])
        ips = [i for i in iocs if i["type"] == "ipv4"]
        self.assertEqual(ips[0]["value"], "203.0.113.42")
        self.assertEqual(ips[0]["confidence"], "boundary-hop")

    def test_hops_below_the_boundary_are_marked_untrusted(self):
        iocs = parse.iocs(load("forged-chain.eml"), trusted=["192.0.2.0/24"])
        ips = {i["value"]: i for i in iocs if i["type"] == "ipv4"}
        self.assertEqual(ips["203.0.113.99"]["confidence"], "boundary-hop")
        self.assertEqual(ips["198.51.100.7"]["confidence"], "untrusted-hop")

    def test_our_own_relays_are_not_reported_as_indicators(self):
        # Inside the boundary is our own infrastructure, not an indicator.
        iocs = parse.iocs(load("forged-chain.eml"), trusted=["192.0.2.0/24"])
        values = [i["value"] for i in iocs]
        self.assertNotIn("192.0.2.11", values)

    def test_urls_carry_their_redacted_form(self):
        iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"])
        urls = [i for i in iocs if i["type"] == "url"]
        self.assertEqual(len(urls), 1)
        self.assertIn("REDACTED", urls[0]["value"])

    def test_a_suspect_path_segment_is_flagged_on_the_ioc(self):
        iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"])
        url = next(i for i in iocs if i["type"] == "url")
        self.assertEqual(url["suspect_path_segments"], ["dGVzdEBleGFtcGxlLm9yZw"])

    def test_a_redirect_target_is_its_own_ioc(self):
        iocs = parse.iocs(load("redirector.eml"), trusted=["192.0.2.0/24"])
        targets = [i for i in iocs if i["origin"] == "redirect-target"]
        self.assertEqual(len(targets), 1)
        self.assertEqual(targets[0]["value"],
                         "http://evil.example.invalid/pay?ref=REDACTED")

    def test_an_unsubscribe_url_is_reported_and_redacted(self):
        # List-Unsubscribe routinely names a domain appearing nowhere else,
        # and an unsubscribe link is a prime carrier of a recipient token,
        # so it is an indicator that must arrive redacted.
        iocs = parse.iocs(load("leaky.eml"), trusted=["192.0.2.0/24"])
        unsub = [i for i in iocs if i["origin"] == "header-list_unsubscribe"]
        self.assertEqual(
            [i["value"] for i in unsub],
            ["http://unsub.example.invalid/u?e=REDACTED"],
        )

    def test_an_attachment_becomes_a_hash_ioc(self):
        iocs = parse.iocs(load("with-attachment.eml"), trusted=["192.0.2.0/24"])
        hashes = [i for i in iocs if i["type"] == "sha256"]
        self.assertEqual(len(hashes), 1)
        self.assertEqual(hashes[0]["filename"], "invoice.pdf")

    def test_no_trusted_relays_still_refuses(self):
        with self.assertRaises(parse.NoTrustBoundary):
            parse.iocs(load("simple.eml"), trusted=[])

    def test_no_ioc_holds_a_recipient_address(self):
        # The safety property, asserted over the whole output.
        for name in ("simple.eml", "forged-chain.eml", "with-attachment.eml",
                     "redirector.eml", "leaky.eml"):
            iocs = parse.iocs(load(name), trusted=["192.0.2.0/24"])
            blob = repr(iocs)
            self.assertNotIn("you@example.org", blob)
            self.assertNotIn("example.org", blob)

    def test_the_address_does_not_survive_any_url_shape(self):
        # leaky.eml carries you@example.org in five placements. Each one has
        # been a real leak in this codebase or is one shape away from it.
        iocs = parse.iocs(load("leaky.eml"), trusted=["192.0.2.0/24"])
        blob = repr(iocs)
        self.assertNotIn("you@example.org", blob)
        self.assertNotIn("you%40example.org", blob)


class ReportHeaders(unittest.TestCase):
    def test_the_whitelist_keeps_what_a_desk_needs(self):
        headers = parse.report_headers(load("reportable.eml"),
                                       trusted=["192.0.2.0/24"])
        names = [name for name, _ in headers]
        for wanted in ("From", "Subject", "Date", "Message-ID", "Reply-To",
                       "Return-Path", "Authentication-Results", "Received-SPF"):
            self.assertIn(wanted, names)

    def test_recipient_headers_never_survive_the_whitelist(self):
        # The first property, at the one place a report reproduces header
        # text verbatim. A blacklist would have to remember each of these;
        # the whitelist never names them at all.
        headers = parse.report_headers(load("reportable.eml"),
                                       trusted=["192.0.2.0/24"])
        names = [name for name, _ in headers]
        blob = repr(headers)
        for name in ("To", "Cc", "Delivered-To", "X-Original-To"):
            self.assertNotIn(name, names)
        self.assertNotIn("victim@example.org", blob)
        self.assertNotIn("colleague@example.org", blob)

    def test_received_stops_at_the_boundary_hop(self):
        # 192.0.2.10 is ours, so its Received line is our own infrastructure
        # and must not be published; the hop below it is the one being
        # reported and is kept.
        headers = parse.report_headers(load("reportable.eml"),
                                       trusted=["192.0.2.0/24"])
        received = [value for name, value in headers if name == "Received"]
        self.assertEqual(len(received), 1)
        self.assertIn("203.0.113.42", received[0])
        self.assertNotIn("mx.example.org with ESMTP id abc123", received[0])

    def test_the_published_hop_carries_no_envelope_recipient(self):
        # The boundary Received line is written by OUR OWN relay, and its
        # optional "for <addr>" clause is the envelope recipient: the
        # victim's address, verbatim, in the one header a report reproduces
        # in full. Truncating the chain is not enough on its own.
        headers = parse.report_headers(load("reportable.eml"),
                                       trusted=["192.0.2.0/24"])
        received = [value for name, value in headers if name == "Received"]
        self.assertNotIn("you@example.org", received[0])
        self.assertNotIn("for <", received[0])
        # The rest of the hop survives; this is a cut, not a blanking.
        self.assertIn("203.0.113.42", received[0])

    def test_every_for_clause_shape_loses_the_address(self):
        # RFC 5321 4.4 puts For inside Opt-info, so With, ID, Via or a CFWS
        # comment may legitimately follow it, and its ABNF is
        # 1*( Path / Mailbox ) where Mailbox carries no angle brackets.
        # Anchoring on "for" being immediately followed by the clause
        # terminator matched only the neatest shape and let four routine
        # ones through, each publishing the victim's address.
        hop = "from a.invalid (a.invalid [203.0.113.5]) by mx.example.org "
        shapes = (
            "for <you@example.org> (envelope-from <b@c.invalid>); Mon, 07 Sep 2026 09:12:40 +0000",
            "for you@example.org; Mon, 07 Sep 2026 09:12:40 +0000",
            "for <you@example.org> with ESMTP; Mon, 07 Sep 2026 09:12:40 +0000",
            "id qq; Mon, 07 Sep 2026 09:12:40 +0000 (for <you@example.org>)",
            "for <you@example.org>; Mon, 07 Sep 2026 09:12:40 +0000",
        )
        for tail in shapes:
            with self.subTest(tail=tail):
                stripped = parse._strip_envelope_recipient(hop + tail)
                self.assertNotIn("you@example.org", stripped)
                # The hop's own evidence survives: this is a cut, not a
                # blanking, and a rule that ate the line would pass the
                # assertion above while destroying the report.
                self.assertIn("203.0.113.5", stripped)
                self.assertIn("mx.example.org", stripped)

    def test_stripping_leaves_no_doubled_space_or_stray_separator(self):
        # Cosmetic in isolation, but the result is published verbatim to a
        # third party, so a mangled line reads as a broken tool.
        hop = ("from a.invalid (a.invalid [203.0.113.5]) by mx.example.org"
               " for <you@example.org>; Mon, 07 Sep 2026 09:12:40 +0000")
        stripped = parse._strip_envelope_recipient(hop)
        self.assertNotIn("  ", stripped)
        self.assertNotIn(" ;", stripped)
        self.assertIn("mx.example.org; Mon", stripped)

    def test_a_comment_holding_only_the_clause_leaves_no_debris(self):
        hop = ("from a.invalid (a.invalid [203.0.113.5]) by mx.example.org"
               " id qq; Mon, 07 Sep 2026 09:12:40 +0000 (for <you@example.org>)")
        stripped = parse._strip_envelope_recipient(hop)
        self.assertNotIn("you@example.org", stripped)
        self.assertFalse(stripped.endswith("("))
        self.assertTrue(stripped.endswith("+0000"))

    def test_the_envelope_sender_comment_survives_the_cut(self):
        # envelope-from is the SENDER, which is what the report is about, so
        # cutting the recipient must not take it along.
        hop = ("from a.invalid (a.invalid [203.0.113.5]) by mx.example.org"
               " for <you@example.org> (envelope-from <bounce@sender.invalid>);"
               " Mon, 07 Sep 2026 09:12:40 +0000")
        stripped = parse._strip_envelope_recipient(hop)
        self.assertNotIn("you@example.org", stripped)
        self.assertIn("bounce@sender.invalid", stripped)

    def test_the_whitelist_does_not_filter_attacker_free_text(self):
        # A DOCUMENTED LIMIT, not a guarantee. The spec keeps Subject and the
        # From display name knowing both are attacker-controlled free text,
        # because they are what lets a desk recognise a campaign. An attacker
        # who writes the recipient's own address into one, obfuscated or not,
        # gets it published: the whitelist governs WHICH headers travel, never
        # what is inside one.
        #
        # This is asserted so the limit is visible and deliberate. Do not
        # "fix" it by filtering free text, which is the judgement-shaped
        # problem AGENTS.md names as the source of every leak here. The
        # sweep over real mail is what covers this class, per AGENTS.md.
        raw = (b"Received: from a.invalid (a.invalid [203.0.113.5])"
               b" by mx.example.org with ESMTP id X;"
               b" Mon, 07 Sep 2026 09:12:40 +0000\r\n"
               b"From: <phish@sender.invalid>\r\n"
               b"Subject: Verify you%40example.org\r\n\r\nbody\r\n")
        headers = parse.report_headers(raw, trusted=["192.0.2.0/24"])
        subject = dict(headers)["Subject"]
        self.assertIn("you%40example.org", subject)

    def test_a_forged_chain_publishes_no_hop_below_the_boundary(self):
        # The same job test_a_forged_chain_stops_at_the_first_untrusted_hop
        # does for sending_ip(), asserted over what actually gets published:
        # 198.51.100.7 is an innocent party the attacker named.
        headers = parse.report_headers(load("forged-chain.eml"),
                                       trusted=["192.0.2.0/24"])
        received = [value for name, value in headers if name == "Received"]
        # Asserted in BOTH directions: dropping Received altogether would
        # satisfy the "not published" half on its own, and a test that
        # passes when the feature is missing protects nothing.
        self.assertEqual(len(received), 1)
        self.assertIn("203.0.113.99", received[0])
        self.assertTrue(all("198.51.100.7" not in value for value in received))
        self.assertTrue(all("198.51.100.8" not in value for value in received))


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