|
| 1 | +""" Codec for the modified UTF-7 encoding used for IMAP4 mailbox names, |
| 2 | +as specified in RFC 3501, section 5.1.3. |
| 3 | +
|
| 4 | +It differs from UTF-7 (RFC 2152) as follows: |
| 5 | +
|
| 6 | +* "&" (not "+") is the shift character introducing a Base64 sequence, |
| 7 | + and "&-" encodes a literal "&"; |
| 8 | +* "," is used instead of "/" in the modified Base64 alphabet; |
| 9 | +* only printable US-ASCII characters (except "&") may be represented |
| 10 | + directly, so all other characters, including other controls, are |
| 11 | + Base64-encoded. |
| 12 | +""" |
| 13 | + |
| 14 | +import binascii |
| 15 | +import codecs |
| 16 | +from base64 import b64encode, b64decode |
| 17 | + |
| 18 | +### Codec APIs |
| 19 | + |
| 20 | +def imap4_utf_7_encode(input, errors='strict'): |
| 21 | + if errors != 'strict': |
| 22 | + raise UnicodeError(f"Unsupported error handling: {errors}") |
| 23 | + res = bytearray() |
| 24 | + start = 0 # start of the current run |
| 25 | + b64run = False # is input[start:i] a Base64 run? |
| 26 | + def flush(end): |
| 27 | + if start < end: |
| 28 | + if b64run: |
| 29 | + b64 = b64encode(input[start:end].encode('utf-16-be'), |
| 30 | + altchars=b'+,', padded=False) |
| 31 | + res.extend(b'&' + b64 + b'-') |
| 32 | + else: |
| 33 | + res.extend(input[start:end].encode('ascii')) |
| 34 | + for i, ch in enumerate(input): |
| 35 | + if ch == '&': |
| 36 | + flush(i) |
| 37 | + res.extend(b'&-') |
| 38 | + start = i + 1 |
| 39 | + b64run = False |
| 40 | + elif ' ' <= ch <= '~': # printable ASCII, represented directly |
| 41 | + if b64run: |
| 42 | + flush(i) |
| 43 | + start = i |
| 44 | + b64run = False |
| 45 | + else: # everything else is Base64-encoded |
| 46 | + if not b64run: |
| 47 | + flush(i) |
| 48 | + start = i |
| 49 | + b64run = True |
| 50 | + flush(len(input)) |
| 51 | + return res.take_bytes(), len(input) |
| 52 | + |
| 53 | +def imap4_utf_7_decode(input, errors='strict'): |
| 54 | + if errors != 'strict': |
| 55 | + raise UnicodeError(f"Unsupported error handling: {errors}") |
| 56 | + input = bytes(input) |
| 57 | + res = [] |
| 58 | + start = 0 # start of the current direct ASCII run |
| 59 | + i = 0 |
| 60 | + n = len(input) |
| 61 | + def flush(end): |
| 62 | + if start < end: |
| 63 | + res.append(input[start:end].decode('ascii')) |
| 64 | + while i < n: |
| 65 | + c = input[i] |
| 66 | + if c == b'&'[0]: |
| 67 | + flush(i) |
| 68 | + j = input.find(b'-', i + 1) |
| 69 | + if j < 0: |
| 70 | + raise UnicodeDecodeError('imap4-utf-7', input, i, n, |
| 71 | + 'unterminated shift sequence') |
| 72 | + if j == i + 1: # '&-' |
| 73 | + res.append('&') |
| 74 | + else: |
| 75 | + b64 = input[i + 1:j] |
| 76 | + try: |
| 77 | + data = b64decode(b64, altchars=b'+,', |
| 78 | + validate=True, padded=False) |
| 79 | + except binascii.Error: |
| 80 | + data = b'' |
| 81 | + if not data or len(data) % 2: |
| 82 | + raise UnicodeDecodeError('imap4-utf-7', input, i, j + 1, |
| 83 | + 'invalid shift sequence') |
| 84 | + res.append(data.decode('utf-16-be')) |
| 85 | + i = j + 1 |
| 86 | + start = i |
| 87 | + elif b' '[0] <= c <= b'~'[0]: |
| 88 | + i += 1 |
| 89 | + else: |
| 90 | + raise UnicodeDecodeError('imap4-utf-7', input, i, i + 1, |
| 91 | + 'unexpected byte') |
| 92 | + flush(n) |
| 93 | + return ''.join(res), len(input) |
| 94 | + |
| 95 | +class Codec(codecs.Codec): |
| 96 | + def encode(self, input, errors='strict'): |
| 97 | + return imap4_utf_7_encode(input, errors) |
| 98 | + def decode(self, input, errors='strict'): |
| 99 | + return imap4_utf_7_decode(input, errors) |
| 100 | + |
| 101 | +class IncrementalEncoder(codecs.IncrementalEncoder): |
| 102 | + def encode(self, input, final=False): |
| 103 | + return imap4_utf_7_encode(input, self.errors)[0] |
| 104 | + |
| 105 | +class IncrementalDecoder(codecs.IncrementalDecoder): |
| 106 | + def decode(self, input, final=False): |
| 107 | + return imap4_utf_7_decode(input, self.errors)[0] |
| 108 | + |
| 109 | +class StreamWriter(Codec, codecs.StreamWriter): |
| 110 | + pass |
| 111 | + |
| 112 | +class StreamReader(Codec, codecs.StreamReader): |
| 113 | + pass |
| 114 | + |
| 115 | +### encodings module API |
| 116 | + |
| 117 | +def getregentry(): |
| 118 | + return codecs.CodecInfo( |
| 119 | + name='imap4-utf-7', |
| 120 | + encode=Codec().encode, |
| 121 | + decode=Codec().decode, |
| 122 | + incrementalencoder=IncrementalEncoder, |
| 123 | + incrementaldecoder=IncrementalDecoder, |
| 124 | + streamwriter=StreamWriter, |
| 125 | + streamreader=StreamReader, |
| 126 | + ) |
0 commit comments