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
|
import unittest
from abusectl import report
class Grouping(unittest.TestCase):
def test_two_contacts_at_one_address_become_one_destination(self):
contacts = [
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["abuse@host.invalid"], "source": "rdap"},
{"iocs": ["ioc-2"], "query": "example.invalid",
"abuse": ["abuse@host.invalid"], "source": "rdap"},
]
destinations = report.email_destinations(contacts)
self.assertEqual(len(destinations), 1)
self.assertEqual(destinations[0]["target"], "abuse@host.invalid")
self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"])
def test_a_contact_with_two_addresses_reaches_both_desks(self):
contacts = [
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["a@host.invalid", "b@host.invalid"], "source": "rdap"},
]
destinations = report.email_destinations(contacts)
self.assertEqual([d["target"] for d in destinations],
["a@host.invalid", "b@host.invalid"])
# Both desks carry the indicator, and each gets its own id: a
# destination that reached only one desk, or two rows sharing an
# id, would pass an assertion on the sorted targets alone.
self.assertEqual([d["iocs"] for d in destinations],
[["ioc-1"], ["ioc-1"]])
self.assertEqual(len({d["id"] for d in destinations}), 2)
for destination in destinations:
self.assertEqual(destination["id"],
report.email_destination_id(
destination["target"]))
def test_a_contact_with_no_address_creates_no_destination(self):
"""The contact that resolved must still produce its destination.
Asserted alongside one that DOES resolve, because "no destination
for this contact" is also what returning nothing at all looks
like, and that is not the behaviour being described.
"""
contacts = [
{"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [],
"source": "rdap", "error": "no abuse role published"},
{"iocs": ["ioc-2"], "query": "198.51.100.7",
"abuse": ["abuse@host.invalid"], "source": "rdap"},
]
destinations = report.email_destinations(contacts)
self.assertEqual([d["target"] for d in destinations],
["abuse@host.invalid"])
self.assertEqual(destinations[0]["iocs"], ["ioc-2"])
def test_destinations_carry_stable_ids_and_pending_status(self):
contacts = [
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["abuse@host.invalid"], "source": "rdap"},
]
destination = report.email_destinations(contacts)[0]
self.assertEqual(destination["id"],
report.email_destination_id("abuse@host.invalid"))
self.assertEqual(destination["kind"], "email")
self.assertEqual(destination["status"], "pending")
def test_an_id_is_the_literal_shape_a_reviewer_will_read(self):
"""Pin the shape, since it becomes a filename in bodies/.
Computed by hand rather than by calling the code under test, so
this fails if the derivation changes rather than following it.
"""
self.assertEqual(report.email_destination_id("abuse@host.invalid"),
"email-bc50e369")
def test_ids_are_derived_per_destination_not_per_contact(self):
"""A contact that resolved to no desk must not shift another's id.
The obvious implementation numbers destinations by position, and a
skipped contact then either burns an id or renumbers the rest.
Both are wrong for the same reason: an id names a desk.
"""
contacts = [
{"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [],
"source": "rdap", "error": "no abuse role published"},
{"iocs": ["ioc-2"], "query": "198.51.100.7",
"abuse": ["a@host.invalid"], "source": "rdap"},
{"iocs": ["ioc-3"], "query": "198.51.100.8",
"abuse": ["b@host.invalid"], "source": "rdap"},
]
destinations = report.email_destinations(contacts)
self.assertEqual([d["id"] for d in destinations],
[report.email_destination_id("a@host.invalid"),
report.email_destination_id("b@host.invalid")])
self.assertEqual([d["target"] for d in destinations],
["a@host.invalid", "b@host.invalid"])
def test_one_desk_listed_twice_by_one_contact_is_one_destination(self):
"""A duplicate in a contact's own abuse list must not duplicate a desk.
RDAP jCards are attacker-adjacent data: an entity can publish the
same address in two vcard rows, and one destination per ADDRESS is
the rule regardless of how many rows produced it.
"""
contacts = [
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["abuse@host.invalid", "abuse@host.invalid"],
"source": "rdap"},
]
destinations = report.email_destinations(contacts)
self.assertEqual(len(destinations), 1)
self.assertEqual(destinations[0]["iocs"], ["ioc-1"])
def test_one_desk_spelled_with_two_domain_cases_is_one_destination(self):
"""A domain is case-insensitive, so two spellings are one desk."""
contacts = [
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["abuse@Host.Invalid"], "source": "rdap"},
{"iocs": ["ioc-2"], "query": "example.invalid",
"abuse": ["abuse@host.invalid"], "source": "rdap"},
]
destinations = report.email_destinations(contacts)
self.assertEqual(len(destinations), 1)
self.assertEqual(destinations[0]["iocs"], ["ioc-1", "ioc-2"])
self.assertEqual(destinations[0]["target"], "abuse@Host.Invalid")
def test_two_spellings_of_one_desk_share_an_id(self):
"""The id derives from the same normalised form the grouping uses.
Otherwise the spelling RDAP happened to publish first would decide
a body's filename, and a re-run that saw the other spelling first
would look like a different desk.
"""
self.assertEqual(report.email_destination_id("abuse@Host.Invalid"),
report.email_destination_id("abuse@host.invalid"))
def test_two_local_part_cases_stay_two_destinations(self):
"""Only the receiving host knows whether its local parts fold.
Folding them here would silently drop a desk that a host genuinely
distinguishes; not folding them costs a duplicate mail at worst.
"""
contacts = [
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["Abuse@host.invalid", "abuse@host.invalid"],
"source": "rdap"},
]
destinations = report.email_destinations(contacts)
self.assertEqual(sorted(d["target"] for d in destinations),
["Abuse@host.invalid", "abuse@host.invalid"])
self.assertNotEqual(destinations[0]["id"], destinations[1]["id"])
def test_a_destination_starts_with_no_body(self):
contacts = [
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["abuse@host.invalid"], "source": "rdap"},
]
self.assertIsNone(report.email_destinations(contacts)[0]["body"])
def test_the_same_contacts_produce_the_same_ids_twice(self):
"""Ids must not depend on dict iteration luck or set ordering."""
contacts = [
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["b@host.invalid", "a@host.invalid"], "source": "rdap"},
{"iocs": ["ioc-2"], "query": "example.invalid",
"abuse": ["c@host.invalid"], "source": "rdap"},
]
first = report.email_destinations(contacts)
second = report.email_destinations(contacts)
self.assertEqual([(d["id"], d["target"]) for d in first],
[(d["id"], d["target"]) for d in second])
self.assertEqual([d["target"] for d in first],
["b@host.invalid", "a@host.invalid",
"c@host.invalid"])
def test_a_desks_id_survives_another_desk_appearing(self):
"""An id names a DESK, not a position in this run's list.
Task 8 writes each body to bodies/<id>.xarf and records its hash
against that id. With a positional id, re-running contacts on a
case that gained an indicator renumbers every desk after the new
one, so bodies/<id>.xarf on disk belongs to a different desk than
the manifest's entry of that id, and the edit check compares one
desk's body against another's.
"""
established = {"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["b@host.invalid"], "source": "rdap"}
first = report.email_destinations([established])
# A later contacts run finds an indicator whose desk sorts ahead.
newcomer = {"iocs": ["ioc-2"], "query": "example.invalid",
"abuse": ["a@new.invalid"], "source": "rdap"}
second = report.email_destinations([newcomer, established])
by_target = {d["target"]: d["id"] for d in second}
self.assertEqual(by_target["b@host.invalid"], first[0]["id"])
self.assertNotEqual(by_target["a@new.invalid"], first[0]["id"])
def test_an_ids_position_does_not_leak_into_it(self):
"""The same desk alone and third in a list gets one id."""
alone = report.email_destinations([
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["desk@host.invalid"], "source": "rdap"},
])
crowded = report.email_destinations([
{"iocs": ["ioc-2"], "query": "198.51.100.8",
"abuse": ["one@host.invalid", "two@host.invalid"],
"source": "rdap"},
{"iocs": ["ioc-1"], "query": "198.51.100.7",
"abuse": ["desk@host.invalid"], "source": "rdap"},
])
self.assertEqual(crowded[2]["target"], "desk@host.invalid")
self.assertEqual(crowded[2]["id"], alone[0]["id"])
if __name__ == "__main__":
unittest.main()
|
