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
|
# dmarc.py - DMARC report parsing.
# Copyright (C) 2016-2017 Tomasz Kramkowski <tk@the-tk.com>
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# 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, see <http://www.gnu.org/licenses/>.
from collections import namedtuple
from defusedxml.ElementTree import fromstring as xmlparse
class MalformedReportException(Exception):
pass
def _text(root, name, mandatory=True, default=None):
elem = root.find(name)
if elem is None or not elem.text:
if mandatory == True:
raise MalformedReportException('element missing: {}'.format(name))
return default
if not elem.text:
return ''
return elem.text
def _integer(root, name, mandatory=True, default=None):
text = _text(root, name, mandatory, default)
try:
return int(text)
except ValueError:
raise MalformedReportException("integer element invalid: {} - '{}'".format(name, text))
def _enum(root, name, values, mandatory=True, default=None):
text = _text(root, name, mandatory, default)
if not text:
return None
if text not in values:
raise MalformedReportException("enum element invalid: {} - '{}' expected: '{}'".format(name, text, values))
return text
def _feedback(root):
Edisposition = ['none', 'quarantine', 'reject']
def reportmeta(root):
def daterange(root):
begin = _integer(root, 'begin')
end = _integer(root, 'end')
nt = namedtuple('DateRangeType', ['begin', 'end'])
return nt(begin, end)
org = _text(root, 'org_name')
email = _text(root, 'email')
extra = _text(root, 'extra_contact_info', mandatory=False)
rid = _text(root, 'report_id')
drange = daterange(root.find('date_range'))
errors = root.findall('error')
if errors:
errors = map(_text, errors)
nt = namedtuple('ReportMetadataType', ['org_name', 'email', 'extra_contact_info',
'report_id', 'date_range', 'error'])
return nt(org, email, extra, rid, drange, errors)
def policypub(root):
Ealignment = ['r', 's']
domain = _text(root, 'domain')
adkim = _enum(root, 'adkim', Ealignment, mandatory=False)
aspf = _enum(root, 'aspf', Ealignment, mandatory=False)
p = _enum(root, 'p', Edisposition)
# common deviation from spec
sp = _enum(root, 'sp', Edisposition, mandatory=False, default=p)
pct = _integer(root, 'pct')
nt = namedtuple('PolicyPublishedType', ['domain', 'adkim', 'aspf', 'p', 'sp',
'pct'])
return nt(domain, adkim, aspf, p, sp, pct)
def record(root):
def row(root):
def policyevaluated(root):
def polreason(root):
Etype = ['forwarded', 'sampled_out', 'trusted_forwarder',
'mailing_list', 'local_policy', 'other']
otype = _enum(root, 'type', Etype)
comm = _text(root, 'comment', mandatory=False)
nt = namedtuple('PolicyOverrideReason', ['type', 'comment'])
return nt(otype, comm)
Eresult = ['pass', 'fail']
disp = _enum(root, 'disposition', Edisposition)
dkim = _enum(root, 'dkim', Eresult)
spf = _enum(root, 'spf', Eresult)
reason = root.findall('reason')
if reason:
reason = map(polreason, reason)
nt = namedtuple('PolicyEvaluatedType', ['disposition', 'dkim', 'spf',
'reason'])
return nt(disp, dkim, spf, reason)
source = _text(root, 'source_ip')
count = _integer(root, 'count')
poleval = root.find('policy_evaluated')
if not poleval:
raise MalformedReportException('policy_evaluated missing')
nt = namedtuple('RowType', ['source_ip', 'count', 'policy_evaluated'])
return nt(source, count, policyevaluated(poleval))
def identifier(root):
envto = _text(root, 'envelope_to', mandatory=False)
hdrfrom = _text(root, 'header_from')
nt = namedtuple('IdentifierType', ['envelope_to', 'header_from'])
return nt(envto, hdrfrom)
def authresult(root):
def dkimauth(root):
Eresult = ['none', 'pass', 'fail', 'policy', 'neutral', 'temperror',
'permerror']
domain = _text(root, 'domain')
selector = _text(root, 'selector', mandatory=False)
result = _enum(root, 'result', Eresult)
human_result = _text(root, 'human_result', mandatory=False)
nt = namedtuple('DKIMAuthResultType', ['domain', 'selector', 'result',
'human_result'])
return nt(domain, selector, result, human_result)
def spfauth(root):
# Escope = ['helo', 'mfrom'] #Deprecated
Eresult = ['none', 'neutral', 'pass', 'fail', 'softfail', 'temperror',
'permerror']
domain = _text(root, 'domain')
result = _enum(root, 'result', Eresult)
nt = namedtuple('SPFAuthResultType', ['domain', 'result'])
return nt(domain, result)
dkims = root.findall('dkim')
spfs = root.findall('spf')
if not spfs:
raise MalformedReportException('auth_results.spf missing')
else:
spfs = map(spfauth, spfs)
if dkims:
dkims = map(dkimauth, dkims)
nt = namedtuple('AuthResultType', ['dkim', 'spf'])
return nt([*dkims], [*spfs])
r = root.find('row')
idents = root.find('identifiers')
results = root.find('auth_results')
if not r or not idents or not results:
raise MalformedReportException('row, identifiers or auth_results missing')
nt = namedtuple('RecordType', ['row', 'identifiers', 'auth_results'])
return nt(row(r), identifier(idents), authresult(results))
meta = root.find('report_metadata')
pol = root.find('policy_published')
recs = root.findall('record')
if meta is None or pol is None or not recs:
raise MalformedReportException('report_metadata, policy_published or record missing')
nt = namedtuple('feedback', ['report_metadata', 'policy_published', 'record'])
return nt(reportmeta(meta), policypub(pol), list(map(record, recs)))
def parse_dmarc(r):
root = xmlparse(r)
if root.tag != 'feedback':
raise MalformedReportException('root is not feedback')
return _feedback(root)
|
