~ar-python-hackers/authentication-results-python/trunk
» Revision
35
(compared to revision 109)
: authres/__init__.py
« back to all changes in this revision
Viewing changes to authres/__init__.py
- browse files at revision 35
- compare with another revision
- download diff
- download tarball
- view history from revision 35
- Committer: Julian Mehnle
- Date: 2012-01-23 02:32:27 UTC
- Revision ID: jmehnle@agari.com-20120123023227-37nqxoxwvnt7zgp8
Convert authres module into a Python package, moving it into an authres/ subdirectory and splitting it into __{init,main}__.py.
- files added:
- authres/__main__.py
- files removed:
- authres/__main__.py
- files modified:
- CHANGES
added
removed
authres/__init__.py
1
1
# coding: utf-8
2
2
3
# Copyright © 2011-2013 Julian Mehnle <julian@mehnle.net>,
4
# Copyright © 2011-2013 Scott Kitterman <scott@kitterman.com>
3
# Copyright © 2011-2012 Julian Mehnle <julian@mehnle.net>,
4
# Copyright © 2011-2012 Scott Kitterman <scott@kitterman.com>
5
5
#
6
6
# Licensed under the Apache License, Version 2.0 (the "License");
7
7
# you may not use this file except in compliance with the License.
16
16
# limitations under the License.
17
17
18
18
"""
19
Package for parsing ``Authentication-Results`` headers as defined in RFC
20
5451/7001/7601. Optional support for authentication methods defined in RFCs
21
5617, 6008, 6212, and 7281.
22
23
Examples:
24
RFC 5451 B.2
25
>>> str(AuthenticationResultsHeader('test.example.org'))
26
'Authentication-Results: test.example.org; none'
27
28
RFC 5451 B.3
29
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
30
... results = [SPFAuthenticationResult(result = 'pass',
31
... smtp_mailfrom = 'example.net')]))
32
'Authentication-Results: example.com; spf=pass smtp.mailfrom=example.net'
33
34
RFC 5451 B.4(1)
35
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
36
... results = [SMTPAUTHAuthenticationResult(result = 'pass', result_comment = 'cram-md5',
37
... smtp_auth = 'sender@example.net'), SPFAuthenticationResult(result = 'pass',
38
... smtp_mailfrom = 'example.net')]))
39
'Authentication-Results: example.com; auth=pass (cram-md5) smtp.auth=sender@example.net; spf=pass smtp.mailfrom=example.net'
40
41
RFC 5451 B.4(2)
42
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
43
... results = [SenderIDAuthenticationResult(result = 'pass',
44
... header_from = 'example.com')]))
45
'Authentication-Results: example.com; sender-id=pass header.from=example.com'
46
47
RFC 5451 B.5(1) # Note: RFC 5451 uses 'hardfail' instead of 'fail' for
48
SPF failures. Hardfail is deprecated. See RFC 6577.
49
Examples here use the correct 'fail'. The authres module does not
50
validate result codes, so either will be processed.
51
52
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
53
... results = [SenderIDAuthenticationResult(result = 'fail',
54
... header_from = 'example.com'), DKIMAuthenticationResult(result = 'pass',
55
... header_i = 'sender@example.com', result_comment = 'good signature')]))
56
'Authentication-Results: example.com; sender-id=fail header.from=example.com; dkim=pass (good signature) header.i=sender@example.com'
57
58
# Missing parsing header comment.
59
#FIXME
60
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.com; sender-id=fail header.from=example.com; dkim=pass (good signature) header.i=sender@example.com')
61
>>> str(arobj.authserv_id)
62
'example.com'
63
>>> str(arobj.results[0])
64
'sender-id=fail header.from=example.com'
65
>>> str(arobj.results[0].method)
66
'sender-id'
67
>>> str(arobj.results[0].result)
68
'fail'
69
>>> str(arobj.results[0].header_from)
70
'example.com'
71
>>> str(arobj.results[0].properties[0].type)
72
'header'
73
>>> str(arobj.results[0].properties[0].name)
74
'from'
75
>>> str(arobj.results[0].properties[0].value)
76
'example.com'
77
>>> str(arobj.results[1])
78
'dkim=pass header.i=sender@example.com'
79
>>> str(arobj.results[1].method)
80
'dkim'
81
>>> str(arobj.results[1].result)
82
'pass'
83
>>> str(arobj.results[1].header_i)
84
'sender@example.com'
85
>>> str(arobj.results[1].properties[0].type)
86
'header'
87
>>> str(arobj.results[1].properties[0].name)
88
'i'
89
>>> str(arobj.results[1].properties[0].value)
90
'sender@example.com'
19
Module for parsing ``Authentication-Results`` headers as defined in RFC 5451.
20
Extended to support additional methods defined in RFCs 5617, 6008, and 6212.
91
21
"""
92
22
93
23
MODULE = 'authres'
94
24
95
25
__author__ = 'Julian Mehnle, Scott Kitterman'
96
26
__email__ = 'julian@mehnle.net'
97
__version__ = '1.0.1'
98
99
import authres.core
100
101
# Backward compatibility: For the benefit of user modules referring to authres.…:
102
from authres.core import *
103
104
# FeatureContext class & convenience methods
105
###############################################################################
106
107
class FeatureContext(object):
108
"""
109
Class representing a "feature context" for the ``authres`` package.
110
A feature context is a collection of extension modules that may override
111
the core AuthenticationResultsHeader class or result classes, or provide
112
additional result classes for new authentication methods.
113
114
To instantiate a feature context, import the desired ``authres.…`` extension
115
modules and pass them to ``FeatureContext()``.
116
117
A ``FeatureContext`` object provides ``parse``, ``parse_value``, ``header``,
118
and ``result`` methods specific to the context's feature set.
119
"""
120
121
def __init__(self, *modules):
122
self.header_class = authres.core.AuthenticationResultsHeader
123
self.result_class_by_auth_method = {}
124
125
modules = [authres.core] + list(modules)
126
for module in modules:
127
try:
128
self.header_class = module.AuthenticationResultsHeader
129
except AttributeError:
130
# Module does not provide new AuthenticationResultsHeader class.
131
pass
132
133
try:
134
for result_class in module.RESULT_CLASSES:
135
self.result_class_by_auth_method[result_class.METHOD] = result_class
136
except AttributeError:
137
# Module does not provide AuthenticationResult subclasses.
138
pass
139
140
def parse(self, string):
141
return self.header_class.parse(self, string)
142
143
def parse_value(self, string):
144
return self.header_class.parse_value(self, string)
145
146
def header(self,
147
authserv_id = None, authserv_id_comment = None,
148
version = None, version_comment = None,
149
results = None
150
):
151
return self.header_class(
152
self, authserv_id, authserv_id_comment, version, version_comment, results)
153
154
def result(self, method, version = None,
155
result = None, result_comment = None,
156
reason = None, reason_comment = None,
27
__version__ = '0.4'
28
29
import re
30
31
# Helper functions
32
###############################################################################
33
34
retype = type(re.compile(''))
35
36
def isre(obj):
37
return isinstance(obj, retype)
38
39
# Patterns
40
###############################################################################
41
42
RFC2045_TOKEN_PATTERN = r"[A-Za-z0-9!#$%&'*+.^_`{|}~-]+" # Printable ASCII w/o tspecials
43
RFC5234_WSP_PATTERN = r'[\t ]'
44
RFC5234_VCHAR_PATTERN = r'[\x21-\x7e]' # Printable ASCII
45
RFC5322_QUOTED_PAIR_PATTERN = r'\\[\t \x21-\x7e]'
46
RFC5322_FWS_PATTERN = r'(?:%s*(?:\r\n|\n))?%s+' % (RFC5234_WSP_PATTERN, RFC5234_WSP_PATTERN)
47
RFC5322_CTEXT_PATTERN = r'[\x21-\x27\x2a-\x5b\x5d-\x7e]' # Printable ASCII w/o ()\
48
RFC5322_ATEXT_PATTERN = r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]" # Printable ASCII w/o specials
49
RFC5322_QTEXT_PATTERN = r'[\x21\x23-\x5b\x5d-\x7e]' # Printable ASCII w/o "\
50
KTEXT_PATTERN = r"[A-Za-z0-9!#$%&'*+?^_`{|}~-]" # Like atext, w/o /=
51
PTEXT_PATTERN = r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~.@-]"
52
53
# Exceptions
54
###############################################################################
55
56
class AuthResError(Exception):
57
"Generic exception generated by the `authres` module"
58
59
def __init__(self, message = None):
60
Exception.__init__(self, message)
61
self.message = message
62
63
class SyntaxError(AuthResError):
64
"Syntax error while parsing ``Authentication-Results`` header"
65
66
def __init__(self, message = None, parse_text = None):
67
AuthResError.__init__(self, message)
68
if parse_text is None or len(parse_text) <= 40:
69
self.parse_text = parse_text
70
else:
71
self.parse_text = parse_text[0:40] + '...'
72
73
def __str__(self):
74
if self.message and self.parse_text:
75
return 'Syntax error: {0} at: {1}'.format(self.message, self.parse_text)
76
elif self.message:
77
return 'Syntax error: {0}'.format(self.message)
78
elif self.parse_text:
79
return 'Syntax error at: {0}'.format(self.parse_text)
80
else:
81
return 'Syntax error'
82
83
class UnsupportedVersionError(AuthResError):
84
"Unsupported ``Authentication-Results`` header version"
85
86
def __init__(self, message = None, version = None):
87
message = message or \
88
'Unsupported Authentication-Results header version: %s' % version
89
AuthResError.__init__(self, message)
90
self.version = version
91
92
class OrphanCommentError(AuthResError):
93
"Comment without associated header element"
94
95
# Main classes
96
###############################################################################
97
98
# AuthenticationResultProperty class
99
# =============================================================================
100
101
class AuthenticationResultProperty(object):
102
"""
103
A property (``type.name=value``) of a result clause of an
104
``Authentication-Results`` header
105
"""
106
107
def __init__(self, type, name, value = None, comment = None):
108
self.type = type.lower()
109
self.name = name.lower()
110
self.value = value
111
self.comment = comment
112
113
def __str__(self):
114
if self.comment:
115
return '%s.%s=%s (%s)' % (self.type, self.name, self.value, self.comment)
116
else:
117
return '%s.%s=%s' % (self.type, self.name, self.value)
118
119
# AuthenticationResult and related classes
120
# =============================================================================
121
122
class BaseAuthenticationResult(object): pass
123
124
class NoneAuthenticationResult(BaseAuthenticationResult):
125
"Sole ``none`` clause of an empty ``Authentication-Results`` header"
126
127
def __init__(self, comment = None):
128
self.comment = comment
129
130
def __str__(self):
131
if self.comment:
132
return 'none (%s)' % self.comment
133
else:
134
return 'none'
135
136
# Clarification of identifier naming:
137
# The following function acts as a factory for Python property attributes to
138
# be bound to a class, so it is named `make_authres_class_properties`. Its
139
# nested `getter` and `setter` functions use the identifier `authres_property`
140
# to refer to an instance of the `AuthenticationResultProperty` class.
141
def make_authres_class_properties(type, name):
142
"""
143
Return a property attribute to be bound to an `AuthenticationResult` class
144
for accessing the `AuthenticationResultProperty` objects in its `properties`
145
attribute.
146
"""
147
148
def value_getter(self, type = type, name = name):
149
authres_property = self._find_first_property(type, name)
150
return authres_property and authres_property.value
151
152
def comment_getter(self, type = type, name = name):
153
authres_property = self._find_first_property(type, name)
154
return authres_property and authres_property.comment
155
156
def value_setter(self, value, type = type, name = name):
157
authres_property = self._find_first_property(type, name)
158
if not authres_property:
159
authres_property = AuthenticationResultProperty(type, name)
160
self.properties.append(authres_property)
161
authres_property.value = value
162
163
def comment_setter(self, comment, type = type, name = name):
164
authres_property = self._find_first_property(type, name)
165
if not authres_property:
166
raise OrphanCommentError(
167
"Cannot include result property comment without associated result property: %s.%s" % (type, name))
168
authres_property.comment = comment
169
170
return property(value_getter, value_setter), property(comment_getter, comment_setter)
171
172
class AuthenticationResult(BaseAuthenticationResult):
173
"Generic result clause of an ``Authentication-Results`` header"
174
175
def __init__(self, method, version = None,
176
result = None, result_comment = None,
177
reason = None, reason_comment = None,
157
178
properties = None
158
179
):
159
try:
160
return self.result_class_by_auth_method[method](version,
161
result, result_comment, reason, reason_comment, properties)
162
except KeyError:
163
return authres.core.AuthenticationResult(method, version,
164
result, result_comment, reason, reason_comment, properties)
165
166
_core_features = None
167
def core_features():
168
"Returns default feature context providing only RFC 5451 core features."
169
global _core_features
170
if not _core_features:
171
_core_features = FeatureContext()
172
return _core_features
173
174
_all_features = None
175
def all_features():
176
"""
177
Returns default feature context providing all features shipped with the
178
``authres`` package.
179
"""
180
global _all_features
181
if not _all_features:
182
import authres.dkim_b
183
import authres.dkim_adsp
184
import authres.vbr
185
import authres.dmarc
186
import authres.smime
187
import authres.rrvs
188
import authres.arc
189
_all_features = FeatureContext(
190
authres.dkim_b,
191
authres.dkim_adsp,
192
authres.vbr,
193
authres.dmarc,
194
authres.smime,
195
authres.rrvs,
196
authres.arc
180
self.method = method.lower()
181
self.version = version and version.lower()
182
self.result = result.lower()
183
if not self.result:
184
raise ArgumentError('Required result argument missing or None or empty')
185
self.result_comment = result_comment
186
self.reason = reason and re.sub(r'[^\x20-\x7e]', '?', reason)
187
# Remove unprintable characters
188
self.reason_comment = reason_comment
189
self.properties = properties or []
190
191
def __str__(self):
192
strs = []
193
strs.append(self.method)
194
if self.version:
195
strs.append('/')
196
strs.append(self.version)
197
strs.append('=')
198
strs.append(self.result)
199
if self.result_comment:
200
strs.append(' (%s)' % self.result_comment)
201
if self.reason:
202
strs.append(' reason="')
203
strs.append(re.sub(r'(["\\])', r'\\\1', self.reason)) # Escape "\
204
strs.append('"')
205
if self.reason_comment:
206
strs.append(' (%s)' % self.reason_comment)
207
for property_ in self.properties:
208
strs.append(' ')
209
strs.append(str(property_))
210
return ''.join(strs)
211
212
def _find_first_property(self, type, name):
213
properties = [
214
property
215
for property
216
in self.properties
217
if property.type == type and property.name == name
218
]
219
return properties[0] if properties else None
220
221
class DKIMAuthenticationResult(AuthenticationResult):
222
"DKIM result clause of an ``Authentication-Results`` header"
223
224
def __init__(self, version = None,
225
result = None, result_comment = None,
226
reason = None, reason_comment = None,
227
properties = None,
228
header_d = None, header_d_comment = None,
229
header_i = None, header_i_comment = None,
230
header_b = None, header_b_comment = None
231
):
232
AuthenticationResult.__init__(self, 'dkim', version,
233
result, result_comment, reason, reason_comment, properties)
234
if header_d: self.header_d = header_d
235
if header_d_comment: self.header_d_comment = header_d_comment
236
if header_i: self.header_i = header_i
237
if header_i_comment: self.header_i_comment = header_i_comment
238
if header_b: self.header_b = header_b
239
if header_b_comment: self.header_b_comment = header_b_comment
240
241
header_d, header_d_comment = make_authres_class_properties('header', 'd')
242
header_i, header_i_comment = make_authres_class_properties('header', 'i')
243
header_b, header_b_comment = make_authres_class_properties('header', 'b')
244
245
class DKIMADSPAuthenticationResult(AuthenticationResult):
246
"DKIM ADSP (RFC 5617) result clause of an ``Authentication-Results`` header"
247
248
def __init__(self, version = None,
249
result = None, result_comment = None,
250
reason = None, reason_comment = None,
251
properties = None,
252
header_from = None, header_from_comment = None
253
):
254
AuthenticationResult.__init__(self, 'dkim-adsp', version,
255
result, result_comment, reason, reason_comment, properties)
256
if header_from: self.header_from = header_from
257
if header_from_comment: self.header_from_comment = header_from_comment
258
259
header_from, header_from_comment = make_authres_class_properties('header', 'from')
260
261
class DomainKeysAuthenticationResult(AuthenticationResult):
262
"DomainKeys result clause of an ``Authentication-Results`` header"
263
264
def __init__(self, version = None,
265
result = None, result_comment = None,
266
reason = None, reason_comment = None,
267
properties = None,
268
header_d = None, header_d_comment = None,
269
header_from = None, header_from_comment = None,
270
header_sender = None, header_sender_comment = None
271
):
272
AuthenticationResult.__init__(self, 'domainkeys', version,
273
result, result_comment, reason, reason_comment, properties)
274
if header_d: self.header_d = header_d
275
if header_d_comment: self.header_d_comment = header_d_comment
276
if header_from: self.header_from = header_from
277
if header_from_comment: self.header_from_comment = header_from_comment
278
if header_sender: self.header_sender = header_sender
279
if header_sender_comment: self.header_sender_comment = header_sender_comment
280
281
header_d, header_d_comment = make_authres_class_properties('header', 'd')
282
header_from, header_from_comment = make_authres_class_properties('header', 'from')
283
header_sender, header_sender_comment = make_authres_class_properties('header', 'sender')
284
285
class SPFAuthenticationResult(AuthenticationResult):
286
"SPF result clause of an ``Authentication-Results`` header"
287
288
def __init__(self, version = None,
289
result = None, result_comment = None,
290
reason = None, reason_comment = None,
291
properties = None,
292
smtp_helo = None, smtp_helo_comment = None,
293
smtp_mailfrom = None, smtp_mailfrom_comment = None
294
):
295
AuthenticationResult.__init__(self, 'spf', version,
296
result, result_comment, reason, reason_comment, properties)
297
if smtp_helo: self.smtp_helo = smtp_helo
298
if smtp_helo_comment: self.smtp_helo_comment = smtp_helo_comment
299
if smtp_mailfrom: self.smtp_mailfrom = smtp_mailfrom
300
if smtp_mailfrom_comment: self.smtp_mailfrom_comment = smtp_mailfrom_comment
301
302
smtp_helo, smtp_helo_comment = make_authres_class_properties('smtp', 'helo')
303
smtp_mailfrom, smtp_mailfrom_comment = make_authres_class_properties('smtp', 'mailfrom')
304
305
class SenderIDAuthenticationResult(AuthenticationResult):
306
"Sender ID result clause of an ``Authentication-Results`` header"
307
308
def __init__(self, version = None,
309
result = None, result_comment = None,
310
reason = None, reason_comment = None,
311
properties = None,
312
header_from = None, header_from_comment = None,
313
header_sender = None, header_sender_comment = None,
314
header_resent_from = None, header_resent_from_comment = None,
315
header_resent_sender = None, header_resent_sender_comment = None
316
):
317
AuthenticationResult.__init__(self, 'sender-id', version,
318
result, result_comment, reason, reason_comment, properties)
319
if header_from: self.header_from = header_from
320
if header_from_comment: self.header_from_comment = header_from_comment
321
if header_sender: self.header_sender = header_sender
322
if header_sender_comment: self.header_sender_comment = header_sender_comment
323
if header_resent_from: self.header_resent_from = header_resent_from
324
if header_resent_from_comment: self.header_resent_from_comment = header_resent_from_comment
325
if header_resent_sender: self.header_resent_sender = header_resent_sender
326
if header_resent_sender_comment: self.header_resent_sender_comment = header_resent_sender_comment
327
328
header_from, header_from_comment = make_authres_class_properties('header', 'from')
329
header_sender, header_sender_comment = make_authres_class_properties('header', 'sender')
330
header_resent_from, header_resent_from_comment = make_authres_class_properties('header', 'resent-from')
331
header_resent_sender, header_resent_sender_comment = make_authres_class_properties('header', 'resent-sender')
332
333
@property
334
def header_pra(self):
335
return (
336
self.header_resent_sender or
337
self.header_resent_from or
338
self.header_sender or
339
self.header_from
197
340
)
198
return _all_features
199
200
# Simple API with implicit core-features-only context
201
###############################################################################
202
203
class AuthenticationResultsHeader(authres.core.AuthenticationResultsHeader):
204
@classmethod
205
def parse(self, string):
206
return authres.core.AuthenticationResultsHeader.parse(core_features(), string)
207
208
@classmethod
209
def parse_value(self, string):
210
return authres.core.AuthenticationResultsHeader.parse_value(core_features(), string)
211
212
def __init__(self,
213
authserv_id = None, authserv_id_comment = None,
214
version = None, version_comment = None,
215
results = None
216
):
217
authres.core.AuthenticationResultsHeader.__init__(self,
218
core_features(), authserv_id, authserv_id_comment, version, version_comment, results)
219
220
def result(method, version = None,
341
342
@property
343
def header_pra_comment(self):
344
if self.header_resent_sender:
345
return self.header_resent_sender_comment
346
elif self.header_resent_from:
347
return self.header_resent_from_comment
348
elif self.header_sender:
349
return self.header_sender_comment
350
elif self.header_from:
351
return self.header_from_comment
352
else:
353
return None
354
355
class IPRevAuthenticationResult(AuthenticationResult):
356
"iprev result clause of an ``Authentication-Results`` header"
357
358
def __init__(self, version = None,
359
result = None, result_comment = None,
360
reason = None, reason_comment = None,
361
properties = None,
362
policy_iprev = None, policy_iprev_comment = None
363
):
364
AuthenticationResult.__init__(self, 'iprev', version,
365
result, result_comment, reason, reason_comment, properties)
366
if policy_iprev: self.policy_iprev = policy_iprev
367
if policy_iprev_comment: self.policy_iprev_comment = policy_iprev_comment
368
369
policy_iprev, policy_iprev_comment = make_authres_class_properties('policy', 'iprev')
370
371
class SMTPAUTHAuthenticationResult(AuthenticationResult):
372
"SMTP AUTH result clause of an ``Authentication-Results`` header"
373
374
def __init__(self, version = None,
375
result = None, result_comment = None,
376
reason = None, reason_comment = None,
377
properties = None,
378
smtp_auth = None, smtp_auth_comment = None
379
):
380
AuthenticationResult.__init__(self, 'auth', version,
381
result, result_comment, reason, reason_comment, properties)
382
if smtp_auth: self.smtp_auth = smtp_auth
383
if smtp_auth_comment: self.smtp_auth_comment = smtp_auth_comment
384
385
smtp_auth, smtp_auth_comment = make_authres_class_properties('smtp', 'auth')
386
387
class VBRAuthenticationResult(AuthenticationResult):
388
"VBR (RFC 6212) result clause of an ``Authentication-Results`` header"
389
390
def __init__(self, version = None,
391
result = None, result_comment = None,
392
reason = None, reason_comment = None,
393
properties = None,
394
header_md = None, header_md_comment = None,
395
header_mv = None, header_mv_comment = None
396
):
397
AuthenticationResult.__init__(self, 'vbr', version,
398
result, result_comment, reason, reason_comment, properties)
399
if header_md: self.header_md = header_md
400
if header_md_comment: self.header_md_comment = header_md_comment
401
if header_mv: self.header_mv = header_mv
402
if header_mv_comment: self.header_mv_comment = header_mv_comment
403
404
header_md, header_md_comment = make_authres_class_properties('header', 'md')
405
header_mv, header_mv_comment = make_authres_class_properties('header', 'mv')
406
407
AUTHRES_CLASS_BY_AUTH_METHOD = {
408
'dkim': DKIMAuthenticationResult,
409
'dkim-adsp': DKIMADSPAuthenticationResult,
410
'domainkeys': DomainKeysAuthenticationResult,
411
'spf': SPFAuthenticationResult,
412
'sender-id': SenderIDAuthenticationResult,
413
'iprev': IPRevAuthenticationResult,
414
'auth': SMTPAUTHAuthenticationResult,
415
'vbr': VBRAuthenticationResult
416
}
417
418
def authres(method, version = None,
221
419
result = None, result_comment = None,
222
420
reason = None, reason_comment = None,
223
421
properties = None
224
422
):
225
return core_features().result(method, version,
226
result, result_comment, reason, reason_comment, properties)
227
228
def header(
229
authserv_id = None, authserv_id_comment = None,
230
version = None, version_comment = None,
231
results = None
232
):
233
return core_features().header(
234
authserv_id, authserv_id_comment, version, version_comment, results)
235
236
def parse(string):
237
return core_features().parse(string)
238
239
def parse_value(string):
240
return core_features().parse_value(string)
423
try:
424
return AUTHRES_CLASS_BY_AUTH_METHOD[method](version,
425
result, result_comment, reason, reason_comment, properties)
426
except KeyError:
427
return AuthenticationResult(method, version,
428
result, result_comment, reason, reason_comment, properties)
429
430
# AuthenticationResultsHeader class
431
# =============================================================================
432
433
class AuthenticationResultsHeader(object):
434
VERSIONS = ['1']
435
436
NONE_RESULT = NoneAuthenticationResult()
437
438
HEADER_FIELD_NAME = 'Authentication-Results'
439
HEADER_FIELD_PATTERN = re.compile(r'^Authentication-Results:\s*', re.I)
440
441
@classmethod
442
def parse(self, string):
443
"""
444
Creates an `AuthenticationResultsHeader` object by parsing an ``Authentication-
445
Results`` header (expecting the field name at the beginning). Expects the
446
header to have been unfolded.
447
"""
448
string, n = self.HEADER_FIELD_PATTERN.subn('', string, 1)
449
if n == 1:
450
return self.parse_value(string)
451
else:
452
raise SyntaxError('parse_with_name', 'Not an "Authentication-Results" header field: {0}'.format(string))
453
454
@classmethod
455
def parse_value(self, string):
456
"""
457
Creates an `AuthenticationResultsHeader` object by parsing an ``Authentication-
458
Results`` header value. Expects the header value to have been unfolded.
459
"""
460
header = self()
461
header._parse_text = string.rstrip('\r\n\t ')
462
header._parse()
463
return header
464
465
def __init__(self,
466
authserv_id = None, authserv_id_comment = None,
467
version = None, version_comment = None,
468
results = None
469
):
470
"""
471
Examples:
472
RFC 5451 B.2
473
>>> str(AuthenticationResultsHeader('test.example.org'))
474
'Authentication-Results: test.example.org; none'
475
476
>>> str(AuthenticationResultsHeader('test.example.org', version=1))
477
'Authentication-Results: test.example.org 1; none'
478
479
None RFC example of no authentication with comment:
480
>>> str(AuthenticationResultsHeader(authserv_id = 'test.example.org',
481
... results = [NoneAuthenticationResult(comment = 'SPF not checked for localhost')]))
482
'Authentication-Results: test.example.org; none (SPF not checked for localhost)'
483
484
RFC 5451 B.3
485
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
486
... results = [SPFAuthenticationResult(result = 'pass',
487
... smtp_mailfrom = 'example.net')]))
488
'Authentication-Results: example.com; spf=pass smtp.mailfrom=example.net'
489
490
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.com; spf=pass smtp.mailfrom=example.net')
491
>>> str(arobj.authserv_id)
492
'example.com'
493
>>> str(arobj.results[0])
494
'spf=pass smtp.mailfrom=example.net'
495
>>> str(arobj.results[0].method)
496
'spf'
497
>>> str(arobj.results[0].result)
498
'pass'
499
>>> str(arobj.results[0].smtp_mailfrom)
500
'example.net'
501
>>> str(arobj.results[0].smtp_helo)
502
'None'
503
>>> str(arobj.results[0].reason)
504
'None'
505
>>> str(arobj.results[0].properties[0].type)
506
'smtp'
507
>>> str(arobj.results[0].properties[0].name)
508
'mailfrom'
509
>>> str(arobj.results[0].properties[0].value)
510
'example.net'
511
512
RFC 5451 B.4(1)
513
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
514
... results = [SMTPAUTHAuthenticationResult(result = 'pass', result_comment = 'cram-md5',
515
... smtp_auth = 'sender@example.net'), SPFAuthenticationResult(result = 'pass',
516
... smtp_mailfrom = 'example.net')]))
517
'Authentication-Results: example.com; auth=pass (cram-md5) smtp.auth=sender@example.net; spf=pass smtp.mailfrom=example.net'
518
519
# Missing parsing header comment.
520
#FIXME
521
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.com; auth=pass (cram-md5) smtp.auth=sender@example.net; spf=pass smtp.mailfrom=example.net')
522
>>> str(arobj.authserv_id)
523
'example.com'
524
>>> str(arobj.results[0])
525
'auth=pass smtp.auth=sender@example.net'
526
>>> str(arobj.results[0].method)
527
'auth'
528
>>> str(arobj.results[0].result)
529
'pass'
530
>>> str(arobj.results[0].smtp_auth)
531
'sender@example.net'
532
>>> str(arobj.results[0].properties[0].type)
533
'smtp'
534
>>> str(arobj.results[0].properties[0].name)
535
'auth'
536
>>> str(arobj.results[0].properties[0].value)
537
'sender@example.net'
538
>>> str(arobj.results[1])
539
'spf=pass smtp.mailfrom=example.net'
540
>>> str(arobj.results[1].method)
541
'spf'
542
>>> str(arobj.results[1].result)
543
'pass'
544
>>> str(arobj.results[1].smtp_mailfrom)
545
'example.net'
546
>>> str(arobj.results[1].properties[0].type)
547
'smtp'
548
>>> str(arobj.results[1].properties[0].name)
549
'mailfrom'
550
>>> str(arobj.results[1].properties[0].value)
551
'example.net'
552
553
RFC 5451 B.4(2)
554
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
555
... results = [SenderIDAuthenticationResult(result = 'pass',
556
... header_from = 'example.com')]))
557
'Authentication-Results: example.com; sender-id=pass header.from=example.com'
558
559
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.com; sender-id=pass header.from=example.com')
560
>>> str(arobj.authserv_id)
561
'example.com'
562
>>> str(arobj.results[0])
563
'sender-id=pass header.from=example.com'
564
>>> str(arobj.results[0].method)
565
'sender-id'
566
>>> str(arobj.results[0].result)
567
'pass'
568
>>> str(arobj.results[0].header_from)
569
'example.com'
570
>>> try:
571
... str(arobj.results[0].smtp_mailfrom)
572
... except AttributeError as x:
573
... print(x)
574
'SenderIDAuthenticationResult' object has no attribute 'smtp_mailfrom'
575
>>> str(arobj.results[0].properties[0].type)
576
'header'
577
>>> str(arobj.results[0].properties[0].name)
578
'from'
579
>>> str(arobj.results[0].properties[0].value)
580
'example.com'
581
582
RFC 5451 B.5(1) # Note: RFC 5451 uses 'hardfail' instead of 'fail' for
583
SPF failures. This erratum is in the process of being corrected.
584
Examples here use the correct 'fail'. The authres module does not
585
validate result codes, so either will be processed.
586
587
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
588
... results = [SenderIDAuthenticationResult(result = 'fail',
589
... header_from = 'example.com'), DKIMAuthenticationResult(result = 'pass',
590
... header_i = 'sender@example.com', result_comment = 'good signature')]))
591
'Authentication-Results: example.com; sender-id=fail header.from=example.com; dkim=pass (good signature) header.i=sender@example.com'
592
593
# Missing parsing header comment.
594
#FIXME
595
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.com; sender-id=fail header.from=example.com; dkim=pass (good signature) header.i=sender@example.com')
596
>>> str(arobj.authserv_id)
597
'example.com'
598
>>> str(arobj.results[0])
599
'sender-id=fail header.from=example.com'
600
>>> str(arobj.results[0].method)
601
'sender-id'
602
>>> str(arobj.results[0].result)
603
'fail'
604
>>> str(arobj.results[0].header_from)
605
'example.com'
606
>>> str(arobj.results[0].properties[0].type)
607
'header'
608
>>> str(arobj.results[0].properties[0].name)
609
'from'
610
>>> str(arobj.results[0].properties[0].value)
611
'example.com'
612
>>> str(arobj.results[1])
613
'dkim=pass header.i=sender@example.com'
614
>>> str(arobj.results[1].method)
615
'dkim'
616
>>> str(arobj.results[1].result)
617
'pass'
618
>>> str(arobj.results[1].header_i)
619
'sender@example.com'
620
>>> str(arobj.results[1].properties[0].type)
621
'header'
622
>>> str(arobj.results[1].properties[0].name)
623
'i'
624
>>> str(arobj.results[1].properties[0].value)
625
'sender@example.com'
626
627
RFC 5451 B.5(2)
628
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
629
... results = [SMTPAUTHAuthenticationResult(result = 'pass', result_comment = 'cram-md5',
630
... smtp_auth = 'sender@example.com'), SPFAuthenticationResult(result = 'fail',
631
... smtp_mailfrom = 'example.com')]))
632
'Authentication-Results: example.com; auth=pass (cram-md5) smtp.auth=sender@example.com; spf=fail smtp.mailfrom=example.com'
633
634
# Missing parsing header comment.
635
#FIXME
636
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.com; auth=pass (cram-md5) smtp.auth=sender@example.com; spf=fail smtp.mailfrom=example.com')
637
>>> str(arobj.results[0])
638
'auth=pass smtp.auth=sender@example.com'
639
>>> str(arobj.results[0].method)
640
'auth'
641
>>> str(arobj.results[0].result)
642
'pass'
643
>>> str(arobj.results[0].smtp_auth)
644
'sender@example.com'
645
>>> str(arobj.results[0].properties[0].type)
646
'smtp'
647
>>> str(arobj.results[0].properties[0].name)
648
'auth'
649
>>> str(arobj.results[0].properties[0].value)
650
'sender@example.com'
651
>>> str(arobj.results[1])
652
'spf=fail smtp.mailfrom=example.com'
653
>>> str(arobj.results[1].method)
654
'spf'
655
>>> str(arobj.results[1].result)
656
'fail'
657
>>> str(arobj.results[1].smtp_mailfrom)
658
'example.com'
659
>>> str(arobj.results[1].properties[0].type)
660
'smtp'
661
>>> str(arobj.results[1].properties[0].name)
662
'mailfrom'
663
>>> str(arobj.results[1].properties[0].value)
664
'example.com'
665
666
RFC 5451 B.6(1)
667
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
668
... results = [DKIMAuthenticationResult(result = 'pass', result_comment = 'good signature',
669
... header_i = '@mail-router.example.net'), DKIMAuthenticationResult(result = 'fail',
670
... header_i = '@newyork.example.com', result_comment = 'bad signature')]))
671
'Authentication-Results: example.com; dkim=pass (good signature) header.i=@mail-router.example.net; dkim=fail (bad signature) header.i=@newyork.example.com'
672
673
# Missing parsing header comment.
674
#FIXME
675
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.com; dkim=pass (good signature) header.i=@mail-router.example.net; dkim=fail (bad signature) header.i=@newyork.example.com')
676
>>> str(arobj.results[0])
677
'dkim=pass header.i=@mail-router.example.net'
678
>>> str(arobj.results[0].method)
679
'dkim'
680
>>> str(arobj.results[0].result)
681
'pass'
682
>>> str(arobj.results[0].header_i)
683
'@mail-router.example.net'
684
>>> str(arobj.results[0].properties[0].type)
685
'header'
686
>>> str(arobj.results[0].properties[0].name)
687
'i'
688
>>> str(arobj.results[0].properties[0].value)
689
'@mail-router.example.net'
690
>>> str(arobj.results[1])
691
'dkim=fail header.i=@newyork.example.com'
692
>>> str(arobj.results[1].method)
693
'dkim'
694
>>> str(arobj.results[1].result)
695
'fail'
696
>>> str(arobj.results[1].header_i)
697
'@newyork.example.com'
698
>>> str(arobj.results[1].properties[0].type)
699
'header'
700
>>> str(arobj.results[1].properties[0].name)
701
'i'
702
>>> str(arobj.results[1].properties[0].value)
703
'@newyork.example.com'
704
705
RFC 5451 B.6(2)
706
>>> str(AuthenticationResultsHeader(authserv_id = 'example.net',
707
... results = [DKIMAuthenticationResult(result = 'pass', result_comment = 'good signature',
708
... header_i = '@newyork.example.com')]))
709
'Authentication-Results: example.net; dkim=pass (good signature) header.i=@newyork.example.com'
710
711
# Missing parsing header comment.
712
#FIXME
713
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.net; dkim=pass (good signature) header.i=@newyork.example.com')
714
>>> str(arobj.results[0])
715
'dkim=pass header.i=@newyork.example.com'
716
>>> str(arobj.results[0].method)
717
'dkim'
718
>>> str(arobj.results[0].result)
719
'pass'
720
>>> str(arobj.results[0].header_i)
721
'@newyork.example.com'
722
>>> str(arobj.results[0].properties[0].type)
723
'header'
724
>>> str(arobj.results[0].properties[0].name)
725
'i'
726
>>> str(arobj.results[0].properties[0].value)
727
'@newyork.example.com'
728
729
RFC 6008 A.1
730
>>> str(AuthenticationResultsHeader(authserv_id = 'mail-router.example.net',
731
... results = [DKIMAuthenticationResult(result = 'pass', result_comment = 'good signature',
732
... header_d = 'newyork.example.com', header_b = 'oINEO8hg'), DKIMAuthenticationResult(result = 'fail',
733
... header_d = 'newyork.example.com', result_comment = 'bad signature', header_b = 'EToRSuvU')]))
734
'Authentication-Results: mail-router.example.net; dkim=pass (good signature) header.d=newyork.example.com header.b=oINEO8hg; dkim=fail (bad signature) header.d=newyork.example.com header.b=EToRSuvU'
735
736
# Missing parsing header comment.
737
#FIXME
738
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: mail-router.example.net; dkim=pass (good signature) header.d=newyork.example.com header.b=oINEO8hg; dkim=fail (bad signature) header.d=newyork.example.com header.b=EToRSuvU')
739
>>> str(arobj.results[0])
740
'dkim=pass header.d=newyork.example.com header.b=oINEO8hg'
741
>>> str(arobj.results[0].method)
742
'dkim'
743
>>> str(arobj.results[0].result)
744
'pass'
745
>>> str(arobj.results[0].header_d)
746
'newyork.example.com'
747
>>> str(arobj.results[0].properties[0].type)
748
'header'
749
>>> str(arobj.results[0].properties[0].name)
750
'd'
751
>>> str(arobj.results[0].properties[0].value)
752
'newyork.example.com'
753
>>> str(arobj.results[0].header_b)
754
'oINEO8hg'
755
>>> str(arobj.results[0].properties[1].type)
756
'header'
757
>>> str(arobj.results[0].properties[1].name)
758
'b'
759
>>> str(arobj.results[0].properties[1].value)
760
'oINEO8hg'
761
>>> str(arobj.results[1].method)
762
'dkim'
763
>>> str(arobj.results[1].result)
764
'fail'
765
>>> str(arobj.results[1].header_d)
766
'newyork.example.com'
767
>>> str(arobj.results[1].properties[0].type)
768
'header'
769
>>> str(arobj.results[1].properties[0].name)
770
'd'
771
>>> str(arobj.results[1].properties[0].value)
772
'newyork.example.com'
773
>>> str(arobj.results[1].header_b)
774
'EToRSuvU'
775
>>> str(arobj.results[1].properties[1].type)
776
'header'
777
>>> str(arobj.results[1].properties[1].name)
778
'b'
779
>>> str(arobj.results[1].properties[1].value)
780
'EToRSuvU'
781
782
# RFC 5617 (based on RFC text, no examples provided)
783
>>> str(AuthenticationResultsHeader(authserv_id = 'example.com',
784
... results = [DKIMAuthenticationResult(result = 'fail', result_comment = 'bad signature',
785
... header_d = 'bank.example.net'), DKIMADSPAuthenticationResult(result = 'discard',
786
... header_from = 'phish@bank.example.com', result_comment = 'From domain and d= domain match')]))
787
'Authentication-Results: example.com; dkim=fail (bad signature) header.d=bank.example.net; dkim-adsp=discard (From domain and d= domain match) header.from=phish@bank.example.com'
788
789
# Missing parsing header comment.
790
#FIXME
791
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: example.com; dkim=fail (bad signature) header.d=bank.example.net; dkim-adsp=discard (From domain and d= domain match) header.from=phish@bank.example.com')
792
>>> str(arobj.results[1])
793
'dkim-adsp=discard header.from=phish@bank.example.com'
794
>>> str(arobj.results[1].method)
795
'dkim-adsp'
796
>>> str(arobj.results[1].result)
797
'discard'
798
>>> str(arobj.results[1].header_from)
799
'phish@bank.example.com'
800
>>> str(arobj.results[1].properties[0].type)
801
'header'
802
>>> str(arobj.results[1].properties[0].name)
803
'from'
804
>>> str(arobj.results[1].properties[0].value)
805
'phish@bank.example.com'
806
807
RFC 6212 A.1
808
>>> str(AuthenticationResultsHeader(authserv_id = 'mail-router.example.net',
809
... results = [DKIMAuthenticationResult(result = 'pass', result_comment = 'good signature',
810
... header_d = 'newyork.example.com', header_b = 'oINEO8hg'), VBRAuthenticationResult(result = 'pass',
811
... header_md = 'newyork.example.com', result_comment = 'voucher.example.net',
812
... header_mv = 'voucher.example.org')]))
813
'Authentication-Results: mail-router.example.net; dkim=pass (good signature) header.d=newyork.example.com header.b=oINEO8hg; vbr=pass (voucher.example.net) header.md=newyork.example.com header.mv=voucher.example.org'
814
815
# Missing parsing header comment.
816
#FIXME
817
>>> arobj = AuthenticationResultsHeader.parse('Authentication-Results: mail-router.example.net; dkim=pass (good signature) header.d=newyork.example.com header.b=oINEO8hg; vbr=pass (voucher.example.net) header.md=newyork.example.com header.mv=voucher.example.org')
818
>>> str(arobj.results[1])
819
'vbr=pass header.md=newyork.example.com header.mv=voucher.example.org'
820
>>> str(arobj.results[1].method)
821
'vbr'
822
>>> str(arobj.results[1].result)
823
'pass'
824
>>> str(arobj.results[1].header_md)
825
'newyork.example.com'
826
>>> str(arobj.results[1].properties[0].type)
827
'header'
828
>>> str(arobj.results[1].properties[0].name)
829
'md'
830
>>> str(arobj.results[1].properties[0].value)
831
'newyork.example.com'
832
>>> str(arobj.results[1].header_mv)
833
'voucher.example.org'
834
>>> str(arobj.results[1].properties[1].type)
835
'header'
836
>>> str(arobj.results[1].properties[1].name)
837
'mv'
838
>>> str(arobj.results[1].properties[1].value)
839
'voucher.example.org'
840
841
"""
842
self.authserv_id = authserv_id and authserv_id.lower()
843
self.authserv_id_comment = authserv_id_comment
844
self.version = version and str(version).lower()
845
if self.version and not self.version in self.VERSIONS:
846
raise UnsupportedVersionError(version = self.version)
847
self.version_comment = version_comment
848
if self.version_comment and not self.version:
849
raise OrphanCommentError('Cannot include header version comment without associated header version')
850
self.results = results or []
851
852
def __str__(self):
853
strs = []
854
strs.append(self.HEADER_FIELD_NAME)
855
strs.append(': ')
856
strs.append(self.authserv_id)
857
if self.authserv_id_comment:
858
strs.append(' (%s)' % self.authserv_id_comment)
859
if self.version:
860
strs.append(' ')
861
strs.append(self.version)
862
if self.version_comment:
863
strs.append(' (%s)' % self.version_comment)
864
if len(self.results):
865
for result in self.results:
866
strs.append('; ')
867
strs.append(str(result))
868
else:
869
strs.append('; ')
870
strs.append(str(self.NONE_RESULT))
871
return ''.join(strs)
872
873
# Principal parser methods
874
# =========================================================================
875
876
def _parse(self):
877
authserv_id = self._parse_authserv_id()
878
if not authserv_id:
879
raise SyntaxError('Expected authserv-id', self._parse_text)
880
881
self._parse_rfc5322_cfws()
882
883
version = self._parse_version()
884
if version and not version in self.VERSIONS:
885
raise UnsupportedVersionError(version = version)
886
887
self._parse_rfc5322_cfws()
888
889
results = []
890
result = True
891
while result:
892
result = self._parse_resinfo()
893
if result:
894
results.append(result)
895
if result == self.NONE_RESULT:
896
break
897
if not len(results):
898
raise SyntaxError('Expected "none" or at least one resinfo', self._parse_text)
899
elif results == [self.NONE_RESULT]:
900
results = []
901
902
self._parse_rfc5322_cfws()
903
self._parse_end()
904
905
self.authserv_id = authserv_id.lower()
906
self.version = version and version.lower()
907
self.results = results
908
909
def _parse_authserv_id(self):
910
return self._parse_rfc5322_dot_atom()
911
912
def _parse_version(self):
913
version_match = self._parse_pattern(r'\d+')
914
self._parse_rfc5322_cfws()
915
return version_match and version_match.group()
916
917
def _parse_resinfo(self):
918
self._parse_rfc5322_cfws()
919
if not self._parse_pattern(r';'):
920
return
921
self._parse_rfc5322_cfws()
922
if self._parse_pattern(r'none'):
923
return self.NONE_RESULT
924
else:
925
method, version, result = self._parse_methodspec()
926
self._parse_rfc5322_cfws()
927
reason = self._parse_reasonspec()
928
properties = []
929
property_ = True
930
while property_:
931
self._parse_rfc5322_cfws()
932
property_ = self._parse_propspec()
933
if property_:
934
properties.append(property_)
935
return authres(method, version, result, None, reason, None, properties)
936
937
def _parse_methodspec(self):
938
self._parse_rfc5322_cfws()
939
method, version = self._parse_method()
940
self._parse_rfc5322_cfws()
941
if not self._parse_pattern(r'='):
942
raise SyntaxError('Expected "="', self._parse_text)
943
self._parse_rfc5322_cfws()
944
result = self._parse_rfc5322_dot_atom()
945
if not result:
946
raise SyntaxError('Expected result', self._parse_text)
947
return (method, version, result)
948
949
def _parse_method(self):
950
method = self._parse_dot_key_atom()
951
if not method:
952
raise SyntaxError('Expected method', self._parse_text)
953
self._parse_rfc5322_cfws()
954
if not self._parse_pattern(r'/'):
955
return (method, None)
956
self._parse_rfc5322_cfws()
957
version_match = self._parse_pattern(r'\d+')
958
if not version_match:
959
raise SyntaxError('Expected version', self._parse_text)
960
return (method, version_match.group())
961
962
def _parse_reasonspec(self):
963
if self._parse_pattern(r'reason'):
964
self._parse_rfc5322_cfws()
965
if not self._parse_pattern(r'='):
966
raise SyntaxError('Expected "="', self._parse_text)
967
self._parse_rfc5322_cfws()
968
reasonspec = self._parse_rfc2045_value()
969
if not reasonspec:
970
raise SyntaxError('Expected reason', self._parse_text)
971
return reasonspec
972
973
def _parse_propspec(self):
974
ptype = self._parse_key_atom()
975
if not ptype:
976
return
977
elif ptype.lower() not in ['smtp', 'header', 'body', 'policy']:
978
raise SyntaxError('Invalid ptype; expected any of "smtp", "header", "body", "policy", got "%s"' % ptype, self._parse_text)
979
self._parse_rfc5322_cfws()
980
if not self._parse_pattern(r'\.'):
981
raise SyntaxError('Expected "."', self._parse_text)
982
self._parse_rfc5322_cfws()
983
property_ = self._parse_dot_key_atom()
984
self._parse_rfc5322_cfws()
985
if not self._parse_pattern(r'='):
986
raise SyntaxError('Expected "="', self._parse_text)
987
pvalue = self._parse_pvalue()
988
if not pvalue:
989
raise SyntaxError('Expected pvalue', self._parse_text)
990
return AuthenticationResultProperty(ptype, property_, pvalue)
991
992
def _parse_pvalue(self):
993
self._parse_rfc5322_cfws()
994
995
# The original rule is (modulo CFWS):
996
#
997
# pvalue = [ [local-part] "@" ] domain-name / value
998
# value = token / quoted-string
999
#
1000
# Distinguishing <token> and <domain-name> may require backtracking,
1001
# and in order to avoid the need for that, the following is a simpli-
1002
# fication of the <pvalue> rule from RFC 5451, erring on the side of
1003
# laxity.
1004
#
1005
# Since <local-part> is either a <quoted-string> or <dot-atom>, and
1006
# <value> is either a <quoted-string> or a <token>, and <dot-atom> and
1007
# <token> are very similar (<dot-atom> is a superset of <token> except
1008
# that it multiple dots may not be adjacent), we allow a union of ".",
1009
# "@" and <atext> characters (jointly denoted <ptext>) in the place of
1010
# <dot-atom> and <token>.
1011
#
1012
# We then allow four patterns:
1013
#
1014
# pvalue = quoted-string /
1015
# quoted-string "@" domain-name /
1016
# "@" domain-name /
1017
# 1*ptext
1018
1019
quoted_string_match = self._parse_rfc5322_quoted_string()
1020
if quoted_string_match:
1021
if self._parse_pattern(r'@'):
1022
# quoted-string "@" domain-name
1023
domain_name = self._parse_rfc5322_dot_atom()
1024
self._parse_rfc5322_cfws()
1025
if domain_name:
1026
return '%s@%s' % (quoted_string, domain_name)
1027
else:
1028
# quoted-string
1029
self._parse_rfc5322_cfws()
1030
# Look ahead to see whether pvalue terminates after quoted-string as expected:
1031
if re.match(r';|$', self._parse_text):
1032
return quoted_string
1033
else:
1034
if self._parse_pattern(r'@'):
1035
# "@" domain-name
1036
domain_name = self._parse_rfc5322_dot_atom()
1037
self._parse_rfc5322_cfws()
1038
if domain_name:
1039
return '@' + domain_name
1040
else:
1041
# 1*ptext
1042
pvalue_match = self._parse_pattern(r'%s+' % PTEXT_PATTERN)
1043
self._parse_rfc5322_cfws()
1044
if pvalue_match:
1045
return pvalue_match.group()
1046
1047
def _parse_end(self):
1048
if self._parse_text == '':
1049
return True
1050
else:
1051
raise SyntaxError('Expected end of text', self._parse_text)
1052
1053
# Generic grammar parser methods
1054
# =========================================================================
1055
1056
def _parse_pattern(self, pattern):
1057
match = [None]
1058
1059
def matched(m):
1060
match[0] = m
1061
return ''
1062
1063
# TODO: This effectively recompiles most patterns on each use, which
1064
# is far from efficient. This should be rearchitected.
1065
regexp = pattern if isre(pattern) else re.compile(r'^' + pattern, re.I)
1066
self._parse_text = regexp.sub(matched, self._parse_text, 1)
1067
return match[0]
1068
1069
def _parse_rfc2045_value(self):
1070
return self._parse_rfc2045_token() or self._parse_rfc5322_quoted_string()
1071
1072
def _parse_rfc2045_token(self):
1073
token_match = self._parse_pattern(RFC2045_TOKEN_PATTERN)
1074
return token_match and token_match.group()
1075
1076
def _parse_rfc5322_quoted_string(self):
1077
self._parse_rfc5322_cfws()
1078
if not self._parse_pattern(r'^"'):
1079
return
1080
all_qcontent = ''
1081
qcontent = True
1082
while qcontent:
1083
fws_match = self._parse_pattern(RFC5322_FWS_PATTERN)
1084
if fws_match:
1085
all_qcontent += fws_match.group()
1086
qcontent = self._parse_rfc5322_qcontent()
1087
if qcontent:
1088
all_qcontent += qcontent
1089
self._parse_pattern(RFC5322_FWS_PATTERN)
1090
if not self._parse_pattern(r'"'):
1091
raise SyntaxError('Expected <">', self._parse_text)
1092
self._parse_rfc5322_cfws()
1093
return all_qcontent
1094
1095
def _parse_rfc5322_qcontent(self):
1096
qtext_match = self._parse_pattern(r'%s+' % RFC5322_QTEXT_PATTERN)
1097
if qtext_match:
1098
return qtext_match.group()
1099
quoted_pair_match = self._parse_pattern(RFC5322_QUOTED_PAIR_PATTERN)
1100
if quoted_pair_match:
1101
return quoted_pair_match.group()
1102
1103
def _parse_rfc5322_dot_atom(self):
1104
self._parse_rfc5322_cfws()
1105
dot_atom_text_match = self._parse_pattern(r'%s+(?:\.%s+)*' %
1106
(RFC5322_ATEXT_PATTERN, RFC5322_ATEXT_PATTERN))
1107
self._parse_rfc5322_cfws()
1108
return dot_atom_text_match and dot_atom_text_match.group()
1109
1110
def _parse_dot_key_atom(self):
1111
# Like _parse_rfc5322_dot_atom, but disallows "/" (forward slash) and
1112
# "=" (equal sign).
1113
self._parse_rfc5322_cfws()
1114
dot_atom_text_match = self._parse_pattern(r'%s+(?:\.%s+)*' %
1115
(KTEXT_PATTERN, KTEXT_PATTERN))
1116
self._parse_rfc5322_cfws()
1117
return dot_atom_text_match and dot_atom_text_match.group()
1118
1119
def _parse_key_atom(self):
1120
# Like _parse_dot_key_atom, but also disallows "." (dot).
1121
self._parse_rfc5322_cfws()
1122
dot_atom_text_match = self._parse_pattern(r'%s+' % KTEXT_PATTERN)
1123
self._parse_rfc5322_cfws()
1124
return dot_atom_text_match and dot_atom_text_match.group()
1125
1126
def _parse_rfc5322_cfws(self):
1127
fws_match = False
1128
comment_match = True
1129
while comment_match:
1130
fws_match = fws_match or self._parse_pattern(RFC5322_FWS_PATTERN)
1131
comment_match = self._parse_rfc5322_comment()
1132
fws_match = fws_match or self._parse_pattern(RFC5322_FWS_PATTERN)
1133
return fws_match or comment_match
1134
1135
def _parse_rfc5322_comment(self):
1136
if self._parse_pattern(r'\('):
1137
while self._parse_pattern(RFC5322_FWS_PATTERN) or self._parse_rfc5322_ccontent(): pass
1138
if self._parse_pattern(r'^\)'):
1139
return True
1140
else:
1141
raise SyntaxError('comment: expected FWS or ccontent or ")"', self._parse_text)
1142
1143
def _parse_rfc5322_ccontent(self):
1144
if self._parse_pattern(r'%s+' % RFC5322_CTEXT_PATTERN):
1145
return True
1146
elif self._parse_pattern(RFC5322_QUOTED_PAIR_PATTERN):
1147
return True
1148
elif self._parse_rfc5322_comment():
1149
return True
241
1150
242
1151
# vim:sw=4 sts=4
Loggerhead is a web-based interface for Breezy
Version: 2.0.1