# Copyright (C) 2002-2006 Python Software Foundation | |
# Author: Ben Gertzfield, Barry Warsaw | |
# Contact: email-sig@python.org | |
"""Header encoding and decoding functionality.""" | |
__all__ = [ | |
'Header', | |
'decode_header', | |
'make_header', | |
] | |
import re | |
import binascii | |
import email.quoprimime | |
import email.base64mime | |
from email.errors import HeaderParseError | |
from email.charset import Charset | |
NL = '\n' | |
SPACE = ' ' | |
USPACE = u' ' | |
SPACE8 = ' ' * 8 | |
UEMPTYSTRING = u'' | |
MAXLINELEN = 76 | |
USASCII = Charset('us-ascii') | |
UTF8 = Charset('utf-8') | |
# Match encoded-word strings in the form =?charset?q?Hello_World?= | |
ecre = re.compile(r''' | |
=\? # literal =? | |
(?P<charset>[^?]*?) # non-greedy up to the next ? is the charset | |
\? # literal ? | |
(?P<encoding>[qb]) # either a "q" or a "b", case insensitive | |
\? # literal ? | |
(?P<encoded>.*?) # non-greedy up to the next ?= is the encoded string | |
\?= # literal ?= | |
(?=[ \t]|$) # whitespace or the end of the string | |
''', re.VERBOSE | re.IGNORECASE | re.MULTILINE) | |
# Field name regexp, including trailing colon, but not separating whitespace, | |
# according to RFC 2822. Character range is from tilde to exclamation mark. | |
# For use with .match() | |
fcre = re.compile(r'[\041-\176]+:$') | |
# Find a header embedded in a putative header value. Used to check for | |
# header injection attack. | |
_embeded_header = re.compile(r'\n[^ \t]+:') | |
# Helpers | |
_max_append = email.quoprimime._max_append | |
def decode_header(header): | |
"""Decode a message header value without converting charset. | |
Returns a list of (decoded_string, charset) pairs containing each of the | |
decoded parts of the header. Charset is None for non-encoded parts of the | |
header, otherwise a lower-case string containing the name of the character | |
set specified in the encoded string. | |
An email.errors.HeaderParseError may be raised when certain decoding error | |
occurs (e.g. a base64 decoding exception). | |
""" | |
# If no encoding, just return the header | |
header = str(header) | |
if not ecre.search(header): | |
return [(header, None)] | |
decoded = [] | |
dec = '' | |
for line in header.splitlines(): | |
# This line might not have an encoding in it | |
if not ecre.search(line): | |
decoded.append((line, None)) | |
continue | |
parts = ecre.split(line) | |
while parts: | |
unenc = parts.pop(0).strip() | |
if unenc: | |
# Should we continue a long line? | |
if decoded and decoded[-1][1] is None: | |
decoded[-1] = (decoded[-1][0] + SPACE + unenc, None) | |
else: | |
decoded.append((unenc, None)) | |
if parts: | |
charset, encoding = [s.lower() for s in parts[0:2]] | |
encoded = parts[2] | |
dec = None | |
if encoding == 'q': | |
dec = email.quoprimime.header_decode(encoded) | |
elif encoding == 'b': | |
paderr = len(encoded) % 4 # Postel's law: add missing padding | |
if paderr: | |
encoded += '==='[:4 - paderr] | |
try: | |
dec = email.base64mime.decode(encoded) | |
except binascii.Error: | |
# Turn this into a higher level exception. BAW: Right | |
# now we throw the lower level exception away but | |
# when/if we get exception chaining, we'll preserve it. | |
raise HeaderParseError | |
if dec is None: | |
dec = encoded | |
if decoded and decoded[-1][1] == charset: | |
decoded[-1] = (decoded[-1][0] + dec, decoded[-1][1]) | |
else: | |
decoded.append((dec, charset)) | |
del parts[0:3] | |
return decoded | |
def make_header(decoded_seq, maxlinelen=None, header_name=None, | |
continuation_ws=' '): | |
"""Create a Header from a sequence of pairs as returned by decode_header() | |
decode_header() takes a header value string and returns a sequence of | |
pairs of the format (decoded_string, charset) where charset is the string | |
name of the character set. | |
This function takes one of those sequence of pairs and returns a Header | |
instance. Optional maxlinelen, header_name, and continuation_ws are as in | |
the Header constructor. | |
""" | |
h = Header(maxlinelen=maxlinelen, header_name=header_name, | |
continuation_ws=continuation_ws) | |
for s, charset in decoded_seq: | |
# None means us-ascii but we can simply pass it on to h.append() | |
if charset is not None and not isinstance(charset, Charset): | |
charset = Charset(charset) | |
h.append(s, charset) | |
return h | |
class Header: | |
def __init__(self, s=None, charset=None, | |
maxlinelen=None, header_name=None, | |
continuation_ws=' ', errors='strict'): | |
"""Create a MIME-compliant header that can contain many character sets. | |
Optional s is the initial header value. If None, the initial header | |
value is not set. You can later append to the header with .append() | |
method calls. s may be a byte string or a Unicode string, but see the | |
.append() documentation for semantics. | |
Optional charset serves two purposes: it has the same meaning as the | |
charset argument to the .append() method. It also sets the default | |
character set for all subsequent .append() calls that omit the charset | |
argument. If charset is not provided in the constructor, the us-ascii | |
charset is used both as s's initial charset and as the default for | |
subsequent .append() calls. | |
The maximum line length can be specified explicit via maxlinelen. For | |
splitting the first line to a shorter value (to account for the field | |
header which isn't included in s, e.g. `Subject') pass in the name of | |
the field in header_name. The default maxlinelen is 76. | |
continuation_ws must be RFC 2822 compliant folding whitespace (usually | |
either a space or a hard tab) which will be prepended to continuation | |
lines. | |
errors is passed through to the .append() call. | |
""" | |
if charset is None: | |
charset = USASCII | |
if not isinstance(charset, Charset): | |
charset = Charset(charset) | |
self._charset = charset | |
self._continuation_ws = continuation_ws | |
cws_expanded_len = len(continuation_ws.replace('\t', SPACE8)) | |
# BAW: I believe `chunks' and `maxlinelen' should be non-public. | |
self._chunks = [] | |
if s is not None: | |
self.append(s, charset, errors) | |
if maxlinelen is None: | |
maxlinelen = MAXLINELEN | |
if header_name is None: | |
# We don't know anything about the field header so the first line | |
# is the same length as subsequent lines. | |
self._firstlinelen = maxlinelen | |
else: | |
# The first line should be shorter to take into account the field | |
# header. Also subtract off 2 extra for the colon and space. | |
self._firstlinelen = maxlinelen - len(header_name) - 2 | |
# Second and subsequent lines should subtract off the length in | |
# columns of the continuation whitespace prefix. | |
self._maxlinelen = maxlinelen - cws_expanded_len | |
def __str__(self): | |
"""A synonym for self.encode().""" | |
return self.encode() | |
def __unicode__(self): | |
"""Helper for the built-in unicode function.""" | |
uchunks = [] | |
lastcs = None | |
for s, charset in self._chunks: | |
# We must preserve spaces between encoded and non-encoded word | |
# boundaries, which means for us we need to add a space when we go | |
# from a charset to None/us-ascii, or from None/us-ascii to a | |
# charset. Only do this for the second and subsequent chunks. | |
nextcs = charset | |
if uchunks: | |
if lastcs not in (None, 'us-ascii'): | |
if nextcs in (None, 'us-ascii'): | |
uchunks.append(USPACE) | |
nextcs = None | |
elif nextcs not in (None, 'us-ascii'): | |
uchunks.append(USPACE) | |
lastcs = nextcs | |
uchunks.append(unicode(s, str(charset))) | |
return UEMPTYSTRING.join(uchunks) | |
# Rich comparison operators for equality only. BAW: does it make sense to | |
# have or explicitly disable <, <=, >, >= operators? | |
def __eq__(self, other): | |
# other may be a Header or a string. Both are fine so coerce | |
# ourselves to a string, swap the args and do another comparison. | |
return other == self.encode() | |
def __ne__(self, other): | |
return not self == other | |
def append(self, s, charset=None, errors='strict'): | |
"""Append a string to the MIME header. | |
Optional charset, if given, should be a Charset instance or the name | |
of a character set (which will be converted to a Charset instance). A | |
value of None (the default) means that the charset given in the | |
constructor is used. | |
s may be a byte string or a Unicode string. If it is a byte string | |
(i.e. isinstance(s, str) is true), then charset is the encoding of | |
that byte string, and a UnicodeError will be raised if the string | |
cannot be decoded with that charset. If s is a Unicode string, then | |
charset is a hint specifying the character set of the characters in | |
the string. In this case, when producing an RFC 2822 compliant header | |
using RFC 2047 rules, the Unicode string will be encoded using the | |
following charsets in order: us-ascii, the charset hint, utf-8. The | |
first character set not to provoke a UnicodeError is used. | |
Optional `errors' is passed as the third argument to any unicode() or | |
ustr.encode() call. | |
""" | |
if charset is None: | |
charset = self._charset | |
elif not isinstance(charset, Charset): | |
charset = Charset(charset) | |
# If the charset is our faux 8bit charset, leave the string unchanged | |
if charset != '8bit': | |
# We need to test that the string can be converted to unicode and | |
# back to a byte string, given the input and output codecs of the | |
# charset. | |
if isinstance(s, str): | |
# Possibly raise UnicodeError if the byte string can't be | |
# converted to a unicode with the input codec of the charset. | |
incodec = charset.input_codec or 'us-ascii' | |
ustr = unicode(s, incodec, errors) | |
# Now make sure that the unicode could be converted back to a | |
# byte string with the output codec, which may be different | |
# than the iput coded. Still, use the original byte string. | |
outcodec = charset.output_codec or 'us-ascii' | |
ustr.encode(outcodec, errors) | |
elif isinstance(s, unicode): | |
# Now we have to be sure the unicode string can be converted | |
# to a byte string with a reasonable output codec. We want to | |
# use the byte string in the chunk. | |
for charset in USASCII, charset, UTF8: | |
try: | |
outcodec = charset.output_codec or 'us-ascii' | |
s = s.encode(outcodec, errors) | |
break | |
except UnicodeError: | |
pass | |
else: | |
assert False, 'utf-8 conversion failed' | |
self._chunks.append((s, charset)) | |
def _split(self, s, charset, maxlinelen, splitchars): | |
# Split up a header safely for use with encode_chunks. | |
splittable = charset.to_splittable(s) | |
encoded = charset.from_splittable(splittable, True) | |
elen = charset.encoded_header_len(encoded) | |
# If the line's encoded length first, just return it | |
if elen <= maxlinelen: | |
return [(encoded, charset)] | |
# If we have undetermined raw 8bit characters sitting in a byte | |
# string, we really don't know what the right thing to do is. We | |
# can't really split it because it might be multibyte data which we | |
# could break if we split it between pairs. The least harm seems to | |
# be to not split the header at all, but that means they could go out | |
# longer than maxlinelen. | |
if charset == '8bit': | |
return [(s, charset)] | |
# BAW: I'm not sure what the right test here is. What we're trying to | |
# do is be faithful to RFC 2822's recommendation that ($2.2.3): | |
# | |
# "Note: Though structured field bodies are defined in such a way that | |
# folding can take place between many of the lexical tokens (and even | |
# within some of the lexical tokens), folding SHOULD be limited to | |
# placing the CRLF at higher-level syntactic breaks." | |
# | |
# For now, I can only imagine doing this when the charset is us-ascii, | |
# although it's possible that other charsets may also benefit from the | |
# higher-level syntactic breaks. | |
elif charset == 'us-ascii': | |
return self._split_ascii(s, charset, maxlinelen, splitchars) | |
# BAW: should we use encoded? | |
elif elen == len(s): | |
# We can split on _maxlinelen boundaries because we know that the | |
# encoding won't change the size of the string | |
splitpnt = maxlinelen | |
first = charset.from_splittable(splittable[:splitpnt], False) | |
last = charset.from_splittable(splittable[splitpnt:], False) | |
else: | |
# Binary search for split point | |
first, last = _binsplit(splittable, charset, maxlinelen) | |
# first is of the proper length so just wrap it in the appropriate | |
# chrome. last must be recursively split. | |
fsplittable = charset.to_splittable(first) | |
fencoded = charset.from_splittable(fsplittable, True) | |
chunk = [(fencoded, charset)] | |
return chunk + self._split(last, charset, self._maxlinelen, splitchars) | |
def _split_ascii(self, s, charset, firstlen, splitchars): | |
chunks = _split_ascii(s, firstlen, self._maxlinelen, | |
self._continuation_ws, splitchars) | |
return zip(chunks, [charset]*len(chunks)) | |
def _encode_chunks(self, newchunks, maxlinelen): | |
# MIME-encode a header with many different charsets and/or encodings. | |
# | |
# Given a list of pairs (string, charset), return a MIME-encoded | |
# string suitable for use in a header field. Each pair may have | |
# different charsets and/or encodings, and the resulting header will | |
# accurately reflect each setting. | |
# | |
# Each encoding can be email.utils.QP (quoted-printable, for | |
# ASCII-like character sets like iso-8859-1), email.utils.BASE64 | |
# (Base64, for non-ASCII like character sets like KOI8-R and | |
# iso-2022-jp), or None (no encoding). | |
# | |
# Each pair will be represented on a separate line; the resulting | |
# string will be in the format: | |
# | |
# =?charset1?q?Mar=EDa_Gonz=E1lez_Alonso?=\n | |
# =?charset2?b?SvxyZ2VuIEL2aW5n?=" | |
chunks = [] | |
for header, charset in newchunks: | |
if not header: | |
continue | |
if charset is None or charset.header_encoding is None: | |
s = header | |
else: | |
s = charset.header_encode(header) | |
# Don't add more folding whitespace than necessary | |
if chunks and chunks[-1].endswith(' '): | |
extra = '' | |
else: | |
extra = ' ' | |
_max_append(chunks, s, maxlinelen, extra) | |
joiner = NL + self._continuation_ws | |
return joiner.join(chunks) | |
def encode(self, splitchars=';, '): | |
"""Encode a message header into an RFC-compliant format. | |
There are many issues involved in converting a given string for use in | |
an email header. Only certain character sets are readable in most | |
email clients, and as header strings can only contain a subset of | |
7-bit ASCII, care must be taken to properly convert and encode (with | |
Base64 or quoted-printable) header strings. In addition, there is a | |
75-character length limit on any given encoded header field, so | |
line-wrapping must be performed, even with double-byte character sets. | |
This method will do its best to convert the string to the correct | |
character set used in email, and encode and line wrap it safely with | |
the appropriate scheme for that character set. | |
If the given charset is not known or an error occurs during | |
conversion, this function will return the header untouched. | |
Optional splitchars is a string containing characters to split long | |
ASCII lines on, in rough support of RFC 2822's `highest level | |
syntactic breaks'. This doesn't affect RFC 2047 encoded lines. | |
""" | |
newchunks = [] | |
maxlinelen = self._firstlinelen | |
lastlen = 0 | |
for s, charset in self._chunks: | |
# The first bit of the next chunk should be just long enough to | |
# fill the next line. Don't forget the space separating the | |
# encoded words. | |
targetlen = maxlinelen - lastlen - 1 | |
if targetlen < charset.encoded_header_len(''): | |
# Stick it on the next line | |
targetlen = maxlinelen | |
newchunks += self._split(s, charset, targetlen, splitchars) | |
lastchunk, lastcharset = newchunks[-1] | |
lastlen = lastcharset.encoded_header_len(lastchunk) | |
value = self._encode_chunks(newchunks, maxlinelen) | |
if _embeded_header.search(value): | |
raise HeaderParseError("header value appears to contain " | |
"an embedded header: {!r}".format(value)) | |
return value | |
def _split_ascii(s, firstlen, restlen, continuation_ws, splitchars): | |
lines = [] | |
maxlen = firstlen | |
for line in s.splitlines(): | |
# Ignore any leading whitespace (i.e. continuation whitespace) already | |
# on the line, since we'll be adding our own. | |
line = line.lstrip() | |
if len(line) < maxlen: | |
lines.append(line) | |
maxlen = restlen | |
continue | |
# Attempt to split the line at the highest-level syntactic break | |
# possible. Note that we don't have a lot of smarts about field | |
# syntax; we just try to break on semi-colons, then commas, then | |
# whitespace. | |
for ch in splitchars: | |
if ch in line: | |
break | |
else: | |
# There's nothing useful to split the line on, not even spaces, so | |
# just append this line unchanged | |
lines.append(line) | |
maxlen = restlen | |
continue | |
# Now split the line on the character plus trailing whitespace | |
cre = re.compile(r'%s\s*' % ch) | |
if ch in ';,': | |
eol = ch | |
else: | |
eol = '' | |
joiner = eol + ' ' | |
joinlen = len(joiner) | |
wslen = len(continuation_ws.replace('\t', SPACE8)) | |
this = [] | |
linelen = 0 | |
for part in cre.split(line): | |
curlen = linelen + max(0, len(this)-1) * joinlen | |
partlen = len(part) | |
onfirstline = not lines | |
# We don't want to split after the field name, if we're on the | |
# first line and the field name is present in the header string. | |
if ch == ' ' and onfirstline and \ | |
len(this) == 1 and fcre.match(this[0]): | |
this.append(part) | |
linelen += partlen | |
elif curlen + partlen > maxlen: | |
if this: | |
lines.append(joiner.join(this) + eol) | |
# If this part is longer than maxlen and we aren't already | |
# splitting on whitespace, try to recursively split this line | |
# on whitespace. | |
if partlen > maxlen and ch != ' ': | |
subl = _split_ascii(part, maxlen, restlen, | |
continuation_ws, ' ') | |
lines.extend(subl[:-1]) | |
this = [subl[-1]] | |
else: | |
this = [part] | |
linelen = wslen + len(this[-1]) | |
maxlen = restlen | |
else: | |
this.append(part) | |
linelen += partlen | |
# Put any left over parts on a line by themselves | |
if this: | |
lines.append(joiner.join(this)) | |
return lines | |
def _binsplit(splittable, charset, maxlinelen): | |
i = 0 | |
j = len(splittable) | |
while i < j: | |
# Invariants: | |
# 1. splittable[:k] fits for all k <= i (note that we *assume*, | |
# at the start, that splittable[:0] fits). | |
# 2. splittable[:k] does not fit for any k > j (at the start, | |
# this means we shouldn't look at any k > len(splittable)). | |
# 3. We don't know about splittable[:k] for k in i+1..j. | |
# 4. We want to set i to the largest k that fits, with i <= k <= j. | |
# | |
m = (i+j+1) >> 1 # ceiling((i+j)/2); i < m <= j | |
chunk = charset.from_splittable(splittable[:m], True) | |
chunklen = charset.encoded_header_len(chunk) | |
if chunklen <= maxlinelen: | |
# m is acceptable, so is a new lower bound. | |
i = m | |
else: | |
# m is not acceptable, so final i must be < m. | |
j = m - 1 | |
# i == j. Invariant #1 implies that splittable[:i] fits, and | |
# invariant #2 implies that splittable[:i+1] does not fit, so i | |
# is what we're looking for. | |
first = charset.from_splittable(splittable[:i], False) | |
last = charset.from_splittable(splittable[i:], False) | |
return first, last |