diff --git a/Lib/imaplib.py b/Lib/imaplib.py index e84ffb2ee..4c0aefd08 100644 --- a/Lib/imaplib.py +++ b/Lib/imaplib.py @@ -1,6 +1,6 @@ """IMAP4 client. -Based on RFC 2060. +Based on RFC 3501. Public class: IMAP4 Public variable: Debug @@ -45,8 +45,8 @@ IMAP4_SSL_PORT = 993 AllowedVersions = ('IMAP4REV1', 'IMAP4') # Most recent first # Maximal line length when calling readline(). This is to prevent -# reading arbitrary length lines. RFC 3501 and 2060 (IMAP 4rev1) -# don't specify a line length. RFC 2683 suggests limiting client +# reading arbitrary length lines. RFC 3501 (IMAP 4rev1) +# doesn't specify a line length. RFC 2683 suggests limiting client # command lines to 1000 octets and that servers should be prepared # to accept command lines up to 8000 octets, so we used to use 10K here. # In the modern world (eg: gmail) the response to, for example, a @@ -131,8 +131,19 @@ Untagged_status = re.compile( # We compile these in _mode_xxx. _Literal = br'.*{(?P\d+)}$' _Untagged_status = br'\* (?P\d+) (?P[A-Z-]+)( (?P.*))?' +# Only NUL, CR and LF are unsafe (they cannot be represented even in +# a quoted string); other control characters are sent quoted. +_control_chars = re.compile(b'[\x00\r\n]') +_non_astring_char = re.compile(br'[(){ \x00-\x1f\x7f-\xff%*\\"]') +_non_list_char = re.compile(br'[(){ \x00-\x1f\x7f-\xff\\"]') +_quoted = re.compile(br'"(?:[^"\\]|\\.)*+"') +def _paren_depth(data, depth=0): + # Net parenthesis nesting of data, ignoring parentheses in quoted strings. + data = _quoted.sub(b'', data) + return depth + data.count(b'(') - data.count(b')') + class IMAP4: @@ -256,9 +267,13 @@ class IMAP4: elif 'OK' in self.untagged_responses: self.state = 'NONAUTH' else: - raise self.error(self.welcome) + # A continuation ('+') greeting is returned as None; report its + # raw line, still held by the last match (gh-108280). + greeting = (self.welcome or self.mo.string).decode( + self._encoding, 'replace') + raise self.error('invalid greeting: ' + greeting) - self._get_capabilities() + self._refresh_capabilities() if __debug__: if self.debug >= 3: self._mesg('CAPABILITIES: %r' % (self.capabilities,)) @@ -488,8 +503,7 @@ class IMAP4: if not mailbox: mailbox = 'INBOX' if flags: - if (flags[0],flags[-1]) != ('(',')'): - flags = '(%s)' % flags + flags = self._set_quote(flags) else: flags = None if date_time: @@ -498,7 +512,7 @@ class IMAP4: date_time = None literal = MapCRLF.sub(CRLF, message) self.literal = literal - return self._simple_command(name, mailbox, flags, date_time) + return self._simple_command(name, self._astring(mailbox), flags, date_time) def authenticate(self, mechanism, authobject): @@ -523,10 +537,11 @@ class IMAP4: #if not cap in self.capabilities: # Let the server decide! # raise self.error("Server doesn't allow %s authentication." % mech) self.literal = _Authenticator(authobject).process - typ, dat = self._simple_command('AUTHENTICATE', mech) + typ, dat = self._simple_command('AUTHENTICATE', self._atom(mech)) if typ != 'OK': raise self.error(dat[-1].decode('utf-8', 'replace')) self.state = 'AUTH' + self._refresh_capabilities() return typ, dat @@ -567,7 +582,8 @@ class IMAP4: (typ, [data]) = .copy(message_set, new_mailbox) """ - return self._simple_command('COPY', message_set, new_mailbox) + return self._simple_command('COPY', self._sequence_set(message_set), + self._astring(new_mailbox)) def create(self, mailbox): @@ -575,7 +591,7 @@ class IMAP4: (typ, [data]) = .create(mailbox) """ - return self._simple_command('CREATE', mailbox) + return self._simple_command('CREATE', self._astring(mailbox)) def delete(self, mailbox): @@ -583,14 +599,15 @@ class IMAP4: (typ, [data]) = .delete(mailbox) """ - return self._simple_command('DELETE', mailbox) + return self._simple_command('DELETE', self._astring(mailbox)) def deleteacl(self, mailbox, who): """Delete the ACLs (remove any rights) set for who on mailbox. (typ, [data]) = .deleteacl(mailbox, who) """ - return self._simple_command('DELETEACL', mailbox, who) + return self._simple_command('DELETEACL', self._astring(mailbox), + self._astring(who)) def enable(self, capability): """Send an RFC5161 enable string to the server. @@ -629,7 +646,8 @@ class IMAP4: 'data' are tuples of message part envelope and data. """ name = 'FETCH' - typ, dat = self._simple_command(name, message_set, message_parts) + typ, dat = self._simple_command(name, self._sequence_set(message_set), + self._fetch_parts(message_parts)) return self._untagged_response(typ, dat, name) @@ -638,7 +656,7 @@ class IMAP4: (typ, [data]) = .getacl(mailbox) """ - typ, dat = self._simple_command('GETACL', mailbox) + typ, dat = self._simple_command('GETACL', self._astring(mailbox)) return self._untagged_response(typ, dat, 'ACL') @@ -646,7 +664,8 @@ class IMAP4: """(typ, [data]) = .getannotation(mailbox, entry, attribute) Retrieve ANNOTATIONs.""" - typ, dat = self._simple_command('GETANNOTATION', mailbox, entry, attribute) + typ, dat = self._simple_command('GETANNOTATION', self._astring(mailbox), + entry, attribute) return self._untagged_response(typ, dat, 'ANNOTATION') @@ -657,7 +676,7 @@ class IMAP4: (typ, [data]) = .getquota(root) """ - typ, dat = self._simple_command('GETQUOTA', root) + typ, dat = self._simple_command('GETQUOTA', self._astring(root)) return self._untagged_response(typ, dat, 'QUOTA') @@ -666,7 +685,7 @@ class IMAP4: (typ, [[QUOTAROOT responses...], [QUOTA responses]]) = .getquotaroot(mailbox) """ - typ, dat = self._simple_command('GETQUOTAROOT', mailbox) + typ, dat = self._simple_command('GETQUOTAROOT', self._astring(mailbox)) typ, quota = self._untagged_response(typ, dat, 'QUOTA') typ, quotaroot = self._untagged_response(typ, dat, 'QUOTAROOT') return typ, [quotaroot, quota] @@ -685,15 +704,16 @@ class IMAP4: return Idler(self, duration) - def list(self, directory='""', pattern='*'): + def list(self, directory='', pattern='*'): """List mailbox names in directory matching pattern. - (typ, [data]) = .list(directory='""', pattern='*') + (typ, [data]) = .list(directory='', pattern='*') 'data' is list of LIST responses. """ name = 'LIST' - typ, dat = self._simple_command(name, directory, pattern) + typ, dat = self._simple_command(name, self._astring(directory), + self._list_mailbox(pattern)) return self._untagged_response(typ, dat, name) @@ -704,10 +724,12 @@ class IMAP4: NB: 'password' will be quoted. """ - typ, dat = self._simple_command('LOGIN', user, self._quote(password)) + typ, dat = self._simple_command('LOGIN', self._astring(user), + self._quote(password)) if typ != 'OK': raise self.error(dat[-1].decode('UTF-8', 'replace')) self.state = 'AUTH' + self._refresh_capabilities() return typ, dat @@ -749,15 +771,16 @@ class IMAP4: return typ, dat - def lsub(self, directory='""', pattern='*'): + def lsub(self, directory='', pattern='*'): """List 'subscribed' mailbox names in directory matching pattern. - (typ, [data, ...]) = .lsub(directory='""', pattern='*') + (typ, [data, ...]) = .lsub(directory='', pattern='*') 'data' are tuples of message part envelope and data. """ name = 'LSUB' - typ, dat = self._simple_command(name, directory, pattern) + typ, dat = self._simple_command(name, self._astring(directory), + self._list_mailbox(pattern)) return self._untagged_response(typ, dat, name) def myrights(self, mailbox): @@ -765,7 +788,7 @@ class IMAP4: (typ, [data]) = .myrights(mailbox) """ - typ,dat = self._simple_command('MYRIGHTS', mailbox) + typ,dat = self._simple_command('MYRIGHTS', self._astring(mailbox)) return self._untagged_response(typ, dat, 'MYRIGHTS') def namespace(self): @@ -811,7 +834,7 @@ class IMAP4: """ name = 'PROXYAUTH' - return self._simple_command('PROXYAUTH', user) + return self._simple_command(name, self._astring(user)) def rename(self, oldmailbox, newmailbox): @@ -819,7 +842,8 @@ class IMAP4: (typ, [data]) = .rename(oldmailbox, newmailbox) """ - return self._simple_command('RENAME', oldmailbox, newmailbox) + return self._simple_command('RENAME', self._astring(oldmailbox), + self._astring(newmailbox)) def search(self, charset, *criteria): @@ -831,10 +855,11 @@ class IMAP4: If UTF8 is enabled, charset MUST be None. """ name = 'SEARCH' - if charset: + if charset is not None: if self.utf8_enabled: raise IMAP4.error("Non-None charset not valid in UTF8 mode") - typ, dat = self._simple_command(name, 'CHARSET', charset, *criteria) + typ, dat = self._simple_command(name, + 'CHARSET', self._astring(charset), *criteria) else: typ, dat = self._simple_command(name, *criteria) return self._untagged_response(typ, dat, name) @@ -858,7 +883,7 @@ class IMAP4: name = 'EXAMINE' else: name = 'SELECT' - typ, dat = self._simple_command(name, mailbox) + typ, dat = self._simple_command(name, self._astring(mailbox)) if typ != 'OK': self.state = 'AUTH' # Might have been 'SELECTED' return typ, dat @@ -868,7 +893,7 @@ class IMAP4: if __debug__: if self.debug >= 1: self._dump_ur(self.untagged_responses) - raise self.readonly('%s is not writable' % mailbox) + raise self.readonly('%r is not writable' % (mailbox,)) return typ, self.untagged_responses.get('EXISTS', [None]) @@ -877,14 +902,15 @@ class IMAP4: (typ, [data]) = .setacl(mailbox, who, what) """ - return self._simple_command('SETACL', mailbox, who, what) + return self._simple_command('SETACL', self._astring(mailbox), + self._astring(who), self._astring(what)) - def setannotation(self, *args): + def setannotation(self, mailbox, *args): """(typ, [data]) = .setannotation(mailbox[, entry, attribute]+) Set ANNOTATIONs.""" - typ, dat = self._simple_command('SETANNOTATION', *args) + typ, dat = self._simple_command('SETANNOTATION', self._astring(mailbox), *args) return self._untagged_response(typ, dat, 'ANNOTATION') @@ -893,7 +919,8 @@ class IMAP4: (typ, [data]) = .setquota(root, limits) """ - typ, dat = self._simple_command('SETQUOTA', root, limits) + typ, dat = self._simple_command('SETQUOTA', self._astring(root), + self._set_quote(limits)) return self._untagged_response(typ, dat, 'QUOTA') @@ -905,8 +932,9 @@ class IMAP4: name = 'SORT' #if not name in self.capabilities: # Let the server decide! # raise self.error('unimplemented extension command: %s' % name) - if (sort_criteria[0],sort_criteria[-1]) != ('(',')'): - sort_criteria = '(%s)' % sort_criteria + sort_criteria = self._set_quote(sort_criteria) + if charset is not None: + charset = self._astring(charset) typ, dat = self._simple_command(name, sort_criteria, charset, *search_criteria) return self._untagged_response(typ, dat, name) @@ -942,7 +970,8 @@ class IMAP4: name = 'STATUS' #if self.PROTOCOL_VERSION == 'IMAP4': # Let the server decide! # raise self.error('%s unimplemented in IMAP4 (obtain IMAP4rev1 server, or re-code)' % name) - typ, dat = self._simple_command(name, mailbox, names) + typ, dat = self._simple_command(name, self._astring(mailbox), + self._set_quote(names)) return self._untagged_response(typ, dat, name) @@ -951,9 +980,9 @@ class IMAP4: (typ, [data]) = .store(message_set, command, flags) """ - if (flags[0],flags[-1]) != ('(',')'): - flags = '(%s)' % flags # Avoid quoting the flags - typ, dat = self._simple_command('STORE', message_set, command, flags) + flags = self._set_quote(flags) + typ, dat = self._simple_command('STORE', self._sequence_set(message_set), + command, flags) return self._untagged_response(typ, dat, 'FETCH') @@ -962,7 +991,7 @@ class IMAP4: (typ, [data]) = .subscribe(mailbox) """ - return self._simple_command('SUBSCRIBE', mailbox) + return self._simple_command('SUBSCRIBE', self._astring(mailbox)) def thread(self, threading_algorithm, charset, *search_criteria): @@ -971,7 +1000,10 @@ class IMAP4: (type, [data]) = .thread(threading_algorithm, charset, search_criteria, ...) """ name = 'THREAD' - typ, dat = self._simple_command(name, threading_algorithm, charset, *search_criteria) + if charset is not None: + charset = self._astring(charset) + typ, dat = self._simple_command(name, self._atom(threading_algorithm), + charset, *search_criteria) return self._untagged_response(typ, dat, name) @@ -985,14 +1017,38 @@ class IMAP4: """ command = command.upper() if not command in Commands: - raise self.error("Unknown IMAP4 UID command: %s" % command) + raise self.error("Unknown IMAP4 UID command: %r" % (command,)) if self.state not in Commands[command]: raise self.error("command %s illegal in state %s, " "only allowed in states %s" % (command, self.state, ', '.join(Commands[command]))) name = 'UID' - typ, dat = self._simple_command(name, command, *args) + if command == 'COPY': + message_set, new_mailbox = args + args = (self._sequence_set(message_set), + self._astring(new_mailbox)) + elif command == 'FETCH': + message_set, message_parts = args + args = (self._sequence_set(message_set), + self._fetch_parts(message_parts)) + elif command == 'STORE': + message_set, op, flags = args + args = (self._sequence_set(message_set), op, + self._set_quote(flags)) + elif command == 'SORT': + sort_criteria, charset, *search_criteria = args + if charset is not None: + charset = self._astring(charset) + args = (self._set_quote(sort_criteria), charset, + *search_criteria) + elif command == 'THREAD': + threading_algorithm, charset, *search_criteria = args + if charset is not None: + charset = self._astring(charset) + args = (self._atom(threading_algorithm), charset, + *search_criteria) + typ, dat = self._simple_command(name, self._atom(command), *args) if command in ('SEARCH', 'SORT', 'THREAD'): name = command else: @@ -1005,7 +1061,7 @@ class IMAP4: (typ, [data]) = .unsubscribe(mailbox) """ - return self._simple_command('UNSUBSCRIBE', mailbox) + return self._simple_command('UNSUBSCRIBE', self._astring(mailbox)) def unselect(self): @@ -1108,6 +1164,8 @@ class IMAP4: if arg is None: continue if isinstance(arg, str): arg = bytes(arg, self._encoding) + if _control_chars.search(arg): + raise ValueError("NUL, CR and LF not allowed in commands") data = data + b' ' + arg literal = self.literal @@ -1192,6 +1250,15 @@ class IMAP4: self.capabilities = tuple(dat.split()) + def _refresh_capabilities(self): + # Use a CAPABILITY response sent by the server, or ask for it. + if 'CAPABILITY' in self.untagged_responses: + dat = self.untagged_responses.pop('CAPABILITY')[-1] + self.capabilities = tuple(str(dat, self._encoding).upper().split()) + else: + self._get_capabilities() + + def _get_response(self, start_timeout=False): # Read response and store. @@ -1216,6 +1283,11 @@ class IMAP4: else: resp = self._get_line() + # Skip spurious blank lines between responses (some servers send one + # after a literal that ends a response). + while resp == b'': + resp = self._get_line() + # Command completion response? if self._match(self.tagre, resp): @@ -1253,6 +1325,7 @@ class IMAP4: # Is there a literal to come? + depth = 0 # open parenthesis nesting so far while self._match(self.Literal, dat): # Read literal direct from connection. @@ -1266,11 +1339,17 @@ class IMAP4: # Store response with literal as tuple self._append_untagged(typ, (dat, data)) + depth = _paren_depth(dat, depth) # Read trailer - possibly containing another literal dat = self._get_line() + # Skip spurious blank lines after a literal, but only inside an + # unclosed parenthesis (at top level they end the response). + while dat == b'' and depth > 0: + dat = self._get_line() + self._append_untagged(typ, dat) # Bracketed response information? @@ -1359,13 +1438,53 @@ class IMAP4: return tag + def _atom(self, arg): + return arg + + def _sequence_set(self, arg): + return arg + + def _set_quote(self, arg): + if arg and arg[0] == '(' and arg[-1] == ')': + return arg + return '(' + arg + ')' + + def _fetch_parts(self, arg): + # "ALL", "FULL" and "FAST" are macros, not data item names; + # they cannot be enclosed in parentheses. + if arg.upper() in ('ALL', 'FULL', 'FAST'): + return arg + return self._set_quote(arg) + def _quote(self, arg): + if isinstance(arg, str): + arg = bytes(arg, self._encoding) + arg = arg.replace(b'\\', br'\\') + arg = arg.replace(b'"', br'\"') + return b'"' + arg + b'"' - arg = arg.replace('\\', '\\\\') - arg = arg.replace('"', '\\"') + # For backward compatibility, an argument already enclosed in double + # quotes is left unquoted, so that code which quotes arguments itself + # keeps working. New code should pass arguments unquoted and let the + # module quote them as needed. - return '"' + arg + '"' + def _astring(self, arg): + if isinstance(arg, str): + arg = bytes(arg, self._encoding) + if _quoted.fullmatch(arg): + return arg + if arg and _non_astring_char.search(arg) is None: + return arg + return self._quote(arg) + def _list_mailbox(self, arg): + if isinstance(arg, str): + arg = bytes(arg, self._encoding) + if _quoted.fullmatch(arg): + return arg + if arg and _non_list_char.search(arg) is None: + return arg + return self._quote(arg) def _simple_command(self, name, *args): @@ -1835,9 +1954,8 @@ def Time2Internaldate(date_time): dt = datetime.fromtimestamp(date_time, timezone.utc).astimezone() elif isinstance(date_time, tuple): - try: - gmtoff = date_time.tm_gmtoff - except AttributeError: + gmtoff = getattr(date_time, "tm_gmtoff", None) + if gmtoff is None: if time.daylight: dst = date_time[8] if dst == -1: diff --git a/Lib/tarfile.py b/Lib/tarfile.py index e6734db24..d3c489997 100644 --- a/Lib/tarfile.py +++ b/Lib/tarfile.py @@ -258,6 +258,32 @@ def copyfileobj(src, dst, length=None, exception=OSError, bufsize=None): dst.write(buf) return +# Maximum number of bytes read in a single call when reading a member's +# extended header (a GNU long name/link or a pax header). The size of such +# a header is taken from the archive and is not trustworthy, so it is read in +# bounded chunks to avoid a huge up-front allocation when a crafted or +# truncated archive claims far more data than the file actually contains +# (gh-151497). +_EXTHEADER_READ_CHUNK = 1024 * 1024 # 1 MiB + +def _safe_read(fileobj, size): + """Read up to *size* bytes from *fileobj* in bounded chunks. + + Returns the same bytes as ``fileobj.read(size)`` would (including a short + result at end of file), but limits pre-allocation, so an + oversized size field in a crafted header cannot force a huge allocation. + """ + if size <= _EXTHEADER_READ_CHUNK: + return fileobj.read(size) + chunks = [] + while size > 0: + chunk = fileobj.read(min(size, _EXTHEADER_READ_CHUNK)) + if not chunk: + break + chunks.append(chunk) + size -= len(chunk) + return b"".join(chunks) + def _safe_print(s): encoding = getattr(sys.stdout, 'encoding', None) if encoding is not None: @@ -524,7 +550,9 @@ class _Stream: if pos - self.pos >= 0: blocks, remainder = divmod(pos - self.pos, self.bufsize) for i in range(blocks): - self.read(self.bufsize) + data = self.read(self.bufsize) + if not data: + break self.read(remainder) else: raise StreamError("seeking backwards is not allowed") @@ -1429,7 +1457,7 @@ class TarInfo(object): """Process the blocks that hold a GNU longname or longlink member. """ - buf = tarfile.fileobj.read(self._block(self.size)) + buf = _safe_read(tarfile.fileobj, self._block(self.size)) # Fetch the next header and process it. try: @@ -1485,7 +1513,7 @@ class TarInfo(object): POSIX.1-2008. """ # Read the header information. - buf = tarfile.fileobj.read(self._block(self.size)) + buf = _safe_read(tarfile.fileobj, self._block(self.size)) # A pax header stores supplemental information for either # the following file (extended) or all following files @@ -2510,7 +2538,8 @@ class TarFile(object): tarinfo, unfiltered = self._get_extract_tarinfo( member, filter_function, path) if tarinfo is not None: - self._extract_one(tarinfo, path, set_attrs, numeric_owner) + self._extract_one(tarinfo, path, set_attrs, numeric_owner, + filter_function=filter_function) def _get_extract_tarinfo(self, member, filter_function, path): """Get (filtered, unfiltered) TarInfos from *member* @@ -2782,6 +2811,9 @@ class TarFile(object): "makelink_with_filter: if filter_function is not None, " + "extraction_root must also not be None") try: + filter_function( + unfiltered.replace(name=tarinfo.name, deep=False), + extraction_root) filtered = filter_function(unfiltered, extraction_root) except _FILTER_ERRORS as cause: raise LinkFallbackError(tarinfo, unfiltered.name) from cause diff --git a/Lib/tempfile.py b/Lib/tempfile.py index a34e062f8..6c036d2fd 100644 --- a/Lib/tempfile.py +++ b/Lib/tempfile.py @@ -915,7 +915,8 @@ class TemporaryDirectory: @classmethod def _rmtree(cls, name, ignore_errors=False, repeated=False): def onexc(func, path, exc): - if isinstance(exc, PermissionError): + # On DragonFly BSD, UF_NOUNLINK removal fails with EISDIR, not EPERM. + if isinstance(exc, (PermissionError, IsADirectoryError)): if repeated and path == name: if ignore_errors: return diff --git a/Lib/test/test_imaplib.py b/Lib/test/test_imaplib.py index f20120618..62d4d6e6e 100644 --- a/Lib/test/test_imaplib.py +++ b/Lib/test/test_imaplib.py @@ -9,6 +9,7 @@ import time import calendar import threading import re +import select import socket from test.support import verbose, run_with_tz, run_with_locale, cpython_only @@ -27,6 +28,80 @@ support.requires_working_socket(module=True) CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "certdata", "keycert3.pem") CAFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "certdata", "pycacert.pem") +_quoted_string = re.compile(r'"(?:[^"\\]|\\.)*"') + + +def splitargs(line): + # Split a command line into IMAP arguments: quoted strings, balanced + # (possibly nested) parenthesized lists, and bare atoms. + args = [] + i, n = 0, len(line) + while i < n: + if line[i] == ' ': + i += 1 + continue + start = i + depth = 0 + while i < n and (depth or line[i] != ' '): + c = line[i] + if c == '"': + i = _quoted_string.match(line, i).end() + continue + elif c == '(': + depth += 1 + elif c == ')': + depth -= 1 + i += 1 + args.append(line[start:i]) + return args + + +def parse_seq_number(s, maxmsg): + if s == '*': + return maxmsg + return int(s) + + +def parse_sequence_set(arg, maxmsg): + for s in arg.split(','): + if ':' in s: + lo, hi = s.split(':') + lo = parse_seq_number(lo, maxmsg) + hi = parse_seq_number(hi, maxmsg) + yield from range(min(lo, hi), max(lo, hi) + 1) + else: + yield parse_seq_number(s, maxmsg) + + +def make_simple_handler(command, untagged_response=(), + completed=None): + if completed is None: + completed = f'{command} completed' + def cmd(self, tag, args): + for msg in untagged_response: + self._send_textline(msg) + self._send_tagged(tag, 'OK', completed) + cmd.__name__ = 'cmd_' + command + class Handler(SimpleIMAPHandler): + pass + Handler.__name__ = command.title() + 'Handler' + setattr(Handler, cmd.__name__, cmd) + Handler.__qualname__ = Handler.__name__ + cmd.__qualname__ = Handler.__qualname__ + '.' + cmd.__name__ + return Handler + + +def _read_literal(handler, marker): + # Read one literal, a raw octet sequence, by its count from the marker + # ('{N}', or '(~{N}' in UTF8 mode). + size = int(re.search(r'\{(\d+)\}', marker).group(1)) + # The client must wait for the continuation, so nothing should be readable. + if select.select([handler.connection], [], [], 0)[0]: + raise AssertionError('client sent the literal before the ' + 'continuation request') + handler._send_textline('+') + return handler.rfile.read(size) + class TestImaplib(unittest.TestCase): @@ -68,6 +143,15 @@ class TestImaplib(unittest.TestCase): internal = imaplib.Time2Internaldate(t) self.assertEqual(internal, expected) + @run_with_tz('STD-1DST,M3.2.0,M11.1.0') + def test_Time2Internaldate_datetime_timetuple(self): + date_time = datetime.fromtimestamp(2000000000).timetuple() + self.assertIsNone(date_time.tm_gmtoff) + self.assertEqual( + imaplib.Time2Internaldate(date_time), + '"18-May-2033 05:33:20 +0200"', + ) + def test_that_Time2Internaldate_returns_a_result(self): # Without tzset, we can check only that it successfully # produces a result, not the correctness of the result itself, @@ -92,6 +176,65 @@ class TestImaplib(unittest.TestCase): imaplib.IMAP4() self.assertIn(cm.exception.errno, expected_errnos) + def test_astring(self): + m = imaplib.IMAP4.__new__(imaplib.IMAP4) + m._encoding = 'ascii' + # Plain atoms are left unquoted. + self.assertEqual(m._astring('INBOX'), b'INBOX') + self.assertEqual(m._astring(b'INBOX'), b'INBOX') + # Names with protocol-sensitive characters are quoted. + self.assertEqual(m._astring('New folder'), b'"New folder"') + self.assertEqual(m._astring('a"b'), b'"a\\"b"') + self.assertEqual(m._astring('a\\b'), b'"a\\\\b"') + self.assertEqual(m._astring(''), b'""') + self.assertEqual(m._astring('*'), b'"*"') + # A well-formed quoted string is passed through unchanged. + self.assertEqual(m._astring('"New folder"'), b'"New folder"') + self.assertEqual(m._astring('""'), b'""') + # Including a lenient (non-RFC) backslash escape, which the server + # may accept. + self.assertEqual(m._astring('"a\\b"'), b'"a\\b"') + # A string that only looks quoted but is not a single token is + # quoted as data, closing the argument injection vector. + self.assertEqual(m._astring('"a" SELECT evil "'), + b'"\\"a\\" SELECT evil \\""') + self.assertEqual(m._astring('"'), b'"\\""') + # Non-ASCII names are only allowed in a quoted string or a + # literal, never in an atom (RFC 6855). + m._encoding = 'utf-8' + self.assertEqual(m._astring('Entwürfe'), '"Entwürfe"'.encode()) + self.assertEqual(m._astring(b'Entw\xc3\xbcrfe'), b'"Entw\xc3\xbcrfe"') + + def test_astring_idempotent(self): + # Quoting an already quoted argument should not change it, so that + # quoting twice gives the same result as quoting once. + m = imaplib.IMAP4.__new__(imaplib.IMAP4) + m._encoding = 'ascii' + for arg in ['INBOX', 'New folder', 'a"b', 'a\\b', '', '*', '%', + '"New folder"', '""', '"a\\b"', '"a" SELECT evil "', + '"', 'a\tb', 'a\rb', '\x7f', '(a)', b'Entw\xc3\xbcrfe']: + with self.subTest(arg=arg): + once = m._astring(arg) + self.assertEqual(m._astring(once), once) + twice = m._list_mailbox(arg) + self.assertEqual(m._list_mailbox(twice), twice) + + def test_list_mailbox(self): + m = imaplib.IMAP4.__new__(imaplib.IMAP4) + m._encoding = 'ascii' + # Wildcards are not quoted in a list pattern. + self.assertEqual(m._list_mailbox('*'), b'*') + self.assertEqual(m._list_mailbox('%'), b'%') + self.assertEqual(m._list_mailbox('foo/%'), b'foo/%') + # But spaces still require quoting. + self.assertEqual(m._list_mailbox('New folder'), b'"New folder"') + self.assertEqual(m._list_mailbox('"New folder"'), b'"New folder"') + # As do non-ASCII names; wildcards keep their meaning inside a + # quoted string. + m._encoding = 'utf-8' + self.assertEqual(m._list_mailbox('Entwürfe/%'), + '"Entwürfe/%"'.encode()) + if ssl: class SecureTCPServer(socketserver.TCPServer): @@ -120,7 +263,7 @@ class SimpleIMAPHandler(socketserver.StreamRequestHandler): def setup(self): super().setup() - self.server.is_selected = False + self.server.is_selected = None self.server.logged = None def _send(self, message): @@ -137,9 +280,11 @@ class SimpleIMAPHandler(socketserver.StreamRequestHandler): def _send_tagged(self, tag, code, message): self._send_textline(' '.join((tag, code, message))) + welcome = '* OK IMAP4rev1' + def handle(self): # Send a welcome message. - self._send_textline('* OK IMAP4rev1') + self._send_textline(self.welcome) while 1: # Gather up input until we receive a line terminator or we timeout. # Accumulate read(1) because it's simpler to handle the differences @@ -166,10 +311,11 @@ class SimpleIMAPHandler(socketserver.StreamRequestHandler): except StopIteration: self.continuation = None continue - splitline = line.decode('ASCII').split() + splitline = splitargs(line.decode().removesuffix('\r\n')) tag = splitline[0] cmd = splitline[1] args = splitline[2:] + self.server.args = args if hasattr(self, 'cmd_' + cmd): continuation = getattr(self, 'cmd_' + cmd)(tag, args) @@ -192,17 +338,22 @@ class SimpleIMAPHandler(socketserver.StreamRequestHandler): self._send_tagged(tag, 'OK', 'LOGOUT completed') def cmd_LOGIN(self, tag, args): - self.server.logged = args[0] + self.server.logged = args self._send_tagged(tag, 'OK', 'LOGIN completed') def cmd_SELECT(self, tag, args): - self.server.is_selected = True + self.server.is_selected = args self._send_line(b'* 2 EXISTS') self._send_tagged(tag, 'OK', '[READ-WRITE] SELECT completed.') + def cmd_EXAMINE(self, tag, args): + self.server.is_selected = args + self._send_line(b'* 2 EXISTS') + self._send_tagged(tag, 'OK', '[READ-ONLY] EXAMINE completed.') + def cmd_UNSELECT(self, tag, args): - if self.server.is_selected: - self.server.is_selected = False + if self.server.is_selected is not None: + self.server.is_selected = None self._send_tagged(tag, 'OK', 'Returned to authenticated state. (Success)') else: self._send_tagged(tag, 'BAD', 'No mailbox selected') @@ -330,6 +481,39 @@ class NewIMAPTestsMixin: self.assertRaises(imaplib.IMAP4.abort, self.imap_class, *server.server_address) + def test_invalid_greeting(self): + # An invalid greeting, e.g. from a POP3 server on the IMAP port, + # must not fail with "error: None" but report the server's line + # (gh-108280). + class Pop3Handler(socketserver.StreamRequestHandler): + def handle(self): + self.wfile.write(b'+OK POP3 server ready\r\n') + _, server = self._setup(Pop3Handler, connect=False) + with self.assertRaisesRegex(imaplib.IMAP4.error, + r'invalid greeting: \+OK POP3 server ready'): + self.imap_class(*server.server_address) + + def test_invalid_greeting_untagged(self): + # An untagged greeting that is neither OK nor PREAUTH (e.g. BYE) + # is reported as is (gh-108280). + class ByeHandler(socketserver.StreamRequestHandler): + def handle(self): + self.wfile.write(b'* BYE Server unavailable\r\n') + _, server = self._setup(ByeHandler, connect=False) + with self.assertRaisesRegex(imaplib.IMAP4.error, + r'invalid greeting: \* BYE Server unavailable'): + self.imap_class(*server.server_address) + + def test_invalid_greeting_bare_continuation(self): + # A bare continuation greeting is still reported (gh-108280). + class BareHandler(socketserver.StreamRequestHandler): + def handle(self): + self.wfile.write(b'+\r\n') + _, server = self._setup(BareHandler, connect=False) + with self.assertRaisesRegex(imaplib.IMAP4.error, + r'invalid greeting: \+'): + self.imap_class(*server.server_address) + def test_line_termination(self): class BadNewlineHandler(SimpleIMAPHandler): def cmd_CAPABILITY(self, tag, args): @@ -372,10 +556,8 @@ class NewIMAPTestsMixin: self.server.response = yield self._send_tagged(tag, 'OK', 'FAKEAUTH successful') def cmd_APPEND(self, tag, args): - self._send_textline('+') self.server.response = args - literal = yield - self.server.response.append(literal) + self.server.response.append(_read_literal(self, args[-1])) literal = yield self.server.response.append(literal) self._send_tagged(tag, 'OK', 'okay') @@ -387,6 +569,7 @@ class NewIMAPTestsMixin: code, _ = client.enable('UTF8=ACCEPT') self.assertEqual(code, 'OK') self.assertEqual(client._encoding, 'utf-8') + self.assertEqual(server.args, ['UTF8=ACCEPT']) msg_string = 'Subject: üñí©öðé' typ, data = client.append( None, None, None, (msg_string + '\n').encode('utf-8')) @@ -414,6 +597,26 @@ class NewIMAPTestsMixin: with self.assertRaisesRegex(imaplib.IMAP4.error, 'charset.*UTF8'): client.search('foo', 'bar') + def test_utf8_mailbox_name(self): + class UTF8Server(SimpleIMAPHandler): + capabilities = 'AUTH ENABLE UTF8=ACCEPT' + def cmd_ENABLE(self, tag, args): + self._send_tagged(tag, 'OK', 'ENABLE successful') + def cmd_AUTHENTICATE(self, tag, args): + self._send_textline('+') + self.server.response = yield + self._send_tagged(tag, 'OK', 'FAKEAUTH successful') + client, server = self._setup(UTF8Server) + typ, _ = client.authenticate('MYAUTH', lambda x: b'fake') + self.assertEqual(typ, 'OK') + typ, _ = client.enable('UTF8=ACCEPT') + self.assertEqual(typ, 'OK') + # A non-ASCII mailbox name is only allowed in a quoted string + # or a literal, never in an atom (RFC 6855). + typ, _ = client.select('Entwürfe') + self.assertEqual(typ, 'OK') + self.assertEqual(server.is_selected, ['"Entwürfe"']) + def test_bad_auth_name(self): class MyServer(SimpleIMAPHandler): def cmd_AUTHENTICATE(self, tag, args): @@ -552,7 +755,7 @@ class NewIMAPTestsMixin: _, server = self._setup(SimpleIMAPHandler, connect=False) with self.imap_class(*server.server_address) as imap: imap.login('user', 'pass') - self.assertEqual(server.logged, 'user') + self.assertEqual(server.logged, ['user', '"pass"']) self.assertIsNone(server.logged) def test_with_statement_logout(self): @@ -560,7 +763,7 @@ class NewIMAPTestsMixin: _, server = self._setup(SimpleIMAPHandler, connect=False) with self.imap_class(*server.server_address) as imap: imap.login('user', 'pass') - self.assertEqual(server.logged, 'user') + self.assertEqual(server.logged, ['user', '"pass"']) imap.logout() self.assertIsNone(server.logged) self.assertIsNone(server.logged) @@ -635,11 +838,142 @@ class NewIMAPTestsMixin: self.fail('multi-packet response was corrupted by idle timeout') def test_login(self): - client, _ = self._setup(SimpleIMAPHandler) + client, server = self._setup(SimpleIMAPHandler) typ, data = client.login('user', 'pass') self.assertEqual(typ, 'OK') self.assertEqual(data[0], b'LOGIN completed') self.assertEqual(client.state, 'AUTH') + # The user name is quoted only when necessary, but the password + # is always quoted. + self.assertEqual(server.logged, ['user', '"pass"']) + self.assertRaises(imaplib.IMAP4.error, client.login, 'user', 'pass') + + def test_login_quoted(self): + client, server = self._setup(SimpleIMAPHandler) + typ, data = client.login('us*r', 'p%ss') + self.assertEqual(typ, 'OK') + self.assertEqual(data[0], b'LOGIN completed') + self.assertEqual(client.state, 'AUTH') + self.assertEqual(server.logged, ['"us*r"', '"p%ss"']) + + def test_login_quoted2(self): + # An already quoted user name is passed through unchanged, rather + # than being quoted a second time; the password is always quoted. + client, server = self._setup(SimpleIMAPHandler) + typ, data = client.login('"user"', '"pass"') + self.assertEqual(typ, 'OK') + self.assertEqual(data[0], b'LOGIN completed') + self.assertEqual(client.state, 'AUTH') + self.assertEqual(server.logged, ['"user"', r'"\"pass\""']) + + def test_append_line_endings(self): + # append() normalizes bare CR and LF in the message to CRLF. + class AppendHandler(SimpleIMAPHandler): + def cmd_APPEND(self, tag, args): + self.server.response = _read_literal(self, args[-1]) + yield # read the trailer line + self._send_tagged(tag, 'OK', 'APPEND completed') + client, server = self._setup(AppendHandler) + client.login('user', 'pass') + message = b'a\rb\nc\r\nd' + client.append('INBOX', None, None, message) + self.assertEqual(server.response, b'a\r\nb\r\nc\r\nd') + + # The mailbox is quoted and the flags are wrapped in parentheses + # when necessary. + client.append('New folder', r'\Seen', None, b'data') + self.assertEqual(server.args, ['"New folder"', r'(\Seen)', '{4}']) + + def test_login_capabilities(self): + # A server may advertise new capabilities after login (as an + # untagged CAPABILITY response); imaplib must refresh its cached + # capability list (gh-63121, gh-103451). + class CapabilityLoginHandler(SimpleIMAPHandler): + def cmd_LOGIN(self, tag, args): + self.server.logged = args[0] + self._send_textline('* CAPABILITY IMAP4rev1 ENABLE UTF8=ACCEPT') + self._send_tagged(tag, 'OK', 'LOGIN completed') + def cmd_ENABLE(self, tag, args): + self._send_tagged(tag, 'OK', 'ENABLE completed') + + client, _ = self._setup(CapabilityLoginHandler) + self.assertNotIn('ENABLE', client.capabilities) + client.login('user', 'pass') + self.assertIn('ENABLE', client.capabilities) + self.assertIn('UTF8=ACCEPT', client.capabilities) + typ, _ = client.enable('UTF8=ACCEPT') + self.assertEqual(typ, 'OK') + + def test_authenticate_capabilities(self): + # Capabilities are also refreshed after AUTHENTICATE, here from a + # CAPABILITY response code in the tagged OK response. + class CapabilityAuthHandler(SimpleIMAPHandler): + def cmd_AUTHENTICATE(self, tag, args): + self._send_textline('+') + self.server.response = yield + self._send_tagged( + tag, 'OK', + '[CAPABILITY IMAP4rev1 ENABLE] AUTHENTICATE completed') + + client, _ = self._setup(CapabilityAuthHandler) + self.assertNotIn('ENABLE', client.capabilities) + client.authenticate('MYAUTH', lambda x: b'fake') + self.assertIn('ENABLE', client.capabilities) + + def test_greeting_capabilities(self): + # Capabilities advertised in the greeting are used directly, + # without sending a separate CAPABILITY command. + class GreetingHandler(SimpleIMAPHandler): + welcome = '* OK [CAPABILITY IMAP4rev1 ENABLE] Server ready' + def cmd_CAPABILITY(self, tag, args): + self.server.capability_queried = True + super().cmd_CAPABILITY(tag, args) + + client, server = self._setup(GreetingHandler) + self.assertEqual(client.capabilities, ('IMAP4REV1', 'ENABLE')) + self.assertFalse(getattr(server, 'capability_queried', False)) + + def test_login_requery_capabilities(self): + # If the server does not advertise capabilities after login, + # imaplib re-queries them (as it does after STARTTLS), so a + # capability that becomes available only after authentication is + # still recognized (gh-63121). + class RequeryHandler(SimpleIMAPHandler): + def cmd_CAPABILITY(self, tag, args): + caps = 'IMAP4rev1 ENABLE' if self.server.logged else 'IMAP4rev1' + self._send_textline('* CAPABILITY ' + caps) + self._send_tagged(tag, 'OK', 'CAPABILITY completed') + + client, _ = self._setup(RequeryHandler) + self.assertNotIn('ENABLE', client.capabilities) + client.login('user', 'pass') + self.assertIn('ENABLE', client.capabilities) + + def test_readonly_error_reports_mailbox(self): + # The read-only error reports the mailbox via repr(), which also + # avoids BytesWarning for a bytes mailbox under -bb. + class ReadOnlyHandler(SimpleIMAPHandler): + def cmd_SELECT(self, tag, args): + self._send_line(b'* 2 EXISTS') + self._send_tagged(tag, 'OK', '[READ-ONLY] SELECT completed.') + client, _ = self._setup(ReadOnlyHandler) + client.login('user', 'pass') + for mailbox, expected in [('INBOX', "'INBOX'"), (b'INBOX', r"b'INBOX'")]: + with self.subTest(mailbox=mailbox): + with self.assertRaisesRegex(imaplib.IMAP4.readonly, + r"%s is not writable" % expected): + client.select(mailbox) + + def test_uid_unknown_command_reports_command(self): + # The unknown-UID-command error reports the command via repr(), which + # also avoids BytesWarning for a bytes command under -bb. + client, _ = self._setup(SimpleIMAPHandler) + for command, expected in [('BOGUS', "'BOGUS'"), (b'BOGUS', r"b'BOGUS'")]: + with self.subTest(command=command): + with self.assertRaisesRegex( + imaplib.IMAP4.error, + r"Unknown IMAP4 UID command: %s" % expected): + client.uid(command, '1') def test_logout(self): client, _ = self._setup(SimpleIMAPHandler) @@ -652,18 +986,88 @@ class NewIMAPTestsMixin: self.assertEqual(client.state, 'LOGOUT') def test_lsub(self): - class LsubCmd(SimpleIMAPHandler): - def cmd_LSUB(self, tag, args): - self._send_textline('* LSUB () "." directoryA') - return self._send_tagged(tag, 'OK', 'LSUB completed') - client, _ = self._setup(LsubCmd) + client, server = self._setup(make_simple_handler('LSUB', + ['* LSUB () "." directoryA', '* LSUB () "." directoryB'])) client.login('user', 'pass') typ, data = client.lsub() self.assertEqual(typ, 'OK') - self.assertEqual(data[0], b'() "." directoryA') + self.assertEqual(data, [b'() "." directoryA', b'() "." directoryB']) + self.assertEqual(server.args, ['""', '*']) + + typ, data = client.lsub('~/Mail/', '%') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['~/Mail/', '%']) + + # The directory is quoted when necessary; wildcards in the pattern + # are preserved. + typ, data = client.lsub('New folder', '%') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"', '%']) + + def test_extra_blank_line_after_literal(self): + # Some buggy servers send an extra blank line after the counted + # literal data. imaplib should skip it instead of failing. + class BlankLineHandler(SimpleIMAPHandler): + def cmd_FETCH(self, tag, args): + self._send(b'* 1 FETCH (BODY[HEADER] {13}\r\n') + self._send(b'Subject: test') # 13-byte literal + self._send(b'\r\n)\r\n') # stray blank line, then ')' + self._send_tagged(tag, 'OK', 'FETCH completed') + client, _ = self._setup(BlankLineHandler) + client.login('user', 'pass') + client.select() + typ, data = client.fetch('1', '(BODY[HEADER])') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [(b'1 (BODY[HEADER] {13}', b'Subject: test'), + b')']) + + def test_literal_terminating_response(self): + # A literal ending a response (a LIST mailbox name sent as a literal) + # has an empty trailer that must not be swallowed. Conforming case: + # no spurious blank lines. + names = [b'My (box)"', b'Another', b'Third'] + class Handler(SimpleIMAPHandler): + def cmd_LIST(self, tag, args): + for name in names: + self._send(b'* LIST (\\HasNoChildren) "/" {%d}\r\n' + % len(name)) + self._send(name) + self._send(b'\r\n') # ends the response, no blank + self._send_tagged(tag, 'OK', 'LIST completed') + client, _ = self._setup(Handler) + client.login('user', 'pass') + typ, data = client.list() + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + (b'(\\HasNoChildren) "/" {9}', b'My (box)"'), b'', + (b'(\\HasNoChildren) "/" {7}', b'Another'), b'', + (b'(\\HasNoChildren) "/" {5}', b'Third'), b'', + ]) + + def test_spurious_blank_lines_between_responses(self): + # A spurious blank line after each terminating literal falls between the + # untagged responses and must be skipped, even several in a row. + names = [b'My (box)"', b'Another', b'Third'] + class Handler(SimpleIMAPHandler): + def cmd_LIST(self, tag, args): + for name in names: + self._send(b'* LIST (\\HasNoChildren) "/" {%d}\r\n' + % len(name)) + self._send(name) + self._send(b'\r\n\r\n') # ends the response, then a blank + self._send_tagged(tag, 'OK', 'LIST completed') + client, _ = self._setup(Handler) + client.login('user', 'pass') + typ, data = client.list() + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + (b'(\\HasNoChildren) "/" {9}', b'My (box)"'), b'', + (b'(\\HasNoChildren) "/" {7}', b'Another'), b'', + (b'(\\HasNoChildren) "/" {5}', b'Third'), b'', + ]) def test_unselect(self): - client, _ = self._setup(SimpleIMAPHandler) + client, server = self._setup(SimpleIMAPHandler) client.login('user', 'pass') typ, data = client.select() self.assertEqual(typ, 'OK') @@ -673,6 +1077,757 @@ class NewIMAPTestsMixin: self.assertEqual(typ, 'OK') self.assertEqual(data[0], b'Returned to authenticated state. (Success)') self.assertEqual(client.state, 'AUTH') + self.assertIsNone(server.is_selected) + + def test_enable(self): + class EnableHandler(SimpleIMAPHandler): + capabilities = 'IMAP4rev1 ID LITERAL+ ENABLE X-GOOD-IDEA' + def cmd_ENABLE(self, tag, args): + capabilities = self.capabilities.split() + for arg in args: + if arg in capabilities: + self._send_textline('* ENABLED ' + arg) + self._send_tagged(tag, 'OK', 'foo') + + client, server = self._setup(EnableHandler) + client.login('user', 'pass') + code, data = client.enable('CONDSTORE X-GOOD-IDEA') + self.assertEqual(code, 'OK') + self.assertEqual(data, [b'foo']) + self.assertEqual(server.args, ['CONDSTORE', 'X-GOOD-IDEA']) + + def test_select(self): + client, server = self._setup(SimpleIMAPHandler) + client.login('user', 'pass') + typ, data = client.select() + self.assertEqual(typ, 'OK') + self.assertEqual(data[0], b'2') + self.assertEqual(server.is_selected, ['INBOX']) + + typ, data = client.select('Archive') + self.assertEqual(typ, 'OK') + self.assertEqual(server.is_selected, ['Archive']) + + # readonly=True issues EXAMINE instead of SELECT + # (there is no separate examine() method). + typ, data = client.select(readonly=True) + self.assertEqual(typ, 'OK') + self.assertEqual(data[0], b'2') + self.assertEqual(server.is_selected, ['INBOX']) + self.assertTrue(client.is_readonly) + + typ, data = client.select('New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.is_selected, ['"New folder"']) + + def test_expunge(self): + client, server = self._setup(make_simple_handler('EXPUNGE', + ['* 3 EXPUNGE', '* 3 EXPUNGE', '* 5 EXPUNGE', '* 8 EXPUNGE'])) + client.login('user', 'pass') + client.select() + typ, data = client.expunge() + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'3', b'3', b'5', b'8']) + + def test_close(self): + client, server = self._setup(make_simple_handler('CLOSE')) + client.login('user', 'pass') + client.select() + typ, data = client.close() + self.assertEqual(typ, 'OK') + self.assertEqual(client.state, 'AUTH') + + def test_check(self): + client, server = self._setup(make_simple_handler('CHECK')) + client.login('user', 'pass') + client.select() + typ, data = client.check() + self.assertEqual(typ, 'OK') + + def test_noop(self): + client, server = self._setup(make_simple_handler('NOOP', + ['* 4 EXISTS'])) + client.login('user', 'pass') + typ, data = client.noop() + self.assertEqual(typ, 'OK') + # NOOP is used to pick up server-pushed untagged responses. + self.assertEqual(client.untagged_responses['EXISTS'], [b'4']) + + def test_namespace(self): + client, server = self._setup(make_simple_handler('NAMESPACE', + ['* NAMESPACE (("" "/")) NIL NIL'])) + client.login('user', 'pass') + typ, data = client.namespace() + self.assertEqual(typ, 'OK') + self.assertEqual(data[0], b'(("" "/")) NIL NIL') + + def test_capability(self): + client, server = self._setup(make_simple_handler('CAPABILITY', + ['* CAPABILITY IMAP4rev1 IDLE NAMESPACE'])) + client.login('user', 'pass') + typ, data = client.capability() + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'IMAP4rev1 IDLE NAMESPACE']) + + def test_recent(self): + # recent() prods the server with NOOP if no RECENT was seen yet. + client, server = self._setup(make_simple_handler('NOOP', + ['* 5 RECENT'])) + client.login('user', 'pass') + typ, data = client.recent() + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'5']) + + def test_response(self): + client, server = self._setup(SimpleIMAPHandler) + client.login('user', 'pass') + client.select() # the handler answers with '* 2 EXISTS' + typ, data = client.response('EXISTS') + self.assertEqual(typ, 'EXISTS') + self.assertEqual(data, [b'2']) + # The value is cleared once read. + typ, data = client.response('EXISTS') + self.assertEqual(data, [None]) + + def test_create(self): + client, server = self._setup(make_simple_handler('CREATE')) + client.login('user', 'pass') + typ, data = client.create('owatagusiam/') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'CREATE completed']) + self.assertEqual(server.args, ['owatagusiam/']) + + typ, data = client.create('owatagusiam/blurdybloop') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'CREATE completed']) + self.assertEqual(server.args, ['owatagusiam/blurdybloop']) + + typ, data = client.create('New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'CREATE completed']) + self.assertEqual(server.args, ['"New folder"']) + + def test_copy(self): + client, server = self._setup(make_simple_handler('COPY')) + client.login('user', 'pass') + client.select() + typ, data = client.copy('2:4', 'MEETING') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'COPY completed']) + self.assertEqual(server.args, ['2:4', 'MEETING']) + + typ, data = client.copy('2:4', 'New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'COPY completed']) + self.assertEqual(server.args, ['2:4', '"New folder"']) + + def test_uid_copy(self): + client, server = self._setup(make_simple_handler('UID', + completed='UID COPY completed')) + client.login('user', 'pass') + client.select() + typ, data = client.uid('copy', '4827313:4828442', 'MEETING') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [None]) + self.assertEqual(server.args, ['COPY', '4827313:4828442', 'MEETING']) + + typ, data = client.uid('copy', '4827313:4828442', 'New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [None]) + self.assertEqual(server.args, ['COPY', '4827313:4828442', '"New folder"']) + + def test_store(self): + client, server = self._setup(make_simple_handler('STORE', [ + r'* 2 FETCH (FLAGS (\Deleted \Seen))', + r'* 3 FETCH (FLAGS (\Deleted))', + r'* 4 FETCH (FLAGS (\Deleted \Flagged \Seen))', + ])) + client.login('user', 'pass') + client.select() + typ, data = client.store('2:4', '+FLAGS', r'(\Deleted)') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + br'2 (FLAGS (\Deleted \Seen))', + br'3 (FLAGS (\Deleted))', + br'4 (FLAGS (\Deleted \Flagged \Seen))', + ]) + self.assertEqual(server.args, ['2:4', '+FLAGS', r'(\Deleted)']) + + typ, data = client.store('2:4', '+FLAGS', r'\Deleted') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['2:4', '+FLAGS', r'(\Deleted)']) + + def test_uid_store(self): + client, server = self._setup(make_simple_handler('UID', [ + r'* 23 FETCH (FLAGS (\Deleted \Seen) UID 4827313)', + r'* 24 FETCH (FLAGS (\Deleted) UID 4827943)', + r'* 25 FETCH (FLAGS (\Deleted \Flagged \Seen) UID 4828442)', + ], 'UID STORE completed')) + client.login('user', 'pass') + client.select() + typ, data = client.uid('store', '4827313:4828442', '+FLAGS', r'(\Deleted)') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + br'23 (FLAGS (\Deleted \Seen) UID 4827313)', + br'24 (FLAGS (\Deleted) UID 4827943)', + br'25 (FLAGS (\Deleted \Flagged \Seen) UID 4828442)', + ]) + self.assertEqual(server.args, ['STORE', '4827313:4828442', '+FLAGS', r'(\Deleted)']) + + typ, data = client.uid('store', '4827313:4828442', '+FLAGS', r'\Deleted') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['STORE', '4827313:4828442', '+FLAGS', r'(\Deleted)']) + + def test_fetch(self): + # The handler expands the requested sequence set and answers for + # exactly those messages, so the test exercises the round trip of + # the message set, not just a canned reply. + class FetchHandler(SimpleIMAPHandler): + messages = 4 + def cmd_SELECT(self, tag, args): + self.server.is_selected = args + self._send_line(b'* %d EXISTS' % self.messages) + self._send_tagged(tag, 'OK', '[READ-WRITE] SELECT completed.') + def cmd_FETCH(self, tag, args): + for n in parse_sequence_set(args[0], self.messages): + self._send_textline(r'* %d FETCH (FLAGS (\Seen))' % n) + self._send_tagged(tag, 'OK', 'FETCH completed') + + client, server = self._setup(FetchHandler) + client.login('user', 'pass') + client.select() + typ, data = client.fetch('2:4', '(FLAGS)') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + br'2 (FLAGS (\Seen))', + br'3 (FLAGS (\Seen))', + br'4 (FLAGS (\Seen))', + ]) + self.assertEqual(server.args, ['2:4', '(FLAGS)']) + + # A comma-separated set with an open range up to '*'. + typ, data = client.fetch('1,3:*', '(FLAGS)') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + br'1 (FLAGS (\Seen))', + br'3 (FLAGS (\Seen))', + br'4 (FLAGS (\Seen))', + ]) + self.assertEqual(server.args, ['1,3:*', '(FLAGS)']) + + # An item with nested parentheses is sent (and parsed) as a + # single argument. + typ, data = client.fetch('1', '(BODY[HEADER.FIELDS (DATE FROM)])') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [br'1 (FLAGS (\Seen))']) + self.assertEqual(server.args, ['1', '(BODY[HEADER.FIELDS (DATE FROM)])']) + + # message_parts is wrapped in parentheses if it is not already. + typ, data = client.fetch('2:4', 'FLAGS') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['2:4', '(FLAGS)']) + + # But the macros are not, as they are not data item names. + typ, data = client.fetch('2:4', 'ALL') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['2:4', 'ALL']) + typ, data = client.fetch('2:4', 'fast') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['2:4', 'fast']) + + def test_uid_fetch(self): + client, server = self._setup(make_simple_handler('UID', [ + r'* 23 FETCH (FLAGS (\Seen) UID 4827313)', + r'* 24 FETCH (FLAGS (\Seen) UID 4827943)', + r'* 25 FETCH (FLAGS (\Seen) UID 4828442)', + ], 'UID FETCH completed')) + client.login('user', 'pass') + client.select() + typ, data = client.uid('fetch', '4827313:4828442', '(FLAGS)') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + br'23 (FLAGS (\Seen) UID 4827313)', + br'24 (FLAGS (\Seen) UID 4827943)', + br'25 (FLAGS (\Seen) UID 4828442)', + ]) + self.assertEqual(server.args, ['FETCH', '4827313:4828442', '(FLAGS)']) + + typ, data = client.uid('fetch', '4827313:4828442', 'FLAGS') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['FETCH', '4827313:4828442', '(FLAGS)']) + + typ, data = client.uid('fetch', '4827313:4828442', 'ALL') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['FETCH', '4827313:4828442', 'ALL']) + + def test_partial(self): + client, server = self._setup(make_simple_handler('PARTIAL', + ['* 1 FETCH (RFC822.TEXT<0.10> "0123456789")'])) + client.login('user', 'pass') + client.select() + typ, data = client.partial('1', 'RFC822.TEXT', '0', '10') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'1 (RFC822.TEXT<0.10> "0123456789")']) + self.assertEqual(server.args, ['1', 'RFC822.TEXT', '0', '10']) + + def test_search(self): + response = [] + client, server = self._setup(make_simple_handler('SEARCH', response)) + client.login('user', 'pass') + client.select() + response[:] = ['* SEARCH 2 84 882'] + typ, data = client.search(None, 'FLAGGED', 'SINCE', '1-Feb-1994', 'NOT', 'FROM', '"Smith"') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'2 84 882']) + self.assertEqual(server.args, ['FLAGGED', 'SINCE', '1-Feb-1994', 'NOT', 'FROM', '"Smith"']) + + response[:] = ['* SEARCH'] + typ, data = client.search(None, 'TEXT', '"string not in mailbox"') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'']) + self.assertEqual(server.args, ['TEXT', '"string not in mailbox"']) + + response[:] = ['* SEARCH 43'] + typ, data = client.search('UTF-8', 'TEXT', 'XXXXXX') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'43']) + self.assertEqual(server.args, ['CHARSET', 'UTF-8', 'TEXT', 'XXXXXX']) + + typ, data = client.search('NF_Z_62-010_(1973)', 'TEXT', 'XXXXXX') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['CHARSET', '"NF_Z_62-010_(1973)"', 'TEXT', 'XXXXXX']) + + def test_uid_search(self): + response = [] + client, server = self._setup(make_simple_handler('UID', response, + 'UID SEARCH completed')) + client.login('user', 'pass') + client.select() + response[:] = ['* SEARCH 2 84 882'] + typ, data = client.uid('SEARCH', 'FLAGGED', 'SINCE', '1-Feb-1994', 'NOT', 'FROM', '"Smith"') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'2 84 882']) + self.assertEqual(server.args, ['SEARCH', 'FLAGGED', 'SINCE', '1-Feb-1994', 'NOT', 'FROM', '"Smith"']) + + response[:] = ['* SEARCH'] + typ, data = client.uid('SEARCH', 'TEXT', '"string not in mailbox"') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'']) + self.assertEqual(server.args, ['SEARCH', 'TEXT', '"string not in mailbox"']) + + response[:] = ['* SEARCH 43'] + typ, data = client.uid('SEARCH', 'CHARSET', 'UTF-8', 'TEXT', 'XXXXXX') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'43']) + self.assertEqual(server.args, ['SEARCH', 'CHARSET', 'UTF-8', 'TEXT', 'XXXXXX']) + + typ, data = client.uid('SEARCH', 'CHARSET', '"NF_Z_62-010_(1973)"', 'TEXT', 'XXXXXX') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['SEARCH', 'CHARSET', '"NF_Z_62-010_(1973)"', 'TEXT', 'XXXXXX']) + + def test_sort(self): + response = [] + client, server = self._setup(make_simple_handler('SORT', response)) + client.login('user', 'pass') + client.select() + response[:] = ['* SORT 2 84 882'] + typ, data = client.sort('(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [br'2 84 882']) + self.assertEqual(server.args, ['(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994']) + + response[:] = ['* SORT 5 3 4 1 2'] + typ, data = client.sort('(SUBJECT REVERSE DATE)', 'UTF-8', 'ALL') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [br'5 3 4 1 2']) + self.assertEqual(server.args, ['(SUBJECT REVERSE DATE)', 'UTF-8', 'ALL']) + + response[:] = ['* SORT'] + typ, data = client.sort('(SUBJECT)', 'US-ASCII', 'TEXT', '"not in mailbox"') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [br'']) + self.assertEqual(server.args, ['(SUBJECT)', 'US-ASCII', 'TEXT', '"not in mailbox"']) + + typ, data = client.sort('SUBJECT', 'NF_Z_62-010_(1973)', 'TEXT', '"not in mailbox"') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['(SUBJECT)', '"NF_Z_62-010_(1973)"', 'TEXT', '"not in mailbox"']) + + def test_uid_sort(self): + response = [] + client, server = self._setup(make_simple_handler('UID', response, + 'UID SORT completed')) + client.login('user', 'pass') + client.select() + response[:] = ['* SORT 2 84 882'] + typ, data = client.uid('sort', '(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [br'2 84 882']) + self.assertEqual(server.args, ['SORT', '(SUBJECT)', 'UTF-8', 'SINCE', '1-Feb-1994']) + + response[:] = ['* SORT 5 3 4 1 2'] + typ, data = client.uid('sort', '(SUBJECT REVERSE DATE)', 'UTF-8', 'ALL') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [br'5 3 4 1 2']) + self.assertEqual(server.args, ['SORT', '(SUBJECT REVERSE DATE)', 'UTF-8', 'ALL']) + + response[:] = ['* SORT'] + typ, data = client.uid('sort', '(SUBJECT)', 'US-ASCII', 'TEXT', '"not in mailbox"') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [br'']) + self.assertEqual(server.args, ['SORT', '(SUBJECT)', 'US-ASCII', 'TEXT', '"not in mailbox"']) + + typ, data = client.uid('sort', 'SUBJECT', 'NF_Z_62-010_(1973)', 'TEXT', '"not in mailbox"') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['SORT', '(SUBJECT)', '"NF_Z_62-010_(1973)"', 'TEXT', '"not in mailbox"']) + + def test_thread(self): + response = [] + client, server = self._setup(make_simple_handler('THREAD', response)) + client.login('user', 'pass') + client.select() + response[:] = [ + '* THREAD (166)(167)(168)(169)(172)(170)(171)' + '(173)(174 (175)(176)(178)(181)(180))(179)(177 ' + '(183)(182)(188)(184)(185)(186)(187)(189))(190)' + '(191)(192)(193)(194 195)(196 (197)(198))(199)' + '(200 202)(201)(203)(204)(205)(206 207)(208)'] + typ, data = client.thread('ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + b'(166)(167)(168)(169)(172)(170)(171)' + b'(173)(174 (175)(176)(178)(181)(180))(179)(177 ' + b'(183)(182)(188)(184)(185)(186)(187)(189))(190)' + b'(191)(192)(193)(194 195)(196 (197)(198))(199)' + b'(200 202)(201)(203)(204)(205)(206 207)(208)']) + self.assertEqual(server.args, ['ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000']) + + response[:] = [ + '* THREAD (166)(167)(168)(169)(172)((170)(179))' + '(171)(173)((174)(175)(176)(178)(181)(180))' + '((177)(183)(182)(188 (184)(189))(185 186)(187))' + '(190)(191)(192)(193)((194)(195 196))(197 198)' + '(199)(200 202)(201)(203)(204)(205 206 207)(208)'] + typ, data = client.thread('ORDEREDSUBJECT', 'US-ASCII', 'TEXT', '"gewp"') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + b'(166)(167)(168)(169)(172)((170)(179))' + b'(171)(173)((174)(175)(176)(178)(181)(180))' + b'((177)(183)(182)(188 (184)(189))(185 186)(187))' + b'(190)(191)(192)(193)((194)(195 196))(197 198)' + b'(199)(200 202)(201)(203)(204)(205 206 207)(208)']) + self.assertEqual(server.args, ['ORDEREDSUBJECT', 'US-ASCII', 'TEXT', '"gewp"']) + + typ, data = client.thread('ORDEREDSUBJECT', 'NF_Z_62-010_(1973)', 'TEXT', '"gewp"') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['ORDEREDSUBJECT', '"NF_Z_62-010_(1973)"', 'TEXT', '"gewp"']) + + def test_uid_thread(self): + response = [] + client, server = self._setup(make_simple_handler('UID', response, + 'UID THREAD completed')) + client.login('user', 'pass') + client.select() + response[:] = [ + '* THREAD (166)(167)(168)(169)(172)(170)(171)' + '(173)(174 (175)(176)(178)(181)(180))(179)(177 ' + '(183)(182)(188)(184)(185)(186)(187)(189))(190)' + '(191)(192)(193)(194 195)(196 (197)(198))(199)' + '(200 202)(201)(203)(204)(205)(206 207)(208)'] + typ, data = client.uid('THREAD', 'ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + b'(166)(167)(168)(169)(172)(170)(171)' + b'(173)(174 (175)(176)(178)(181)(180))(179)(177 ' + b'(183)(182)(188)(184)(185)(186)(187)(189))(190)' + b'(191)(192)(193)(194 195)(196 (197)(198))(199)' + b'(200 202)(201)(203)(204)(205)(206 207)(208)']) + self.assertEqual(server.args, ['THREAD', 'ORDEREDSUBJECT', 'UTF-8', 'SINCE', '5-MAR-2000']) + + response[:] = [ + '* THREAD (166)(167)(168)(169)(172)((170)(179))' + '(171)(173)((174)(175)(176)(178)(181)(180))' + '((177)(183)(182)(188 (184)(189))(185 186)(187))' + '(190)(191)(192)(193)((194)(195 196))(197 198)' + '(199)(200 202)(201)(203)(204)(205 206 207)(208)'] + typ, data = client.uid('THREAD', 'ORDEREDSUBJECT', 'US-ASCII', 'TEXT', '"gewp"') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [ + b'(166)(167)(168)(169)(172)((170)(179))' + b'(171)(173)((174)(175)(176)(178)(181)(180))' + b'((177)(183)(182)(188 (184)(189))(185 186)(187))' + b'(190)(191)(192)(193)((194)(195 196))(197 198)' + b'(199)(200 202)(201)(203)(204)(205 206 207)(208)']) + self.assertEqual(server.args, ['THREAD', 'ORDEREDSUBJECT', 'US-ASCII', 'TEXT', '"gewp"']) + + typ, data = client.uid('THREAD', 'ORDEREDSUBJECT', 'NF_Z_62-010_(1973)', 'TEXT', '"gewp"') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['THREAD', 'ORDEREDSUBJECT', '"NF_Z_62-010_(1973)"', 'TEXT', '"gewp"']) + + def test_delete(self): + client, server = self._setup(make_simple_handler('DELETE')) + client.login('user', 'pass') + typ, data = client.delete('blurdybloop') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'DELETE completed']) + self.assertEqual(server.args, ['blurdybloop']) + + typ, data = client.delete('foo/bar') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['foo/bar']) + + typ, data = client.delete('New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"']) + + def test_rename(self): + client, server = self._setup(make_simple_handler('RENAME')) + client.login('user', 'pass') + typ, data = client.rename('blurdybloop', 'sarasoop') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'RENAME completed']) + self.assertEqual(server.args, ['blurdybloop', 'sarasoop']) + + typ, data = client.rename('Old folder', 'New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"Old folder"', '"New folder"']) + + def test_subscribe(self): + client, server = self._setup(make_simple_handler('SUBSCRIBE')) + client.login('user', 'pass') + typ, data = client.subscribe('#news.comp.mail.mime') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'SUBSCRIBE completed']) + self.assertEqual(server.args, ['#news.comp.mail.mime']) + + typ, data = client.subscribe('New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"']) + + def test_unsubscribe(self): + client, server = self._setup(make_simple_handler('UNSUBSCRIBE')) + client.login('user', 'pass') + typ, data = client.unsubscribe('#news.comp.mail.mime') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'UNSUBSCRIBE completed']) + self.assertEqual(server.args, ['#news.comp.mail.mime']) + + typ, data = client.unsubscribe('New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"']) + + def test_list(self): + client, server = self._setup(make_simple_handler('LIST', + [r'* LIST (\Noselect) "/" ""', + r'* LIST (\Unmarked) "/" "~/Mail/foo"'])) + client.login('user', 'pass') + typ, data = client.list() + self.assertEqual(typ, 'OK') + self.assertEqual(data, [br'(\Noselect) "/" ""', + br'(\Unmarked) "/" "~/Mail/foo"']) + self.assertEqual(server.args, ['""', '*']) + + typ, data = client.list('~/Mail/', '%') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['~/Mail/', '%']) + + typ, data = client.list('New folder', '*') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"', '*']) + + # A pattern without wildcards is quoted when necessary. + typ, data = client.list('~/Mail/', 'My Folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['~/Mail/', '"My Folder"']) + + def test_status(self): + client, server = self._setup(make_simple_handler('STATUS', + ['* STATUS blurdybloop (MESSAGES 231 UIDNEXT 44292)'])) + client.login('user', 'pass') + typ, data = client.status('blurdybloop', '(UIDNEXT MESSAGES)') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'blurdybloop (MESSAGES 231 UIDNEXT 44292)']) + self.assertEqual(server.args, ['blurdybloop', '(UIDNEXT MESSAGES)']) + + # The names argument is wrapped in parentheses if it is not already. + typ, data = client.status('New folder', 'UIDNEXT MESSAGES') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"', '(UIDNEXT MESSAGES)']) + + def test_getacl(self): + client, server = self._setup(make_simple_handler('GETACL', + ['* ACL INBOX Fred rwipslxetad'])) + client.login('user', 'pass') + typ, data = client.getacl('INBOX') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'INBOX Fred rwipslxetad']) + self.assertEqual(server.args, ['INBOX']) + + typ, data = client.getacl('New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"']) + + def test_setacl(self): + client, server = self._setup(make_simple_handler('SETACL')) + client.login('user', 'pass') + typ, data = client.setacl('INBOX', 'Fred', 'rwipslxetad') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'SETACL completed']) + self.assertEqual(server.args, ['INBOX', 'Fred', 'rwipslxetad']) + + typ, data = client.setacl('New folder', 'Fred', '+lr') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"', 'Fred', '+lr']) + + # The identifier and the rights are quoted when necessary too. + typ, data = client.setacl('INBOX', 'John Doe', 'a b') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['INBOX', '"John Doe"', '"a b"']) + + def test_deleteacl(self): + client, server = self._setup(make_simple_handler('DELETEACL')) + client.login('user', 'pass') + typ, data = client.deleteacl('INBOX', 'Fred') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'DELETEACL completed']) + self.assertEqual(server.args, ['INBOX', 'Fred']) + + # The identifier is quoted when necessary too. + typ, data = client.deleteacl('New folder', 'John Doe') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"', '"John Doe"']) + + def test_myrights(self): + client, server = self._setup(make_simple_handler('MYRIGHTS', + ['* MYRIGHTS INBOX rwiptsldaex'])) + client.login('user', 'pass') + typ, data = client.myrights('INBOX') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'INBOX rwiptsldaex']) + self.assertEqual(server.args, ['INBOX']) + + typ, data = client.myrights('New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"']) + + def test_getquota(self): + client, server = self._setup(make_simple_handler('GETQUOTA', + ['* QUOTA "" (STORAGE 10 512)'])) + client.login('user', 'pass') + typ, data = client.getquota('#news') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'"" (STORAGE 10 512)']) + self.assertEqual(server.args, ['#news']) + + typ, data = client.getquota('') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['""']) + + def test_getquotaroot(self): + client, server = self._setup(make_simple_handler('GETQUOTAROOT', + ['* QUOTAROOT INBOX ""', '* QUOTA "" (STORAGE 10 512)'])) + client.login('user', 'pass') + typ, data = client.getquotaroot('INBOX') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [[b'INBOX ""'], [b'"" (STORAGE 10 512)']]) + self.assertEqual(server.args, ['INBOX']) + + typ, data = client.getquotaroot('New folder') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"']) + + def test_setquota(self): + client, server = self._setup(make_simple_handler('SETQUOTA', + ['* QUOTA "" (STORAGE 512)'])) + client.login('user', 'pass') + typ, data = client.setquota('#news', '(STORAGE 512)') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'"" (STORAGE 512)']) + self.assertEqual(server.args, ['#news', '(STORAGE 512)']) + + typ, data = client.setquota('', '(STORAGE 512)') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['""', '(STORAGE 512)']) + + # The limits argument is wrapped in parentheses if it is not already. + typ, data = client.setquota('', 'STORAGE 512') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['""', '(STORAGE 512)']) + + def test_getannotation(self): + client, server = self._setup(make_simple_handler('GETANNOTATION', + ['* ANNOTATION INBOX "/comment" ("value.shared" "Hello")'])) + client.login('user', 'pass') + typ, data = client.getannotation('INBOX', '/comment', 'value.shared') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'INBOX "/comment" ("value.shared" "Hello")']) + self.assertEqual(server.args, ['INBOX', '/comment', 'value.shared']) + + typ, data = client.getannotation('New folder', '/comment', 'value.shared') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"New folder"', '/comment', 'value.shared']) + + def test_setannotation(self): + client, server = self._setup(make_simple_handler('SETANNOTATION')) + client.login('user', 'pass') + typ, data = client.setannotation('INBOX', '/comment', + '("value.shared" "My comment")') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, + ['INBOX', '/comment', '("value.shared" "My comment")']) + + typ, data = client.setannotation('New folder', '/comment', + '("value.shared" "My comment")') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, + ['"New folder"', '/comment', '("value.shared" "My comment")']) + + def test_proxyauth(self): + client, server = self._setup(make_simple_handler('PROXYAUTH')) + client.login('user', 'pass') + typ, data = client.proxyauth('user') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'PROXYAUTH completed']) + self.assertEqual(server.args, ['user']) + + typ, data = client.proxyauth('us er') + self.assertEqual(typ, 'OK') + self.assertEqual(server.args, ['"us er"']) + + def test_xatom(self): + client, server = self._setup(make_simple_handler('MYCOMMAND', + completed='MYCOMMAND completed')) + client.login('user', 'pass') + self.addCleanup(imaplib.Commands.pop, 'MYCOMMAND', None) + typ, data = client.xatom('MYCOMMAND', 'arg1', 'arg2') + self.assertEqual(typ, 'OK') + self.assertEqual(data, [b'MYCOMMAND completed']) + self.assertEqual(server.args, ['arg1', 'arg2']) + + def test_uppercase_command_names(self): + client, server = self._setup(SimpleIMAPHandler) + client.login('user', 'pass') + self.assertEqual(client.CAPABILITY, client.capability) + self.assertEqual(client.SELECT, client.select) + typ, data = client.CAPABILITY() + self.assertEqual(typ, 'OK') + with self.assertRaises(AttributeError): + client.NONEXISTENT + + def test_control_characters(self): + client, server = self._setup(SimpleIMAPHandler) + client.login('user', 'pass') + for c in '\0\r\n': + with self.assertRaises(ValueError): + client.select(f'a{c}b') + # Other control characters are valid in a quoted string and can + # occur in mailbox names returned by the server, so the client + # must be able to send them back. + for c in support.control_characters_c0(): + if c in '\0\r\n': + continue + typ, _ = client.select(f'a{c}b') + self.assertEqual(typ, 'OK') + self.assertEqual(server.is_selected, [f'"a{c}b"']) # property tests @@ -791,8 +1946,8 @@ class ThreadedNetworkedTests(unittest.TestCase): @threading_helper.reap_threads def test_bracket_flags(self): - # This violates RFC 3501, which disallows ']' characters in tag names, - # but imaplib has allowed producing such tags forever, other programs + # This violates RFC 3501, which disallows ']' characters in flags, + # but imaplib has allowed producing such flags forever, other programs # also produce them (eg: OtherInbox's Organizer app as of 20140716), # and Gmail, for example, accepts them and produces them. So we # support them. See issue #21815. @@ -904,10 +2059,8 @@ class ThreadedNetworkedTests(unittest.TestCase): class UTF8AppendServer(self.UTF8Server): def cmd_APPEND(self, tag, args): - self._send_textline('+') self.server.response = args - literal = yield - self.server.response.append(literal) + self.server.response.append(_read_literal(self, args[-1])) literal = yield self.server.response.append(literal) self._send_tagged(tag, 'OK', 'okay') @@ -921,6 +2074,7 @@ class ThreadedNetworkedTests(unittest.TestCase): code, _ = client.enable('UTF8=ACCEPT') self.assertEqual(code, 'OK') self.assertEqual(client._encoding, 'utf-8') + self.assertEqual(server.args, ['UTF8=ACCEPT']) msg_string = 'Subject: üñí©öðé' typ, data = client.append( None, None, None, (msg_string + '\n').encode('utf-8')) @@ -1076,7 +2230,7 @@ class ThreadedNetworkedTests(unittest.TestCase): with self.reaped_server(SimpleIMAPHandler) as server: with self.imap_class(*server.server_address) as imap: imap.login('user', 'pass') - self.assertEqual(server.logged, 'user') + self.assertEqual(server.logged, ['user', '"pass"']) self.assertIsNone(server.logged) @threading_helper.reap_threads @@ -1085,7 +2239,7 @@ class ThreadedNetworkedTests(unittest.TestCase): with self.reaped_server(SimpleIMAPHandler) as server: with self.imap_class(*server.server_address) as imap: imap.login('user', 'pass') - self.assertEqual(server.logged, 'user') + self.assertEqual(server.logged, ['user', '"pass"']) imap.logout() self.assertIsNone(server.logged) self.assertIsNone(server.logged) diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py index d974c7d46..f3b61d9fb 100644 --- a/Lib/test/test_tarfile.py +++ b/Lib/test/test_tarfile.py @@ -549,6 +549,53 @@ class CommonReadTest(ReadTest): self.assertIs(fobj.seekable(), True) +class ReadSizeRecorder(io.BytesIO): + # Records the largest size ever passed to read(), so a test can check + # that tarfile does not request far more data than the archive holds + # (which on a real file would pre-allocate it). + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.max_read_size = 0 + + def read(self, size=-1): + if size is not None and size >= 0: + self.max_read_size = max(self.max_read_size, size) + return super().read(size) + + +@support.cpython_only +class ExtendedHeaderMemoryTest(unittest.TestCase): + # gh-151497: the size of a GNU long name/link or a pax extended header is + # read from the archive and is untrusted. A crafted header can claim a + # size far larger than the file actually contains; opening such an archive + # must not try to read (and so pre-allocate) the claimed size in one go. + + def crafted_archive(self, hdrtype): + tarinfo = tarfile.TarInfo("A") + tarinfo.type = hdrtype + tarinfo.size = 0xFFFFFFFF # ~4 GiB claimed in a 512-byte header + return tarinfo.tobuf(format=tarfile.GNU_FORMAT) + + def check(self, hdrtype): + fobj = ReadSizeRecorder(self.crafted_archive(hdrtype)) + try: + with tarfile.open(fileobj=fobj, mode="r:") as tar: + tar.getmembers() + except tarfile.ReadError: + pass # a truncated header is fine; we only check the allocation + # The bogus ~4 GiB size must never reach a single read() call. + self.assertLessEqual(fobj.max_read_size, tarfile._EXTHEADER_READ_CHUNK) + + def test_gnu_longname_oversized_size(self): + self.check(tarfile.GNUTYPE_LONGNAME) + + def test_gnu_longlink_oversized_size(self): + self.check(tarfile.GNUTYPE_LONGLINK) + + def test_pax_header_oversized_size(self): + self.check(tarfile.XHDTYPE) + + class MiscReadTestBase(CommonReadTest): is_stream = False @@ -4344,6 +4391,30 @@ class TestExtractionFilters(unittest.TestCase): self.expect_file("boom", symlink_to='../../link_here') self.expect_file("c", symlink_to='b') + @symlink_test + def test_sneaky_hardlink_fallback_deep(self): + # (CVE-2026-11940) + with ArchiveMaker() as arc: + arc.add("a/b/s", symlink_to=os.path.join("..", "escape")) + arc.add("s", hardlink_to=os.path.join("a", "b", "s")) + + with self.check_context(arc.open(), 'data'): + e = self.expect_exception( + tarfile.LinkFallbackError, + "link 's' would be extracted as a copy of " + + "'a/b/s', which was rejected") + self.assertIsInstance(e.__cause__, + tarfile.LinkOutsideDestinationError) + + for filter in 'tar', 'fully_trusted': + with self.subTest(filter), self.check_context(arc.open(), filter): + if not os_helper.can_symlink(): + self.expect_file("a/") + self.expect_file("a/b/") + else: + self.expect_file("a/b/s", symlink_to=os.path.join('..', 'escape')) + self.expect_file("s", symlink_to=os.path.join('..', 'escape')) + @symlink_test def test_exfiltration_via_symlink(self): # (CVE-2025-4138) @@ -4399,6 +4470,98 @@ class TestExtractionFilters(unittest.TestCase): st_mode = cc.outerdir.stat().st_mode self.assertNotEqual(st_mode & 0o777, 0o777) + @symlink_test + @unittest.skipUnless(hasattr(os, 'chown'), "missing os.chown") + @unittest.skipUnless(hasattr(os, 'lchown'), "missing os.lchown") + @unittest.skipUnless(hasattr(os, 'geteuid'), "missing os.geteuid") + @support.subTests('link_type', (tarfile.SYMTYPE, tarfile.LNKTYPE)) + def test_chown_links_on_extract(self, link_type): + with ArchiveMaker() as arc: + arc.add("test.txt", + uid=1337, gid=1337, uname="", gname="", mode='-rwxr-xr-x') + arc.add("link", + type=link_type, + linkname='test.txt', + uid=1337, gid=1337, uname="", gname="", mode='-rwxr-xr-x') + + with ( + os_helper.temp_dir() as tmpdir, + arc.open() as tar, + unittest.mock.patch("os.chown") as mock_chown, + unittest.mock.patch("os.lchown") as mock_lchown, + unittest.mock.patch("os.geteuid") as mock_geteuid, + ): + # Set UID to 0 so chown() is attempted. + mock_geteuid.return_value = 0 + tar.extract("link", path=tmpdir, filter='data') + extract_path = os.path.join(tmpdir, "link") + + if link_type == tarfile.SYMTYPE: + mock_chown.assert_not_called() + mock_lchown.assert_called_once_with(extract_path, -1, -1) + else: + mock_chown.assert_has_calls([ + unittest.mock.call(extract_path, -1, -1), + unittest.mock.call(extract_path, -1, -1) + ]) + mock_lchown.assert_not_called() + + @symlink_test + @unittest.skipUnless(hasattr(os, 'chown'), "missing os.chown") + @unittest.skipUnless(hasattr(os, 'lchown'), "missing os.lchown") + @unittest.skipUnless(hasattr(os, 'geteuid'), "missing os.geteuid") + @support.subTests('link_type', (tarfile.SYMTYPE, tarfile.LNKTYPE)) + def test_chown_links_on_extractall(self, link_type): + with ArchiveMaker() as arc: + arc.add("test.txt", + uid=1337, gid=1337, uname="", gname="", mode='-rwxr-xr-x') + arc.add("link", + type=link_type, + linkname='test.txt', + uid=1337, gid=1337, uname="", gname="", mode='-rwxr-xr-x') + + with ( + os_helper.temp_dir() as tmpdir, + arc.open() as tar, + unittest.mock.patch("os.chown") as mock_chown, + unittest.mock.patch("os.lchown") as mock_lchown, + unittest.mock.patch("os.geteuid") as mock_geteuid, + ): + # Set UID to 0 so chown() is attempted. + mock_geteuid.return_value = 0 + tar.extractall(path=tmpdir, filter='data') + extract_link_path = os.path.join(tmpdir, "link") + extract_file_path = os.path.join(tmpdir, "test.txt") + + if link_type == tarfile.SYMTYPE: + mock_chown.assert_called_once_with(extract_file_path, -1, -1) + mock_lchown.assert_called_once_with(extract_link_path, -1, -1) + else: + mock_chown.assert_has_calls([ + unittest.mock.call(extract_file_path, -1, -1), + unittest.mock.call(extract_link_path, -1, -1) + ]) + mock_lchown.assert_not_called() + + def test_extract_filters_target(self): + # Test that when extract() falls back to extracting (rather than + # linking) a hardlink target, it filters the target. + with ArchiveMaker() as arc: + arc.add("target") + arc.add("link", hardlink_to="target") + def testing_filter(member, path): + if member.name == 'target': + # target: set read-only + return member.replace(mode=stat.S_IRUSR) + # link: don't overwrite the mode + return member.replace(mode=None) + tempdir = pathlib.Path(TEMPDIR) / 'extract' + with os_helper.temp_dir(tempdir), arc.open() as tar: + tar.extract("link", path=tempdir, filter=testing_filter) + path = tempdir / 'link' + if os_helper.can_chmod(): + self.assertFalse(path.stat().st_mode & stat.S_IWUSR) + def test_link_fallback_normalizes(self): # Make sure hardlink fallbacks work for non-normalized paths for all # filters @@ -4762,6 +4925,22 @@ class TestExtractionFilters(unittest.TestCase): with self.check_context(arc.open(errorlevel='boo!'), filtererror_filter): self.expect_exception(TypeError) # errorlevel is not int + @support.subTests('format', [tarfile.GNU_FORMAT, tarfile.PAX_FORMAT]) + def test_getmembers_big_size(self, format): + # gh-151981: A loop in seek() for streaming files tried to read the + # declared number of blocks even at EOF + tinfo = tarfile.TarInfo("huge-file") + tinfo.size = 1 << 64 + bio = io.BytesIO() + # Write header without data + bio.write(tinfo.tobuf(format)) + + # Reset & try to get contents + bio.seek(0) + with tarfile.open(fileobj=bio, mode="r|") as tar: + with self.assertRaises(tarfile.ReadError): + tar.getmembers() + class OverwriteTests(archiver_tests.OverwriteTests, unittest.TestCase): testdir = os.path.join(TEMPDIR, "testoverwrite") diff --git a/Lib/test/test_type_params.py b/Lib/test/test_type_params.py index 65261fcb6..cf9c336f1 100644 --- a/Lib/test/test_type_params.py +++ b/Lib/test/test_type_params.py @@ -1,4 +1,5 @@ import annotationlib +import inspect import textwrap import types import unittest @@ -1447,6 +1448,30 @@ class TestEvaluateFunctions(unittest.TestCase): self.assertIs(annotationlib.call_evaluate_function(case, annotationlib.Format.FORWARDREF), int) self.assertEqual(annotationlib.call_evaluate_function(case, annotationlib.Format.STRING), 'int') + def test_signature(self): + # gh-151665: the ".format" parameter of compiler-generated evaluators + # used to break inspect.signature(). It should show up as "format". + type Alias = int + def f[T: int = int, **P = int, *Ts = int](): pass + T, P, Ts = f.__type_params__ + def g[T: (int, str)](): pass + T3, = g.__type_params__ + cases = [ + Alias.evaluate_value, + T.evaluate_bound, + T.evaluate_default, + P.evaluate_default, + Ts.evaluate_default, + T3.evaluate_constraints, + ] + for case in cases: + with self.subTest(case=case): + sig = inspect.signature(case) + self.assertEqual(str(sig), '(format=1, /)') + param, = sig.parameters.values() + self.assertEqual(param.name, 'format') + self.assertIs(param.kind, inspect.Parameter.POSITIONAL_ONLY) + def test_constraints(self): def f[T: (int, str)](): pass T, = f.__type_params__ diff --git a/Lib/test/test_typing.py b/Lib/test/test_typing.py index ec91f818c..c10fa37be 100644 --- a/Lib/test/test_typing.py +++ b/Lib/test/test_typing.py @@ -52,6 +52,8 @@ import types from test.support import ( captured_stderr, cpython_only, infinite_recursion, requires_docstrings, import_helper, run_code, EqualToForwardRef, + exceeds_recursion_limit, run_with_limited_c_stack, + skip_wasi_stack_overflow, skip_emscripten_stack_overflow, ) from test.typinganndata import ( ann_module695, mod_generics_cache, _typed_dict_helper, @@ -2742,8 +2744,14 @@ class LiteralTests(BaseTestCase): self.assertEqual(Literal[1, 2, 3].__args__, (1, 2, 3)) self.assertEqual(Literal[1, 2, 3, 3].__args__, (1, 2, 3)) self.assertEqual(Literal[1, Literal[2], Literal[3, 4]].__args__, (1, 2, 3, 4)) - # Mutable arguments will not be deduplicated - self.assertEqual(Literal[[], []].__args__, ([], [])) + # Unhashable arguments will be deduplicated too + self.assertEqual(Literal[[], []].__args__, ([],)) + self.assertEqual(Literal[{"a": 1}, {"a": 1}].__args__, ({"a": 1},)) + self.assertEqual( + Literal[1, {'a': 'b'}, 2, {'a': 'b'}, 3].__args__, + (1, {'a': 'b'}, 2, 3), + ) + self.assertEqual(Literal[{1}, {1}, {2}, {2}].__args__, ({1}, {2})) def test_flatten(self): l1 = Literal[Literal[1], Literal[2], Literal[3]] @@ -4923,6 +4931,17 @@ class GenericTests(BaseTestCase): pass self.assertEqual(MM2.__bases__, (collections.abc.MutableMapping, Generic)) + @cpython_only + @run_with_limited_c_stack() + @skip_wasi_stack_overflow() + @skip_emscripten_stack_overflow() + def test_parameters_deep_recursion(self): + x = [0] + for _ in range(exceeds_recursion_limit()): + x = [x] + with self.assertRaisesRegex(RecursionError, "in __parameter__ calculation"): + list[x].__parameters__ + def test_orig_bases(self): T = TypeVar('T') class C(typing.Dict[str, T]): ... @@ -5688,6 +5707,27 @@ class GenericTests(BaseTestCase): foo(42) + def test_genericalias_instance_isclass(self): + # test against user-defined generic classes + T = TypeVar('T') + + class Node(Generic[T]): + def __init__(self, label: T, + left: 'Node[T] | None' = None, + right: 'Node[T] | None' = None): + self.label = label + self.left = left + self.right = right + + self.assertTrue(inspect.isclass(Node)) + self.assertFalse(inspect.isclass(Node[int])) + self.assertFalse(inspect.isclass(Node[str])) + + # test against standard generic classes + self.assertFalse(inspect.isclass(set[int])) + self.assertFalse(inspect.isclass(list[bytes])) + self.assertFalse(inspect.isclass(dict[str, str])) + def test_implicit_any(self): T = TypeVar('T') @@ -7429,6 +7469,18 @@ class EvaluateForwardRefTests(BaseTestCase): typing.evaluate_forward_ref( fwdref_module.fw,) + def test_evaluate_forward_ref_string_format(self): + # Test evaluating forward references in STRING format + # does not 'leak' internal names + # See https://github.com/python/cpython/issues/150641 + + def f(arg: unknown | str | int | list[str] | tuple[int, ...]): ... + + ref = annotationlib.get_annotations(f, format=annotationlib.Format.FORWARDREF)['arg'] + self.assertEqual( + typing.evaluate_forward_ref(ref, format=annotationlib.Format.STRING), + "unknown | str | int | list[str] | tuple[int, ...]", + ) class CollectionsAbcTests(BaseTestCase): diff --git a/Lib/typing.py b/Lib/typing.py index 8c36399a6..2b4389b29 100644 --- a/Lib/typing.py +++ b/Lib/typing.py @@ -772,13 +772,16 @@ def Literal(self, *parameters): # There is no '_type_check' call because arguments to Literal[...] are # values, not types. parameters = _flatten_literal_params(parameters) + value_and_type_parameters = list(_value_and_type_iter(parameters)) + deduplicated_parameters = tuple( + p + for p, _ in _deduplicate( + value_and_type_parameters, + unhashable_fallback=True, + ) + ) - try: - parameters = tuple(p for p, _ in _deduplicate(list(_value_and_type_iter(parameters)))) - except TypeError: # unhashable parameters - pass - - return _LiteralGenericAlias(self, parameters) + return _LiteralGenericAlias(self, deduplicated_parameters) @_SpecialForm @@ -991,7 +994,7 @@ def evaluate_forward_ref( """ if format == _lazy_annotationlib.Format.STRING: - return forward_ref.__forward_arg__ + return forward_ref.__resolved_str__ if forward_ref.__forward_arg__ in _recursive_guard: return forward_ref @@ -1313,31 +1316,35 @@ class _BaseGenericAlias(_Final, _root=True): class _GenericAlias(_BaseGenericAlias, _root=True): - # The type of parameterized generics. - # - # That is, for example, `type(List[int])` is `_GenericAlias`. - # - # Objects which are instances of this class include: - # * Parameterized container types, e.g. `Tuple[int]`, `List[int]`. - # * Note that native container types, e.g. `tuple`, `list`, use - # `types.GenericAlias` instead. - # * Parameterized classes: - # class C[T]: pass - # # C[int] is a _GenericAlias - # * `Callable` aliases, generic `Callable` aliases, and - # parameterized `Callable` aliases: - # T = TypeVar('T') - # # _CallableGenericAlias inherits from _GenericAlias. - # A = Callable[[], None] # _CallableGenericAlias - # B = Callable[[T], None] # _CallableGenericAlias - # C = B[int] # _CallableGenericAlias - # * Parameterized `Final`, `ClassVar`, `TypeGuard`, and `TypeIs`: - # # All _GenericAlias - # Final[int] - # ClassVar[float] - # TypeGuard[bool] - # TypeIs[range] + """The type of parameterized generics. + That is, for example, `type(List[int])` is `_GenericAlias`. + + Objects which are instances of this class include: + * Parameterized container types, e.g. `Tuple[int]`, `List[int]`. + * Note that native container types, e.g. `tuple`, `list`, use + `types.GenericAlias` instead. + * Parameterized classes: + class C[T]: pass + # C[int] is a _GenericAlias + * `Callable` aliases, generic `Callable` aliases, and + parameterized `Callable` aliases: + T = TypeVar('T') + # _CallableGenericAlias inherits from _GenericAlias. + A = Callable[[], None] # _CallableGenericAlias + B = Callable[[T], None] # _CallableGenericAlias + C = B[int] # _CallableGenericAlias + * Parameterized `Final`, `ClassVar`, `TypeForm`, `TypeGuard`, and `TypeIs`: + # All _GenericAlias + Final[int] + ClassVar[float] + TypeForm[bytearray] + TypeGuard[bool] + TypeIs[range] + + Note that instances of this class are not classes (e.g by `inspect.isclass`), + even though they behave like them. + """ def __init__(self, origin, args, *, inst=True, name=None): super().__init__(origin, inst=inst, name=name) if not isinstance(args, tuple): @@ -1369,20 +1376,21 @@ class _GenericAlias(_BaseGenericAlias, _root=True): @_tp_cache def __getitem__(self, args): - # Parameterizes an already-parameterized object. - # - # For example, we arrive here doing something like: - # T1 = TypeVar('T1') - # T2 = TypeVar('T2') - # T3 = TypeVar('T3') - # class A(Generic[T1]): pass - # B = A[T2] # B is a _GenericAlias - # C = B[T3] # Invokes _GenericAlias.__getitem__ - # - # We also arrive here when parameterizing a generic `Callable` alias: - # T = TypeVar('T') - # C = Callable[[T], None] - # C[int] # Invokes _GenericAlias.__getitem__ + """Parameterizes an already-parameterized object. + + For example, we arrive here doing something like: + T1 = TypeVar('T1') + T2 = TypeVar('T2') + T3 = TypeVar('T3') + class A(Generic[T1]): pass + B = A[T2] # B is a _GenericAlias + C = B[T3] # Invokes _GenericAlias.__getitem__ + + We also arrive here when parameterizing a generic `Callable` alias: + T = TypeVar('T') + C = Callable[[T], None] + C[int] # Invokes _GenericAlias.__getitem__ + """ if self.__origin__ in (Generic, Protocol): # Can't subscript Generic[...] or Protocol[...]. @@ -1399,20 +1407,20 @@ class _GenericAlias(_BaseGenericAlias, _root=True): return r def _determine_new_args(self, args): - # Determines new __args__ for __getitem__. - # - # For example, suppose we had: - # T1 = TypeVar('T1') - # T2 = TypeVar('T2') - # class A(Generic[T1, T2]): pass - # T3 = TypeVar('T3') - # B = A[int, T3] - # C = B[str] - # `B.__args__` is `(int, T3)`, so `C.__args__` should be `(int, str)`. - # Unfortunately, this is harder than it looks, because if `T3` is - # anything more exotic than a plain `TypeVar`, we need to consider - # edge cases. + """Determines new __args__ for __getitem__. + For example, suppose we had: + T1 = TypeVar('T1') + T2 = TypeVar('T2') + class A(Generic[T1, T2]): pass + T3 = TypeVar('T3') + B = A[int, T3] + C = B[str] + `B.__args__` is `(int, T3)`, so `C.__args__` should be `(int, str)`. + Unfortunately, this is harder than it looks, because if `T3` is + anything more exotic than a plain `TypeVar`, we need to consider + edge cases. + """ params = self.__parameters__ # In the example above, this would be {T3: str} for param in params: