"""Header value parser implementing various email-related RFC parsing rules.

The parsing methods defined in this module implement various email related

parsing rules. Principal among them is RFC 5322, which is the followon

to RFC 2822 and primarily a clarification of the former. It also implements

RFC 2047 encoded word decoding.

RFC 5322 goes to considerable trouble to maintain backward compatibility with

RFC 822 in the parse phase, while cleaning up the structure on the generation

phase. This parser supports correct RFC 5322 generation by tagging white space

as folding white space only when folding is allowed in the non-obsolete rule

sets. Actually, the parser is even more generous when accepting input than RFC

5322 mandates, following the spirit of Postel's Law, which RFC 5322 encourages.

Where possible deviations from the standard are annotated on the 'defects'

attribute of tokens that deviate.

The general structure of the parser follows RFC 5322, and uses its terminology

where there is a direct correspondence. Where the implementation requires a

somewhat different structure than that used by the formal grammar, new terms

that mimic the closest existing terms are used. Thus, it really helps to have

a copy of RFC 5322 handy when studying this code.

Input to the parser is a string that has already been unfolded according to

RFC 5322 rules. According to the RFC this unfolding is the very first step, and

this parser leaves the unfolding step to a higher level message parser, which

will have already detected the line breaks that need unfolding while

determining the beginning and end of each header.

The output of the parser is a TokenList object, which is a list subclass. A

TokenList is a recursive data structure. The terminal nodes of the structure

are Terminal objects, which are subclasses of str. These do not correspond

directly to terminal objects in the formal grammar, but are instead more

practical higher level combinations of true terminals.

All TokenList and Terminal objects have a 'value' attribute, which produces the

semantically meaningful value of that part of the parse subtree. The value of

all whitespace tokens (no matter how many sub-tokens they may contain) is a

single space, as per the RFC rules. This includes 'CFWS', which is herein

included in the general class of whitespace tokens. There is one exception to

the rule that whitespace tokens are collapsed into single spaces in values: in

the value of a 'bare-quoted-string' (a quoted-string with no leading or

trailing whitespace), any whitespace that appeared between the quotation marks

is preserved in the returned value. Note that in all Terminal strings quoted

pairs are turned into their unquoted values.

All TokenList and Terminal objects also have a string value, which attempts to

be a "canonical" representation of the RFC-compliant form of the substring that

produced the parsed subtree, including minimal use of quoted pair quoting.

Whitespace runs are not collapsed.

Comment tokens also have a 'content' attribute providing the string found

between the parens (including any nested comments) with whitespace preserved.

All TokenList and Terminal objects have a 'defects' attribute which is a

possibly empty list all of the defects found while creating the token. Defects

may appear on any token in the tree, and a composite list of all defects in the

subtree is available through the 'all_defects' attribute of any node. (For

Terminal notes x.defects == x.all_defects.)

Each object in a parse tree is called a 'token', and each has a 'token_type'

attribute that gives the name from the RFC 5322 grammar that it represents.

Not all RFC 5322 nodes are produced, and there is one non-RFC 5322 node that

may be produced: 'ptext'. A 'ptext' is a string of printable ascii characters.

It is returned in place of lists of (ctext/quoted-pair) and

(qtext/quoted-pair).

XXX: provide complete list of token types.

"""

import re

import sys

import urllib # For urllib.parse.unquote

from string import hexdigits

from operator import itemgetter

from email import _encoded_words as _ew

from email import errors

from email import utils

#

# Useful constants and functions

#

WSP = set(' \t')

CFWS_LEADER = WSP | set('(')

SPECIALS = set(r'()<>@,:;.\"[]')

ATOM_ENDS = SPECIALS | WSP

DOT_ATOM_ENDS = ATOM_ENDS - set('.')

# '.', '"', and '(' do not end phrases in order to support obs-phrase

PHRASE_ENDS = SPECIALS - set('."(')

TSPECIALS = (SPECIALS | set('/?=')) - set('.')

TOKEN_ENDS = TSPECIALS | WSP

ASPECIALS = TSPECIALS | set("*'%")

ATTRIBUTE_ENDS = ASPECIALS | WSP

EXTENDED_ATTRIBUTE_ENDS = ATTRIBUTE_ENDS - set('%')

NLSET = {'\n', '\r'}

SPECIALSNL = SPECIALS | NLSET

def make_quoted_pairs(value):

"""Escape dquote and backslash for use within a quoted-string."""

return str(value).replace('\\', '\\\\').replace('"', '\\"')

def make_parenthesis_pairs(value):

"""Escape parenthesis and backslash for use within a comment."""

return str(value).replace('\\', '\\\\') \

.replace('(', '\\(').replace(')', '\\)')

def quote_string(value):

escaped = make_quoted_pairs(value)

return f'"{escaped}"'

# 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)

