# 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) 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()) 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")), []) if __name__ == "__main__": unittest.main()