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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
|
# 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")), [])
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"):
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)
if __name__ == "__main__":
unittest.main()
|
