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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
|
# Copyright (C) 2016 by the Free Software Foundation, Inc.
#
# This file is part of GNU Mailman.
#
# GNU Mailman 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.
#
# GNU Mailman 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
# GNU Mailman. If not, see <http://www.gnu.org/licenses/>.
"""DMARC mitigation rule."""
import re
import logging
import dns.resolver
from dns.exception import DNSException
from email.utils import parseaddr
from lazr.config import as_timedelta
from mailman.config import config
from mailman.core.i18n import _
from mailman.interfaces.mailinglist import DMARCMitigateAction
from mailman.interfaces.rules import IRule
from mailman.utilities.string import wrap
from public import public
from urllib import error, request
from zope.interface import implementer
elog = logging.getLogger('mailman.error')
vlog = logging.getLogger('mailman.vette')
s_dict = dict()
KEEP_LOOKING = object()
def _get_suffixes(url):
# This loads and parses the data from the url argument into s_dict for
# use by _get_org_dom.
global s_dict
if not url:
return
try:
d = request.urlopen(url)
except error.URLError as e:
elog.error('Unable to retrieve data from %s: %s', url, e.reason)
return
for line in d.readlines():
line = str(line, encoding='utf-8')
if not line.strip() or line.startswith('//'):
continue
line = re.sub('\s.*', '', line)
if not line:
continue
parts = line.lower().split('.')
if parts[0].startswith('!'):
exc = True
parts = [parts[0][1:]] + parts[1:]
else:
exc = False
parts.reverse()
k = '.'.join(parts)
s_dict[k] = exc
def _get_dom(d, l):
# A helper to get a domain name consisting of the first l+1 labels
# in d.
dom = d[:min(l+1, len(d))]
dom.reverse()
return '.'.join(dom)
def _get_org_dom(domain):
# Given a domain name, this returns the corresponding Organizational
# Domain which may be the same as the input.
global s_dict
if not s_dict:
_get_suffixes(config.dmarc.org_domain_data_url)
hits = []
d = domain.lower().split('.')
d.reverse()
for k in s_dict.keys():
ks = k.split('.')
if len(d) >= len(ks):
for i in range(len(ks)-1):
if d[i] != ks[i] and ks[i] != '*':
break
else:
if d[len(ks)-1] == ks[-1] or ks[-1] == '*':
hits.append(k)
if not hits:
return _get_dom(d, 1)
l = 0
for k in hits:
if s_dict[k]:
# It's an exception
return _get_dom(d, len(k.split('.'))-1)
if len(k.split('.')) > l:
l = len(k.split('.'))
return _get_dom(d, l)
def _DMARCProhibited(mlist, email, dmarc_domain, org=False):
resolver = dns.resolver.Resolver()
resolver.timeout = as_timedelta(
config.dmarc.resolver_timeout).total_seconds()
resolver.lifetime = as_timedelta(
config.dmarc.resolver_lifetime).total_seconds()
try:
txt_recs = resolver.query(dmarc_domain, dns.rdatatype.TXT)
except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
return KEEP_LOOKING
except DNSException as e:
elog.error(
'DNSException: Unable to query DMARC policy for %s (%s). %s',
email, dmarc_domain, e.__doc__)
return KEEP_LOOKING
# Be as robust as possible in parsing the result.
results_by_name = {}
cnames = {}
want_names = set([dmarc_domain + '.'])
for txt_rec in txt_recs.response.answer:
if txt_rec.rdtype == dns.rdatatype.CNAME:
cnames[txt_rec.name.to_text()] = (
txt_rec.items[0].target.to_text())
if txt_rec.rdtype != dns.rdatatype.TXT:
continue
results_by_name.setdefault(
txt_rec.name.to_text(), []).append(
"".join(
[str(x, encoding='utf-8')
for x in txt_rec.items[0].strings]))
expands = list(want_names)
seen = set(expands)
while expands:
item = expands.pop(0)
if item in cnames:
if cnames[item] in seen:
continue # cname loop
expands.append(cnames[item])
seen.add(cnames[item])
want_names.add(cnames[item])
want_names.discard(item)
if len(want_names) != 1:
elog.error(
"""multiple DMARC entries in results for %s,
processing each to be strict""",
dmarc_domain)
for name in want_names:
if name not in results_by_name:
continue
dmarcs = [x for x in results_by_name[name]
if x.startswith('v=DMARC1;')]
if len(dmarcs) == 0:
return KEEP_LOOKING
if len(dmarcs) > 1:
elog.error(
"""RRset of TXT records for %s has %d v=DMARC1 entries;
testing them all""",
dmarc_domain, len(dmarcs))
for entry in dmarcs:
mo = re.search(r'\bsp=(\w*)\b', entry, re.IGNORECASE)
if org and mo:
policy = mo.group(1).lower()
else:
mo = re.search(r'\bp=(\w*)\b', entry, re.IGNORECASE)
if mo:
policy = mo.group(1).lower()
else:
continue
if policy in ('reject', 'quarantine'):
vlog.info(
"""%s: DMARC lookup for %s (%s)
found p=%s in %s = %s""",
mlist.list_name,
email,
dmarc_domain,
policy,
name,
entry)
return True
return False
def _IsDMARCProhibited(mlist, email):
# This takes an email address, and returns True if DMARC policy is
# p=reject or quarantine.
email = email.lower()
# Scan from the right in case quoted local part has an '@'.
local, at, from_domain = email.rpartition('@')
if at != '@':
return False
x = _DMARCProhibited(mlist, email, '_dmarc.{}'.format(from_domain))
if x is not KEEP_LOOKING:
return x
org_dom = _get_org_dom(from_domain)
if org_dom != from_domain:
x = _DMARCProhibited(
mlist, email, '_dmarc.{}'.format(org_dom), org=True)
if x is not KEEP_LOOKING:
return x
return False
@public
@implementer(IRule)
class DMARCMitigation:
"""The DMARC mitigation rule."""
name = 'dmarc-mitigation'
description = _('Find DMARC policy of From: domain.')
record = True
def check(self, mlist, msg, msgdata):
"""See `IRule`."""
if mlist.dmarc_mitigate_action is DMARCMitigateAction.no_mitigation:
# Don't bother to check if we're not going to do anything.
return False
dn, addr = parseaddr(msg.get('from'))
if _IsDMARCProhibited(mlist, addr):
# If dmarc_mitigate_action is discard or reject, this rule fires
# and jumps to the 'moderation' chain to do the actual discard.
# Otherwise, the rule misses but sets a flag for the dmarc handler
# to do the appropriate action.
msgdata['dmarc'] = True
if mlist.dmarc_mitigate_action is DMARCMitigateAction.discard:
msgdata['moderation_action'] = 'discard'
msgdata['moderation_reasons'] = [_('DMARC moderation')]
elif mlist.dmarc_mitigate_action is DMARCMitigateAction.reject:
listowner = mlist.owner_address # noqa F841
reason = (mlist.dmarc_moderation_notice or
_('You are not allowed to post to this mailing '
'list From: a domain which publishes a DMARC '
'policy of reject or quarantine, and your message'
' has been automatically rejected. If you think '
'that your messages are being rejected in error, '
'contact the mailing list owner at ${listowner}.'))
msgdata['moderation_reasons'] = [wrap(reason)]
msgdata['moderation_action'] = 'reject'
else:
return False
msgdata['moderation_sender'] = addr
return True
return False
|
