# Copyright (C) 2026 Danilo M. # # 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: \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: \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_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") 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: \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_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) if __name__ == "__main__": unittest.main()