#

# TokenList and its subclasses

#

class TokenList(list):

token_type = None

syntactic_break = True

ew_combine_allowed = True

def __init__(self, *args, **kw):

super().__init__(*args, **kw)

self.defects = []

def __str__(self):

return ''.join(str(x) for x in self)

def __repr__(self):

return '{}({})'.format(self.__class__.__name__,

super().__repr__())

@property

def value(self):

return ''.join(x.value for x in self if x.value)

@property

def all_defects(self):

return sum((x.all_defects for x in self), self.defects)

def startswith_fws(self):

return self[0].startswith_fws()

@property

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):

comments = []

for token in self:

comments.extend(token.comments)

return comments

def fold(self, *, policy):

return _refold_parse_tree(self, policy=policy)

def pprint(self, indent=''):

print(self.ppstr(indent=indent))

def ppstr(self, indent=''):

return '\n'.join(self._pp(indent=indent))

def _pp(self, indent=''):

yield '{}{}/{}('.format(

indent,

self.__class__.__name__,

self.token_type)

for token in self:

if not hasattr(token, '_pp'):

yield (indent + ' !! invalid element in token '

'list: {!r}'.format(token))

else:

yield from token._pp(indent+' ')

if self.defects:

extra = ' Defects: {}'.format(self.defects)

else:

extra = ''

yield '{}){}'.format(indent, extra)

class WhiteSpaceTokenList(TokenList):

@property

def value(self):

return ' '

@property

def comments(self):

return [x.content for x in self if x.token_type=='comment']

class UnstructuredTokenList(TokenList):

token_type = 'unstructured'

class Phrase(TokenList):

token_type = 'phrase'

class Word(TokenList):

token_type = 'word'

class CFWSList(WhiteSpaceTokenList):

token_type = 'cfws'

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

class QuotedString(TokenList):

token_type = 'quoted-string'

@property

def content(self):

for x in self:

if x.token_type == 'bare-quoted-string':

return x.value

@property

def quoted_value(self):

res = []

for x in self:

if x.token_type == 'bare-quoted-string':

res.append(str(x))

else:

res.append(x.value)

return ''.join(res)

@property

def stripped_value(self):

for token in self:

if token.token_type == 'bare-quoted-string':

return token.value

class BareQuotedString(QuotedString):

token_type = 'bare-quoted-string'

def __str__(self):

return quote_string(''.join(str(x) for x in self))

@property

def value(self):

return ''.join(str(x) for x in self)

class Comment(WhiteSpaceTokenList):

token_type = 'comment'

def __str__(self):

return ''.join(sum([

["("],

[self.quote(x) for x in self],

[")"],

], []))

def quote(self, value):

if value.token_type == 'comment':

return str(value)

return str(value).replace('\\', '\\\\').replace(

'(', r'\(').replace(

')', r'\)')

@property

def content(self):

return ''.join(str(x) for x in self)

@property

def comments(self):

return [self.content]

class AddressList(TokenList):

token_type = 'address-list'

@property

def addresses(self):

return [x for x in self if x.token_type=='address']

@property

def mailboxes(self):

return sum((x.mailboxes

for x in self if x.token_type=='address'), [])

@property

def all_mailboxes(self):

return sum((x.all_mailboxes

for x in self if x.token_type=='address'), [])

class Address(TokenList):

token_type = 'address'

@property

def display_name(self):

if self[0].token_type == 'group':

return self[0].display_name

@property

def mailboxes(self):

if self[0].token_type == 'mailbox':

return [self[0]]

elif self[0].token_type == 'invalid-mailbox':

return []

return self[0].mailboxes

@property

def all_mailboxes(self):

if self[0].token_type == 'mailbox':

return [self[0]]

elif self[0].token_type == 'invalid-mailbox':

return [self[0]]

return self[0].all_mailboxes

class MailboxList(TokenList):

token_type = 'mailbox-list'

@property

def mailboxes(self):

return [x for x in self if x.token_type=='mailbox']

@property

def all_mailboxes(self):

return [x for x in self

if x.token_type in ('mailbox', 'invalid-mailbox')]

class GroupList(TokenList):

token_type = 'group-list'

@property

def mailboxes(self):

if not self or self[0].token_type != 'mailbox-list':

return []

return self[0].mailboxes

@property

def all_mailboxes(self):

if not self or self[0].token_type != 'mailbox-list':

return []

return self[0].all_mailboxes

class Group(TokenList):

token_type = "group"

@property

def mailboxes(self):

if self[2].token_type != 'group-list':

return []

return self[2].mailboxes

@property

def all_mailboxes(self):

if self[2].token_type != 'group-list':

return []

return self[2].all_mailboxes

@property

def display_name(self):

return self[0].display_name

class NameAddr(TokenList):

token_type = 'name-addr'

@property

