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
|
# 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.
"""IOCs and abuse contacts to report bodies: the last step before anything
irreversible happens.
This module is PURE and OFFLINE. It opens no socket, sends no mail and reads
no file outside the case directory. What it produces is a document the user
reads, edits and approves, so the output is written for a human first and a
parser second.
It takes the reporting identity as an ARGUMENT rather than reading the
config, the way parse.py takes the trust boundary. The identity is the one
thing in a report disclosed deliberately, and a module that reaches for it
itself is a module that can disclose it in a code path nobody reviewed.
"""
def _group_key(address: str) -> str:
"""The key two spellings of one desk must share, and no more than that.
The DOMAIN is case-insensitive by every standard that touches it, so
"abuse@Host.Invalid" and "abuse@host.invalid" are one desk and must
not be mailed twice about one incident.
The LOCAL PART is left exactly as published. RFC 5321 leaves its
interpretation to the receiving host, and only that host knows whether
it folds case. In practice it almost always does, but "almost always"
is the wrong standard for the one field that decides whether a report
arrives: folding two desks that a host genuinely distinguishes would
silently drop one of them, and the cost of being wrong the other way
is a duplicate mail. A dropped desk is the worse failure, so the
conservative direction is to fold only what is defined to fold.
"""
local, at, domain = address.rpartition("@")
if not at:
# Not an address shape we can split. RDAP data is third-party and
# occasionally malformed; group it by its literal text rather than
# inventing a domain for it.
return address
return f"{local}@{domain.lower()}"
def email_destinations(contacts: list[dict]) -> list[dict]:
"""Group contacts into one destination per abuse ADDRESS.
Contacts already fold by host, but two different contacts can still
resolve to the same address, an IP and a domain at one hoster being the
common case. One mail per address rather than per contact is what stops
a desk receiving two mails about one incident.
Ids are numbered over the DESTINATIONS produced, not over the contacts
read, so a contact that resolved to no desk leaves no hole: a reviewer
who sees "email-2" and "email-4" reasonably reads two reports as
missing.
Insertion order carries the numbering, so the same contacts in the same
order always produce the same ids. That matters downstream: bodies are
written to bodies/<id>.xarf and hashed against that id, so an id that
wandered between two runs over one input would compare one desk's body
against another's. It is NOT a promise that an id survives a change in
the contacts themselves; adding a desk earlier in the list renumbers
every desk after it.
"""
by_address: dict[str, dict] = {}
for contact in contacts:
for address in contact.get("abuse", []):
key = _group_key(address)
# First spelling seen wins the target. Any spelling reaches the
# desk, and picking one keeps the report stable across a re-run.
destination = by_address.setdefault(key, {"target": address,
"iocs": []})
for ioc in contact.get("iocs", []):
if ioc not in destination["iocs"]:
destination["iocs"].append(ioc)
return [
{
"id": f"email-{index}",
"kind": "email",
"target": destination["target"],
"iocs": destination["iocs"],
"body": None,
"status": "pending",
}
for index, destination in enumerate(by_address.values(), start=1)
]
|
