1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
# 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)
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: <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")), [])
if __name__ == "__main__":
unittest.main()
|