def display_name(self):

if len(self) == 1:

return None

return self[0].display_name

@property

def local_part(self):

return self[-1].local_part

@property

def domain(self):

return self[-1].domain

@property

def route(self):

return self[-1].route

@property

def addr_spec(self):

return self[-1].addr_spec

class AngleAddr(TokenList):

token_type = 'angle-addr'

@property

def local_part(self):

for x in self:

if x.token_type == 'addr-spec':

return x.local_part

@property

def domain(self):

for x in self:

if x.token_type == 'addr-spec':

return x.domain

@property

def route(self):

for x in self:

if x.token_type == 'obs-route':

return x.domains

@property

def addr_spec(self):

for x in self:

if x.token_type == 'addr-spec':

if x.local_part:

return x.addr_spec

else:

return quote_string(x.local_part) + x.addr_spec

else:

return '<>'

class ObsRoute(TokenList):

token_type = 'obs-route'

@property

def domains(self):

return [x.domain for x in self if x.token_type == 'domain']

class Mailbox(TokenList):

token_type = 'mailbox'

@property

def display_name(self):

if self[0].token_type == 'name-addr':

return self[0].display_name

@property

def local_part(self):

return self[0].local_part

@property

def domain(self):

return self[0].domain

@property

def route(self):

if self[0].token_type == 'name-addr':

return self[0].route

@property

def addr_spec(self):

return self[0].addr_spec

class InvalidMailbox(TokenList):

token_type = 'invalid-mailbox'

@property

def display_name(self):

return None

local_part = domain = route = addr_spec = display_name

class Domain(TokenList):

token_type = 'domain'

as_ew_allowed = False

@property

def domain(self):

return ''.join(super().value.split())

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):

return self[0].local_part

@property

def domain(self):

if len(self) < 3:

return None

return self[-1].domain

@property

def value(self):

if len(self) < 3:

return self[0].value

return self[0].value.rstrip()+self[1].value+self[2].value.lstrip()

@property

def addr_spec(self):

nameset = set(self.local_part)

if len(nameset) > len(nameset-DOT_ATOM_ENDS):

lp = quote_string(self.local_part)

else:

lp = self.local_part

if self.domain is not None:

return lp + '@' + self.domain

return lp

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 (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 (isinstance(res[-1], TokenList) and

res[-1][-1].token_type == 'cfws'):

res[-1] = TokenList(res[-1][:-1])

return res.value

@property

def value(self):

quote = False

if self.defects:

quote = True

else:

for x in self:

if x.token_type == 'quoted-string':

quote = True

if len(self) != 0 and quote:

pre = post = ''

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

isinstance(self[-1], TokenList) and

self[-1][-1].token_type == 'cfws'):

post = ' '

return pre+quote_string(self.display_name)+post

else:

return super().value

class LocalPart(TokenList):

token_type = 'local-part'

as_ew_allowed = False

@property

def value(self):

if self[0].token_type == "quoted-string":

return self[0].quoted_value

else:

return self[0].value

@property

def local_part(self):

# Strip whitespace from front, back, and around dots.

res = [DOT]

last = DOT

last_is_tl = False

for tok in self[0] + [DOT]:

if tok.token_type == 'cfws':

continue

if (last_is_tl and tok.token_type == 'dot' and

last[-1].token_type == 'cfws'):

res[-1] = TokenList(last[:-1])

is_tl = isinstance(tok, TokenList)

if (is_tl and last.token_type == 'dot' and

tok[0].token_type == 'cfws'):

res.append(TokenList(tok[1:]))

else:

res.append(tok)

last = res[-1]

last_is_tl = is_tl

res = TokenList(res[1:-1])

return res.value

class DomainLiteral(TokenList):

token_type = 'domain-literal'

as_ew_allowed = False

@property

def domain(self):

return ''.join(super().value.split())

@property

def ip(self):

for x in self:

if x.token_type == 'ptext':

return x.value

class MIMEVersion(TokenList):

token_type = 'mime-version'

major = None

minor = None

class Parameter(TokenList):

token_type = 'parameter'

sectioned = False

extended = False

charset = 'us-ascii'

@property

def section_number(self):

# Because the first token, the attribute (name) eats CFWS, the second

# token is always the section if there is one.

return self[1].number if self.sectioned else 0

@property

def param_value(self):

# This is part of the "handle quoted extended parameters" hack.

for token in self:

if token.token_type == 'value':

return token.stripped_value

if token.token_type == 'quoted-string':

for token in token:

if token.token_type == 'bare-quoted-string':

for token in token:

if token.token_type == 'value':

return token.stripped_value

return ''

class InvalidParameter(Parameter):

token_type = 'invalid-parameter'

class Attribute(TokenList):

token_type = 'attribute'

@property

def stripped_value(self):

for token in self:

