mirror of
https://github.com/python/cpython.git
synced 2025-11-03 19:34:08 +00:00
bef9d21d14
line splitting examples don't split things the way they used to -- or should. In these cases, change the test case and add an XXX. The final failure was in Charset.body_encode() with euc-jp charset. These return the original string unencoded, which isn't right. XXX and comment this out for now; we'll fix it after a1.
578 lines
24 KiB
Python
578 lines
24 KiB
Python
# Copyright (C) 2002-2007 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 = ' '
|
||
BSPACE = b' '
|
||
SPACE8 = ' ' * 8
|
||
EMPTYSTRING = ''
|
||
MAXLINELEN = 78
|
||
|
||
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]+:$')
|
||
|
||
|
||
|
||
# Helpers
|
||
_max_append = email.quoprimime._max_append
|
||
|
||
|
||
|
||
def decode_header(header):
|
||
"""Decode a message header value without converting charset.
|
||
|
||
Returns a list of (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 with no charset.
|
||
if not ecre.search(header):
|
||
return [(header, None)]
|
||
# First step is to parse all the encoded parts into triplets of the form
|
||
# (encoded_string, encoding, charset). For unencoded strings, the last
|
||
# two parts will be None.
|
||
words = []
|
||
for line in header.splitlines():
|
||
parts = ecre.split(line)
|
||
while parts:
|
||
unencoded = parts.pop(0).strip()
|
||
if unencoded:
|
||
words.append((unencoded, None, None))
|
||
if parts:
|
||
charset = parts.pop(0).lower()
|
||
encoding = parts.pop(0).lower()
|
||
encoded = parts.pop(0)
|
||
words.append((encoded, encoding, charset))
|
||
# The next step is to decode each encoded word by applying the reverse
|
||
# base64 or quopri transformation. decoded_words is now a list of the
|
||
# form (decoded_word, charset).
|
||
decoded_words = []
|
||
for encoded_string, encoding, charset in words:
|
||
if encoding is None:
|
||
# This is an unencoded word.
|
||
decoded_words.append((encoded_string, charset))
|
||
elif encoding == 'q':
|
||
word = email.quoprimime.header_decode(encoded_string)
|
||
decoded_words.append((word, charset))
|
||
elif encoding == 'b':
|
||
try:
|
||
word = email.base64mime.decode(encoded_string)
|
||
except binascii.Error:
|
||
raise HeaderParseError('Base64 decoding error')
|
||
else:
|
||
decoded_words.append((word, charset))
|
||
else:
|
||
raise AssertionError('Unexpected encoding: ' + encoding)
|
||
# Now convert all words to bytes and collapse consecutive runs of
|
||
# similarly encoded words.
|
||
collapsed = []
|
||
last_word = last_charset = None
|
||
for word, charset in decoded_words:
|
||
if isinstance(word, str):
|
||
word = bytes(word, 'raw-unicode-escape')
|
||
if last_word is None:
|
||
last_word = word
|
||
last_charset = charset
|
||
elif charset != last_charset:
|
||
collapsed.append((last_word, last_charset))
|
||
last_word = word
|
||
last_charset = charset
|
||
elif last_charset is None:
|
||
last_word += BSPACE + word
|
||
else:
|
||
last_word += word
|
||
collapsed.append((last_word, last_charset))
|
||
return collapsed
|
||
|
||
|
||
|
||
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 78 as recommended
|
||
by RFC 2822.
|
||
|
||
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
|
||
elif not isinstance(charset, Charset):
|
||
charset = Charset(charset)
|
||
self._charset = charset
|
||
self._continuation_ws = continuation_ws
|
||
self._chunks = []
|
||
if s is not None:
|
||
self.append(s, charset, errors)
|
||
if maxlinelen is None:
|
||
maxlinelen = MAXLINELEN
|
||
self._maxlinelen = maxlinelen
|
||
if header_name is None:
|
||
self._headerlen = 0
|
||
else:
|
||
# Take the separating colon and space into account.
|
||
self._headerlen = len(header_name) + 2
|
||
|
||
def __str__(self):
|
||
"""Return the string value of the header."""
|
||
self._normalize()
|
||
uchunks = []
|
||
lastcs = None
|
||
for string, 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(SPACE)
|
||
nextcs = None
|
||
elif nextcs not in (None, 'us-ascii'):
|
||
uchunks.append(SPACE)
|
||
lastcs = nextcs
|
||
uchunks.append(string)
|
||
return EMPTYSTRING.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 unicode (of the unencoded header value), swap the
|
||
# args and do another comparison.
|
||
return other == str(self)
|
||
|
||
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 isinstance(s, str):
|
||
# Convert the string from the input character set to the output
|
||
# character set and store the resulting bytes and the charset for
|
||
# composition later.
|
||
input_charset = charset.input_codec or 'us-ascii'
|
||
input_bytes = s.encode(input_charset, errors)
|
||
else:
|
||
# We already have the bytes we will store internally.
|
||
input_bytes = s
|
||
# Ensure that the bytes we're storing can be decoded to the output
|
||
# character set, otherwise an early error is thrown.
|
||
output_charset = charset.output_codec or 'us-ascii'
|
||
output_string = input_bytes.decode(output_charset, errors)
|
||
self._chunks.append((output_string, charset))
|
||
|
||
def encode(self, splitchars=';, \t', maxlinelen=None):
|
||
"""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.
|
||
"""
|
||
self._normalize()
|
||
if maxlinelen is None:
|
||
maxlinelen = self._maxlinelen
|
||
# A maxlinelen of 0 means don't wrap. For all practical purposes,
|
||
# choosing a huge number here accomplishes that and makes the
|
||
# _ValueFormatter algorithm much simpler.
|
||
if maxlinelen == 0:
|
||
maxlinelen = 1000000
|
||
formatter = _ValueFormatter(self._headerlen, maxlinelen,
|
||
self._continuation_ws, splitchars)
|
||
for string, charset in self._chunks:
|
||
lines = string.splitlines()
|
||
for line in lines:
|
||
formatter.feed(line, charset)
|
||
if len(lines) > 1:
|
||
formatter.newline()
|
||
formatter.add_transition()
|
||
return str(formatter)
|
||
|
||
def _normalize(self):
|
||
# Step 1: Normalize the chunks so that all runs of identical charsets
|
||
# get collapsed into a single unicode string.
|
||
chunks = []
|
||
last_charset = None
|
||
last_chunk = []
|
||
for string, charset in self._chunks:
|
||
if charset == last_charset:
|
||
last_chunk.append(string)
|
||
else:
|
||
if last_charset is not None:
|
||
chunks.append((SPACE.join(last_chunk), last_charset))
|
||
last_chunk = [string]
|
||
last_charset = charset
|
||
if last_chunk:
|
||
chunks.append((SPACE.join(last_chunk), last_charset))
|
||
self._chunks = chunks
|
||
|
||
|
||
|
||
class _ValueFormatter:
|
||
def __init__(self, headerlen, maxlen, continuation_ws, splitchars):
|
||
self._maxlen = maxlen
|
||
self._continuation_ws = continuation_ws
|
||
self._continuation_ws_len = len(continuation_ws.replace('\t', SPACE8))
|
||
self._splitchars = splitchars
|
||
self._lines = []
|
||
self._current_line = _Accumulator(headerlen)
|
||
|
||
def __str__(self):
|
||
self.newline()
|
||
return NL.join(self._lines)
|
||
|
||
def newline(self):
|
||
end_of_line = self._current_line.pop()
|
||
if end_of_line is not None:
|
||
self._current_line.push(end_of_line)
|
||
if len(self._current_line) > 0:
|
||
self._lines.append(str(self._current_line))
|
||
self._current_line.reset()
|
||
|
||
def add_transition(self):
|
||
self._current_line.push(None)
|
||
|
||
def feed(self, string, charset):
|
||
# If the string itself fits on the current line in its encoded format,
|
||
# then add it now and be done with it.
|
||
encoded_string = charset.header_encode(string)
|
||
if len(encoded_string) + len(self._current_line) <= self._maxlen:
|
||
self._current_line.push(encoded_string)
|
||
return
|
||
# If the charset has no header encoding (i.e. it is an ASCII encoding)
|
||
# then we must split the header 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. Eventually, this should be pluggable.
|
||
if charset.header_encoding is None:
|
||
for ch in self._splitchars:
|
||
if ch in string:
|
||
break
|
||
else:
|
||
ch = None
|
||
# If there's no available split character then regardless of
|
||
# whether the string fits on the line, we have to put it on a line
|
||
# by itself.
|
||
if ch is None:
|
||
if not self._current_line.is_onlyws():
|
||
self._lines.append(str(self._current_line))
|
||
self._current_line.reset(self._continuation_ws)
|
||
self._current_line.push(encoded_string)
|
||
else:
|
||
self._ascii_split(string, ch)
|
||
return
|
||
# Otherwise, we're doing either a Base64 or a quoted-printable
|
||
# encoding which means we don't need to split the line on syntactic
|
||
# breaks. We can basically just find enough characters to fit on the
|
||
# current line, minus the RFC 2047 chrome. What makes this trickier
|
||
# though is that we have to split at octet boundaries, not character
|
||
# boundaries but it's only safe to split at character boundaries so at
|
||
# best we can only get close.
|
||
encoded_lines = charset.header_encode_lines(string, self._maxlengths())
|
||
# The first element extends the current line, but if it's None then
|
||
# nothing more fit on the current line so start a new line.
|
||
try:
|
||
first_line = encoded_lines.pop(0)
|
||
except IndexError:
|
||
# There are no encoded lines, so we're done.
|
||
return
|
||
if first_line is not None:
|
||
self._current_line.push(first_line)
|
||
self._lines.append(str(self._current_line))
|
||
self._current_line.reset(self._continuation_ws)
|
||
try:
|
||
last_line = encoded_lines.pop()
|
||
except IndexError:
|
||
# There was only one line.
|
||
return
|
||
self._current_line.push(last_line)
|
||
# Everything else are full lines in themselves.
|
||
for line in encoded_lines:
|
||
self._lines.append(self._continuation_ws + line)
|
||
|
||
def _maxlengths(self):
|
||
# The first line's length.
|
||
yield self._maxlen - len(self._current_line)
|
||
while True:
|
||
yield self._maxlen - self._continuation_ws_len
|
||
|
||
def _ascii_split(self, string, ch):
|
||
holding = _Accumulator()
|
||
# Split the line on the split character, preserving it. If the split
|
||
# character is whitespace RFC 2822 $2.2.3 requires us to fold on the
|
||
# whitespace, so that the line leads with the original whitespace we
|
||
# split on. However, if a higher syntactic break is used instead
|
||
# (e.g. comma or semicolon), the folding should happen after the split
|
||
# character. But then in that case, we need to add our own
|
||
# continuation whitespace -- although won't that break unfolding?
|
||
for part, splitpart, nextpart in _spliterator(ch, string):
|
||
if not splitpart:
|
||
# No splitpart means this is the last chunk. Put this part
|
||
# either on the current line or the next line depending on
|
||
# whether it fits.
|
||
holding.push(part)
|
||
if len(holding) + len(self._current_line) <= self._maxlen:
|
||
# It fits, but we're done.
|
||
self._current_line.push(str(holding))
|
||
else:
|
||
# It doesn't fit, but we're done. Before pushing a new
|
||
# line, watch out for the current line containing only
|
||
# whitespace.
|
||
holding.pop()
|
||
if self._current_line.is_onlyws() and holding.is_onlyws():
|
||
# Don't start a new line.
|
||
holding.push(part)
|
||
part = None
|
||
self._current_line.push(str(holding))
|
||
self._lines.append(str(self._current_line))
|
||
if part is None:
|
||
self._current_line.reset()
|
||
else:
|
||
holding.reset(part)
|
||
self._current_line.reset(str(holding))
|
||
return
|
||
elif not nextpart:
|
||
# There must be some trailing split characters because we
|
||
# found a split character but no next part. In this case we
|
||
# must treat the thing to fit as the part + splitpart because
|
||
# if splitpart is whitespace it's not allowed to be the only
|
||
# thing on the line, and if it's not whitespace we must split
|
||
# after the syntactic break. In either case, we're done.
|
||
holding_prelen = len(holding)
|
||
holding.push(part + splitpart)
|
||
if len(holding) + len(self._current_line) <= self._maxlen:
|
||
self._current_line.push(str(holding))
|
||
elif holding_prelen == 0:
|
||
# This is the only chunk left so it has to go on the
|
||
# current line.
|
||
self._current_line.push(str(holding))
|
||
else:
|
||
save_part = holding.pop()
|
||
self._current_line.push(str(holding))
|
||
self._lines.append(str(self._current_line))
|
||
holding.reset(save_part)
|
||
self._current_line.reset(str(holding))
|
||
return
|
||
elif not part:
|
||
# We're leading with a split character. See if the splitpart
|
||
# and nextpart fits on the current line.
|
||
holding.push(splitpart + nextpart)
|
||
holding_len = len(holding)
|
||
# We know we're not leaving the nextpart on the stack.
|
||
holding.pop()
|
||
if holding_len + len(self._current_line) <= self._maxlen:
|
||
holding.push(splitpart)
|
||
else:
|
||
# It doesn't fit. Since there's no current part really
|
||
# the best we can do is start a new line and push the
|
||
# split part onto it.
|
||
self._current_line.push(str(holding))
|
||
holding.reset()
|
||
if len(self._current_line) > 0 and self._lines:
|
||
self._lines.append(str(self._current_line))
|
||
self._current_line.reset()
|
||
holding.push(splitpart)
|
||
else:
|
||
# All three parts are present. First let's see if all three
|
||
# parts will fit on the current line. If so, we don't need to
|
||
# split it.
|
||
holding.push(part + splitpart + nextpart)
|
||
holding_len = len(holding)
|
||
# Pop the part because we'll push nextpart on the next
|
||
# iteration through the loop.
|
||
holding.pop()
|
||
if holding_len + len(self._current_line) <= self._maxlen:
|
||
holding.push(part + splitpart)
|
||
else:
|
||
# The entire thing doesn't fit. See if we need to split
|
||
# before or after the split characters.
|
||
if splitpart.isspace():
|
||
# Split before whitespace. Remember that the
|
||
# whitespace becomes the continuation whitespace of
|
||
# the next line so it goes to current_line not holding.
|
||
holding.push(part)
|
||
self._current_line.push(str(holding))
|
||
holding.reset()
|
||
self._lines.append(str(self._current_line))
|
||
self._current_line.reset(splitpart)
|
||
else:
|
||
# Split after non-whitespace. The continuation
|
||
# whitespace comes from the instance variable.
|
||
holding.push(part + splitpart)
|
||
self._current_line.push(str(holding))
|
||
holding.reset()
|
||
self._lines.append(str(self._current_line))
|
||
if nextpart[0].isspace():
|
||
self._current_line.reset()
|
||
else:
|
||
self._current_line.reset(self._continuation_ws)
|
||
# Get the last of the holding part
|
||
self._current_line.push(str(holding))
|
||
|
||
|
||
|
||
def _spliterator(character, string):
|
||
parts = list(reversed(re.split('(%s)' % character, string)))
|
||
while parts:
|
||
part = parts.pop()
|
||
splitparts = (parts.pop() if parts else None)
|
||
nextpart = (parts.pop() if parts else None)
|
||
yield (part, splitparts, nextpart)
|
||
if nextpart is not None:
|
||
parts.append(nextpart)
|
||
|
||
|
||
class _Accumulator:
|
||
def __init__(self, initial_size=0):
|
||
self._initial_size = initial_size
|
||
self._current = []
|
||
|
||
def push(self, string):
|
||
self._current.append(string)
|
||
|
||
def pop(self):
|
||
if not self._current:
|
||
return None
|
||
return self._current.pop()
|
||
|
||
def __len__(self):
|
||
return sum(((1 if string is None else len(string))
|
||
for string in self._current),
|
||
self._initial_size)
|
||
|
||
def __str__(self):
|
||
if self._current and self._current[-1] is None:
|
||
self._current.pop()
|
||
return EMPTYSTRING.join((' ' if string is None else string)
|
||
for string in self._current)
|
||
|
||
def reset(self, string=None):
|
||
self._current = []
|
||
self._initial_size = 0
|
||
if string is not None:
|
||
self.push(string)
|
||
|
||
def is_onlyws(self):
|
||
return len(self) == 0 or str(self).isspace()
|