diff --git a/Lib/email/__init__.py b/Lib/email/__init__.py index fae872439..9fa477830 100644 --- a/Lib/email/__init__.py +++ b/Lib/email/__init__.py @@ -25,7 +25,6 @@ __all__ = [ ] - # Some convenience routines. Don't import Parser and Message as side-effects # of importing email since those cascadingly import most of the rest of the # email package. diff --git a/Lib/email/_encoded_words.py b/Lib/email/_encoded_words.py index 5eaab36ed..6795a606d 100644 --- a/Lib/email/_encoded_words.py +++ b/Lib/email/_encoded_words.py @@ -62,7 +62,7 @@ __all__ = ['decode_q', # regex based decoder. _q_byte_subber = functools.partial(re.compile(br'=([a-fA-F0-9]{2})').sub, - lambda m: bytes([int(m.group(1), 16)])) + lambda m: bytes.fromhex(m.group(1).decode())) def decode_q(encoded): encoded = encoded.replace(b'_', b' ') @@ -98,30 +98,42 @@ def len_q(bstring): # def decode_b(encoded): - defects = [] + # First try encoding with validate=True, fixing the padding if needed. + # This will succeed only if encoded includes no invalid characters. pad_err = len(encoded) % 4 - if pad_err: - defects.append(errors.InvalidBase64PaddingDefect()) - padded_encoded = encoded + b'==='[:4-pad_err] - else: - padded_encoded = encoded + missing_padding = b'==='[:4-pad_err] if pad_err else b'' try: - return base64.b64decode(padded_encoded, validate=True), defects + return ( + base64.b64decode(encoded + missing_padding, validate=True), + [errors.InvalidBase64PaddingDefect()] if pad_err else [], + ) except binascii.Error: - # Since we had correct padding, this must an invalid char error. - defects = [errors.InvalidBase64CharactersDefect()] + # Since we had correct padding, this is likely an invalid char error. + # # The non-alphabet characters are ignored as far as padding - # goes, but we don't know how many there are. So we'll just - # try various padding lengths until something works. - for i in 0, 1, 2, 3: + # goes, but we don't know how many there are. So try without adding + # padding to see if it works. + try: + return ( + base64.b64decode(encoded, validate=False), + [errors.InvalidBase64CharactersDefect()], + ) + except binascii.Error: + # Add as much padding as could possibly be necessary (extra padding + # is ignored). try: - return base64.b64decode(encoded+b'='*i, validate=False), defects + return ( + base64.b64decode(encoded + b'==', validate=False), + [errors.InvalidBase64CharactersDefect(), + errors.InvalidBase64PaddingDefect()], + ) except binascii.Error: - if i==0: - defects.append(errors.InvalidBase64PaddingDefect()) - else: - # This should never happen. - raise AssertionError("unexpected binascii.Error") + # This only happens when the encoded string's length is 1 more + # than a multiple of 4, which is invalid. + # + # bpo-27397: Just return the encoded string since there's no + # way to decode. + return encoded, [errors.InvalidBase64LengthDefect()] def encode_b(bstring): return base64.b64encode(bstring).decode('ascii') @@ -167,15 +179,15 @@ def decode(ew): # Turn the CTE decoded bytes into unicode. try: string = bstring.decode(charset) - except UnicodeError: + except UnicodeDecodeError: defects.append(errors.UndecodableBytesDefect("Encoded word " - "contains bytes not decodable using {} charset".format(charset))) + f"contains bytes not decodable using {charset!r} charset")) string = bstring.decode(charset, 'surrogateescape') - except LookupError: + except (LookupError, UnicodeEncodeError): string = bstring.decode('ascii', 'surrogateescape') if charset.lower() != 'unknown-8bit': - defects.append(errors.CharsetError("Unknown charset {} " - "in encoded word; decoded as unknown bytes".format(charset))) + defects.append(errors.CharsetError(f"Unknown charset {charset!r} " + f"in encoded word; decoded as unknown bytes")) return string, charset, lang, defects diff --git a/Lib/email/_header_value_parser.py b/Lib/email/_header_value_parser.py index 57d01fbcb..ec2215a5e 100644 --- a/Lib/email/_header_value_parser.py +++ b/Lib/email/_header_value_parser.py @@ -68,9 +68,9 @@ XXX: provide complete list of token types. """ import re +import sys import urllib # For urllib.parse.unquote from string import hexdigits -from collections import OrderedDict from operator import itemgetter from email import _encoded_words as _ew from email import errors @@ -92,93 +92,23 @@ TOKEN_ENDS = TSPECIALS | WSP ASPECIALS = TSPECIALS | set("*'%") ATTRIBUTE_ENDS = ASPECIALS | WSP EXTENDED_ATTRIBUTE_ENDS = ATTRIBUTE_ENDS - set('%') +NLSET = {'\n', '\r'} +SPECIALSNL = SPECIALS | NLSET def quote_string(value): return '"'+str(value).replace('\\', '\\\\').replace('"', r'\"')+'"' -# -# Accumulator for header folding -# +# Match a RFC 2047 word, looks like =?utf-8?q?someword?= +rfc2047_matcher = re.compile(r''' + =\? # literal =? + [^?]* # charset + \? # literal ? + [qQbB] # literal 'q' or 'b', case insensitive + \? # literal ? + .*? # encoded word + \?= # literal ?= +''', re.VERBOSE | re.MULTILINE) -class _Folded: - - def __init__(self, maxlen, policy): - self.maxlen = maxlen - self.policy = policy - self.lastlen = 0 - self.stickyspace = None - self.firstline = True - self.done = [] - self.current = [] - - def newline(self): - self.done.extend(self.current) - self.done.append(self.policy.linesep) - self.current.clear() - self.lastlen = 0 - - def finalize(self): - if self.current: - self.newline() - - def __str__(self): - return ''.join(self.done) - - def append(self, stoken): - self.current.append(stoken) - - def append_if_fits(self, token, stoken=None): - if stoken is None: - stoken = str(token) - l = len(stoken) - if self.stickyspace is not None: - stickyspace_len = len(self.stickyspace) - if self.lastlen + stickyspace_len + l <= self.maxlen: - self.current.append(self.stickyspace) - self.lastlen += stickyspace_len - self.current.append(stoken) - self.lastlen += l - self.stickyspace = None - self.firstline = False - return True - if token.has_fws: - ws = token.pop_leading_fws() - if ws is not None: - self.stickyspace += str(ws) - stickyspace_len += len(ws) - token._fold(self) - return True - if stickyspace_len and l + 1 <= self.maxlen: - margin = self.maxlen - l - if 0 < margin < stickyspace_len: - trim = stickyspace_len - margin - self.current.append(self.stickyspace[:trim]) - self.stickyspace = self.stickyspace[trim:] - stickyspace_len = trim - self.newline() - self.current.append(self.stickyspace) - self.current.append(stoken) - self.lastlen = l + stickyspace_len - self.stickyspace = None - self.firstline = False - return True - if not self.firstline: - self.newline() - self.current.append(self.stickyspace) - self.current.append(stoken) - self.stickyspace = None - self.firstline = False - return True - if self.lastlen + l <= self.maxlen: - self.current.append(stoken) - self.lastlen += l - return True - if l < self.maxlen: - self.newline() - self.current.append(stoken) - self.lastlen = l - return True - return False # # TokenList and its subclasses @@ -187,6 +117,8 @@ class _Folded: class TokenList(list): token_type = None + syntactic_break = True + ew_combine_allowed = True def __init__(self, *args, **kw): super().__init__(*args, **kw) @@ -207,84 +139,13 @@ class TokenList(list): def all_defects(self): return sum((x.all_defects for x in self), self.defects) - # - # Folding API - # - # parts(): - # - # return a list of objects that constitute the "higher level syntactic - # objects" specified by the RFC as the best places to fold a header line. - # The returned objects must include leading folding white space, even if - # this means mutating the underlying parse tree of the object. Each object - # is only responsible for returning *its* parts, and should not drill down - # to any lower level except as required to meet the leading folding white - # space constraint. - # - # _fold(folded): - # - # folded: the result accumulator. This is an instance of _Folded. - # (XXX: I haven't finished factoring this out yet, the folding code - # pretty much uses this as a state object.) When the folded.current - # contains as much text as will fit, the _fold method should call - # folded.newline. - # folded.lastlen: the current length of the test stored in folded.current. - # folded.maxlen: The maximum number of characters that may appear on a - # folded line. Differs from the policy setting in that "no limit" is - # represented by +inf, which means it can be used in the trivially - # logical fashion in comparisons. - # - # Currently no subclasses implement parts, and I think this will remain - # true. A subclass only needs to implement _fold when the generic version - # isn't sufficient. _fold will need to be implemented primarily when it is - # possible for encoded words to appear in the specialized token-list, since - # there is no generic algorithm that can know where exactly the encoded - # words are allowed. A _fold implementation is responsible for filling - # lines in the same general way that the top level _fold does. It may, and - # should, call the _fold method of sub-objects in a similar fashion to that - # of the top level _fold. - # - # XXX: I'm hoping it will be possible to factor the existing code further - # to reduce redundancy and make the logic clearer. - - @property - def parts(self): - klass = self.__class__ - this = [] - for token in self: - if token.startswith_fws(): - if this: - yield this[0] if len(this)==1 else klass(this) - this.clear() - end_ws = token.pop_trailing_ws() - this.append(token) - if end_ws: - yield klass(this) - this = [end_ws] - if this: - yield this[0] if len(this)==1 else klass(this) - def startswith_fws(self): return self[0].startswith_fws() - def pop_leading_fws(self): - if self[0].token_type == 'fws': - return self.pop(0) - return self[0].pop_leading_fws() - - def pop_trailing_ws(self): - if self[-1].token_type == 'cfws': - return self.pop(-1) - return self[-1].pop_trailing_ws() - @property - def has_fws(self): - for part in self: - if part.has_fws: - return True - return False - - def has_leading_comment(self): - return self[0].has_leading_comment() + def as_ew_allowed(self): + """True if all top level tokens of this part may be RFC2047 encoded.""" + return all(part.as_ew_allowed for part in self) @property def comments(self): @@ -294,71 +155,13 @@ class TokenList(list): return comments def fold(self, *, policy): - # max_line_length 0/None means no limit, ie: infinitely long. - maxlen = policy.max_line_length or float("+inf") - folded = _Folded(maxlen, policy) - self._fold(folded) - folded.finalize() - return str(folded) - - def as_encoded_word(self, charset): - # This works only for things returned by 'parts', which include - # the leading fws, if any, that should be used. - res = [] - ws = self.pop_leading_fws() - if ws: - res.append(ws) - trailer = self.pop(-1) if self[-1].token_type=='fws' else '' - res.append(_ew.encode(str(self), charset)) - res.append(trailer) - return ''.join(res) - - def cte_encode(self, charset, policy): - res = [] - for part in self: - res.append(part.cte_encode(charset, policy)) - return ''.join(res) - - def _fold(self, folded): - encoding = 'utf-8' if folded.policy.utf8 else 'ascii' - for part in self.parts: - tstr = str(part) - tlen = len(tstr) - try: - str(part).encode(encoding) - except UnicodeEncodeError: - if any(isinstance(x, errors.UndecodableBytesDefect) - for x in part.all_defects): - charset = 'unknown-8bit' - else: - # XXX: this should be a policy setting when utf8 is False. - charset = 'utf-8' - tstr = part.cte_encode(charset, folded.policy) - tlen = len(tstr) - if folded.append_if_fits(part, tstr): - continue - # Peel off the leading whitespace if any and make it sticky, to - # avoid infinite recursion. - ws = part.pop_leading_fws() - if ws is not None: - # Peel off the leading whitespace and make it sticky, to - # avoid infinite recursion. - folded.stickyspace = str(part.pop(0)) - if folded.append_if_fits(part): - continue - if part.has_fws: - part._fold(folded) - continue - # There are no fold points in this one; it is too long for a single - # line and can't be split...we just have to put it on its own line. - folded.append(tstr) - folded.newline() + return _refold_parse_tree(self, policy=policy) def pprint(self, indent=''): - print('\n'.join(self._pp(indent=''))) + print(self.ppstr(indent=indent)) def ppstr(self, indent=''): - return '\n'.join(self._pp(indent='')) + return '\n'.join(self._pp(indent=indent)) def _pp(self, indent=''): yield '{}{}/{}('.format( @@ -390,213 +193,35 @@ class WhiteSpaceTokenList(TokenList): class UnstructuredTokenList(TokenList): - token_type = 'unstructured' - def _fold(self, folded): - last_ew = None - encoding = 'utf-8' if folded.policy.utf8 else 'ascii' - for part in self.parts: - tstr = str(part) - is_ew = False - try: - str(part).encode(encoding) - except UnicodeEncodeError: - if any(isinstance(x, errors.UndecodableBytesDefect) - for x in part.all_defects): - charset = 'unknown-8bit' - else: - charset = 'utf-8' - if last_ew is not None: - # We've already done an EW, combine this one with it - # if there's room. - chunk = get_unstructured( - ''.join(folded.current[last_ew:]+[tstr])).as_encoded_word(charset) - oldlastlen = sum(len(x) for x in folded.current[:last_ew]) - schunk = str(chunk) - lchunk = len(schunk) - if oldlastlen + lchunk <= folded.maxlen: - del folded.current[last_ew:] - folded.append(schunk) - folded.lastlen = oldlastlen + lchunk - continue - tstr = part.as_encoded_word(charset) - is_ew = True - if folded.append_if_fits(part, tstr): - if is_ew: - last_ew = len(folded.current) - 1 - continue - if is_ew or last_ew: - # It's too big to fit on the line, but since we've - # got encoded words we can use encoded word folding. - part._fold_as_ew(folded) - continue - # Peel off the leading whitespace if any and make it sticky, to - # avoid infinite recursion. - ws = part.pop_leading_fws() - if ws is not None: - folded.stickyspace = str(ws) - if folded.append_if_fits(part): - continue - if part.has_fws: - part._fold(folded) - continue - # It can't be split...we just have to put it on its own line. - folded.append(tstr) - folded.newline() - last_ew = None - - def cte_encode(self, charset, policy): - res = [] - last_ew = None - for part in self: - spart = str(part) - try: - spart.encode('us-ascii') - res.append(spart) - except UnicodeEncodeError: - if last_ew is None: - res.append(part.cte_encode(charset, policy)) - last_ew = len(res) - else: - tl = get_unstructured(''.join(res[last_ew:] + [spart])) - res.append(tl.as_encoded_word(charset)) - return ''.join(res) - class Phrase(TokenList): - token_type = 'phrase' - def _fold(self, folded): - # As with Unstructured, we can have pure ASCII with or without - # surrogateescape encoded bytes, or we could have unicode. But this - # case is more complicated, since we have to deal with the various - # sub-token types and how they can be composed in the face of - # unicode-that-needs-CTE-encoding, and the fact that if a token a - # comment that becomes a barrier across which we can't compose encoded - # words. - last_ew = None - encoding = 'utf-8' if folded.policy.utf8 else 'ascii' - for part in self.parts: - tstr = str(part) - tlen = len(tstr) - has_ew = False - try: - str(part).encode(encoding) - except UnicodeEncodeError: - if any(isinstance(x, errors.UndecodableBytesDefect) - for x in part.all_defects): - charset = 'unknown-8bit' - else: - charset = 'utf-8' - if last_ew is not None and not part.has_leading_comment(): - # We've already done an EW, let's see if we can combine - # this one with it. The last_ew logic ensures that all we - # have at this point is atoms, no comments or quoted - # strings. So we can treat the text between the last - # encoded word and the content of this token as - # unstructured text, and things will work correctly. But - # we have to strip off any trailing comment on this token - # first, and if it is a quoted string we have to pull out - # the content (we're encoding it, so it no longer needs to - # be quoted). - if part[-1].token_type == 'cfws' and part.comments: - remainder = part.pop(-1) - else: - remainder = '' - for i, token in enumerate(part): - if token.token_type == 'bare-quoted-string': - part[i] = UnstructuredTokenList(token[:]) - chunk = get_unstructured( - ''.join(folded.current[last_ew:]+[tstr])).as_encoded_word(charset) - schunk = str(chunk) - lchunk = len(schunk) - if last_ew + lchunk <= folded.maxlen: - del folded.current[last_ew:] - folded.append(schunk) - folded.lastlen = sum(len(x) for x in folded.current) - continue - tstr = part.as_encoded_word(charset) - tlen = len(tstr) - has_ew = True - if folded.append_if_fits(part, tstr): - if has_ew and not part.comments: - last_ew = len(folded.current) - 1 - elif part.comments or part.token_type == 'quoted-string': - # If a comment is involved we can't combine EWs. And if a - # quoted string is involved, it's not worth the effort to - # try to combine them. - last_ew = None - continue - part._fold(folded) - - def cte_encode(self, charset, policy): - res = [] - last_ew = None - is_ew = False - for part in self: - spart = str(part) - try: - spart.encode('us-ascii') - res.append(spart) - except UnicodeEncodeError: - is_ew = True - if last_ew is None: - if not part.comments: - last_ew = len(res) - res.append(part.cte_encode(charset, policy)) - elif not part.has_leading_comment(): - if part[-1].token_type == 'cfws' and part.comments: - remainder = part.pop(-1) - else: - remainder = '' - for i, token in enumerate(part): - if token.token_type == 'bare-quoted-string': - part[i] = UnstructuredTokenList(token[:]) - tl = get_unstructured(''.join(res[last_ew:] + [spart])) - res[last_ew:] = [tl.as_encoded_word(charset)] - if part.comments or (not is_ew and part.token_type == 'quoted-string'): - last_ew = None - return ''.join(res) - class Word(TokenList): - token_type = 'word' class CFWSList(WhiteSpaceTokenList): - token_type = 'cfws' - def has_leading_comment(self): - return bool(self.comments) - class Atom(TokenList): - token_type = 'atom' class Token(TokenList): - token_type = 'token' + encode_as_ew = False class EncodedWord(TokenList): - token_type = 'encoded-word' cte = None charset = None lang = None - @property - def encoded(self): - if self.cte is not None: - return self.cte - _ew.encode(str(self), self.charset) - - class QuotedString(TokenList): @@ -812,7 +437,10 @@ class AngleAddr(TokenList): def addr_spec(self): for x in self: if x.token_type == 'addr-spec': - return x.addr_spec + if x.local_part: + return x.addr_spec + else: + return quote_string(x.local_part) + x.addr_spec else: return '<>' @@ -867,6 +495,7 @@ class InvalidMailbox(TokenList): class Domain(TokenList): token_type = 'domain' + as_ew_allowed = False @property def domain(self): @@ -874,18 +503,23 @@ class Domain(TokenList): class DotAtom(TokenList): - token_type = 'dot-atom' class DotAtomText(TokenList): - token_type = 'dot-atom-text' + as_ew_allowed = True + + +class NoFoldLiteral(TokenList): + token_type = 'no-fold-literal' + as_ew_allowed = False class AddrSpec(TokenList): token_type = 'addr-spec' + as_ew_allowed = False @property def local_part(self): @@ -918,24 +552,30 @@ class AddrSpec(TokenList): class ObsLocalPart(TokenList): token_type = 'obs-local-part' + as_ew_allowed = False class DisplayName(Phrase): token_type = 'display-name' + ew_combine_allowed = False @property def display_name(self): res = TokenList(self) + if len(res) == 0: + return res.value if res[0].token_type == 'cfws': res.pop(0) else: - if res[0][0].token_type == 'cfws': + if (isinstance(res[0], TokenList) and + res[0][0].token_type == 'cfws'): res[0] = TokenList(res[0][1:]) if res[-1].token_type == 'cfws': res.pop() else: - if res[-1][-1].token_type == 'cfws': + if (isinstance(res[-1], TokenList) and + res[-1][-1].token_type == 'cfws'): res[-1] = TokenList(res[-1][:-1]) return res.value @@ -948,11 +588,15 @@ class DisplayName(Phrase): for x in self: if x.token_type == 'quoted-string': quote = True - if quote: + if len(self) != 0 and quote: pre = post = '' - if self[0].token_type=='cfws' or self[0][0].token_type=='cfws': + if (self[0].token_type == 'cfws' or + isinstance(self[0], TokenList) and + self[0][0].token_type == 'cfws'): pre = ' ' - if self[-1].token_type=='cfws' or self[-1][-1].token_type=='cfws': + if (self[-1].token_type == 'cfws' or + isinstance(self[-1], TokenList) and + self[-1][-1].token_type == 'cfws'): post = ' ' return pre+quote_string(self.display_name)+post else: @@ -962,6 +606,7 @@ class DisplayName(Phrase): class LocalPart(TokenList): token_type = 'local-part' + as_ew_allowed = False @property def value(self): @@ -997,6 +642,7 @@ class LocalPart(TokenList): class DomainLiteral(TokenList): token_type = 'domain-literal' + as_ew_allowed = False @property def domain(self): @@ -1083,6 +729,7 @@ class Value(TokenList): class MimeParameters(TokenList): token_type = 'mime-parameters' + syntactic_break = False @property def params(self): @@ -1091,7 +738,7 @@ class MimeParameters(TokenList): # to assume the RFC 2231 pieces can come in any order. However, we # output them in the order that we first see a given name, which gives # us a stable __str__. - params = OrderedDict() + params = {} # Using order preserving dict from Python 3.7+ for token in self: if not token.token_type.endswith('parameter'): continue @@ -1142,7 +789,7 @@ class MimeParameters(TokenList): else: try: value = value.decode(charset, 'surrogateescape') - except LookupError: + except (LookupError, UnicodeEncodeError): # XXX: there should really be a custom defect for # unknown character set to make it easy to find, # because otherwise unknown charset is a silent @@ -1167,6 +814,10 @@ class MimeParameters(TokenList): class ParameterizedHeaderValue(TokenList): + # Set this false so that the value doesn't wind up on a new line even + # if it and the parameters would fit there but not on the first line. + syntactic_break = False + @property def params(self): for token in reversed(self): @@ -1174,59 +825,51 @@ class ParameterizedHeaderValue(TokenList): return token.params return {} - @property - def parts(self): - if self and self[-1].token_type == 'mime-parameters': - # We don't want to start a new line if all of the params don't fit - # after the value, so unwrap the parameter list. - return TokenList(self[:-1] + self[-1]) - return TokenList(self).parts - class ContentType(ParameterizedHeaderValue): - token_type = 'content-type' + as_ew_allowed = False maintype = 'text' subtype = 'plain' class ContentDisposition(ParameterizedHeaderValue): - token_type = 'content-disposition' + as_ew_allowed = False content_disposition = None class ContentTransferEncoding(TokenList): - token_type = 'content-transfer-encoding' + as_ew_allowed = False cte = '7bit' class HeaderLabel(TokenList): - token_type = 'header-label' + as_ew_allowed = False + + +class MsgID(TokenList): + token_type = 'msg-id' + as_ew_allowed = False + + def fold(self, policy): + # message-id tokens may not be folded. + return str(self) + policy.linesep + + +class MessageID(MsgID): + token_type = 'message-id' + + +class InvalidMessageID(MessageID): + token_type = 'invalid-message-id' class Header(TokenList): - token_type = 'header' - def _fold(self, folded): - folded.append(str(self.pop(0))) - folded.lastlen = len(folded.current[0]) - # The first line of the header is different from all others: we don't - # want to start a new object on a new line if it has any fold points in - # it that would allow part of it to be on the first header line. - # Further, if the first fold point would fit on the new line, we want - # to do that, but if it doesn't we want to put it on the first line. - # Folded supports this via the stickyspace attribute. If this - # attribute is not None, it does the special handling. - folded.stickyspace = str(self.pop(0)) if self[0].token_type == 'cfws' else '' - rest = self.pop(0) - if self: - raise ValueError("Malformed Header token list") - rest._fold(folded) - # # Terminal classes and instances @@ -1234,6 +877,10 @@ class Header(TokenList): class Terminal(str): + as_ew_allowed = True + ew_combine_allowed = True + syntactic_break = True + def __new__(cls, value, token_type): self = super().__new__(cls, value) self.token_type = token_type @@ -1243,6 +890,9 @@ class Terminal(str): def __repr__(self): return "{}({})".format(self.__class__.__name__, super().__repr__()) + def pprint(self): + print(self.__class__.__name__ + '/' + self.token_type) + @property def all_defects(self): return list(self.defects) @@ -1256,29 +906,14 @@ class Terminal(str): '' if not self.defects else ' {}'.format(self.defects), )] - def cte_encode(self, charset, policy): - value = str(self) - try: - value.encode('us-ascii') - return value - except UnicodeEncodeError: - return _ew.encode(value, charset) - def pop_trailing_ws(self): # This terminates the recursion. return None - def pop_leading_fws(self): - # This terminates the recursion. - return None - @property def comments(self): return [] - def has_leading_comment(self): - return False - def __getnewargs__(self): return(str(self), self.token_type) @@ -1292,8 +927,6 @@ class WhiteSpaceTerminal(Terminal): def startswith_fws(self): return True - has_fws = True - class ValueTerminal(Terminal): @@ -1304,11 +937,6 @@ class ValueTerminal(Terminal): def startswith_fws(self): return False - has_fws = False - - def as_encoded_word(self, charset): - return _ew.encode(str(self), charset) - class EWWhiteSpaceTerminal(WhiteSpaceTerminal): @@ -1316,14 +944,12 @@ class EWWhiteSpaceTerminal(WhiteSpaceTerminal): def value(self): return '' - @property - def encoded(self): - return self[:] - def __str__(self): return '' - has_fws = True + +class _InvalidEwError(errors.HeaderParseError): + """Invalid encoded word found while parsing headers.""" # XXX these need to become classes and used as instances so @@ -1331,6 +957,8 @@ class EWWhiteSpaceTerminal(WhiteSpaceTerminal): # up other parse trees. Maybe should have tests for that, too. DOT = ValueTerminal('.', 'dot') ListSeparator = ValueTerminal(',', 'list-separator') +ListSeparator.as_ew_allowed = False +ListSeparator.syntactic_break = False RouteComponentMarker = ValueTerminal('@', 'route-component-marker') # @@ -1356,15 +984,14 @@ RouteComponentMarker = ValueTerminal('@', 'route-component-marker') _wsp_splitter = re.compile(r'([{}]+)'.format(''.join(WSP))).split _non_atom_end_matcher = re.compile(r"[^{}]+".format( - ''.join(ATOM_ENDS).replace('\\','\\\\').replace(']',r'\]'))).match + re.escape(''.join(ATOM_ENDS)))).match _non_printable_finder = re.compile(r"[\x00-\x20\x7F]").findall _non_token_end_matcher = re.compile(r"[^{}]+".format( - ''.join(TOKEN_ENDS).replace('\\','\\\\').replace(']',r'\]'))).match + re.escape(''.join(TOKEN_ENDS)))).match _non_attribute_end_matcher = re.compile(r"[^{}]+".format( - ''.join(ATTRIBUTE_ENDS).replace('\\','\\\\').replace(']',r'\]'))).match + re.escape(''.join(ATTRIBUTE_ENDS)))).match _non_extended_attribute_end_matcher = re.compile(r"[^{}]+".format( - ''.join(EXTENDED_ATTRIBUTE_ENDS).replace( - '\\','\\\\').replace(']',r'\]'))).match + re.escape(''.join(EXTENDED_ATTRIBUTE_ENDS)))).match def _validate_xtext(xtext): """If input token contains ASCII non-printables, register a defect.""" @@ -1431,7 +1058,10 @@ def get_encoded_word(value): raise errors.HeaderParseError( "expected encoded word but found {}".format(value)) remstr = ''.join(remainder) - if len(remstr) > 1 and remstr[0] in hexdigits and remstr[1] in hexdigits: + if (len(remstr) > 1 and + remstr[0] in hexdigits and + remstr[1] in hexdigits and + tok.count('?') < 2): # The ? after the CTE was followed by an encoded word escape (=XX). rest, *remainder = remstr.split('?=', 1) tok = tok + '?=' + rest @@ -1442,8 +1072,8 @@ def get_encoded_word(value): value = ''.join(remainder) try: text, charset, lang, defects = _ew.decode('=?' + tok + '?=') - except ValueError: - raise errors.HeaderParseError( + except (ValueError, KeyError): + raise _InvalidEwError( "encoded word format invalid: '{}'".format(ew.cte)) ew.charset = charset ew.lang = lang @@ -1458,6 +1088,10 @@ def get_encoded_word(value): _validate_xtext(vtext) ew.append(vtext) text = ''.join(remainder) + # Encoded words should be followed by a WS + if value and value[0] not in WSP: + ew.defects.append(errors.InvalidHeaderDefect( + "missing trailing whitespace after encoded-word")) return ew, value def get_unstructured(value): @@ -1489,9 +1123,12 @@ def get_unstructured(value): token, value = get_fws(value) unstructured.append(token) continue + valid_ew = True if value.startswith('=?'): try: token, value = get_encoded_word(value) + except _InvalidEwError: + valid_ew = False except errors.HeaderParseError: # XXX: Need to figure out how to register defects when # appropriate here. @@ -1510,6 +1147,14 @@ def get_unstructured(value): unstructured.append(token) continue tok, *remainder = _wsp_splitter(value, 1) + # Split in the middle of an atom if there is a rfc2047 encoded word + # which does not have WSP on both sides. The defect will be registered + # the next time through the loop. + # This needs to only be performed when the encoded word is valid; + # otherwise, performing it on an invalid encoded word can cause + # the parser to go in an infinite loop. + if valid_ew and rfc2047_matcher.search(tok): + tok, *remainder = value.partition('=?') vtext = ValueTerminal(tok, 'vtext') _validate_xtext(vtext) unstructured.append(vtext) @@ -1571,21 +1216,33 @@ def get_bare_quoted_string(value): value is the text between the quote marks, with whitespace preserved and quoted pairs decoded. """ - if value[0] != '"': + if not value or value[0] != '"': raise errors.HeaderParseError( "expected '\"' but found '{}'".format(value)) bare_quoted_string = BareQuotedString() value = value[1:] + if value and value[0] == '"': + token, value = get_qcontent(value) + bare_quoted_string.append(token) while value and value[0] != '"': if value[0] in WSP: token, value = get_fws(value) elif value[:2] == '=?': + valid_ew = False try: token, value = get_encoded_word(value) bare_quoted_string.defects.append(errors.InvalidHeaderDefect( "encoded word inside quoted string")) + valid_ew = True except errors.HeaderParseError: token, value = get_qcontent(value) + # Collapse the whitespace between two encoded words that occur in a + # bare-quoted-string. + if valid_ew and len(bare_quoted_string) > 1: + if (bare_quoted_string[-1].token_type == 'fws' and + bare_quoted_string[-2].token_type == 'encoded-word'): + bare_quoted_string[-1] = EWWhiteSpaceTerminal( + bare_quoted_string[-1], 'fws') else: token, value = get_qcontent(value) bare_quoted_string.append(token) @@ -1742,6 +1399,9 @@ def get_word(value): leader, value = get_cfws(value) else: leader = None + if not value: + raise errors.HeaderParseError( + "Expected 'atom' or 'quoted-string' but found nothing.") if value[0]=='"': token, value = get_quoted_string(value) elif value[0] in SPECIALS: @@ -1797,7 +1457,7 @@ def get_local_part(value): """ local_part = LocalPart() leader = None - if value[0] in CFWS_LEADER: + if value and value[0] in CFWS_LEADER: leader, value = get_cfws(value) if not value: raise errors.HeaderParseError( @@ -1863,13 +1523,18 @@ def get_obs_local_part(value): raise token, value = get_cfws(value) obs_local_part.append(token) + if not obs_local_part: + raise errors.HeaderParseError( + "expected obs-local-part but found '{}'".format(value)) if (obs_local_part[0].token_type == 'dot' or obs_local_part[0].token_type=='cfws' and + len(obs_local_part) > 1 and obs_local_part[1].token_type=='dot'): obs_local_part.defects.append(errors.InvalidHeaderDefect( "Invalid leading '.' in local part")) if (obs_local_part[-1].token_type == 'dot' or obs_local_part[-1].token_type=='cfws' and + len(obs_local_part) > 1 and obs_local_part[-2].token_type=='dot'): obs_local_part.defects.append(errors.InvalidHeaderDefect( "Invalid trailing '.' in local part")) @@ -1951,7 +1616,7 @@ def get_domain(value): """ domain = Domain() leader = None - if value[0] in CFWS_LEADER: + if value and value[0] in CFWS_LEADER: leader, value = get_cfws(value) if not value: raise errors.HeaderParseError( @@ -1966,6 +1631,8 @@ def get_domain(value): token, value = get_dot_atom(value) except errors.HeaderParseError: token, value = get_atom(value) + if value and value[0] == '@': + raise errors.HeaderParseError('Invalid Domain') if leader is not None: token[:0] = [leader] domain.append(token) @@ -1989,7 +1656,7 @@ def get_addr_spec(value): addr_spec.append(token) if not value or value[0] != '@': addr_spec.defects.append(errors.InvalidHeaderDefect( - "add-spec local part with no domain")) + "addr-spec local part with no domain")) return addr_spec, value addr_spec.append(ValueTerminal('@', 'address-at-symbol')) token, value = get_domain(value[1:]) @@ -2025,6 +1692,8 @@ def get_obs_route(value): if value[0] in CFWS_LEADER: token, value = get_cfws(value) obs_route.append(token) + if not value: + break if value[0] == '@': obs_route.append(RouteComponentMarker) token, value = get_domain(value[1:]) @@ -2043,7 +1712,7 @@ def get_angle_addr(value): """ angle_addr = AngleAddr() - if value[0] in CFWS_LEADER: + if value and value[0] in CFWS_LEADER: token, value = get_cfws(value) angle_addr.append(token) if not value or value[0] != '<': @@ -2053,7 +1722,7 @@ def get_angle_addr(value): value = value[1:] # Although it is not legal per RFC5322, SMTP uses '<>' in certain # circumstances. - if value[0] == '>': + if value and value[0] == '>': angle_addr.append(ValueTerminal('>', 'angle-addr-end')) angle_addr.defects.append(errors.InvalidHeaderDefect( "null addr-spec in angle-addr")) @@ -2105,6 +1774,9 @@ def get_name_addr(value): name_addr = NameAddr() # Both the optional display name and the angle-addr can start with cfws. leader = None + if not value: + raise errors.HeaderParseError( + "expected name-addr but found '{}'".format(value)) if value[0] in CFWS_LEADER: leader, value = get_cfws(value) if not value: @@ -2119,7 +1791,10 @@ def get_name_addr(value): raise errors.HeaderParseError( "expected name-addr but found '{}'".format(token)) if leader is not None: - token[0][:0] = [leader] + if isinstance(token[0], TokenList): + token[0][:0] = [leader] + else: + token[:0] = [leader] leader = None name_addr.append(token) token, value = get_angle_addr(value) @@ -2281,7 +1956,7 @@ def get_group(value): if not value: group.defects.append(errors.InvalidHeaderDefect( "end of header in group")) - if value[0] != ';': + elif value[0] != ';': raise errors.HeaderParseError( "expected ';' at end of group but found {}".format(value)) group.append(ValueTerminal(';', 'group-terminator')) @@ -2335,7 +2010,7 @@ def get_address_list(value): try: token, value = get_address(value) address_list.append(token) - except errors.HeaderParseError as err: + except errors.HeaderParseError: leader = None if value[0] in CFWS_LEADER: leader, value = get_cfws(value) @@ -2370,10 +2045,122 @@ def get_address_list(value): address_list.defects.append(errors.InvalidHeaderDefect( "invalid address in address-list")) if value: # Must be a , at this point. - address_list.append(ValueTerminal(',', 'list-separator')) + address_list.append(ListSeparator) value = value[1:] return address_list, value + +def get_no_fold_literal(value): + """ no-fold-literal = "[" *dtext "]" + """ + no_fold_literal = NoFoldLiteral() + if not value: + raise errors.HeaderParseError( + "expected no-fold-literal but found '{}'".format(value)) + if value[0] != '[': + raise errors.HeaderParseError( + "expected '[' at the start of no-fold-literal " + "but found '{}'".format(value)) + no_fold_literal.append(ValueTerminal('[', 'no-fold-literal-start')) + value = value[1:] + token, value = get_dtext(value) + no_fold_literal.append(token) + if not value or value[0] != ']': + raise errors.HeaderParseError( + "expected ']' at the end of no-fold-literal " + "but found '{}'".format(value)) + no_fold_literal.append(ValueTerminal(']', 'no-fold-literal-end')) + return no_fold_literal, value[1:] + +def get_msg_id(value): + """msg-id = [CFWS] "<" id-left '@' id-right ">" [CFWS] + id-left = dot-atom-text / obs-id-left + id-right = dot-atom-text / no-fold-literal / obs-id-right + no-fold-literal = "[" *dtext "]" + """ + msg_id = MsgID() + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + msg_id.append(token) + if not value or value[0] != '<': + raise errors.HeaderParseError( + "expected msg-id but found '{}'".format(value)) + msg_id.append(ValueTerminal('<', 'msg-id-start')) + value = value[1:] + # Parse id-left. + try: + token, value = get_dot_atom_text(value) + except errors.HeaderParseError: + try: + # obs-id-left is same as local-part of add-spec. + token, value = get_obs_local_part(value) + msg_id.defects.append(errors.ObsoleteHeaderDefect( + "obsolete id-left in msg-id")) + except errors.HeaderParseError: + raise errors.HeaderParseError( + "expected dot-atom-text or obs-id-left" + " but found '{}'".format(value)) + msg_id.append(token) + if not value or value[0] != '@': + msg_id.defects.append(errors.InvalidHeaderDefect( + "msg-id with no id-right")) + # Even though there is no id-right, if the local part + # ends with `>` let's just parse it too and return + # along with the defect. + if value and value[0] == '>': + msg_id.append(ValueTerminal('>', 'msg-id-end')) + value = value[1:] + return msg_id, value + msg_id.append(ValueTerminal('@', 'address-at-symbol')) + value = value[1:] + # Parse id-right. + try: + token, value = get_dot_atom_text(value) + except errors.HeaderParseError: + try: + token, value = get_no_fold_literal(value) + except errors.HeaderParseError: + try: + token, value = get_domain(value) + msg_id.defects.append(errors.ObsoleteHeaderDefect( + "obsolete id-right in msg-id")) + except errors.HeaderParseError: + raise errors.HeaderParseError( + "expected dot-atom-text, no-fold-literal or obs-id-right" + " but found '{}'".format(value)) + msg_id.append(token) + if value and value[0] == '>': + value = value[1:] + else: + msg_id.defects.append(errors.InvalidHeaderDefect( + "missing trailing '>' on msg-id")) + msg_id.append(ValueTerminal('>', 'msg-id-end')) + if value and value[0] in CFWS_LEADER: + token, value = get_cfws(value) + msg_id.append(token) + return msg_id, value + + +def parse_message_id(value): + """message-id = "Message-ID:" msg-id CRLF + """ + message_id = MessageID() + try: + token, value = get_msg_id(value) + message_id.append(token) + except errors.HeaderParseError as ex: + token = get_unstructured(value) + message_id = InvalidMessageID(token) + message_id.defects.append( + errors.InvalidHeaderDefect("Invalid msg-id: {!r}".format(ex))) + else: + # Value after parsing a valid msg_id should be None. + if value: + message_id.defects.append(errors.InvalidHeaderDefect( + "Unexpected {!r}".format(value))) + + return message_id + # # XXX: As I begin to add additional header parsers, I'm realizing we probably # have two level of parser routines: the get_XXX methods that get a token in @@ -2615,8 +2402,8 @@ def get_section(value): digits += value[0] value = value[1:] if digits[0] == '0' and digits != '0': - section.defects.append(errors.InvalidHeaderError("section number" - "has an invalid leading 0")) + section.defects.append(errors.InvalidHeaderDefect( + "section number has an invalid leading 0")) section.number = int(digits) section.append(ValueTerminal(digits, 'digits')) return section, value @@ -2679,7 +2466,6 @@ def get_parameter(value): raise errors.HeaderParseError("Parameter not followed by '='") param.append(ValueTerminal('=', 'parameter-separator')) value = value[1:] - leader = None if value and value[0] in CFWS_LEADER: token, value = get_cfws(value) param.append(token) @@ -2754,7 +2540,7 @@ def get_parameter(value): if value[0] != "'": raise errors.HeaderParseError("Expected RFC2231 char/lang encoding " "delimiter, but found {!r}".format(value)) - appendto.append(ValueTerminal("'", 'RFC2231 delimiter')) + appendto.append(ValueTerminal("'", 'RFC2231-delimiter')) value = value[1:] if value and value[0] != "'": token, value = get_attrtext(value) @@ -2763,7 +2549,7 @@ def get_parameter(value): if not value or value[0] != "'": raise errors.HeaderParseError("Expected RFC2231 char/lang encoding " "delimiter, but found {}".format(value)) - appendto.append(ValueTerminal("'", 'RFC2231 delimiter')) + appendto.append(ValueTerminal("'", 'RFC2231-delimiter')) value = value[1:] if remainder is not None: # Treat the rest of value as bare quoted string content. @@ -2771,6 +2557,9 @@ def get_parameter(value): while value: if value[0] in WSP: token, value = get_fws(value) + elif value[0] == '"': + token = ValueTerminal('"', 'DQUOTE') + value = value[1:] else: token, value = get_qcontent(value) v.append(token) @@ -2791,7 +2580,7 @@ def parse_mime_parameters(value): the formal RFC grammar, but it is more convenient for us for the set of parameters to be treated as its own TokenList. - This is 'parse' routine because it consumes the reminaing value, but it + This is 'parse' routine because it consumes the remaining value, but it would never be called to parse a full header. Instead it is called to parse everything after the non-parameter value of a specific MIME header. @@ -2801,7 +2590,7 @@ def parse_mime_parameters(value): try: token, value = get_parameter(value) mime_parameters.append(token) - except errors.HeaderParseError as err: + except errors.HeaderParseError: leader = None if value[0] in CFWS_LEADER: leader, value = get_cfws(value) @@ -2859,7 +2648,6 @@ def parse_content_type_header(value): don't do that. """ ctype = ContentType() - recover = False if not value: ctype.defects.append(errors.HeaderMissingRequiredValue( "Missing content type specification")) @@ -2968,3 +2756,323 @@ def parse_content_transfer_encoding_header(value): token, value = get_phrase(value) cte_header.append(token) return cte_header + + +# +# Header folding +# +# Header folding is complex, with lots of rules and corner cases. The +# following code does its best to obey the rules and handle the corner +# cases, but you can be sure there are few bugs:) +# +# This folder generally canonicalizes as it goes, preferring the stringified +# version of each token. The tokens contain information that supports the +# folder, including which tokens can be encoded in which ways. +# +# Folded text is accumulated in a simple list of strings ('lines'), each +# one of which should be less than policy.max_line_length ('maxlen'). +# + +def _steal_trailing_WSP_if_exists(lines): + wsp = '' + if lines and lines[-1] and lines[-1][-1] in WSP: + wsp = lines[-1][-1] + lines[-1] = lines[-1][:-1] + return wsp + +def _refold_parse_tree(parse_tree, *, policy): + """Return string of contents of parse_tree folded according to RFC rules. + + """ + # max_line_length 0/None means no limit, ie: infinitely long. + maxlen = policy.max_line_length or sys.maxsize + encoding = 'utf-8' if policy.utf8 else 'us-ascii' + lines = [''] # Folded lines to be output + leading_whitespace = '' # When we have whitespace between two encoded + # words, we may need to encode the whitespace + # at the beginning of the second word. + last_ew = None # Points to the last encoded character if there's an ew on + # the line + last_charset = None + wrap_as_ew_blocked = 0 + want_encoding = False # This is set to True if we need to encode this part + end_ew_not_allowed = Terminal('', 'wrap_as_ew_blocked') + parts = list(parse_tree) + while parts: + part = parts.pop(0) + if part is end_ew_not_allowed: + wrap_as_ew_blocked -= 1 + continue + tstr = str(part) + if not want_encoding: + if part.token_type == 'ptext': + # Encode if tstr contains special characters. + want_encoding = not SPECIALSNL.isdisjoint(tstr) + else: + # Encode if tstr contains newlines. + want_encoding = not NLSET.isdisjoint(tstr) + try: + tstr.encode(encoding) + charset = encoding + except UnicodeEncodeError: + if any(isinstance(x, errors.UndecodableBytesDefect) + for x in part.all_defects): + charset = 'unknown-8bit' + else: + # If policy.utf8 is false this should really be taken from a + # 'charset' property on the policy. + charset = 'utf-8' + want_encoding = True + + if part.token_type == 'mime-parameters': + # Mime parameter folding (using RFC2231) is extra special. + _fold_mime_parameters(part, lines, maxlen, encoding) + continue + + if want_encoding and not wrap_as_ew_blocked: + if not part.as_ew_allowed: + want_encoding = False + last_ew = None + if part.syntactic_break: + encoded_part = part.fold(policy=policy)[:-len(policy.linesep)] + if policy.linesep not in encoded_part: + # It fits on a single line + if len(encoded_part) > maxlen - len(lines[-1]): + # But not on this one, so start a new one. + newline = _steal_trailing_WSP_if_exists(lines) + # XXX what if encoded_part has no leading FWS? + lines.append(newline) + lines[-1] += encoded_part + continue + # Either this is not a major syntactic break, so we don't + # want it on a line by itself even if it fits, or it + # doesn't fit on a line by itself. Either way, fall through + # to unpacking the subparts and wrapping them. + if not hasattr(part, 'encode'): + # It's not a Terminal, do each piece individually. + parts = list(part) + parts + want_encoding = False + continue + elif part.as_ew_allowed: + # It's a terminal, wrap it as an encoded word, possibly + # combining it with previously encoded words if allowed. + if (last_ew is not None and + charset != last_charset and + (last_charset == 'unknown-8bit' or + last_charset == 'utf-8' and charset != 'us-ascii')): + last_ew = None + last_ew = _fold_as_ew(tstr, lines, maxlen, last_ew, + part.ew_combine_allowed, charset, leading_whitespace) + # This whitespace has been added to the lines in _fold_as_ew() + # so clear it now. + leading_whitespace = '' + last_charset = charset + want_encoding = False + continue + else: + # It's a terminal which should be kept non-encoded + # (e.g. a ListSeparator). + last_ew = None + want_encoding = False + # fall through + + if len(tstr) <= maxlen - len(lines[-1]): + lines[-1] += tstr + continue + + # This part is too long to fit. The RFC wants us to break at + # "major syntactic breaks", so unless we don't consider this + # to be one, check if it will fit on the next line by itself. + leading_whitespace = '' + if (part.syntactic_break and + len(tstr) + 1 <= maxlen): + newline = _steal_trailing_WSP_if_exists(lines) + if newline or part.startswith_fws(): + # We're going to fold the data onto a new line here. Due to + # the way encoded strings handle continuation lines, we need to + # be prepared to encode any whitespace if the next line turns + # out to start with an encoded word. + lines.append(newline + tstr) + + whitespace_accumulator = [] + for char in lines[-1]: + if char not in WSP: + break + whitespace_accumulator.append(char) + leading_whitespace = ''.join(whitespace_accumulator) + last_ew = None + continue + if not hasattr(part, 'encode'): + # It's not a terminal, try folding the subparts. + newparts = list(part) + if not part.as_ew_allowed: + wrap_as_ew_blocked += 1 + newparts.append(end_ew_not_allowed) + parts = newparts + parts + continue + if part.as_ew_allowed and not wrap_as_ew_blocked: + # It doesn't need CTE encoding, but encode it anyway so we can + # wrap it. + parts.insert(0, part) + want_encoding = True + continue + # We can't figure out how to wrap, it, so give up. + newline = _steal_trailing_WSP_if_exists(lines) + if newline or part.startswith_fws(): + lines.append(newline + tstr) + else: + # We can't fold it onto the next line either... + lines[-1] += tstr + + return policy.linesep.join(lines) + policy.linesep + +def _fold_as_ew(to_encode, lines, maxlen, last_ew, ew_combine_allowed, charset, leading_whitespace): + """Fold string to_encode into lines as encoded word, combining if allowed. + Return the new value for last_ew, or None if ew_combine_allowed is False. + + If there is already an encoded word in the last line of lines (indicated by + a non-None value for last_ew) and ew_combine_allowed is true, decode the + existing ew, combine it with to_encode, and re-encode. Otherwise, encode + to_encode. In either case, split to_encode as necessary so that the + encoded segments fit within maxlen. + + """ + if last_ew is not None and ew_combine_allowed: + to_encode = str( + get_unstructured(lines[-1][last_ew:] + to_encode)) + lines[-1] = lines[-1][:last_ew] + elif to_encode[0] in WSP: + # We're joining this to non-encoded text, so don't encode + # the leading blank. + leading_wsp = to_encode[0] + to_encode = to_encode[1:] + if (len(lines[-1]) == maxlen): + lines.append(_steal_trailing_WSP_if_exists(lines)) + lines[-1] += leading_wsp + + trailing_wsp = '' + if to_encode[-1] in WSP: + # Likewise for the trailing space. + trailing_wsp = to_encode[-1] + to_encode = to_encode[:-1] + new_last_ew = len(lines[-1]) if last_ew is None else last_ew + + encode_as = 'utf-8' if charset == 'us-ascii' else charset + + # The RFC2047 chrome takes up 7 characters plus the length + # of the charset name. + chrome_len = len(encode_as) + 7 + + if (chrome_len + 1) >= maxlen: + raise errors.HeaderParseError( + "max_line_length is too small to fit an encoded word") + + while to_encode: + remaining_space = maxlen - len(lines[-1]) + text_space = remaining_space - chrome_len - len(leading_whitespace) + if text_space <= 0: + lines.append(' ') + continue + + # If we are at the start of a continuation line, prepend whitespace + # (we only want to do this when the line starts with an encoded word + # but if we're folding in this helper function, then we know that we + # are going to be writing out an encoded word.) + if len(lines) > 1 and len(lines[-1]) == 1 and leading_whitespace: + encoded_word = _ew.encode(leading_whitespace, charset=encode_as) + lines[-1] += encoded_word + leading_whitespace = '' + + to_encode_word = to_encode[:text_space] + encoded_word = _ew.encode(to_encode_word, charset=encode_as) + excess = len(encoded_word) - remaining_space + while excess > 0: + # Since the chunk to encode is guaranteed to fit into less than 100 characters, + # shrinking it by one at a time shouldn't take long. + to_encode_word = to_encode_word[:-1] + encoded_word = _ew.encode(to_encode_word, charset=encode_as) + excess = len(encoded_word) - remaining_space + lines[-1] += encoded_word + to_encode = to_encode[len(to_encode_word):] + leading_whitespace = '' + + if to_encode: + lines.append(' ') + new_last_ew = len(lines[-1]) + lines[-1] += trailing_wsp + return new_last_ew if ew_combine_allowed else None + +def _fold_mime_parameters(part, lines, maxlen, encoding): + """Fold TokenList 'part' into the 'lines' list as mime parameters. + + Using the decoded list of parameters and values, format them according to + the RFC rules, including using RFC2231 encoding if the value cannot be + expressed in 'encoding' and/or the parameter+value is too long to fit + within 'maxlen'. + + """ + # Special case for RFC2231 encoding: start from decoded values and use + # RFC2231 encoding iff needed. + # + # Note that the 1 and 2s being added to the length calculations are + # accounting for the possibly-needed spaces and semicolons we'll be adding. + # + for name, value in part.params: + # XXX What if this ';' puts us over maxlen the first time through the + # loop? We should split the header value onto a newline in that case, + # but to do that we need to recognize the need earlier or reparse the + # header, so I'm going to ignore that bug for now. It'll only put us + # one character over. + if not lines[-1].rstrip().endswith(';'): + lines[-1] += ';' + charset = encoding + error_handler = 'strict' + try: + value.encode(encoding) + encoding_required = False + except UnicodeEncodeError: + encoding_required = True + if utils._has_surrogates(value): + charset = 'unknown-8bit' + error_handler = 'surrogateescape' + else: + charset = 'utf-8' + if encoding_required: + encoded_value = urllib.parse.quote( + value, safe='', errors=error_handler) + tstr = "{}*={}''{}".format(name, charset, encoded_value) + else: + tstr = '{}={}'.format(name, quote_string(value)) + if len(lines[-1]) + len(tstr) + 1 < maxlen: + lines[-1] = lines[-1] + ' ' + tstr + continue + elif len(tstr) + 2 <= maxlen: + lines.append(' ' + tstr) + continue + # We need multiple sections. We are allowed to mix encoded and + # non-encoded sections, but we aren't going to. We'll encode them all. + section = 0 + extra_chrome = charset + "''" + while value: + chrome_len = len(name) + len(str(section)) + 3 + len(extra_chrome) + if maxlen <= chrome_len + 3: + # We need room for the leading blank, the trailing semicolon, + # and at least one character of the value. If we don't + # have that, we'd be stuck, so in that case fall back to + # the RFC standard width. + maxlen = 78 + splitpoint = maxchars = maxlen - chrome_len - 2 + while True: + partial = value[:splitpoint] + encoded_value = urllib.parse.quote( + partial, safe='', errors=error_handler) + if len(encoded_value) <= maxchars: + break + splitpoint -= 1 + lines.append(" {}*{}*={}{}".format( + name, section, extra_chrome, encoded_value)) + extra_chrome = '' + section += 1 + value = value[splitpoint:] + if value: + lines[-1] += ';' diff --git a/Lib/email/_parseaddr.py b/Lib/email/_parseaddr.py index cdfa3729a..0f1bf8e42 100644 --- a/Lib/email/_parseaddr.py +++ b/Lib/email/_parseaddr.py @@ -13,7 +13,7 @@ __all__ = [ 'quote', ] -import time, calendar +import time SPACE = ' ' EMPTYSTRING = '' @@ -65,8 +65,10 @@ def _parsedate_tz(data): """ if not data: - return + return None data = data.split() + if not data: # This happens for whitespace-only input. + return None # The FWS after the comma after the day-of-week is optional, so search and # adjust for this. if data[0].endswith(',') or data[0].lower() in _daynames: @@ -93,6 +95,8 @@ def _parsedate_tz(data): return None data = data[:5] [dd, mm, yy, tm, tz] = data + if not (dd and mm and yy): + return None mm = mm.lower() if mm not in _monthnames: dd, mm = mm, dd.lower() @@ -108,6 +112,8 @@ def _parsedate_tz(data): yy, tm = tm, yy if yy[-1] == ',': yy = yy[:-1] + if not yy: + return None if not yy[0].isdigit(): yy, tz = tz, yy if tm[-1] == ',': @@ -126,6 +132,8 @@ def _parsedate_tz(data): tss = 0 elif len(tm) == 3: [thh, tmm, tss] = tm + else: + return None else: return None try: @@ -186,6 +194,9 @@ def mktime_tz(data): # No zone info, so localtime is better assumption than GMT return time.mktime(data[:8] + (-1,)) else: + # Delay the import, since mktime_tz is rarely used + import calendar + t = calendar.timegm(data) return t - data[9] @@ -379,7 +390,12 @@ class AddrlistClass: aslist.append('@') self.pos += 1 self.gotonext() - return EMPTYSTRING.join(aslist) + self.getdomain() + domain = self.getdomain() + if not domain: + # Invalid domain, return an empty address instead of returning a + # local part to denote failed parsing. + return EMPTYSTRING + return EMPTYSTRING.join(aslist) + domain def getdomain(self): """Get the complete domain name from an address.""" @@ -394,6 +410,10 @@ class AddrlistClass: elif self.field[self.pos] == '.': self.pos += 1 sdlist.append('.') + elif self.field[self.pos] == '@': + # bpo-34155: Don't parse domains with two `@` like + # `a@malicious.org@important.com`. + return EMPTYSTRING elif self.field[self.pos] in self.atomends: break else: diff --git a/Lib/email/_policybase.py b/Lib/email/_policybase.py index df4649676..c9f0d7430 100644 --- a/Lib/email/_policybase.py +++ b/Lib/email/_policybase.py @@ -152,11 +152,18 @@ class Policy(_PolicyBase, metaclass=abc.ABCMeta): mangle_from_ -- a flag that, when True escapes From_ lines in the body of the message by putting a `>' in front of them. This is used when the message is being - serialized by a generator. Default: True. + serialized by a generator. Default: False. message_factory -- the class to use to create new message objects. If the value is None, the default is Message. + verify_generated_headers + -- if true, the generator verifies that each header + they are properly folded, so that a parser won't + treat it as multiple headers, start-of-body, or + part of another header. + This is a check against custom Header & fold() + implementations. """ raise_on_defect = False @@ -165,6 +172,7 @@ class Policy(_PolicyBase, metaclass=abc.ABCMeta): max_line_length = 78 mangle_from_ = False message_factory = None + verify_generated_headers = True def handle_defect(self, obj, defect): """Based on policy, either raise defect or call register_defect. @@ -294,12 +302,12 @@ class Compat32(Policy): """+ The name is parsed as everything up to the ':' and returned unmodified. The value is determined by stripping leading whitespace off the - remainder of the first line, joining all subsequent lines together, and + remainder of the first line joined with all subsequent lines, and stripping any trailing carriage return or linefeed characters. """ name, value = sourcelines[0].split(':', 1) - value = value.lstrip(' \t') + ''.join(sourcelines[1:]) + value = ''.join((value, *sourcelines[1:])).lstrip(' \t\r\n') return (name, value.rstrip('\r\n')) def header_store_parse(self, name, value): @@ -361,8 +369,12 @@ class Compat32(Policy): # Assume it is a Header-like object. h = value if h is not None: - parts.append(h.encode(linesep=self.linesep, - maxlinelen=self.max_line_length)) + # The Header class interprets a value of None for maxlinelen as the + # default value of 78, as recommended by RFC 2822. + maxlinelen = 0 + if self.max_line_length is not None: + maxlinelen = self.max_line_length + parts.append(h.encode(linesep=self.linesep, maxlinelen=maxlinelen)) parts.append(self.linesep) return ''.join(parts) diff --git a/Lib/email/architecture.rst b/Lib/email/architecture.rst index 78572ae63..fcd10bde1 100644 --- a/Lib/email/architecture.rst +++ b/Lib/email/architecture.rst @@ -66,7 +66,7 @@ data payloads. Message Lifecycle ----------------- -The general lifecyle of a message is: +The general lifecycle of a message is: Creation A `Message` object can be created by a Parser, or it can be diff --git a/Lib/email/base64mime.py b/Lib/email/base64mime.py index 17f0818f6..4cdf22666 100644 --- a/Lib/email/base64mime.py +++ b/Lib/email/base64mime.py @@ -45,7 +45,6 @@ EMPTYSTRING = '' MISC_LEN = 7 - # Helpers def header_length(bytearray): """Return the length of s when it is encoded with base64.""" @@ -57,7 +56,6 @@ def header_length(bytearray): return n - def header_encode(header_bytes, charset='iso-8859-1'): """Encode a single header line with Base64 encoding in a given charset. @@ -72,7 +70,6 @@ def header_encode(header_bytes, charset='iso-8859-1'): return '=?%s?b?%s?=' % (charset, encoded) - def body_encode(s, maxlinelen=76, eol=NL): r"""Encode a string with base64. @@ -84,7 +81,7 @@ def body_encode(s, maxlinelen=76, eol=NL): in an email. """ if not s: - return s + return "" encvec = [] max_unencoded = maxlinelen * 3 // 4 @@ -98,7 +95,6 @@ def body_encode(s, maxlinelen=76, eol=NL): return EMPTYSTRING.join(encvec) - def decode(string): """Decode a raw base64 string, returning a bytes object. diff --git a/Lib/email/charset.py b/Lib/email/charset.py index ee564040c..043801107 100644 --- a/Lib/email/charset.py +++ b/Lib/email/charset.py @@ -18,7 +18,6 @@ from email import errors from email.encoders import encode_7or8bit - # Flags for types of header encodings QP = 1 # Quoted-Printable BASE64 = 2 # Base64 @@ -32,7 +31,6 @@ UNKNOWN8BIT = 'unknown-8bit' EMPTYSTRING = '' - # Defaults CHARSETS = { # input header enc body enc output conv @@ -104,7 +102,6 @@ CODEC_MAP = { } - # Convenience functions for extending the above mappings def add_charset(charset, header_enc=None, body_enc=None, output_charset=None): """Add character set properties to the global registry. @@ -112,8 +109,8 @@ def add_charset(charset, header_enc=None, body_enc=None, output_charset=None): charset is the input character set, and must be the canonical name of a character set. - Optional header_enc and body_enc is either Charset.QP for - quoted-printable, Charset.BASE64 for base64 encoding, Charset.SHORTEST for + Optional header_enc and body_enc is either charset.QP for + quoted-printable, charset.BASE64 for base64 encoding, charset.SHORTEST for the shortest of qp or base64 encoding, or None for no encoding. SHORTEST is only valid for header_enc. It describes how message headers and message bodies in the input charset are to be encoded. Default is no @@ -153,7 +150,6 @@ def add_codec(charset, codecname): CODEC_MAP[charset] = codecname - # Convenience function for encoding strings, taking into account # that they might be unknown-8bit (ie: have surrogate-escaped bytes) def _encode(string, codec): @@ -163,7 +159,6 @@ def _encode(string, codec): return string.encode(codec) - class Charset: """Map character sets to their email properties. @@ -185,13 +180,13 @@ class Charset: header_encoding: If the character set must be encoded before it can be used in an email header, this attribute will be set to - Charset.QP (for quoted-printable), Charset.BASE64 (for - base64 encoding), or Charset.SHORTEST for the shortest of + charset.QP (for quoted-printable), charset.BASE64 (for + base64 encoding), or charset.SHORTEST for the shortest of QP or BASE64 encoding. Otherwise, it will be None. body_encoding: Same as header_encoding, but describes the encoding for the mail message's body, which indeed may be different than the - header encoding. Charset.SHORTEST is not allowed for + header encoding. charset.SHORTEST is not allowed for body_encoding. output_charset: Some character sets must be converted before they can be @@ -241,11 +236,9 @@ class Charset: self.output_codec = CODEC_MAP.get(self.output_charset, self.output_charset) - def __str__(self): + def __repr__(self): return self.input_charset.lower() - __repr__ = __str__ - def __eq__(self, other): return str(self) == str(other).lower() @@ -348,7 +341,6 @@ class Charset: if not lines and not current_line: lines.append(None) else: - separator = (' ' if lines else '') joined_line = EMPTYSTRING.join(current_line) header_bytes = _encode(joined_line, codec) lines.append(encoder(header_bytes)) diff --git a/Lib/email/contentmanager.py b/Lib/email/contentmanager.py index b904ded94..b4f5830be 100644 --- a/Lib/email/contentmanager.py +++ b/Lib/email/contentmanager.py @@ -72,12 +72,14 @@ def get_non_text_content(msg): return msg.get_payload(decode=True) for maintype in 'audio image video application'.split(): raw_data_manager.add_get_handler(maintype, get_non_text_content) +del maintype def get_message_content(msg): return msg.get_payload(0) for subtype in 'rfc822 external-body'.split(): raw_data_manager.add_get_handler('message/'+subtype, get_message_content) +del subtype def get_and_fixup_unknown_message_content(msg): @@ -144,15 +146,15 @@ def _encode_text(string, charset, cte, policy): linesep = policy.linesep.encode('ascii') def embedded_body(lines): return linesep.join(lines) + linesep def normal_body(lines): return b'\n'.join(lines) + b'\n' - if cte==None: + if cte is None: # Use heuristics to decide on the "best" encoding. - try: - return '7bit', normal_body(lines).decode('ascii') - except UnicodeDecodeError: - pass - if (policy.cte_type == '8bit' and - max(len(x) for x in lines) <= policy.max_line_length): - return '8bit', normal_body(lines).decode('ascii', 'surrogateescape') + if max((len(x) for x in lines), default=0) <= policy.max_line_length: + try: + return '7bit', normal_body(lines).decode('ascii') + except UnicodeDecodeError: + pass + if policy.cte_type == '8bit': + return '8bit', normal_body(lines).decode('ascii', 'surrogateescape') sniff = embedded_body(lines[:10]) sniff_qp = quoprimime.body_encode(sniff.decode('latin-1'), policy.max_line_length) @@ -238,9 +240,7 @@ def set_bytes_content(msg, data, maintype, subtype, cte='base64', data = binascii.b2a_qp(data, istext=False, header=False, quotetabs=True) data = data.decode('ascii') elif cte == '7bit': - # Make sure it really is only ASCII. The early warning here seems - # worth the overhead...if you care write your own content manager :). - data.encode('ascii') + data = data.decode('ascii') elif cte in ('8bit', 'binary'): data = data.decode('ascii', 'surrogateescape') msg.set_payload(data) @@ -248,3 +248,4 @@ def set_bytes_content(msg, data, maintype, subtype, cte='base64', _finalize_set(msg, disposition, filename, cid, params) for typ in (bytes, bytearray, memoryview): raw_data_manager.add_set_handler(typ, set_bytes_content) +del typ diff --git a/Lib/email/encoders.py b/Lib/email/encoders.py index 0a66acb62..17bd1ab7b 100644 --- a/Lib/email/encoders.py +++ b/Lib/email/encoders.py @@ -16,7 +16,6 @@ from base64 import encodebytes as _bencode from quopri import encodestring as _encodestring - def _qencode(s): enc = _encodestring(s, quotetabs=True) # Must encode spaces, which quopri.encodestring() doesn't do @@ -34,7 +33,6 @@ def encode_base64(msg): msg['Content-Transfer-Encoding'] = 'base64' - def encode_quopri(msg): """Encode the message's payload in quoted-printable. @@ -46,7 +44,6 @@ def encode_quopri(msg): msg['Content-Transfer-Encoding'] = 'quoted-printable' - def encode_7or8bit(msg): """Set the Content-Transfer-Encoding header to 7bit or 8bit.""" orig = msg.get_payload(decode=True) @@ -64,6 +61,5 @@ def encode_7or8bit(msg): msg['Content-Transfer-Encoding'] = '7bit' - def encode_noop(msg): """Do nothing.""" diff --git a/Lib/email/errors.py b/Lib/email/errors.py index 791239fa6..02aa5eced 100644 --- a/Lib/email/errors.py +++ b/Lib/email/errors.py @@ -29,6 +29,10 @@ class CharsetError(MessageError): """An illegal charset was given.""" +class HeaderWriteError(MessageError): + """Error while writing headers.""" + + # These are parsing defects which the parser was able to work around. class MessageDefect(ValueError): """Base class for a message defect.""" @@ -73,6 +77,9 @@ class InvalidBase64PaddingDefect(MessageDefect): class InvalidBase64CharactersDefect(MessageDefect): """base64 encoded sequence had characters not in base64 alphabet""" +class InvalidBase64LengthDefect(MessageDefect): + """base64 encoded sequence had invalid length (1 mod 4)""" + # These errors are specific to header parsing. class HeaderDefect(MessageDefect): @@ -105,3 +112,6 @@ class NonASCIILocalPartDefect(HeaderDefect): """local_part contains non-ASCII characters""" # This defect only occurs during unicode parsing, not when # parsing messages decoded from binary. + +class InvalidDateDefect(HeaderDefect): + """Header has unparsable or invalid date""" diff --git a/Lib/email/feedparser.py b/Lib/email/feedparser.py index 7c07ca864..06d6b4a3a 100644 --- a/Lib/email/feedparser.py +++ b/Lib/email/feedparser.py @@ -37,11 +37,12 @@ NLCRE_crack = re.compile(r'(\r\n|\r|\n)') headerRE = re.compile(r'^(From |[\041-\071\073-\176]*:|[\t ])') EMPTYSTRING = '' NL = '\n' +boundaryendRE = re.compile( + r'(?P--)?(?P[ \t]*)(?P\r\n|\r|\n)?$') NeedMoreData = object() - class BufferedSubFile(object): """A file-ish object that can have new data loaded into it. @@ -132,7 +133,6 @@ class BufferedSubFile(object): return line - class FeedParser: """A feed-style parser of email.""" @@ -189,7 +189,7 @@ class FeedParser: assert not self._msgstack # Look for final set of defects if root.get_content_maintype() == 'multipart' \ - and not root.is_multipart(): + and not root.is_multipart() and not self._headersonly: defect = errors.MultipartInvariantViolationDefect() self.policy.handle_defect(root, defect) return root @@ -266,7 +266,7 @@ class FeedParser: yield NeedMoreData continue break - msg = self._pop_message() + self._pop_message() # We need to pop the EOF matcher in order to tell if we're at # the end of the current file, not the end of the last block # of message headers. @@ -320,7 +320,7 @@ class FeedParser: self._cur.set_payload(EMPTYSTRING.join(lines)) return # Make sure a valid content type was specified per RFC 2045:6.4. - if (self._cur.get('content-transfer-encoding', '8bit').lower() + if (str(self._cur.get('content-transfer-encoding', '8bit')).lower() not in ('7bit', '8bit', 'binary')): defect = errors.InvalidMultipartContentTransferEncodingDefect() self.policy.handle_defect(self._cur, defect) @@ -329,9 +329,10 @@ class FeedParser: # this onto the input stream until we've scanned past the # preamble. separator = '--' + boundary - boundaryre = re.compile( - '(?P' + re.escape(separator) + - r')(?P--)?(?P[ \t]*)(?P\r\n|\r|\n)?$') + def boundarymatch(line): + if not line.startswith(separator): + return None + return boundaryendRE.match(line, len(separator)) capturing_preamble = True preamble = [] linesep = False @@ -343,7 +344,7 @@ class FeedParser: continue if line == '': break - mo = boundaryre.match(line) + mo = boundarymatch(line) if mo: # If we're looking at the end boundary, we're done with # this multipart. If there was a newline at the end of @@ -375,13 +376,13 @@ class FeedParser: if line is NeedMoreData: yield NeedMoreData continue - mo = boundaryre.match(line) + mo = boundarymatch(line) if not mo: self._input.unreadline(line) break # Recurse to parse this subpart; the input stream points # at the subpart's first line. - self._input.push_eof_matcher(boundaryre.match) + self._input.push_eof_matcher(boundarymatch) for retval in self._parsegen(): if retval is NeedMoreData: yield NeedMoreData diff --git a/Lib/email/generator.py b/Lib/email/generator.py index ae670c235..47b9df8f4 100644 --- a/Lib/email/generator.py +++ b/Lib/email/generator.py @@ -14,15 +14,16 @@ import random from copy import deepcopy from io import StringIO, BytesIO from email.utils import _has_surrogates +from email.errors import HeaderWriteError UNDERSCORE = '_' NL = '\n' # XXX: no longer used by the code below. NLCRE = re.compile(r'\r\n|\r|\n') fcre = re.compile(r'^From ', re.MULTILINE) +NEWLINE_WITHOUT_FWSP = re.compile(r'\r\n[^ \t]|\r[^ \n\t]|\n[^ \t]') - class Generator: """Generates output from a Message object tree. @@ -170,7 +171,7 @@ class Generator: # parameter. # # The way we do this, so as to make the _handle_*() methods simpler, - # is to cache any subpart writes into a buffer. The we write the + # is to cache any subpart writes into a buffer. Then we write the # headers and the buffer contents. That way, subpart handlers can # Do The Right Thing, and can still modify the Content-Type: header if # necessary. @@ -186,7 +187,11 @@ class Generator: # If we munged the cte, copy the message again and re-fix the CTE. if munge_cte: msg = deepcopy(msg) - msg.replace_header('content-transfer-encoding', munge_cte[0]) + # Preserve the header order if the CTE header already exists. + if msg.get('content-transfer-encoding') is None: + msg['Content-Transfer-Encoding'] = munge_cte[0] + else: + msg.replace_header('content-transfer-encoding', munge_cte[0]) msg.replace_header('content-type', munge_cte[1]) # Write the headers. First we see if the message object wants to # handle that itself. If not, we'll do it generically. @@ -219,7 +224,16 @@ class Generator: def _write_headers(self, msg): for h, v in msg.raw_items(): - self.write(self.policy.fold(h, v)) + folded = self.policy.fold(h, v) + if self.policy.verify_generated_headers: + linesep = self.policy.linesep + if not folded.endswith(self.policy.linesep): + raise HeaderWriteError( + f'folded header does not end with {linesep!r}: {folded!r}') + if NEWLINE_WITHOUT_FWSP.search(folded.removesuffix(linesep)): + raise HeaderWriteError( + f'folded header contains newline: {folded!r}') + self.write(folded) # A blank line always separates headers from body self.write(self._NL) @@ -240,7 +254,7 @@ class Generator: # existing message. msg = deepcopy(msg) del msg['content-transfer-encoding'] - msg.set_payload(payload, charset) + msg.set_payload(msg._payload, charset) payload = msg.get_payload() self._munge_cte = (msg['content-transfer-encoding'], msg['content-type']) @@ -388,7 +402,7 @@ class Generator: def _compile_re(cls, s, flags): return re.compile(s, flags) - + class BytesGenerator(Generator): """Generates a bytes version of a Message object tree. @@ -439,7 +453,6 @@ class BytesGenerator(Generator): return re.compile(s.encode('ascii'), flags) - _FMT = '[Non-text (%(type)s) part of message omitted, filename %(filename)s]' class DecodedGenerator(Generator): @@ -499,7 +512,6 @@ class DecodedGenerator(Generator): }, file=self) - # Helper used by Generator._make_boundary _width = len(repr(sys.maxsize-1)) _fmt = '%%0%dd' % _width diff --git a/Lib/email/header.py b/Lib/email/header.py index c7b2dd9f3..984851a7d 100644 --- a/Lib/email/header.py +++ b/Lib/email/header.py @@ -36,11 +36,11 @@ ecre = re.compile(r''' =\? # literal =? (?P[^?]*?) # non-greedy up to the next ? is the charset \? # literal ? - (?P[qb]) # either a "q" or a "b", case insensitive + (?P[qQbB]) # either a "q" or a "b", case insensitive \? # literal ? (?P.*?) # non-greedy up to the next ?= is the encoded string \?= # literal ?= - ''', re.VERBOSE | re.IGNORECASE | re.MULTILINE) + ''', re.VERBOSE | re.MULTILINE) # Field name regexp, including trailing colon, but not separating whitespace, # according to RFC 2822. Character range is from tilde to exclamation mark. @@ -52,12 +52,10 @@ fcre = re.compile(r'[\041-\176]+:$') _embedded_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. @@ -152,7 +150,6 @@ def decode_header(header): 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() @@ -175,7 +172,6 @@ def make_header(decoded_seq, maxlinelen=None, header_name=None, return h - class Header: def __init__(self, s=None, charset=None, maxlinelen=None, header_name=None, @@ -409,7 +405,6 @@ class Header: self._chunks = chunks - class _ValueFormatter: def __init__(self, headerlen, maxlen, continuation_ws, splitchars): self._maxlen = maxlen @@ -431,7 +426,7 @@ class _ValueFormatter: if end_of_line != (' ', ''): self._current_line.push(*end_of_line) if len(self._current_line) > 0: - if self._current_line.is_onlyws(): + if self._current_line.is_onlyws() and self._lines: self._lines[-1] += str(self._current_line) else: self._lines.append(str(self._current_line)) diff --git a/Lib/email/headerregistry.py b/Lib/email/headerregistry.py index 0fc2231e5..543141dc4 100644 --- a/Lib/email/headerregistry.py +++ b/Lib/email/headerregistry.py @@ -2,10 +2,6 @@ This module provides an implementation of the HeaderRegistry API. The implementation is designed to flexibly follow RFC5322 rules. - -Eventually HeaderRegistry will be a public API, but it isn't yet, -and will probably change some before that happens. - """ from types import MappingProxyType @@ -31,6 +27,11 @@ class Address: without any Content Transfer Encoding. """ + + inputs = ''.join(filter(None, (display_name, username, domain, addr_spec))) + if '\r' in inputs or '\n' in inputs: + raise ValueError("invalid arguments; address parts cannot contain CR or LF") + # This clause with its potential 'raise' may only happen when an # application program creates an Address object using an addr_spec # keyword. The email library code itself must always supply username @@ -69,11 +70,9 @@ class Address: """The addr_spec (username@domain) portion of the address, quoted according to RFC 5322 rules, but with no Content Transfer Encoding. """ - nameset = set(self.username) - if len(nameset) > len(nameset-parser.DOT_ATOM_ENDS): - lp = parser.quote_string(self.username) - else: - lp = self.username + lp = self.username + if not parser.DOT_ATOM_ENDS.isdisjoint(lp): + lp = parser.quote_string(lp) if self.domain: return lp + '@' + self.domain if not lp: @@ -86,19 +85,17 @@ class Address: self.display_name, self.username, self.domain) def __str__(self): - nameset = set(self.display_name) - if len(nameset) > len(nameset-parser.SPECIALS): - disp = parser.quote_string(self.display_name) - else: - disp = self.display_name + disp = self.display_name + if not parser.SPECIALS.isdisjoint(disp): + disp = parser.quote_string(disp) if disp: addr_spec = '' if self.addr_spec=='<>' else self.addr_spec return "{} <{}>".format(disp, addr_spec) return self.addr_spec def __eq__(self, other): - if type(other) != type(self): - return False + if not isinstance(other, Address): + return NotImplemented return (self.display_name == other.display_name and self.username == other.username and self.domain == other.domain) @@ -141,17 +138,15 @@ class Group: if self.display_name is None and len(self.addresses)==1: return str(self.addresses[0]) disp = self.display_name - if disp is not None: - nameset = set(disp) - if len(nameset) > len(nameset-parser.SPECIALS): - disp = parser.quote_string(disp) + if disp is not None and not parser.SPECIALS.isdisjoint(disp): + disp = parser.quote_string(disp) adrstr = ", ".join(str(x) for x in self.addresses) adrstr = ' ' + adrstr if adrstr else adrstr return "{}:{};".format(disp, adrstr) def __eq__(self, other): - if type(other) != type(self): - return False + if not isinstance(other, Group): + return NotImplemented return (self.display_name == other.display_name and self.addresses == other.addresses) @@ -223,7 +218,7 @@ class BaseHeader(str): self.__class__.__bases__, str(self), ), - self.__dict__) + self.__getstate__()) @classmethod def _reconstruct(cls, value): @@ -245,13 +240,16 @@ class BaseHeader(str): the header name and the ': ' separator. """ - # At some point we need to only put fws here if it was in the source. + # At some point we need to put fws here if it was in the source. header = parser.Header([ parser.HeaderLabel([ parser.ValueTerminal(self.name, 'header-name'), parser.ValueTerminal(':', 'header-sep')]), - parser.CFWSList([parser.WhiteSpaceTerminal(' ', 'fws')]), - self._parse_tree]) + ]) + if self._parse_tree: + header.append( + parser.CFWSList([parser.WhiteSpaceTerminal(' ', 'fws')])) + header.append(self._parse_tree) return header.fold(policy=policy) @@ -300,7 +298,14 @@ class DateHeader: kwds['parse_tree'] = parser.TokenList() return if isinstance(value, str): - value = utils.parsedate_to_datetime(value) + kwds['decoded'] = value + try: + value = utils.parsedate_to_datetime(value) + except ValueError: + kwds['defects'].append(errors.InvalidDateDefect('Invalid date value or format')) + kwds['datetime'] = None + kwds['parse_tree'] = parser.TokenList() + return kwds['datetime'] = value kwds['decoded'] = utils.format_datetime(kwds['datetime']) kwds['parse_tree'] = cls.value_parser(kwds['decoded']) @@ -369,8 +374,8 @@ class AddressHeader: @property def addresses(self): if self._addresses is None: - self._addresses = tuple([address for group in self._groups - for address in group.addresses]) + self._addresses = tuple(address for group in self._groups + for address in group.addresses) return self._addresses @@ -517,6 +522,18 @@ class ContentTransferEncodingHeader: return self._cte +class MessageIDHeader: + + max_count = 1 + value_parser = staticmethod(parser.parse_message_id) + + @classmethod + def parse(cls, value, kwds): + kwds['parse_tree'] = parse_tree = cls.value_parser(value) + kwds['decoded'] = str(parse_tree) + kwds['defects'].extend(parse_tree.all_defects) + + # The header factory # _default_header_map = { @@ -539,6 +556,7 @@ _default_header_map = { 'content-type': ContentTypeHeader, 'content-disposition': ContentDispositionHeader, 'content-transfer-encoding': ContentTransferEncodingHeader, + 'message-id': MessageIDHeader, } class HeaderRegistry: diff --git a/Lib/email/iterators.py b/Lib/email/iterators.py index b5502ee97..3410935e3 100644 --- a/Lib/email/iterators.py +++ b/Lib/email/iterators.py @@ -15,7 +15,6 @@ import sys from io import StringIO - # This function will become a method of the Message class def walk(self): """Walk over the message tree, yielding each subpart. @@ -29,7 +28,6 @@ def walk(self): yield from subpart.walk() - # These two functions are imported into the Iterators.py interface module. def body_line_iterator(msg, decode=False): """Iterate over the parts, returning string payloads line-by-line. @@ -55,7 +53,6 @@ def typed_subpart_iterator(msg, maintype='text', subtype=None): yield subpart - def _structure(msg, fp=None, level=0, include_default=False): """A handy debugging aid""" if fp is None: diff --git a/Lib/email/message.py b/Lib/email/message.py index f93218687..46bb8c219 100644 --- a/Lib/email/message.py +++ b/Lib/email/message.py @@ -6,6 +6,7 @@ __all__ = ['Message', 'EmailMessage'] +import binascii import re import quopri from io import BytesIO, StringIO @@ -13,7 +14,7 @@ from io import BytesIO, StringIO # Intrapackage imports from email import utils from email import errors -from email._policybase import Policy, compat32 +from email._policybase import compat32 from email import charset as _charset from email._encoded_words import decode_b Charset = _charset.Charset @@ -34,7 +35,7 @@ def _splitparam(param): if not sep: return a.strip(), None return a.strip(), b.strip() - + def _formatparam(param, value=None, quote=True): """Convenience function to format and return a key=value pair. @@ -129,7 +130,8 @@ def _decode_uu(encoded): decoded_lines.append(decoded_line) return b''.join(decoded_lines) - + + class Message: """Basic message object. @@ -169,7 +171,7 @@ class Message: header. For backward compatibility reasons, if maxheaderlen is not specified it defaults to 0, so you must override it explicitly if you want a different maxheaderlen. 'policy' is passed to the - Generator instance used to serialize the mesasge; if it is not + Generator instance used to serialize the message; if it is not specified the policy associated with the message instance is used. If the message object contains binary data that is not encoded @@ -287,25 +289,26 @@ class Message: # cte might be a Header, so for now stringify it. cte = str(self.get('content-transfer-encoding', '')).lower() # payload may be bytes here. - if isinstance(payload, str): - if utils._has_surrogates(payload): - bpayload = payload.encode('ascii', 'surrogateescape') - if not decode: + if not decode: + if isinstance(payload, str) and utils._has_surrogates(payload): + try: + bpayload = payload.encode('ascii', 'surrogateescape') try: - payload = bpayload.decode(self.get_param('charset', 'ascii'), 'replace') + payload = bpayload.decode(self.get_content_charset('ascii'), 'replace') except LookupError: payload = bpayload.decode('ascii', 'replace') - elif decode: - try: - bpayload = payload.encode('ascii') - except UnicodeError: - # This won't happen for RFC compliant messages (messages - # containing only ASCII code points in the unicode input). - # If it does happen, turn the string into bytes in a way - # guaranteed not to fail. - bpayload = payload.encode('raw-unicode-escape') - if not decode: + except UnicodeEncodeError: + pass return payload + if isinstance(payload, str): + try: + bpayload = payload.encode('ascii', 'surrogateescape') + except UnicodeEncodeError: + # This won't happen for RFC compliant messages (messages + # containing only ASCII code points in the unicode input). + # If it does happen, turn the string into bytes in a way + # guaranteed not to fail. + bpayload = payload.encode('raw-unicode-escape') if cte == 'quoted-printable': return quopri.decodestring(bpayload) elif cte == 'base64': @@ -337,7 +340,7 @@ class Message: return if not isinstance(charset, Charset): charset = Charset(charset) - payload = payload.encode(charset.output_charset) + payload = payload.encode(charset.output_charset, 'surrogateescape') if hasattr(payload, 'decode'): self._payload = payload.decode('ascii', 'surrogateescape') else: @@ -446,7 +449,11 @@ class Message: self._headers = newheaders def __contains__(self, name): - return name.lower() in [k.lower() for k, v in self._headers] + name_lower = name.lower() + for k, v in self._headers: + if name_lower == k.lower(): + return True + return False def __iter__(self): for field, value in self._headers: @@ -973,7 +980,7 @@ class MIMEPart(Message): if policy is None: from email.policy import default policy = default - Message.__init__(self, policy) + super().__init__(policy) def as_string(self, unixfrom=False, maxheaderlen=None, policy=None): @@ -983,14 +990,14 @@ class MIMEPart(Message): header. maxheaderlen is retained for backward compatibility with the base Message class, but defaults to None, meaning that the policy value for max_line_length controls the header maximum length. 'policy' is - passed to the Generator instance used to serialize the mesasge; if it + passed to the Generator instance used to serialize the message; if it is not specified the policy associated with the message instance is used. """ policy = self.policy if policy is None else policy if maxheaderlen is None: maxheaderlen = policy.max_line_length - return super().as_string(maxheaderlen=maxheaderlen, policy=policy) + return super().as_string(unixfrom, maxheaderlen, policy) def __str__(self): return self.as_string(policy=self.policy.clone(utf8=True)) @@ -1007,7 +1014,7 @@ class MIMEPart(Message): if subtype in preferencelist: yield (preferencelist.index(subtype), part) return - if maintype != 'multipart': + if maintype != 'multipart' or not self.is_multipart(): return if subtype != 'related': for subpart in part.iter_parts(): @@ -1066,7 +1073,16 @@ class MIMEPart(Message): maintype, subtype = self.get_content_type().split('/') if maintype != 'multipart' or subtype == 'alternative': return - parts = self.get_payload().copy() + payload = self.get_payload() + # Certain malformed messages can have content type set to `multipart/*` + # but still have single part body, in which case payload.copy() can + # fail with AttributeError. + try: + parts = payload.copy() + except AttributeError: + # payload is not a list, it is most probably a string. + return + if maintype == 'multipart' and subtype == 'related': # For related, we treat everything but the root as an attachment. # The root may be indicated by 'start'; if there's no start or we @@ -1103,7 +1119,7 @@ class MIMEPart(Message): Return an empty iterator for a non-multipart. """ - if self.get_content_maintype() == 'multipart': + if self.is_multipart(): yield from self.get_payload() def get_content(self, *args, content_manager=None, **kw): diff --git a/Lib/email/mime/application.py b/Lib/email/mime/application.py index 6877e554e..f67cbad3f 100644 --- a/Lib/email/mime/application.py +++ b/Lib/email/mime/application.py @@ -17,7 +17,7 @@ class MIMEApplication(MIMENonMultipart): _encoder=encoders.encode_base64, *, policy=None, **_params): """Create an application/* type MIME document. - _data is a string containing the raw application data. + _data contains the bytes for the raw application data. _subtype is the MIME content type subtype, defaulting to 'octet-stream'. diff --git a/Lib/email/mime/audio.py b/Lib/email/mime/audio.py index 4bcd7b224..aa0c4905c 100644 --- a/Lib/email/mime/audio.py +++ b/Lib/email/mime/audio.py @@ -6,39 +6,10 @@ __all__ = ['MIMEAudio'] -import sndhdr - -from io import BytesIO from email import encoders from email.mime.nonmultipart import MIMENonMultipart - -_sndhdr_MIMEmap = {'au' : 'basic', - 'wav' :'x-wav', - 'aiff':'x-aiff', - 'aifc':'x-aiff', - } - -# There are others in sndhdr that don't have MIME types. :( -# Additional ones to be added to sndhdr? midi, mp3, realaudio, wma?? -def _whatsnd(data): - """Try to identify a sound file type. - - sndhdr.what() has a pretty cruddy interface, unfortunately. This is why - we re-do it here. It would be easier to reverse engineer the Unix 'file' - command and use the standard 'magic' file, as shipped with a modern Unix. - """ - hdr = data[:512] - fakefile = BytesIO(hdr) - for testfn in sndhdr.tests: - res = testfn(hdr, fakefile) - if res is not None: - return _sndhdr_MIMEmap.get(res[0]) - return None - - - class MIMEAudio(MIMENonMultipart): """Class for generating audio/* MIME documents.""" @@ -46,8 +17,8 @@ class MIMEAudio(MIMENonMultipart): _encoder=encoders.encode_base64, *, policy=None, **_params): """Create an audio/* type MIME document. - _audiodata is a string containing the raw audio data. If this data - can be decoded by the standard Python `sndhdr' module, then the + _audiodata contains the bytes for the raw audio data. If this data + can be decoded as au, wav, aiff, or aifc, then the subtype will be automatically included in the Content-Type header. Otherwise, you can specify the specific audio subtype via the _subtype parameter. If _subtype is not given, and no subtype can be @@ -65,10 +36,62 @@ class MIMEAudio(MIMENonMultipart): header. """ if _subtype is None: - _subtype = _whatsnd(_audiodata) + _subtype = _what(_audiodata) if _subtype is None: raise TypeError('Could not find audio MIME subtype') MIMENonMultipart.__init__(self, 'audio', _subtype, policy=policy, **_params) self.set_payload(_audiodata) _encoder(self) + + +_rules = [] + + +# Originally from the sndhdr module. +# +# There are others in sndhdr that don't have MIME types. :( +# Additional ones to be added to sndhdr? midi, mp3, realaudio, wma?? +def _what(data): + # Try to identify a sound file type. + # + # sndhdr.what() had a pretty cruddy interface, unfortunately. This is why + # we re-do it here. It would be easier to reverse engineer the Unix 'file' + # command and use the standard 'magic' file, as shipped with a modern Unix. + for testfn in _rules: + if res := testfn(data): + return res + else: + return None + + +def rule(rulefunc): + _rules.append(rulefunc) + return rulefunc + + +@rule +def _aiff(h): + if not h.startswith(b'FORM'): + return None + if h[8:12] in {b'AIFC', b'AIFF'}: + return 'x-aiff' + else: + return None + + +@rule +def _au(h): + if h.startswith(b'.snd'): + return 'basic' + else: + return None + + +@rule +def _wav(h): + # 'RIFF' 'WAVE' 'fmt ' + if not h.startswith(b'RIFF') or h[8:12] != b'WAVE' or h[12:16] != b'fmt ': + return None + else: + return "x-wav" diff --git a/Lib/email/mime/base.py b/Lib/email/mime/base.py index 1a3f9b51f..f601f621c 100644 --- a/Lib/email/mime/base.py +++ b/Lib/email/mime/base.py @@ -11,7 +11,6 @@ import email.policy from email import message - class MIMEBase(message.Message): """Base class for MIME specializations.""" diff --git a/Lib/email/mime/image.py b/Lib/email/mime/image.py index 92724643c..4b7f2f9cb 100644 --- a/Lib/email/mime/image.py +++ b/Lib/email/mime/image.py @@ -6,13 +6,10 @@ __all__ = ['MIMEImage'] -import imghdr - from email import encoders from email.mime.nonmultipart import MIMENonMultipart - class MIMEImage(MIMENonMultipart): """Class for generating image/* type MIME documents.""" @@ -20,11 +17,11 @@ class MIMEImage(MIMENonMultipart): _encoder=encoders.encode_base64, *, policy=None, **_params): """Create an image/* type MIME document. - _imagedata is a string containing the raw image data. If this data - can be decoded by the standard Python `imghdr' module, then the - subtype will be automatically included in the Content-Type header. - Otherwise, you can specify the specific image subtype via the _subtype - parameter. + _imagedata contains the bytes for the raw image data. If the data + type can be detected (jpeg, png, gif, tiff, rgb, pbm, pgm, ppm, + rast, xbm, bmp, webp, and exr attempted), then the subtype will be + automatically included in the Content-Type header. Otherwise, you can + specify the specific image subtype via the _subtype parameter. _encoder is a function which will perform the actual encoding for transport of the image data. It takes one argument, which is this @@ -37,11 +34,119 @@ class MIMEImage(MIMENonMultipart): constructor, which turns them into parameters on the Content-Type header. """ - if _subtype is None: - _subtype = imghdr.what(None, _imagedata) + _subtype = _what(_imagedata) if _subtype is None else _subtype if _subtype is None: raise TypeError('Could not guess image MIME subtype') MIMENonMultipart.__init__(self, 'image', _subtype, policy=policy, **_params) self.set_payload(_imagedata) _encoder(self) + + +_rules = [] + + +# Originally from the imghdr module. +def _what(data): + for rule in _rules: + if res := rule(data): + return res + else: + return None + + +def rule(rulefunc): + _rules.append(rulefunc) + return rulefunc + + +@rule +def _jpeg(h): + """JPEG data with JFIF or Exif markers; and raw JPEG""" + if h[6:10] in (b'JFIF', b'Exif'): + return 'jpeg' + elif h[:4] == b'\xff\xd8\xff\xdb': + return 'jpeg' + + +@rule +def _png(h): + if h.startswith(b'\211PNG\r\n\032\n'): + return 'png' + + +@rule +def _gif(h): + """GIF ('87 and '89 variants)""" + if h[:6] in (b'GIF87a', b'GIF89a'): + return 'gif' + + +@rule +def _tiff(h): + """TIFF (can be in Motorola or Intel byte order)""" + if h[:2] in (b'MM', b'II'): + return 'tiff' + + +@rule +def _rgb(h): + """SGI image library""" + if h.startswith(b'\001\332'): + return 'rgb' + + +@rule +def _pbm(h): + """PBM (portable bitmap)""" + if len(h) >= 3 and \ + h[0] == ord(b'P') and h[1] in b'14' and h[2] in b' \t\n\r': + return 'pbm' + + +@rule +def _pgm(h): + """PGM (portable graymap)""" + if len(h) >= 3 and \ + h[0] == ord(b'P') and h[1] in b'25' and h[2] in b' \t\n\r': + return 'pgm' + + +@rule +def _ppm(h): + """PPM (portable pixmap)""" + if len(h) >= 3 and \ + h[0] == ord(b'P') and h[1] in b'36' and h[2] in b' \t\n\r': + return 'ppm' + + +@rule +def _rast(h): + """Sun raster file""" + if h.startswith(b'\x59\xA6\x6A\x95'): + return 'rast' + + +@rule +def _xbm(h): + """X bitmap (X10 or X11)""" + if h.startswith(b'#define '): + return 'xbm' + + +@rule +def _bmp(h): + if h.startswith(b'BM'): + return 'bmp' + + +@rule +def _webp(h): + if h.startswith(b'RIFF') and h[8:12] == b'WEBP': + return 'webp' + + +@rule +def _exr(h): + if h.startswith(b'\x76\x2f\x31\x01'): + return 'exr' diff --git a/Lib/email/mime/message.py b/Lib/email/mime/message.py index 07e4f2d11..61836b5a7 100644 --- a/Lib/email/mime/message.py +++ b/Lib/email/mime/message.py @@ -10,7 +10,6 @@ from email import message from email.mime.nonmultipart import MIMENonMultipart - class MIMEMessage(MIMENonMultipart): """Class representing message/* MIME documents.""" diff --git a/Lib/email/mime/multipart.py b/Lib/email/mime/multipart.py index 2d3f28881..94d81c771 100644 --- a/Lib/email/mime/multipart.py +++ b/Lib/email/mime/multipart.py @@ -9,7 +9,6 @@ __all__ = ['MIMEMultipart'] from email.mime.base import MIMEBase - class MIMEMultipart(MIMEBase): """Base class for MIME multipart/* type messages.""" diff --git a/Lib/email/mime/nonmultipart.py b/Lib/email/mime/nonmultipart.py index e1f51968b..a41386eb1 100644 --- a/Lib/email/mime/nonmultipart.py +++ b/Lib/email/mime/nonmultipart.py @@ -10,7 +10,6 @@ from email import errors from email.mime.base import MIMEBase - class MIMENonMultipart(MIMEBase): """Base class for MIME non-multipart type messages.""" diff --git a/Lib/email/mime/text.py b/Lib/email/mime/text.py index 35b442383..7672b7891 100644 --- a/Lib/email/mime/text.py +++ b/Lib/email/mime/text.py @@ -6,11 +6,9 @@ __all__ = ['MIMEText'] -from email.charset import Charset from email.mime.nonmultipart import MIMENonMultipart - class MIMEText(MIMENonMultipart): """Class for generating text/* type MIME documents.""" @@ -37,6 +35,6 @@ class MIMEText(MIMENonMultipart): _charset = 'utf-8' MIMENonMultipart.__init__(self, 'text', _subtype, policy=policy, - **{'charset': str(_charset)}) + charset=str(_charset)) self.set_payload(_text, _charset) diff --git a/Lib/email/parser.py b/Lib/email/parser.py index 555b17256..06d99b17f 100644 --- a/Lib/email/parser.py +++ b/Lib/email/parser.py @@ -13,7 +13,6 @@ from email.feedparser import FeedParser, BytesFeedParser from email._policybase import compat32 - class Parser: def __init__(self, _class=None, *, policy=compat32): """Parser of RFC 2822 and MIME email messages. @@ -50,10 +49,7 @@ class Parser: feedparser = FeedParser(self._class, policy=self.policy) if headersonly: feedparser._set_headersonly() - while True: - data = fp.read(8192) - if not data: - break + while data := fp.read(8192): feedparser.feed(data) return feedparser.close() @@ -68,7 +64,6 @@ class Parser: return self.parse(StringIO(text), headersonly=headersonly) - class HeaderParser(Parser): def parse(self, fp, headersonly=True): return Parser.parse(self, fp, True) @@ -76,7 +71,7 @@ class HeaderParser(Parser): def parsestr(self, text, headersonly=True): return Parser.parsestr(self, text, True) - + class BytesParser: def __init__(self, *args, **kw): diff --git a/Lib/email/policy.py b/Lib/email/policy.py index 5131311ac..6e109b650 100644 --- a/Lib/email/policy.py +++ b/Lib/email/policy.py @@ -3,6 +3,7 @@ code that adds all the email6 features. """ import re +import sys from email._policybase import Policy, Compat32, compat32, _extend_docstrings from email.utils import _has_surrogates from email.headerregistry import HeaderRegistry as HeaderRegistry @@ -20,7 +21,7 @@ __all__ = [ 'HTTP', ] -linesep_splitter = re.compile(r'\n|\r') +linesep_splitter = re.compile(r'\n|\r\n?') @_extend_docstrings class EmailPolicy(Policy): @@ -118,13 +119,13 @@ class EmailPolicy(Policy): """+ The name is parsed as everything up to the ':' and returned unmodified. The value is determined by stripping leading whitespace off the - remainder of the first line, joining all subsequent lines together, and + remainder of the first line joined with all subsequent lines, and stripping any trailing carriage return or linefeed characters. (This is the same as Compat32). """ name, value = sourcelines[0].split(':', 1) - value = value.lstrip(' \t') + ''.join(sourcelines[1:]) + value = ''.join((value, *sourcelines[1:])).lstrip(' \t\r\n') return (name, value.rstrip('\r\n')) def header_store_parse(self, name, value): @@ -203,14 +204,22 @@ class EmailPolicy(Policy): def _fold(self, name, value, refold_binary=False): if hasattr(value, 'name'): return value.fold(policy=self) - maxlen = self.max_line_length if self.max_line_length else float('inf') - lines = value.splitlines() + maxlen = self.max_line_length if self.max_line_length else sys.maxsize + # We can't use splitlines here because it splits on more than \r and \n. + lines = linesep_splitter.split(value) refold = (self.refold_source == 'all' or self.refold_source == 'long' and (lines and len(lines[0])+len(name)+2 > maxlen or any(len(x) > maxlen for x in lines[1:]))) - if refold or refold_binary and _has_surrogates(value): + + if not refold: + if not self.utf8: + refold = not value.isascii() + elif refold_binary: + refold = _has_surrogates(value) + if refold: return self.header_factory(name, ''.join(lines)).fold(policy=self) + return name + ': ' + self.linesep.join(lines) + self.linesep diff --git a/Lib/email/quoprimime.py b/Lib/email/quoprimime.py index c543eb59a..27fcbb5a2 100644 --- a/Lib/email/quoprimime.py +++ b/Lib/email/quoprimime.py @@ -148,6 +148,7 @@ def header_encode(header_bytes, charset='iso-8859-1'): _QUOPRI_BODY_ENCODE_MAP = _QUOPRI_BODY_MAP[:] for c in b'\r\n': _QUOPRI_BODY_ENCODE_MAP[c] = chr(c) +del c def body_encode(body, maxlinelen=76, eol=NL): """Encode with quoted-printable, wrapping at maxlinelen characters. @@ -173,7 +174,7 @@ def body_encode(body, maxlinelen=76, eol=NL): if not body: return body - # quote speacial characters + # quote special characters body = body.translate(_QUOPRI_BODY_ENCODE_MAP) soft_break = '=' + eol diff --git a/Lib/email/utils.py b/Lib/email/utils.py index a759d2330..e42674fa4 100644 --- a/Lib/email/utils.py +++ b/Lib/email/utils.py @@ -25,8 +25,6 @@ __all__ = [ import os import re import time -import random -import socket import datetime import urllib.parse @@ -36,9 +34,6 @@ from email._parseaddr import mktime_tz from email._parseaddr import parsedate, parsedate_tz, _parsedate_tz -# Intrapackage imports -from email.charset import Charset - COMMASPACE = ', ' EMPTYSTRING = '' UEMPTYSTRING = '' @@ -48,11 +43,12 @@ TICK = "'" specialsre = re.compile(r'[][\\()<>@,:;".]') escapesre = re.compile(r'[\\"]') + def _has_surrogates(s): - """Return True if s contains surrogate-escaped binary data.""" + """Return True if s may contain surrogate-escaped binary data.""" # This check is based on the fact that unless there are surrogates, utf8 # (Python's default encoding) can encode any string. This is the fastest - # way to check for surrogates, see issue 11454 for timings. + # way to check for surrogates, see bpo-11454 (moved to gh-55663) for timings. try: s.encode() return False @@ -81,7 +77,7 @@ def formataddr(pair, charset='utf-8'): If the first element of pair is false, then the second element is returned unmodified. - Optional charset if given is the character set that is used to encode + The optional charset is the character set that is used to encode realname in case realname is not ASCII safe. Can be an instance of str or a Charset-like object which has a header_encode method. Default is 'utf-8'. @@ -94,6 +90,8 @@ def formataddr(pair, charset='utf-8'): name.encode('ascii') except UnicodeEncodeError: if isinstance(charset, str): + # lazy import to improve module import time + from email.charset import Charset charset = Charset(charset) encoded_name = charset.header_encode(name) return "%s <%s>" % (encoded_name, address) @@ -106,24 +104,127 @@ def formataddr(pair, charset='utf-8'): return address - -def getaddresses(fieldvalues): - """Return a list of (REALNAME, EMAIL) for each fieldvalue.""" - all = COMMASPACE.join(fieldvalues) - a = _AddressList(all) - return a.addresslist +def _iter_escaped_chars(addr): + pos = 0 + escape = False + for pos, ch in enumerate(addr): + if escape: + yield (pos, '\\' + ch) + escape = False + elif ch == '\\': + escape = True + else: + yield (pos, ch) + if escape: + yield (pos, '\\') +def _strip_quoted_realnames(addr): + """Strip real names between quotes.""" + if '"' not in addr: + # Fast path + return addr -ecre = re.compile(r''' - =\? # literal =? - (?P[^?]*?) # non-greedy up to the next ? is the charset - \? # literal ? - (?P[qb]) # either a "q" or a "b", case insensitive - \? # literal ? - (?P.*?) # non-greedy up to the next ?= is the atom - \?= # literal ?= - ''', re.VERBOSE | re.IGNORECASE) + start = 0 + open_pos = None + result = [] + for pos, ch in _iter_escaped_chars(addr): + if ch == '"': + if open_pos is None: + open_pos = pos + else: + if start != open_pos: + result.append(addr[start:open_pos]) + start = pos + 1 + open_pos = None + + if start < len(addr): + result.append(addr[start:]) + + return ''.join(result) + + +supports_strict_parsing = True + +def getaddresses(fieldvalues, *, strict=True): + """Return a list of (REALNAME, EMAIL) or ('','') for each fieldvalue. + + When parsing fails for a fieldvalue, a 2-tuple of ('', '') is returned in + its place. + + If strict is true, use a strict parser which rejects malformed inputs. + """ + + # If strict is true, if the resulting list of parsed addresses is greater + # than the number of fieldvalues in the input list, a parsing error has + # occurred and consequently a list containing a single empty 2-tuple [('', + # '')] is returned in its place. This is done to avoid invalid output. + # + # Malformed input: getaddresses(['alice@example.com ']) + # Invalid output: [('', 'alice@example.com'), ('', 'bob@example.com')] + # Safe output: [('', '')] + + if not strict: + all = COMMASPACE.join(str(v) for v in fieldvalues) + a = _AddressList(all) + return a.addresslist + + fieldvalues = [str(v) for v in fieldvalues] + fieldvalues = _pre_parse_validation(fieldvalues) + addr = COMMASPACE.join(fieldvalues) + a = _AddressList(addr) + result = _post_parse_validation(a.addresslist) + + # Treat output as invalid if the number of addresses is not equal to the + # expected number of addresses. + n = 0 + for v in fieldvalues: + # When a comma is used in the Real Name part it is not a deliminator. + # So strip those out before counting the commas. + v = _strip_quoted_realnames(v) + # Expected number of addresses: 1 + number of commas + n += 1 + v.count(',') + if len(result) != n: + return [('', '')] + + return result + + +def _check_parenthesis(addr): + # Ignore parenthesis in quoted real names. + addr = _strip_quoted_realnames(addr) + + opens = 0 + for pos, ch in _iter_escaped_chars(addr): + if ch == '(': + opens += 1 + elif ch == ')': + opens -= 1 + if opens < 0: + return False + return (opens == 0) + + +def _pre_parse_validation(email_header_fields): + accepted_values = [] + for v in email_header_fields: + if not _check_parenthesis(v): + v = "('', '')" + accepted_values.append(v) + + return accepted_values + + +def _post_parse_validation(parsed_email_header_tuples): + accepted_values = [] + # The parser would have parsed a correctly formatted domain-literal + # The existence of an [ after parsing indicates a parsing failure + for v in parsed_email_header_tuples: + if '[' in v[1]: + v = ('', '') + accepted_values.append(v) + + return accepted_values def _format_timetuple_and_zone(timetuple, zone): @@ -140,7 +241,7 @@ def formatdate(timeval=None, localtime=False, usegmt=False): Fri, 09 Nov 2001 01:08:47 -0000 - Optional timeval if given is a floating point time value as accepted by + Optional timeval if given is a floating-point time value as accepted by gmtime() and localtime(), otherwise the current time is used. Optional localtime is a flag that when True, interprets timeval, and @@ -155,13 +256,13 @@ def formatdate(timeval=None, localtime=False, usegmt=False): # 2822 requires that day and month names be the English abbreviations. if timeval is None: timeval = time.time() - if localtime or usegmt: - dt = datetime.datetime.fromtimestamp(timeval, datetime.timezone.utc) - else: - dt = datetime.datetime.utcfromtimestamp(timeval) + dt = datetime.datetime.fromtimestamp(timeval, datetime.timezone.utc) + if localtime: dt = dt.astimezone() usegmt = False + elif not usegmt: + dt = dt.replace(tzinfo=None) return format_datetime(dt, usegmt) def format_datetime(dt, usegmt=False): @@ -193,6 +294,11 @@ def make_msgid(idstring=None, domain=None): portion of the message id after the '@'. It defaults to the locally defined hostname. """ + # Lazy imports to speedup module import time + # (no other functions in email.utils need these modules) + import random + import socket + timeval = int(time.time()*100) pid = os.getpid() randint = random.getrandbits(64) @@ -207,17 +313,43 @@ def make_msgid(idstring=None, domain=None): def parsedate_to_datetime(data): - *dtuple, tz = _parsedate_tz(data) + parsed_date_tz = _parsedate_tz(data) + if parsed_date_tz is None: + raise ValueError('Invalid date value or format "%s"' % str(data)) + *dtuple, tz = parsed_date_tz if tz is None: return datetime.datetime(*dtuple[:6]) return datetime.datetime(*dtuple[:6], tzinfo=datetime.timezone(datetime.timedelta(seconds=tz))) -def parseaddr(addr): - addrs = _AddressList(addr).addresslist - if not addrs: - return '', '' +def parseaddr(addr, *, strict=True): + """ + Parse addr into its constituent realname and email address parts. + + Return a tuple of realname and email address, unless the parse fails, in + which case return a 2-tuple of ('', ''). + + If strict is True, use a strict parser which rejects malformed inputs. + """ + if not strict: + addrs = _AddressList(addr).addresslist + if not addrs: + return ('', '') + return addrs[0] + + if isinstance(addr, list): + addr = addr[0] + + if not isinstance(addr, str): + return ('', '') + + addr = _pre_parse_validation([addr])[0] + addrs = _post_parse_validation(_AddressList(addr).addresslist) + + if not addrs or len(addrs) > 1: + return ('', '') + return addrs[0] @@ -265,21 +397,13 @@ def decode_params(params): params is a sequence of 2-tuples containing (param name, string value). """ - # Copy params so we don't mess with the original - params = params[:] - new_params = [] + new_params = [params[0]] # Map parameter's name to a list of continuations. The values are a # 3-tuple of the continuation number, the string value, and a flag # specifying whether a particular segment is %-encoded. rfc2231_params = {} - name, value = params.pop(0) - new_params.append((name, value)) - while params: - name, value = params.pop(0) - if name.endswith('*'): - encoded = True - else: - encoded = False + for name, value in params[1:]: + encoded = name.endswith('*') value = unquote(value) mo = rfc2231_continuation.match(name) if mo: @@ -342,41 +466,23 @@ def collapse_rfc2231_value(value, errors='replace', # better than not having it. # -def localtime(dt=None, isdst=-1): +def localtime(dt=None, isdst=None): """Return local time as an aware datetime object. If called without arguments, return current time. Otherwise *dt* argument should be a datetime instance, and it is converted to the local time zone according to the system time zone database. If *dt* is naive (that is, dt.tzinfo is None), it is assumed to be in local time. - In this case, a positive or zero value for *isdst* causes localtime to - presume initially that summer time (for example, Daylight Saving Time) - is or is not (respectively) in effect for the specified time. A - negative value for *isdst* causes the localtime() function to attempt - to divine whether summer time is in effect for the specified time. + The isdst parameter is ignored. """ + if isdst is not None: + import warnings + warnings._deprecated( + "The 'isdst' parameter to 'localtime'", + message='{name} is deprecated and slated for removal in Python {remove}', + remove=(3, 14), + ) if dt is None: - return datetime.datetime.now(datetime.timezone.utc).astimezone() - if dt.tzinfo is not None: - return dt.astimezone() - # We have a naive datetime. Convert to a (localtime) timetuple and pass to - # system mktime together with the isdst hint. System mktime will return - # seconds since epoch. - tm = dt.timetuple()[:-1] + (isdst,) - seconds = time.mktime(tm) - localtm = time.localtime(seconds) - try: - delta = datetime.timedelta(seconds=localtm.tm_gmtoff) - tz = datetime.timezone(delta, localtm.tm_zone) - except AttributeError: - # Compute UTC offset and compare with the value implied by tm_isdst. - # If the values match, use the zone name implied by tm_isdst. - delta = dt - datetime.datetime(*time.gmtime(seconds)[:6]) - dst = time.daylight and localtm.tm_isdst > 0 - gmtoff = -(time.altzone if dst else time.timezone) - if delta == datetime.timedelta(seconds=gmtoff): - tz = datetime.timezone(delta, time.tzname[dst]) - else: - tz = datetime.timezone(delta) - return dt.replace(tzinfo=tz) + dt = datetime.datetime.now() + return dt.astimezone()