if token.token_type.endswith('attrtext'):

return token.value

class Section(TokenList):

token_type = 'section'

number = None

class Value(TokenList):

token_type = 'value'

@property

def stripped_value(self):

token = self[0]

if token.token_type == 'cfws':

token = self[1]

if token.token_type.endswith(

('quoted-string', 'attribute', 'extended-attribute')):

return token.stripped_value

return self.value

class MimeParameters(TokenList):

token_type = 'mime-parameters'

syntactic_break = False

@property

def params(self):

# The RFC specifically states that the ordering of parameters is not

# guaranteed and may be reordered by the transport layer. So we have

# 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 = {} # Using order preserving dict from Python 3.7+

for token in self:

if not token.token_type.endswith('parameter'):

continue

if token[0].token_type != 'attribute':

continue

name = token[0].value.strip()

if name not in params:

params[name] = []

params[name].append((token.section_number, token))

for name, parts in params.items():

parts = sorted(parts, key=itemgetter(0))

first_param = parts[0][1]

charset = first_param.charset

# Our arbitrary error recovery is to ignore duplicate parameters,

# to use appearance order if there are duplicate rfc 2231 parts,

# and to ignore gaps. This mimics the error recovery of get_param.

if not first_param.extended and len(parts) > 1:

if parts[1][0] == 0:

parts[1][1].defects.append(errors.InvalidHeaderDefect(

'duplicate parameter name; duplicate(s) ignored'))

parts = parts[:1]

# Else assume the *0* was missing...note that this is different

# from get_param, but we registered a defect for this earlier.

value_parts = []

i = 0

for section_number, param in parts:

if section_number != i:

# We could get fancier here and look for a complete

# duplicate extended parameter and ignore the second one

# seen. But we're not doing that. The old code didn't.

if not param.extended:

param.defects.append(errors.InvalidHeaderDefect(

'duplicate parameter name; duplicate ignored'))

continue

else:

param.defects.append(errors.InvalidHeaderDefect(

"inconsistent RFC2231 parameter numbering"))

i += 1

value = param.param_value

if param.extended:

try:

value = urllib.parse.unquote_to_bytes(value)

except UnicodeEncodeError:

# source had surrogate escaped bytes. What we do now

# is a bit of an open question. I'm not sure this is

# the best choice, but it is what the old algorithm did

value = urllib.parse.unquote(value, encoding='latin-1')

else:

try:

value = value.decode(charset, 'surrogateescape')

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

# failure.

value = value.decode('us-ascii', 'surrogateescape')

if utils._has_surrogates(value):

param.defects.append(errors.UndecodableBytesDefect())

value_parts.append(value)

value = ''.join(value_parts)

yield name, value

def __str__(self):

params = []

for name, value in self.params:

if value:

params.append('{}={}'.format(name, quote_string(value)))

else:

params.append(name)

params = '; '.join(params)

return ' ' + params if params else ''

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):

if token.token_type == 'mime-parameters':

return token.params

return {}

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'

#

# Terminal classes and instances

#

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

self.defects = []

return self

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)

def _pp(self, indent=''):

return ["{}{}/{}({}){}".format(

indent,

self.__class__.__name__,

self.token_type,

super().__repr__(),

'' if not self.defects else ' {}'.format(self.defects),

)]

def pop_trailing_ws(self):

# This terminates the recursion.

return None

@property

def comments(self):

return []

def __getnewargs__(self):

return(str(self), self.token_type)

class WhiteSpaceTerminal(Terminal):

@property

def value(self):

return ' '

def startswith_fws(self):

return self and self[0] in WSP

class ValueTerminal(Terminal):

@property

def value(self):

return self

def startswith_fws(self):

return False

class EWWhiteSpaceTerminal(WhiteSpaceTerminal):

@property

def value(self):

return ''

def __str__(self):

return ''

class _InvalidEwError(errors.HeaderParseError):

"""Invalid encoded word found while parsing headers."""

# XXX these need to become classes and used as instances so

# that a program can't change them in a parse tree and screw

# 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')

#

# Parser

#

# Parse strings according to RFC822/2047/2822/5322 rules.

#

# This is a stateless parser. Each get_XXX function accepts a string and

# returns either a Terminal or a TokenList representing the RFC object named

# by the method and a string containing the remaining unparsed characters

# from the input. Thus a parser method consumes the next syntactic construct

# of a given type and returns a token representing the construct plus the

# unparsed remainder of the input string.

#

# For example, if the first element of a structured header is a 'phrase',

# then:

#

# phrase, value = get_phrase(value)

#

# returns the complete phrase from the start of the string value, plus any

# characters left in the string after the phrase is removed.

_wsp_splitter = re.compile(r'([{}]+)'.format(''.join(WSP))).split

_non_atom_end_matcher = re.compile(r"[^{}]+".format(