Update some libs and tests to 3.14.7 (#8603)

* `typing.py`

* tempfile

* tarfile

* imaplib

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<size>\d+)}$'
 _Untagged_status = br'\* (?P<data>\d+) (?P<type>[A-Z-]+)( (?P<data2>.*))?'
+# 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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]]) = <instance>.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]) = <instance>.list(directory='""', pattern='*')
+        (typ, [data]) = <instance>.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, ...]) = <instance>.lsub(directory='""', pattern='*')
+        (typ, [data, ...]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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]) = <instance>.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:

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

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

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")

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__

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):
 

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: