Mercurial > repos > rliterman > csp2

annotate CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/email/header.py @ 69:33d812a61356

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit 2e9511a184a1ca667c7be0c6321a36dc4e3d116d
author jpayne
date Tue, 18 Mar 2025 17:55:14 -0400
parents
children
rev   line source
jpayne@69 1 # Copyright (C) 2002-2007 Python Software Foundation
jpayne@69 2 # Author: Ben Gertzfield, Barry Warsaw
jpayne@69 3 # Contact: email-sig@python.org
jpayne@69 4
jpayne@69 5 """Header encoding and decoding functionality."""
jpayne@69 6
jpayne@69 7 __all__ = [
jpayne@69 8 'Header',
jpayne@69 9 'decode_header',
jpayne@69 10 'make_header',
jpayne@69 11 ]
jpayne@69 12
jpayne@69 13 import re
jpayne@69 14 import binascii
jpayne@69 15
jpayne@69 16 import email.quoprimime
jpayne@69 17 import email.base64mime
jpayne@69 18
jpayne@69 19 from email.errors import HeaderParseError
jpayne@69 20 from email import charset as _charset
jpayne@69 21 Charset = _charset.Charset
jpayne@69 22
jpayne@69 23 NL = '\n'
jpayne@69 24 SPACE = ' '
jpayne@69 25 BSPACE = b' '
jpayne@69 26 SPACE8 = ' ' * 8
jpayne@69 27 EMPTYSTRING = ''
jpayne@69 28 MAXLINELEN = 78
jpayne@69 29 FWS = ' \t'
jpayne@69 30
jpayne@69 31 USASCII = Charset('us-ascii')
jpayne@69 32 UTF8 = Charset('utf-8')
jpayne@69 33
jpayne@69 34 # Match encoded-word strings in the form =?charset?q?Hello_World?=
jpayne@69 35 ecre = re.compile(r'''
jpayne@69 36 =\? # literal =?
jpayne@69 37 (?P<charset>[^?]*?) # non-greedy up to the next ? is the charset
jpayne@69 38 \? # literal ?
jpayne@69 39 (?P<encoding>[qQbB]) # either a "q" or a "b", case insensitive
jpayne@69 40 \? # literal ?
jpayne@69 41 (?P<encoded>.*?) # non-greedy up to the next ?= is the encoded string
jpayne@69 42 \?= # literal ?=
jpayne@69 43 ''', re.VERBOSE | re.MULTILINE)
jpayne@69 44
jpayne@69 45 # Field name regexp, including trailing colon, but not separating whitespace,
jpayne@69 46 # according to RFC 2822. Character range is from tilde to exclamation mark.
jpayne@69 47 # For use with .match()
jpayne@69 48 fcre = re.compile(r'[\041-\176]+:$')
jpayne@69 49
jpayne@69 50 # Find a header embedded in a putative header value. Used to check for
jpayne@69 51 # header injection attack.
jpayne@69 52 _embedded_header = re.compile(r'\n[^ \t]+:')
jpayne@69 53
jpayne@69 54
jpayne@69 55
jpayne@69 56 # Helpers
jpayne@69 57 _max_append = email.quoprimime._max_append
jpayne@69 58
jpayne@69 59
jpayne@69 60
jpayne@69 61 def decode_header(header):
jpayne@69 62 """Decode a message header value without converting charset.
jpayne@69 63
jpayne@69 64 Returns a list of (string, charset) pairs containing each of the decoded
jpayne@69 65 parts of the header. Charset is None for non-encoded parts of the header,
jpayne@69 66 otherwise a lower-case string containing the name of the character set
jpayne@69 67 specified in the encoded string.
jpayne@69 68
jpayne@69 69 header may be a string that may or may not contain RFC2047 encoded words,
jpayne@69 70 or it may be a Header object.
jpayne@69 71
jpayne@69 72 An email.errors.HeaderParseError may be raised when certain decoding error
jpayne@69 73 occurs (e.g. a base64 decoding exception).
jpayne@69 74 """
jpayne@69 75 # If it is a Header object, we can just return the encoded chunks.
jpayne@69 76 if hasattr(header, '_chunks'):
jpayne@69 77 return [(_charset._encode(string, str(charset)), str(charset))
jpayne@69 78 for string, charset in header._chunks]
jpayne@69 79 # If no encoding, just return the header with no charset.
jpayne@69 80 if not ecre.search(header):
jpayne@69 81 return [(header, None)]
jpayne@69 82 # First step is to parse all the encoded parts into triplets of the form
jpayne@69 83 # (encoded_string, encoding, charset). For unencoded strings, the last
jpayne@69 84 # two parts will be None.
jpayne@69 85 words = []
jpayne@69 86 for line in header.splitlines():
jpayne@69 87 parts = ecre.split(line)
jpayne@69 88 first = True
jpayne@69 89 while parts:
jpayne@69 90 unencoded = parts.pop(0)
jpayne@69 91 if first:
jpayne@69 92 unencoded = unencoded.lstrip()
jpayne@69 93 first = False
jpayne@69 94 if unencoded:
jpayne@69 95 words.append((unencoded, None, None))
jpayne@69 96 if parts:
jpayne@69 97 charset = parts.pop(0).lower()
jpayne@69 98 encoding = parts.pop(0).lower()
jpayne@69 99 encoded = parts.pop(0)
jpayne@69 100 words.append((encoded, encoding, charset))
jpayne@69 101 # Now loop over words and remove words that consist of whitespace
jpayne@69 102 # between two encoded strings.
jpayne@69 103 droplist = []
jpayne@69 104 for n, w in enumerate(words):
jpayne@69 105 if n>1 and w[1] and words[n-2][1] and words[n-1][0].isspace():
jpayne@69 106 droplist.append(n-1)
jpayne@69 107 for d in reversed(droplist):
jpayne@69 108 del words[d]
jpayne@69 109
jpayne@69 110 # The next step is to decode each encoded word by applying the reverse
jpayne@69 111 # base64 or quopri transformation. decoded_words is now a list of the
jpayne@69 112 # form (decoded_word, charset).
jpayne@69 113 decoded_words = []
jpayne@69 114 for encoded_string, encoding, charset in words:
jpayne@69 115 if encoding is None:
jpayne@69 116 # This is an unencoded word.
jpayne@69 117 decoded_words.append((encoded_string, charset))
jpayne@69 118 elif encoding == 'q':
jpayne@69 119 word = email.quoprimime.header_decode(encoded_string)
jpayne@69 120 decoded_words.append((word, charset))
jpayne@69 121 elif encoding == 'b':
jpayne@69 122 paderr = len(encoded_string) % 4 # Postel's law: add missing padding
jpayne@69 123 if paderr:
jpayne@69 124 encoded_string += '==='[:4 - paderr]
jpayne@69 125 try:
jpayne@69 126 word = email.base64mime.decode(encoded_string)
jpayne@69 127 except binascii.Error:
jpayne@69 128 raise HeaderParseError('Base64 decoding error')
jpayne@69 129 else:
jpayne@69 130 decoded_words.append((word, charset))
jpayne@69 131 else:
jpayne@69 132 raise AssertionError('Unexpected encoding: ' + encoding)
jpayne@69 133 # Now convert all words to bytes and collapse consecutive runs of
jpayne@69 134 # similarly encoded words.
jpayne@69 135 collapsed = []
jpayne@69 136 last_word = last_charset = None
jpayne@69 137 for word, charset in decoded_words:
jpayne@69 138 if isinstance(word, str):
jpayne@69 139 word = bytes(word, 'raw-unicode-escape')
jpayne@69 140 if last_word is None:
jpayne@69 141 last_word = word
jpayne@69 142 last_charset = charset
jpayne@69 143 elif charset != last_charset:
jpayne@69 144 collapsed.append((last_word, last_charset))
jpayne@69 145 last_word = word
jpayne@69 146 last_charset = charset
jpayne@69 147 elif last_charset is None:
jpayne@69 148 last_word += BSPACE + word
jpayne@69 149 else:
jpayne@69 150 last_word += word
jpayne@69 151 collapsed.append((last_word, last_charset))
jpayne@69 152 return collapsed
jpayne@69 153
jpayne@69 154
jpayne@69 155
jpayne@69 156 def make_header(decoded_seq, maxlinelen=None, header_name=None,
jpayne@69 157 continuation_ws=' '):
jpayne@69 158 """Create a Header from a sequence of pairs as returned by decode_header()
jpayne@69 159
jpayne@69 160 decode_header() takes a header value string and returns a sequence of
jpayne@69 161 pairs of the format (decoded_string, charset) where charset is the string
jpayne@69 162 name of the character set.
jpayne@69 163
jpayne@69 164 This function takes one of those sequence of pairs and returns a Header
jpayne@69 165 instance. Optional maxlinelen, header_name, and continuation_ws are as in
jpayne@69 166 the Header constructor.
jpayne@69 167 """
jpayne@69 168 h = Header(maxlinelen=maxlinelen, header_name=header_name,
jpayne@69 169 continuation_ws=continuation_ws)
jpayne@69 170 for s, charset in decoded_seq:
jpayne@69 171 # None means us-ascii but we can simply pass it on to h.append()
jpayne@69 172 if charset is not None and not isinstance(charset, Charset):
jpayne@69 173 charset = Charset(charset)
jpayne@69 174 h.append(s, charset)
jpayne@69 175 return h
jpayne@69 176
jpayne@69 177
jpayne@69 178
jpayne@69 179 class Header:
jpayne@69 180 def __init__(self, s=None, charset=None,
jpayne@69 181 maxlinelen=None, header_name=None,
jpayne@69 182 continuation_ws=' ', errors='strict'):
jpayne@69 183 """Create a MIME-compliant header that can contain many character sets.
jpayne@69 184
jpayne@69 185 Optional s is the initial header value. If None, the initial header
jpayne@69 186 value is not set. You can later append to the header with .append()
jpayne@69 187 method calls. s may be a byte string or a Unicode string, but see the
jpayne@69 188 .append() documentation for semantics.
jpayne@69 189
jpayne@69 190 Optional charset serves two purposes: it has the same meaning as the
jpayne@69 191 charset argument to the .append() method. It also sets the default
jpayne@69 192 character set for all subsequent .append() calls that omit the charset
jpayne@69 193 argument. If charset is not provided in the constructor, the us-ascii
jpayne@69 194 charset is used both as s's initial charset and as the default for
jpayne@69 195 subsequent .append() calls.
jpayne@69 196
jpayne@69 197 The maximum line length can be specified explicitly via maxlinelen. For
jpayne@69 198 splitting the first line to a shorter value (to account for the field
jpayne@69 199 header which isn't included in s, e.g. `Subject') pass in the name of
jpayne@69 200 the field in header_name. The default maxlinelen is 78 as recommended
jpayne@69 201 by RFC 2822.
jpayne@69 202
jpayne@69 203 continuation_ws must be RFC 2822 compliant folding whitespace (usually
jpayne@69 204 either a space or a hard tab) which will be prepended to continuation
jpayne@69 205 lines.
jpayne@69 206
jpayne@69 207 errors is passed through to the .append() call.
jpayne@69 208 """
jpayne@69 209 if charset is None:
jpayne@69 210 charset = USASCII
jpayne@69 211 elif not isinstance(charset, Charset):
jpayne@69 212 charset = Charset(charset)
jpayne@69 213 self._charset = charset
jpayne@69 214 self._continuation_ws = continuation_ws
jpayne@69 215 self._chunks = []
jpayne@69 216 if s is not None:
jpayne@69 217 self.append(s, charset, errors)
jpayne@69 218 if maxlinelen is None:
jpayne@69 219 maxlinelen = MAXLINELEN
jpayne@69 220 self._maxlinelen = maxlinelen
jpayne@69 221 if header_name is None:
jpayne@69 222 self._headerlen = 0
jpayne@69 223 else:
jpayne@69 224 # Take the separating colon and space into account.
jpayne@69 225 self._headerlen = len(header_name) + 2
jpayne@69 226
jpayne@69 227 def __str__(self):
jpayne@69 228 """Return the string value of the header."""
jpayne@69 229 self._normalize()
jpayne@69 230 uchunks = []
jpayne@69 231 lastcs = None
jpayne@69 232 lastspace = None
jpayne@69 233 for string, charset in self._chunks:
jpayne@69 234 # We must preserve spaces between encoded and non-encoded word
jpayne@69 235 # boundaries, which means for us we need to add a space when we go
jpayne@69 236 # from a charset to None/us-ascii, or from None/us-ascii to a
jpayne@69 237 # charset. Only do this for the second and subsequent chunks.
jpayne@69 238 # Don't add a space if the None/us-ascii string already has
jpayne@69 239 # a space (trailing or leading depending on transition)
jpayne@69 240 nextcs = charset
jpayne@69 241 if nextcs == _charset.UNKNOWN8BIT:
jpayne@69 242 original_bytes = string.encode('ascii', 'surrogateescape')
jpayne@69 243 string = original_bytes.decode('ascii', 'replace')
jpayne@69 244 if uchunks:
jpayne@69 245 hasspace = string and self._nonctext(string[0])
jpayne@69 246 if lastcs not in (None, 'us-ascii'):
jpayne@69 247 if nextcs in (None, 'us-ascii') and not hasspace:
jpayne@69 248 uchunks.append(SPACE)
jpayne@69 249 nextcs = None
jpayne@69 250 elif nextcs not in (None, 'us-ascii') and not lastspace:
jpayne@69 251 uchunks.append(SPACE)
jpayne@69 252 lastspace = string and self._nonctext(string[-1])
jpayne@69 253 lastcs = nextcs
jpayne@69 254 uchunks.append(string)
jpayne@69 255 return EMPTYSTRING.join(uchunks)
jpayne@69 256
jpayne@69 257 # Rich comparison operators for equality only. BAW: does it make sense to
jpayne@69 258 # have or explicitly disable <, <=, >, >= operators?
jpayne@69 259 def __eq__(self, other):
jpayne@69 260 # other may be a Header or a string. Both are fine so coerce
jpayne@69 261 # ourselves to a unicode (of the unencoded header value), swap the
jpayne@69 262 # args and do another comparison.
jpayne@69 263 return other == str(self)
jpayne@69 264
jpayne@69 265 def append(self, s, charset=None, errors='strict'):
jpayne@69 266 """Append a string to the MIME header.
jpayne@69 267
jpayne@69 268 Optional charset, if given, should be a Charset instance or the name
jpayne@69 269 of a character set (which will be converted to a Charset instance). A
jpayne@69 270 value of None (the default) means that the charset given in the
jpayne@69 271 constructor is used.
jpayne@69 272
jpayne@69 273 s may be a byte string or a Unicode string. If it is a byte string
jpayne@69 274 (i.e. isinstance(s, str) is false), then charset is the encoding of
jpayne@69 275 that byte string, and a UnicodeError will be raised if the string
jpayne@69 276 cannot be decoded with that charset. If s is a Unicode string, then
jpayne@69 277 charset is a hint specifying the character set of the characters in
jpayne@69 278 the string. In either case, when producing an RFC 2822 compliant
jpayne@69 279 header using RFC 2047 rules, the string will be encoded using the
jpayne@69 280 output codec of the charset. If the string cannot be encoded to the
jpayne@69 281 output codec, a UnicodeError will be raised.
jpayne@69 282
jpayne@69 283 Optional `errors' is passed as the errors argument to the decode
jpayne@69 284 call if s is a byte string.
jpayne@69 285 """
jpayne@69 286 if charset is None:
jpayne@69 287 charset = self._charset
jpayne@69 288 elif not isinstance(charset, Charset):
jpayne@69 289 charset = Charset(charset)
jpayne@69 290 if not isinstance(s, str):
jpayne@69 291 input_charset = charset.input_codec or 'us-ascii'
jpayne@69 292 if input_charset == _charset.UNKNOWN8BIT:
jpayne@69 293 s = s.decode('us-ascii', 'surrogateescape')
jpayne@69 294 else:
jpayne@69 295 s = s.decode(input_charset, errors)
jpayne@69 296 # Ensure that the bytes we're storing can be decoded to the output
jpayne@69 297 # character set, otherwise an early error is raised.
jpayne@69 298 output_charset = charset.output_codec or 'us-ascii'
jpayne@69 299 if output_charset != _charset.UNKNOWN8BIT:
jpayne@69 300 try:
jpayne@69 301 s.encode(output_charset, errors)
jpayne@69 302 except UnicodeEncodeError:
jpayne@69 303 if output_charset!='us-ascii':
jpayne@69 304 raise
jpayne@69 305 charset = UTF8
jpayne@69 306 self._chunks.append((s, charset))
jpayne@69 307
jpayne@69 308 def _nonctext(self, s):
jpayne@69 309 """True if string s is not a ctext character of RFC822.
jpayne@69 310 """
jpayne@69 311 return s.isspace() or s in ('(', ')', '\\')
jpayne@69 312
jpayne@69 313 def encode(self, splitchars=';, \t', maxlinelen=None, linesep='\n'):
jpayne@69 314 r"""Encode a message header into an RFC-compliant format.
jpayne@69 315
jpayne@69 316 There are many issues involved in converting a given string for use in
jpayne@69 317 an email header. Only certain character sets are readable in most
jpayne@69 318 email clients, and as header strings can only contain a subset of
jpayne@69 319 7-bit ASCII, care must be taken to properly convert and encode (with
jpayne@69 320 Base64 or quoted-printable) header strings. In addition, there is a
jpayne@69 321 75-character length limit on any given encoded header field, so
jpayne@69 322 line-wrapping must be performed, even with double-byte character sets.
jpayne@69 323
jpayne@69 324 Optional maxlinelen specifies the maximum length of each generated
jpayne@69 325 line, exclusive of the linesep string. Individual lines may be longer
jpayne@69 326 than maxlinelen if a folding point cannot be found. The first line
jpayne@69 327 will be shorter by the length of the header name plus ": " if a header
jpayne@69 328 name was specified at Header construction time. The default value for
jpayne@69 329 maxlinelen is determined at header construction time.
jpayne@69 330
jpayne@69 331 Optional splitchars is a string containing characters which should be
jpayne@69 332 given extra weight by the splitting algorithm during normal header
jpayne@69 333 wrapping. This is in very rough support of RFC 2822's `higher level
jpayne@69 334 syntactic breaks': split points preceded by a splitchar are preferred
jpayne@69 335 during line splitting, with the characters preferred in the order in
jpayne@69 336 which they appear in the string. Space and tab may be included in the
jpayne@69 337 string to indicate whether preference should be given to one over the
jpayne@69 338 other as a split point when other split chars do not appear in the line
jpayne@69 339 being split. Splitchars does not affect RFC 2047 encoded lines.
jpayne@69 340
jpayne@69 341 Optional linesep is a string to be used to separate the lines of
jpayne@69 342 the value. The default value is the most useful for typical
jpayne@69 343 Python applications, but it can be set to \r\n to produce RFC-compliant
jpayne@69 344 line separators when needed.
jpayne@69 345 """
jpayne@69 346 self._normalize()
jpayne@69 347 if maxlinelen is None:
jpayne@69 348 maxlinelen = self._maxlinelen
jpayne@69 349 # A maxlinelen of 0 means don't wrap. For all practical purposes,
jpayne@69 350 # choosing a huge number here accomplishes that and makes the
jpayne@69 351 # _ValueFormatter algorithm much simpler.
jpayne@69 352 if maxlinelen == 0:
jpayne@69 353 maxlinelen = 1000000
jpayne@69 354 formatter = _ValueFormatter(self._headerlen, maxlinelen,
jpayne@69 355 self._continuation_ws, splitchars)
jpayne@69 356 lastcs = None
jpayne@69 357 hasspace = lastspace = None
jpayne@69 358 for string, charset in self._chunks:
jpayne@69 359 if hasspace is not None:
jpayne@69 360 hasspace = string and self._nonctext(string[0])
jpayne@69 361 if lastcs not in (None, 'us-ascii'):
jpayne@69 362 if not hasspace or charset not in (None, 'us-ascii'):
jpayne@69 363 formatter.add_transition()
jpayne@69 364 elif charset not in (None, 'us-ascii') and not lastspace:
jpayne@69 365 formatter.add_transition()
jpayne@69 366 lastspace = string and self._nonctext(string[-1])
jpayne@69 367 lastcs = charset
jpayne@69 368 hasspace = False
jpayne@69 369 lines = string.splitlines()
jpayne@69 370 if lines:
jpayne@69 371 formatter.feed('', lines[0], charset)
jpayne@69 372 else:
jpayne@69 373 formatter.feed('', '', charset)
jpayne@69 374 for line in lines[1:]:
jpayne@69 375 formatter.newline()
jpayne@69 376 if charset.header_encoding is not None:
jpayne@69 377 formatter.feed(self._continuation_ws, ' ' + line.lstrip(),
jpayne@69 378 charset)
jpayne@69 379 else:
jpayne@69 380 sline = line.lstrip()
jpayne@69 381 fws = line[:len(line)-len(sline)]
jpayne@69 382 formatter.feed(fws, sline, charset)
jpayne@69 383 if len(lines) > 1:
jpayne@69 384 formatter.newline()
jpayne@69 385 if self._chunks:
jpayne@69 386 formatter.add_transition()
jpayne@69 387 value = formatter._str(linesep)
jpayne@69 388 if _embedded_header.search(value):
jpayne@69 389 raise HeaderParseError("header value appears to contain "
jpayne@69 390 "an embedded header: {!r}".format(value))
jpayne@69 391 return value
jpayne@69 392
jpayne@69 393 def _normalize(self):
jpayne@69 394 # Step 1: Normalize the chunks so that all runs of identical charsets
jpayne@69 395 # get collapsed into a single unicode string.
jpayne@69 396 chunks = []
jpayne@69 397 last_charset = None
jpayne@69 398 last_chunk = []
jpayne@69 399 for string, charset in self._chunks:
jpayne@69 400 if charset == last_charset:
jpayne@69 401 last_chunk.append(string)
jpayne@69 402 else:
jpayne@69 403 if last_charset is not None:
jpayne@69 404 chunks.append((SPACE.join(last_chunk), last_charset))
jpayne@69 405 last_chunk = [string]
jpayne@69 406 last_charset = charset
jpayne@69 407 if last_chunk:
jpayne@69 408 chunks.append((SPACE.join(last_chunk), last_charset))
jpayne@69 409 self._chunks = chunks
jpayne@69 410
jpayne@69 411
jpayne@69 412
jpayne@69 413 class _ValueFormatter:
jpayne@69 414 def __init__(self, headerlen, maxlen, continuation_ws, splitchars):
jpayne@69 415 self._maxlen = maxlen
jpayne@69 416 self._continuation_ws = continuation_ws
jpayne@69 417 self._continuation_ws_len = len(continuation_ws)
jpayne@69 418 self._splitchars = splitchars
jpayne@69 419 self._lines = []
jpayne@69 420 self._current_line = _Accumulator(headerlen)
jpayne@69 421
jpayne@69 422 def _str(self, linesep):
jpayne@69 423 self.newline()
jpayne@69 424 return linesep.join(self._lines)
jpayne@69 425
jpayne@69 426 def __str__(self):
jpayne@69 427 return self._str(NL)
jpayne@69 428
jpayne@69 429 def newline(self):
jpayne@69 430 end_of_line = self._current_line.pop()
jpayne@69 431 if end_of_line != (' ', ''):
jpayne@69 432 self._current_line.push(*end_of_line)
jpayne@69 433 if len(self._current_line) > 0:
jpayne@69 434 if self._current_line.is_onlyws() and self._lines:
jpayne@69 435 self._lines[-1] += str(self._current_line)
jpayne@69 436 else:
jpayne@69 437 self._lines.append(str(self._current_line))
jpayne@69 438 self._current_line.reset()
jpayne@69 439
jpayne@69 440 def add_transition(self):
jpayne@69 441 self._current_line.push(' ', '')
jpayne@69 442
jpayne@69 443 def feed(self, fws, string, charset):
jpayne@69 444 # If the charset has no header encoding (i.e. it is an ASCII encoding)
jpayne@69 445 # then we must split the header at the "highest level syntactic break"
jpayne@69 446 # possible. Note that we don't have a lot of smarts about field
jpayne@69 447 # syntax; we just try to break on semi-colons, then commas, then
jpayne@69 448 # whitespace. Eventually, this should be pluggable.
jpayne@69 449 if charset.header_encoding is None:
jpayne@69 450 self._ascii_split(fws, string, self._splitchars)
jpayne@69 451 return
jpayne@69 452 # Otherwise, we're doing either a Base64 or a quoted-printable
jpayne@69 453 # encoding which means we don't need to split the line on syntactic
jpayne@69 454 # breaks. We can basically just find enough characters to fit on the
jpayne@69 455 # current line, minus the RFC 2047 chrome. What makes this trickier
jpayne@69 456 # though is that we have to split at octet boundaries, not character
jpayne@69 457 # boundaries but it's only safe to split at character boundaries so at
jpayne@69 458 # best we can only get close.
jpayne@69 459 encoded_lines = charset.header_encode_lines(string, self._maxlengths())
jpayne@69 460 # The first element extends the current line, but if it's None then
jpayne@69 461 # nothing more fit on the current line so start a new line.
jpayne@69 462 try:
jpayne@69 463 first_line = encoded_lines.pop(0)
jpayne@69 464 except IndexError:
jpayne@69 465 # There are no encoded lines, so we're done.
jpayne@69 466 return
jpayne@69 467 if first_line is not None:
jpayne@69 468 self._append_chunk(fws, first_line)
jpayne@69 469 try:
jpayne@69 470 last_line = encoded_lines.pop()
jpayne@69 471 except IndexError:
jpayne@69 472 # There was only one line.
jpayne@69 473 return
jpayne@69 474 self.newline()
jpayne@69 475 self._current_line.push(self._continuation_ws, last_line)
jpayne@69 476 # Everything else are full lines in themselves.
jpayne@69 477 for line in encoded_lines:
jpayne@69 478 self._lines.append(self._continuation_ws + line)
jpayne@69 479
jpayne@69 480 def _maxlengths(self):
jpayne@69 481 # The first line's length.
jpayne@69 482 yield self._maxlen - len(self._current_line)
jpayne@69 483 while True:
jpayne@69 484 yield self._maxlen - self._continuation_ws_len
jpayne@69 485
jpayne@69 486 def _ascii_split(self, fws, string, splitchars):
jpayne@69 487 # The RFC 2822 header folding algorithm is simple in principle but
jpayne@69 488 # complex in practice. Lines may be folded any place where "folding
jpayne@69 489 # white space" appears by inserting a linesep character in front of the
jpayne@69 490 # FWS. The complication is that not all spaces or tabs qualify as FWS,
jpayne@69 491 # and we are also supposed to prefer to break at "higher level
jpayne@69 492 # syntactic breaks". We can't do either of these without intimate
jpayne@69 493 # knowledge of the structure of structured headers, which we don't have
jpayne@69 494 # here. So the best we can do here is prefer to break at the specified
jpayne@69 495 # splitchars, and hope that we don't choose any spaces or tabs that
jpayne@69 496 # aren't legal FWS. (This is at least better than the old algorithm,
jpayne@69 497 # where we would sometimes *introduce* FWS after a splitchar, or the
jpayne@69 498 # algorithm before that, where we would turn all white space runs into
jpayne@69 499 # single spaces or tabs.)
jpayne@69 500 parts = re.split("(["+FWS+"]+)", fws+string)
jpayne@69 501 if parts[0]:
jpayne@69 502 parts[:0] = ['']
jpayne@69 503 else:
jpayne@69 504 parts.pop(0)
jpayne@69 505 for fws, part in zip(*[iter(parts)]*2):
jpayne@69 506 self._append_chunk(fws, part)
jpayne@69 507
jpayne@69 508 def _append_chunk(self, fws, string):
jpayne@69 509 self._current_line.push(fws, string)
jpayne@69 510 if len(self._current_line) > self._maxlen:
jpayne@69 511 # Find the best split point, working backward from the end.
jpayne@69 512 # There might be none, on a long first line.
jpayne@69 513 for ch in self._splitchars:
jpayne@69 514 for i in range(self._current_line.part_count()-1, 0, -1):
jpayne@69 515 if ch.isspace():
jpayne@69 516 fws = self._current_line[i][0]
jpayne@69 517 if fws and fws[0]==ch:
jpayne@69 518 break
jpayne@69 519 prevpart = self._current_line[i-1][1]
jpayne@69 520 if prevpart and prevpart[-1]==ch:
jpayne@69 521 break
jpayne@69 522 else:
jpayne@69 523 continue
jpayne@69 524 break
jpayne@69 525 else:
jpayne@69 526 fws, part = self._current_line.pop()
jpayne@69 527 if self._current_line._initial_size > 0:
jpayne@69 528 # There will be a header, so leave it on a line by itself.
jpayne@69 529 self.newline()
jpayne@69 530 if not fws:
jpayne@69 531 # We don't use continuation_ws here because the whitespace
jpayne@69 532 # after a header should always be a space.
jpayne@69 533 fws = ' '
jpayne@69 534 self._current_line.push(fws, part)
jpayne@69 535 return
jpayne@69 536 remainder = self._current_line.pop_from(i)
jpayne@69 537 self._lines.append(str(self._current_line))
jpayne@69 538 self._current_line.reset(remainder)
jpayne@69 539
jpayne@69 540
jpayne@69 541 class _Accumulator(list):
jpayne@69 542
jpayne@69 543 def __init__(self, initial_size=0):
jpayne@69 544 self._initial_size = initial_size
jpayne@69 545 super().__init__()
jpayne@69 546
jpayne@69 547 def push(self, fws, string):
jpayne@69 548 self.append((fws, string))
jpayne@69 549
jpayne@69 550 def pop_from(self, i=0):
jpayne@69 551 popped = self[i:]
jpayne@69 552 self[i:] = []
jpayne@69 553 return popped
jpayne@69 554
jpayne@69 555 def pop(self):
jpayne@69 556 if self.part_count()==0:
jpayne@69 557 return ('', '')
jpayne@69 558 return super().pop()
jpayne@69 559
jpayne@69 560 def __len__(self):
jpayne@69 561 return sum((len(fws)+len(part) for fws, part in self),
jpayne@69 562 self._initial_size)
jpayne@69 563
jpayne@69 564 def __str__(self):
jpayne@69 565 return EMPTYSTRING.join((EMPTYSTRING.join((fws, part))
jpayne@69 566 for fws, part in self))
jpayne@69 567
jpayne@69 568 def reset(self, startval=None):
jpayne@69 569 if startval is None:
jpayne@69 570 startval = []
jpayne@69 571 self[:] = startval
jpayne@69 572 self._initial_size = 0
jpayne@69 573
jpayne@69 574 def is_onlyws(self):
jpayne@69 575 return self._initial_size==0 and (not self or str(self).isspace())
jpayne@69 576
jpayne@69 577 def part_count(self):
jpayne@69 578 return super().__len__()