Tachyon /Users/simon/.local/share/uv/python/cpython-3.15.0-macos-aarch64-none/lib/python3.15/email/utils.py
2
Self Samples
2
Cumulative
1
Lines Hit
0.01%
% of Total
2
Max Self
2
Max Total
Intensity:
Cold Hot
Self Time
Total Time
Show All
Hot Only
Heat
Specialization
Line
Self
Total
Code
1
# Copyright (C) 2001 Python Software Foundation
2
# Author: Barry Warsaw
3
# Contact: email-sig@python.org
4
5
"""Miscellaneous utilities."""
6
7
__all__ = [
8
'collapse_rfc2231_value',
9
'decode_params',
10
'decode_rfc2231',
11
'encode_rfc2231',
12
'formataddr',
13
'formatdate',
14
'format_datetime',
15
'getaddresses',
16
'make_msgid',
17
'mktime_tz',
18
'parseaddr',
19
'parsedate',
20
'parsedate_tz',
21
'parsedate_to_datetime',
22
'unquote',
23
]
24
25
import os
26
import re
27
import time
28
import datetime
29
import urllib.parse
30
31
from email._parseaddr import quote
32
from email._parseaddr import AddressList as _AddressList
33
from email._parseaddr import mktime_tz
34
35
from email._parseaddr import parsedate, parsedate_tz, _parsedate_tz
36
37
COMMASPACE = ', '
38
EMPTYSTRING = ''
39
UEMPTYSTRING = ''
40
CRLF = '\r\n'
41
TICK = "'"
42
43
specialsre = re.compile(r'[][\\()<>@,:;".]')
44
escapesre = re.compile(r'[\\"]')
45
46
47
def _has_surrogates(s):
48
"""Return True if s may contain surrogate-escaped binary data."""
49
# This check is based on the fact that unless there are surrogates, utf8
50
# (Python's default encoding) can encode any string. This is the fastest
51
# way to check for surrogates, see bpo-11454 (moved to gh-55663) for timings.
52
try:
53
s.encode()
54
return False
55
except UnicodeEncodeError:
56
return True
57
58
# How to deal with a string containing bytes before handing it to the
59
# application through the 'normal' interface.
60
def _sanitize(string):
61
# Turn any escaped bytes into the Unicode 'unknown' char. If the escaped
62
# bytes happen to be utf-8 they will instead get decoded, even if they
63
# were invalid in the charset the source was supposed to be in. This
64
# seems like it is not a bad thing; a defect was still registered.
65
original_bytes = string.encode('utf-8', 'surrogateescape')
66
return original_bytes.decode('utf-8', 'replace')
67
68
69
70
# Helpers
71
72
def formataddr(pair, charset='utf-8'):
73
"""The inverse of parseaddr(), this takes a 2-tuple of the form
74
(realname, email_address) and returns the string value suitable
75
for an RFC 2822 From, To or Cc header.
76
77
If the first element of pair is false, then the second element is
78
returned unmodified.
79
80
The optional charset is the character set that is used to encode
81
realname in case realname is not ASCII safe. Can be an instance of str or
82
a Charset-like object which has a header_encode method. Default is
83
'utf-8'.
84
"""
85
name, address = pair
86
# The address MUST (per RFC) be ascii, so raise a UnicodeError if it isn't.
87
address.encode('ascii')
88
if name:
89
try:
90
name.encode('ascii')
91
except UnicodeEncodeError:
92
if isinstance(charset, str):
93
# lazy import to improve module import time
94
from email.charset import Charset
95
charset = Charset(charset)
96
encoded_name = charset.header_encode(name)
97
return "%s <%s>" % (encoded_name, address)
98
else:
99
quotes = ''
100
if specialsre.search(name):
101
quotes = '"'
102
name = escapesre.sub(r'\\\g<0>', name)
103
return '%s%s%s <%s>' % (quotes, name, quotes, address)
104
return address
105
106
107
def _iter_escaped_chars(addr):
108
pos = 0
109
escape = False
110
for pos, ch in enumerate(addr):
111
if escape:
112
yield (pos, '\\' + ch)
113
escape = False
114
elif ch == '\\':
115
escape = True
116
else:
117
yield (pos, ch)
118
if escape:
119
yield (pos, '\\')
120
121
122
def _strip_quoted_realnames(addr):
123
"""Strip real names between quotes."""
124
if '"' not in addr:
125
# Fast path
126
return addr
127
128
start = 0
129
open_pos = None
130
result = []
131
for pos, ch in _iter_escaped_chars(addr):
132
if ch == '"':
133
if open_pos is None:
134
open_pos = pos
135
else:
136
if start != open_pos:
137
result.append(addr[start:open_pos])
138
start = pos + 1
139
open_pos = None
140
141
if start < len(addr):
142
result.append(addr[start:])
143
144
return ''.join(result)
145
146
147
supports_strict_parsing = True
148
149
def getaddresses(fieldvalues, *, strict=True):
150
"""Return a list of (REALNAME, EMAIL) or ('','') for each fieldvalue.
151
152
When parsing fails for a fieldvalue, a 2-tuple of ('', '') is returned in
153
its place.
154
155
If strict is true, use a strict parser which rejects malformed inputs.
156
"""
157
158
# If strict is true, if the resulting list of parsed addresses is greater
159
# than the number of fieldvalues in the input list, a parsing error has
160
# occurred and consequently a list containing a single empty 2-tuple [('',
161
# '')] is returned in its place. This is done to avoid invalid output.
162
#
163
# Malformed input: getaddresses(['alice@example.com <bob@example.com>'])
164
# Invalid output: [('', 'alice@example.com'), ('', 'bob@example.com')]
165
# Safe output: [('', '')]
166
167
if not strict:
168
all = COMMASPACE.join(str(v) for v in fieldvalues)
169
a = _AddressList(all)
170
return a.addresslist
171
172
fieldvalues = [str(v) for v in fieldvalues]
173
fieldvalues = _pre_parse_validation(fieldvalues)
174
addr = COMMASPACE.join(fieldvalues)
175
a = _AddressList(addr)
176
result = _post_parse_validation(a.addresslist)
177
178
# Treat output as invalid if the number of addresses is not equal to the
179
# expected number of addresses.
180
n = 0
181
for v in fieldvalues:
182
# When a comma is used in the Real Name part it is not a deliminator.
183
# So strip those out before counting the commas.
184
v = _strip_quoted_realnames(v)
185
# Expected number of addresses: 1 + number of commas
186
n += 1 + v.count(',')
187
if len(result) != n:
188
return [('', '')]
189
190
return result
191
192
193
def _check_parenthesis(addr):
194
# Ignore parenthesis in quoted real names.
195
addr = _strip_quoted_realnames(addr)
196
197
opens = 0
198
for pos, ch in _iter_escaped_chars(addr):
199
if ch == '(':
200
opens += 1
201
elif ch == ')':
202
opens -= 1
203
if opens < 0:
204
return False
205
return (opens == 0)
206
207
208
def _pre_parse_validation(email_header_fields):
209
accepted_values = []
210
for v in email_header_fields:
211
if not _check_parenthesis(v):
212
v = "('', '')"
213
accepted_values.append(v)
214
215
return accepted_values
216
217
218
def _post_parse_validation(parsed_email_header_tuples):
219
accepted_values = []
220
# The parser would have parsed a correctly formatted domain-literal
221
# The existence of an [ after parsing indicates a parsing failure
222
for v in parsed_email_header_tuples:
223
if '[' in v[1]:
224
v = ('', '')
225
accepted_values.append(v)
226
227
return accepted_values
228
229
230
def _format_timetuple_and_zone(timetuple, zone):
231
return '%s, %02d %s %04d %02d:%02d:%02d %s' % (
232
['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'][timetuple[6]],
233
timetuple[2],
234
['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
235
'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec'][timetuple[1] - 1],
236
timetuple[0], timetuple[3], timetuple[4], timetuple[5],
237
zone)
238
239
def formatdate(timeval=None, localtime=False, usegmt=False):
240
"""Returns a date string as specified by RFC 2822, e.g.:
241
242
Fri, 09 Nov 2001 01:08:47 -0000
243
244
Optional timeval if given is a floating-point time value as accepted by
245
gmtime() and localtime(), otherwise the current time is used.
246
247
Optional localtime is a flag that when True, interprets timeval, and
248
returns a date relative to the local timezone instead of UTC, properly
249
taking daylight savings time into account.
250
251
Optional argument usegmt means that the timezone is written out as
252
an ascii string, not numeric one (so "GMT" instead of "+0000"). This
253
is needed for HTTP, and is only used when localtime==False.
254
"""
255
# Note: we cannot use strftime() because that honors the locale and RFC
256
# 2822 requires that day and month names be the English abbreviations.
257
if timeval is None:
258
timeval = time.time()
259
dt = datetime.datetime.fromtimestamp(timeval, datetime.timezone.utc)
260
261
if localtime:
262
dt = dt.astimezone()
263
usegmt = False
264
elif not usegmt:
265
dt = dt.replace(tzinfo=None)
266
2
2
return format_datetime(dt, usegmt)
267
268
def format_datetime(dt, usegmt=False):
269
"""Turn a datetime into a date string as specified in RFC 2822.
270
271
If usegmt is True, dt must be an aware datetime with an offset of zero. In
272
this case 'GMT' will be rendered instead of the normal +0000 required by
273
RFC2822. This is to support HTTP headers involving date stamps.
274
"""
275
now = dt.timetuple()
276
if usegmt:
277
if dt.tzinfo is None or dt.tzinfo != datetime.timezone.utc:
278
raise ValueError("usegmt option requires a UTC datetime")
279
zone = 'GMT'
280
elif dt.tzinfo is None:
281
zone = '-0000'
282
else:
283
zone = dt.strftime("%z")
284
return _format_timetuple_and_zone(now, zone)
285
286
287
def make_msgid(idstring=None, domain=None):
288
"""Returns a string suitable for RFC 2822 compliant Message-ID, e.g:
289
290
<142480216486.20800.16526388040877946887@nightshade.la.mastaler.com>
291
292
Optional idstring if given is a string used to strengthen the
293
uniqueness of the message id. Optional domain if given provides the
294
portion of the message id after the '@'. It defaults to the locally
295
defined hostname.
296
"""
297
# Lazy imports to speedup module import time
298
# (no other functions in email.utils need these modules)
299
import random
300
import socket
301
302
timeval = int(time.time()*100)
303
pid = os.getpid()
304
randint = random.getrandbits(64)
305
if idstring is None:
306
idstring = ''
307
else:
308
idstring = '.' + idstring
309
if domain is None:
310
domain = socket.getfqdn()
311
msgid = '<%d.%d.%d%s@%s>' % (timeval, pid, randint, idstring, domain)
312
return msgid
313
314
315
def parsedate_to_datetime(data):
316
parsed_date_tz = _parsedate_tz(data)
317
if parsed_date_tz is None:
318
raise ValueError('Invalid date value or format "%s"' % str(data))
319
*dtuple, tz = parsed_date_tz
320
try:
321
if tz is None:
322
return datetime.datetime(*dtuple[:6])
323
return datetime.datetime(*dtuple[:6],
324
tzinfo=datetime.timezone(datetime.timedelta(seconds=tz)))
325
except OverflowError as exc:
326
raise ValueError('Invalid date value or format "%s"' % str(data)) from exc
327
328
329
def parseaddr(addr, *, strict=True):
330
"""
331
Parse addr into its constituent realname and email address parts.
332
333
Return a tuple of realname and email address, unless the parse fails, in
334
which case return a 2-tuple of ('', '').
335
336
If strict is True, use a strict parser which rejects malformed inputs.
337
"""
338
if not strict:
339
addrs = _AddressList(addr).addresslist
340
if not addrs:
341
return ('', '')
342
return addrs[0]
343
344
if isinstance(addr, list):
345
addr = addr[0]
346
347
if not isinstance(addr, str):
348
return ('', '')
349
350
addr = _pre_parse_validation([addr])[0]
351
addrs = _post_parse_validation(_AddressList(addr).addresslist)
352
353
if not addrs or len(addrs) > 1:
354
return ('', '')
355
356
return addrs[0]
357
358
359
# rfc822.unquote() doesn't properly de-backslash-ify in Python pre-2.3.
360
def unquote(str):
361
"""Remove quotes from a string."""
362
if len(str) > 1:
363
if str.startswith('"') and str.endswith('"'):
364
return str[1:-1].replace('\\\\', '\\').replace('\\"', '"')
365
if str.startswith('<') and str.endswith('>'):
366
return str[1:-1]
367
return str
368
369
370
371
# RFC2231-related functions - parameter encoding and decoding
372
def decode_rfc2231(s):
373
"""Decode string according to RFC 2231"""
374
parts = s.split(TICK, 2)
375
if len(parts) <= 2:
376
return None, None, s
377
return parts
378
379
380
def encode_rfc2231(s, charset=None, language=None):
381
"""Encode string according to RFC 2231.
382
383
If neither charset nor language is given, then s is returned as-is. If
384
charset is given but not language, the string is encoded using the empty
385
string for language.
386
"""
387
s = urllib.parse.quote(s, safe='', encoding=charset or 'ascii')
388
if charset is None and language is None:
389
return s
390
if language is None:
391
language = ''
392
return "%s'%s'%s" % (charset, language, s)
393
394
395
rfc2231_continuation = re.compile(r'^(?P<name>\w+)\*((?P<num>[0-9]+)\*?)?$',
396
re.ASCII)
397
398
def decode_params(params):
399
"""Decode parameters list according to RFC 2231.
400
401
params is a sequence of 2-tuples containing (param name, string value).
402
"""
403
new_params = [params[0]]
404
# Map parameter's name to a list of continuations. The values are a
405
# 3-tuple of the continuation number, the string value, and a flag
406
# specifying whether a particular segment is %-encoded.
407
rfc2231_params = {}
408
for name, value in params[1:]:
409
encoded = name.endswith('*')
410
value = unquote(value)
411
mo = rfc2231_continuation.match(name)
412
if mo:
413
name, num = mo.group('name', 'num')
414
if num is not None:
415
num = int(num)
416
rfc2231_params.setdefault(name, []).append((num, value, encoded))
417
else:
418
new_params.append((name, '"%s"' % quote(value)))
419
if rfc2231_params:
420
for name, continuations in rfc2231_params.items():
421
value = []
422
extended = False
423
# Sort by number, treating None as 0 if there is no 0,
424
# and ignore it if there is already a 0.
425
has_zero = any(x[0] == 0 for x in continuations)
426
if has_zero:
427
continuations = [x for x in continuations if x[0] is not None]
428
else:
429
continuations = [(x[0] or 0, x[1], x[2]) for x in continuations]
430
continuations.sort(key=lambda x: x[0])
431
# And now append all values in numerical order, converting
432
# %-encodings for the encoded segments. If any of the
433
# continuation names ends in a *, then the entire string, after
434
# decoding segments and concatenating, must have the charset and
435
# language specifiers at the beginning of the string.
436
for num, s, encoded in continuations:
437
if encoded:
438
# Decode as "latin-1", so the characters in s directly
439
# represent the percent-encoded octet values.
440
# collapse_rfc2231_value treats this as an octet sequence.
441
s = urllib.parse.unquote(s, encoding="latin-1")
442
extended = True
443
value.append(s)
444
value = quote(EMPTYSTRING.join(value))
445
if extended:
446
charset, language, value = decode_rfc2231(value)
447
new_params.append((name, (charset, language, '"%s"' % value)))
448
else:
449
new_params.append((name, '"%s"' % value))
450
return new_params
451
452
def collapse_rfc2231_value(value, errors='replace',
453
fallback_charset='us-ascii'):
454
if not isinstance(value, tuple) or len(value) != 3:
455
return unquote(value)
456
# While value comes to us as a string, we need it to be a bytes
457
# object. We do not want bytes() normal utf-8 decoder, we want a straight
458
# interpretation of the string as character bytes.
459
charset, language, text = value
460
if charset is None:
461
# Issue 17369: if charset/lang is None, decode_rfc2231 couldn't parse
462
# the value, so use the fallback_charset.
463
charset = fallback_charset
464
rawbytes = bytes(text, 'raw-unicode-escape')
465
try:
466
# Explicitly look up the codec for warning generation, see gh-140030
467
# Can be removed in 3.17
468
import codecs
469
codecs.lookup(charset)
470
return str(rawbytes, charset, errors)
471
except LookupError:
472
# charset is not a known codec.
473
return unquote(text)
474
475
476
#
477
# datetime doesn't provide a localtime function yet, so provide one. Code
478
# adapted from the patch in issue 9527. This may not be perfect, but it is
479
# better than not having it.
480
#
481
482
def localtime(dt=None):
483
"""Return local time as an aware datetime object.
484
485
If called without arguments, return current time. Otherwise *dt*
486
argument should be a datetime instance, and it is converted to the
487
local time zone according to the system time zone database. If *dt* is
488
naive (that is, dt.tzinfo is None), it is assumed to be in local time.
489
490
"""
491
if dt is None:
492
dt = datetime.datetime.now()
493
return dt.astimezone()