Mercurial > repos > rliterman > csp2

annotate CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/email/header.py @ 68:5028fdace37b

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