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
|
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(
sorted(d["target"] for d in destinations),
["a@host.invalid", "b@host.invalid"],
)
def test_a_contact_with_no_address_creates_no_destination(self):
contacts = [
{"iocs": ["ioc-1"], "query": "example.invalid", "abuse": [],
"source": "rdap", "error": "no abuse role published"},
]
self.assertEqual(report.email_destinations(contacts), [])
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"], "email-1")
self.assertEqual(destination["kind"], "email")
self.assertEqual(destination["status"], "pending")
def test_ids_are_numbered_per_destination_not_per_contact(self):
"""A skipped contact must not leave a hole in the numbering.
The obvious implementation enumerates the contacts, and a contact
with no abuse address then burns an id: the desks come back as
"email-2" and "email-4", which reads to a reviewer as two reports
having gone missing.
"""
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],
["email-1", "email-2"])
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_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"])
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.
Task 8 writes each body to bodies/<id>.xarf and records its hash
against that id, so an id that moved between two runs over the
same input would compare one desk's body against another's.
"""
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"])
if __name__ == "__main__":
unittest.main()
|
