Spaces:
Running
Running
| # Copyright (C) 2002-2007 Python Software Foundation | |
| # Author: Ben Gertzfield, Barry Warsaw | |
| # Contact: [email protected] | |
| """Header encoding and decoding functionality.""" | |
| __all__ = [ | |
| 'Header', | |
| 'decode_header', | |
| 'make_header', | |
| ] | |
| import re | |
| import binascii | |
| import email.quoprimime | |
| import email.base64mime | |
| from email.errors import HeaderParseError | |
| from email import charset as _charset | |
| Charset = _charset.Charset | |
| NL = '\n' | |
| SPACE = ' ' | |
| BSPACE = b' ' | |
| SPACE8 = ' ' * 8 | |
| EMPTYSTRING = '' | |
| MAXLINELEN = 78 | |
| FWS = ' \t' | |
| USASCII = Charset('us-ascii') | |
| UTF8 = Charset('utf-8') | |
| # Match encoded-word strings in the form =?charset?q?Hello_World?= | |
| ecre = re.compile(r''' | |
| =\? # literal =? | |
| (?P<charset>[^?]*?) # non-greedy up to the next ? is the charset | |
| \? # literal ? | |
| (?P<encoding>[qQbB]) # either a "q" or a "b", case insensitive | |
| \? # literal ? | |
| (?P<encoded>.*?) # non-greedy up to the next ?= is the encoded string | |
| \?= # literal ?= | |
| ''', 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. | |
| # For use with .match() | |
| fcre = re.compile(r'[\041-\176]+:$') | |
| # Find a header embedded in a putative header value. Used to check for | |
| # header injection attack. | |
| _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. | |
| Returns a list of (string, charset) pairs containing each of the decoded | |
| parts of the header. Charset is None for non-encoded parts of the header, | |
| otherwise a lower-case string containing the name of the character set | |
| specified in the encoded string. | |
| header may be a string that may or may not contain RFC2047 encoded words, | |
| or it may be a Header object. | |
| An email.errors.HeaderParseError may be raised when certain decoding error | |
| occurs (e.g. a base64 decoding exception). | |
| """ | |
| # If it is a Header object, we can just return the encoded chunks. | |
| if hasattr(header, '_chunks'): | |
| return [(_charset._encode(string, str(charset)), str(charset)) | |
| for string, charset in header._chunks] | |
| # If no encoding, just return the header with no charset. | |
| if not ecre.search(header): | |
| return [(header, None)] | |
| # First step is to parse all the encoded parts into triplets of the form | |
| # (encoded_string, encoding, charset). For unencoded strings, the last | |
| # two parts will be None. | |
| words = [] | |
| for line in header.splitlines(): | |
| parts = ecre.split(line) | |
| first = True | |
| while parts: | |
| unencoded = parts.pop(0) | |
| if first: | |
| unencoded = unencoded.lstrip() | |
| first = False | |
| if unencoded: | |
| words.append((unencoded, None, None)) | |
| if parts: | |
| charset = parts.pop(0).lower() | |
| encoding = parts.pop(0).lower() | |
| encoded = parts.pop(0) | |
| words.append((encoded, encoding, charset)) | |
| # Now loop over words and remove words that consist of whitespace | |
| # between two encoded strings. | |
| droplist = [] | |
| for n, w in enumerate(words): | |
| if n>1 and w[1] and words[n-2][1] and words[n-1][0].isspace(): | |
| droplist.append(n-1) | |
| for d in reversed(droplist): | |
| del words[d] | |
| # The next step is to decode each encoded word by applying the reverse | |
| # base64 or quopri transformation. decoded_words is now a list of the | |
| # form (decoded_word, charset). | |
| decoded_words = [] | |
| for encoded_string, encoding, charset in words: | |
| if encoding is None: | |
| # This is an unencoded word. | |
| decoded_words.append((encoded_string, charset)) | |
| elif encoding == 'q': | |
| word = email.quoprimime.header_decode(encoded_string) | |
| decoded_words.append((word, charset)) | |
| elif encoding == 'b': | |
| paderr = len(encoded_string) % 4 # Postel's law: add missing padding | |
| if paderr: | |
| encoded_string += '==='[:4 - paderr] | |
| try: | |
| word = email.base64mime.decode(encoded_string) | |
| except binascii.Error: | |
| raise HeaderParseError('Base64 decoding error') | |
| else: | |
| decoded_words.append((word, charset)) | |
| else: | |
| raise AssertionError('Unexpected encoding: ' + encoding) | |
| # Now convert all words to bytes and collapse consecutive runs of | |
| # similarly encoded words. | |
| collapsed = [] | |
| last_word = last_charset = None | |
| for word, charset in decoded_words: | |
| if isinstance(word, str): | |
| word = bytes(word, 'raw-unicode-escape') | |
| if last_word is None: | |
| last_word = word | |
| last_charset = charset | |
| elif charset != last_charset: | |
| collapsed.append((last_word, last_charset)) | |
| last_word = word | |
| last_charset = charset | |
| elif last_charset is None: | |
| last_word += BSPACE + word | |
| else: | |
| last_word += word | |
| collapsed.append((last_word, last_charset)) | |
| return collapsed | |
| def make_header(decoded_seq, maxlinelen=None, header_name=None, | |
| continuation_ws=' '): | |
| """Create a Header from a sequence of pairs as returned by decode_header() | |
| decode_header() takes a header value string and returns a sequence of | |
| pairs of the format (decoded_string, charset) where charset is the string | |
| name of the character set. | |
| This function takes one of those sequence of pairs and returns a Header | |
| instance. Optional maxlinelen, header_name, and continuation_ws are as in | |
| the Header constructor. | |
| """ | |
| h = Header(maxlinelen=maxlinelen, header_name=header_name, | |
| continuation_ws=continuation_ws) | |
| for s, charset in decoded_seq: | |
| # None means us-ascii but we can simply pass it on to h.append() | |
| if charset is not None and not isinstance(charset, Charset): | |
| charset = Charset(charset) | |
| h.append(s, charset) | |
| return h | |
| class Header: | |
| def __init__(self, s=None, charset=None, | |
| maxlinelen=None, header_name=None, | |
| continuation_ws=' ', errors='strict'): | |
| """Create a MIME-compliant header that can contain many character sets. | |
| Optional s is the initial header value. If None, the initial header | |
| value is not set. You can later append to the header with .append() | |
| method calls. s may be a byte string or a Unicode string, but see the | |
| .append() documentation for semantics. | |
| Optional charset serves two purposes: it has the same meaning as the | |
| charset argument to the .append() method. It also sets the default | |
| character set for all subsequent .append() calls that omit the charset | |
| argument. If charset is not provided in the constructor, the us-ascii | |
| charset is used both as s's initial charset and as the default for | |
| subsequent .append() calls. | |
| The maximum line length can be specified explicitly via maxlinelen. For | |
| splitting the first line to a shorter value (to account for the field | |
| header which isn't included in s, e.g. `Subject') pass in the name of | |
| the field in header_name. The default maxlinelen is 78 as recommended | |
| by RFC 2822. | |
| continuation_ws must be RFC 2822 compliant folding whitespace (usually | |
| either a space or a hard tab) which will be prepended to continuation | |
| lines. | |
| errors is passed through to the .append() call. | |
| """ | |
| if charset is None: | |
| charset = USASCII | |
| elif not isinstance(charset, Charset): | |
| charset = Charset(charset) | |
| self._charset = charset | |
| self._continuation_ws = continuation_ws | |
| self._chunks = [] | |
| if s is not None: | |
| self.append(s, charset, errors) | |
| if maxlinelen is None: | |
| maxlinelen = MAXLINELEN | |
| self._maxlinelen = maxlinelen | |
| if header_name is None: | |
| self._headerlen = 0 | |
| else: | |
| # Take the separating colon and space into account. | |
| self._headerlen = len(header_name) + 2 | |
| def __str__(self): | |
| """Return the string value of the header.""" | |
| self._normalize() | |
| uchunks = [] | |
| lastcs = None | |
| lastspace = None | |
| for string, charset in self._chunks: | |
| # We must preserve spaces between encoded and non-encoded word | |
| # boundaries, which means for us we need to add a space when we go | |
| # from a charset to None/us-ascii, or from None/us-ascii to a | |
| # charset. Only do this for the second and subsequent chunks. | |
| # Don't add a space if the None/us-ascii string already has | |
| # a space (trailing or leading depending on transition) | |
| nextcs = charset | |
| if nextcs == _charset.UNKNOWN8BIT: | |
| original_bytes = string.encode('ascii', 'surrogateescape') | |
| string = original_bytes.decode('ascii', 'replace') | |
| if uchunks: | |
| hasspace = string and self._nonctext(string[0]) | |
| if lastcs not in (None, 'us-ascii'): | |
| if nextcs in (None, 'us-ascii') and not hasspace: | |
| uchunks.append(SPACE) | |
| nextcs = None | |
| elif nextcs not in (None, 'us-ascii') and not lastspace: | |
| uchunks.append(SPACE) | |
| lastspace = string and self._nonctext(string[-1]) | |
| lastcs = nextcs | |
| uchunks.append(string) | |
| return EMPTYSTRING.join(uchunks) | |
| # Rich comparison operators for equality only. BAW: does it make sense to | |
| # have or explicitly disable <, <=, >, >= operators? | |
| def __eq__(self, other): | |
| # other may be a Header or a string. Both are fine so coerce | |
| # ourselves to a unicode (of the unencoded header value), swap the | |
| # args and do another comparison. | |
| return other == str(self) | |
| def append(self, s, charset=None, errors='strict'): | |
| """Append a string to the MIME header. | |
| Optional charset, if given, should be a Charset instance or the name | |
| of a character set (which will be converted to a Charset instance). A | |
| value of None (the default) means that the charset given in the | |
| constructor is used. | |
| s may be a byte string or a Unicode string. If it is a byte string | |
| (i.e. isinstance(s, str) is false), then charset is the encoding of | |
| that byte string, and a UnicodeError will be raised if the string | |
| cannot be decoded with that charset. If s is a Unicode string, then | |
| charset is a hint specifying the character set of the characters in | |
| the string. In either case, when producing an RFC 2822 compliant | |
| header using RFC 2047 rules, the string will be encoded using the | |
| output codec of the charset. If the string cannot be encoded to the | |
| output codec, a UnicodeError will be raised. | |
| Optional `errors' is passed as the errors argument to the decode | |
| call if s is a byte string. | |
| """ | |
| if charset is None: | |
| charset = self._charset | |
| elif not isinstance(charset, Charset): | |
| charset = Charset(charset) | |
| if not isinstance(s, str): | |
| input_charset = charset.input_codec or 'us-ascii' | |
| if input_charset == _charset.UNKNOWN8BIT: | |
| s = s.decode('us-ascii', 'surrogateescape') | |
| else: | |
| s = s.decode(input_charset, errors) | |
| # Ensure that the bytes we're storing can be decoded to the output | |
| # character set, otherwise an early error is raised. | |
| output_charset = charset.output_codec or 'us-ascii' | |
| if output_charset != _charset.UNKNOWN8BIT: | |
| try: | |
| s.encode(output_charset, errors) | |
| except UnicodeEncodeError: | |
| if output_charset!='us-ascii': | |
| raise | |
| charset = UTF8 | |
| self._chunks.append((s, charset)) | |
| def _nonctext(self, s): | |
| """True if string s is not a ctext character of RFC822. | |
| """ | |
| return s.isspace() or s in ('(', ')', '\\') | |
| def encode(self, splitchars=';, \t', maxlinelen=None, linesep='\n'): | |
| r"""Encode a message header into an RFC-compliant format. | |
| There are many issues involved in converting a given string for use in | |
| an email header. Only certain character sets are readable in most | |
| email clients, and as header strings can only contain a subset of | |
| 7-bit ASCII, care must be taken to properly convert and encode (with | |
| Base64 or quoted-printable) header strings. In addition, there is a | |
| 75-character length limit on any given encoded header field, so | |
| line-wrapping must be performed, even with double-byte character sets. | |
| Optional maxlinelen specifies the maximum length of each generated | |
| line, exclusive of the linesep string. Individual lines may be longer | |
| than maxlinelen if a folding point cannot be found. The first line | |
| will be shorter by the length of the header name plus ": " if a header | |
| name was specified at Header construction time. The default value for | |
| maxlinelen is determined at header construction time. | |
| Optional splitchars is a string containing characters which should be | |
| given extra weight by the splitting algorithm during normal header | |
| wrapping. This is in very rough support of RFC 2822's `higher level | |
| syntactic breaks': split points preceded by a splitchar are preferred | |
| during line splitting, with the characters preferred in the order in | |
| which they appear in the string. Space and tab may be included in the | |
| string to indicate whether preference should be given to one over the | |
| other as a split point when other split chars do not appear in the line | |
| being split. Splitchars does not affect RFC 2047 encoded lines. | |
| Optional linesep is a string to be used to separate the lines of | |
| the value. The default value is the most useful for typical | |
| Python applications, but it can be set to \r\n to produce RFC-compliant | |
| line separators when needed. | |
| """ | |
| self._normalize() | |
| if maxlinelen is None: | |
| maxlinelen = self._maxlinelen | |
| # A maxlinelen of 0 means don't wrap. For all practical purposes, | |
| # choosing a huge number here accomplishes that and makes the | |
| # _ValueFormatter algorithm much simpler. | |
| if maxlinelen == 0: | |
| maxlinelen = 1000000 | |
| formatter = _ValueFormatter(self._headerlen, maxlinelen, | |
| self._continuation_ws, splitchars) | |
| lastcs = None | |
| hasspace = lastspace = None | |
| for string, charset in self._chunks: | |
| if hasspace is not None: | |
| hasspace = string and self._nonctext(string[0]) | |
| if lastcs not in (None, 'us-ascii'): | |
| if not hasspace or charset not in (None, 'us-ascii'): | |
| formatter.add_transition() | |
| elif charset not in (None, 'us-ascii') and not lastspace: | |
| formatter.add_transition() | |
| lastspace = string and self._nonctext(string[-1]) | |
| lastcs = charset | |
| hasspace = False | |
| lines = string.splitlines() | |
| if lines: | |
| formatter.feed('', lines[0], charset) | |
| else: | |
| formatter.feed('', '', charset) | |
| for line in lines[1:]: | |
| formatter.newline() | |
| if charset.header_encoding is not None: | |
| formatter.feed(self._continuation_ws, ' ' + line.lstrip(), | |
| charset) | |
| else: | |
| sline = line.lstrip() | |
| fws = line[:len(line)-len(sline)] | |
| formatter.feed(fws, sline, charset) | |
| if len(lines) > 1: | |
| formatter.newline() | |
| if self._chunks: | |
| formatter.add_transition() | |
| value = formatter._str(linesep) | |
| if _embedded_header.search(value): | |
| raise HeaderParseError("header value appears to contain " | |
| "an embedded header: {!r}".format(value)) | |
| return value | |
| def _normalize(self): | |
| # Step 1: Normalize the chunks so that all runs of identical charsets | |
| # get collapsed into a single unicode string. | |
| chunks = [] | |
| last_charset = None | |
| last_chunk = [] | |
| for string, charset in self._chunks: | |
| if charset == last_charset: | |
| last_chunk.append(string) | |
| else: | |
| if last_charset is not None: | |
| chunks.append((SPACE.join(last_chunk), last_charset)) | |
| last_chunk = [string] | |
| last_charset = charset | |
| if last_chunk: | |
| chunks.append((SPACE.join(last_chunk), last_charset)) | |
| self._chunks = chunks | |
| class _ValueFormatter: | |
| def __init__(self, headerlen, maxlen, continuation_ws, splitchars): | |
| self._maxlen = maxlen | |
| self._continuation_ws = continuation_ws | |
| self._continuation_ws_len = len(continuation_ws) | |
| self._splitchars = splitchars | |
| self._lines = [] | |
| self._current_line = _Accumulator(headerlen) | |
| def _str(self, linesep): | |
| self.newline() | |
| return linesep.join(self._lines) | |
| def __str__(self): | |
| return self._str(NL) | |
| def newline(self): | |
| end_of_line = self._current_line.pop() | |
| if end_of_line != (' ', ''): | |
| self._current_line.push(*end_of_line) | |
| if len(self._current_line) > 0: | |
| if self._current_line.is_onlyws() and self._lines: | |
| self._lines[-1] += str(self._current_line) | |
| else: | |
| self._lines.append(str(self._current_line)) | |
| self._current_line.reset() | |
| def add_transition(self): | |
| self._current_line.push(' ', '') | |
| def feed(self, fws, string, charset): | |
| # If the charset has no header encoding (i.e. it is an ASCII encoding) | |
| # then we must split the header at the "highest level syntactic break" | |
| # possible. Note that we don't have a lot of smarts about field | |
| # syntax; we just try to break on semi-colons, then commas, then | |
| # whitespace. Eventually, this should be pluggable. | |
| if charset.header_encoding is None: | |
| self._ascii_split(fws, string, self._splitchars) | |
| return | |
| # Otherwise, we're doing either a Base64 or a quoted-printable | |
| # encoding which means we don't need to split the line on syntactic | |
| # breaks. We can basically just find enough characters to fit on the | |
| # current line, minus the RFC 2047 chrome. What makes this trickier | |
| # though is that we have to split at octet boundaries, not character | |
| # boundaries but it's only safe to split at character boundaries so at | |
| # best we can only get close. | |
| encoded_lines = charset.header_encode_lines(string, self._maxlengths()) | |
| # The first element extends the current line, but if it's None then | |
| # nothing more fit on the current line so start a new line. | |
| try: | |
| first_line = encoded_lines.pop(0) | |
| except IndexError: | |
| # There are no encoded lines, so we're done. | |
| return | |
| if first_line is not None: | |
| self._append_chunk(fws, first_line) | |
| try: | |
| last_line = encoded_lines.pop() | |
| except IndexError: | |
| # There was only one line. | |
| return | |
| self.newline() | |
| self._current_line.push(self._continuation_ws, last_line) | |
| # Everything else are full lines in themselves. | |
| for line in encoded_lines: | |
| self._lines.append(self._continuation_ws + line) | |
| def _maxlengths(self): | |
| # The first line's length. | |
| yield self._maxlen - len(self._current_line) | |
| while True: | |
| yield self._maxlen - self._continuation_ws_len | |
| def _ascii_split(self, fws, string, splitchars): | |
| # The RFC 2822 header folding algorithm is simple in principle but | |
| # complex in practice. Lines may be folded any place where "folding | |
| # white space" appears by inserting a linesep character in front of the | |
| # FWS. The complication is that not all spaces or tabs qualify as FWS, | |
| # and we are also supposed to prefer to break at "higher level | |
| # syntactic breaks". We can't do either of these without intimate | |
| # knowledge of the structure of structured headers, which we don't have | |
| # here. So the best we can do here is prefer to break at the specified | |
| # splitchars, and hope that we don't choose any spaces or tabs that | |
| # aren't legal FWS. (This is at least better than the old algorithm, | |
| # where we would sometimes *introduce* FWS after a splitchar, or the | |
| # algorithm before that, where we would turn all white space runs into | |
| # single spaces or tabs.) | |
| parts = re.split("(["+FWS+"]+)", fws+string) | |
| if parts[0]: | |
| parts[:0] = [''] | |
| else: | |
| parts.pop(0) | |
| for fws, part in zip(*[iter(parts)]*2): | |
| self._append_chunk(fws, part) | |
| def _append_chunk(self, fws, string): | |
| self._current_line.push(fws, string) | |
| if len(self._current_line) > self._maxlen: | |
| # Find the best split point, working backward from the end. | |
| # There might be none, on a long first line. | |
| for ch in self._splitchars: | |
| for i in range(self._current_line.part_count()-1, 0, -1): | |
| if ch.isspace(): | |
| fws = self._current_line[i][0] | |
| if fws and fws[0]==ch: | |
| break | |
| prevpart = self._current_line[i-1][1] | |
| if prevpart and prevpart[-1]==ch: | |
| break | |
| else: | |
| continue | |
| break | |
| else: | |
| fws, part = self._current_line.pop() | |
| if self._current_line._initial_size > 0: | |
| # There will be a header, so leave it on a line by itself. | |
| self.newline() | |
| if not fws: | |
| # We don't use continuation_ws here because the whitespace | |
| # after a header should always be a space. | |
| fws = ' ' | |
| self._current_line.push(fws, part) | |
| return | |
| remainder = self._current_line.pop_from(i) | |
| self._lines.append(str(self._current_line)) | |
| self._current_line.reset(remainder) | |
| class _Accumulator(list): | |
| def __init__(self, initial_size=0): | |
| self._initial_size = initial_size | |
| super().__init__() | |
| def push(self, fws, string): | |
| self.append((fws, string)) | |
| def pop_from(self, i=0): | |
| popped = self[i:] | |
| self[i:] = [] | |
| return popped | |
| def pop(self): | |
| if self.part_count()==0: | |
| return ('', '') | |
| return super().pop() | |
| def __len__(self): | |
| return sum((len(fws)+len(part) for fws, part in self), | |
| self._initial_size) | |
| def __str__(self): | |
| return EMPTYSTRING.join((EMPTYSTRING.join((fws, part)) | |
| for fws, part in self)) | |
| def reset(self, startval=None): | |
| if startval is None: | |
| startval = [] | |
| self[:] = startval | |
| self._initial_size = 0 | |
| def is_onlyws(self): | |
| return self._initial_size==0 and (not self or str(self).isspace()) | |
| def part_count(self): | |
| return super().__len__() | |