Align OS argument conversion and syscall handling with Python 3.15 (#9146)

* Align OS argument conversion and syscall handling with Python 3.15

Import canonical os tests through update_lib, retaining existing guards. Preserve raw descriptor validation, errno, audit ordering, signal retries and unlocked sendfile buffers.

Validation: independent source review passed; matching release build and nine bounded canonical cases pending.

Assisted-by: OpenAI Codex:model version unavailable

* Import canonical root-user test support for OS package

Add hash-pinned selective assignment migration within test.support and import the three Python 3.15.0 privilege helpers plus their public exports. Preserve unrelated support helpers and existing test markers.

Validation: 177 host updater tests pass; byte-identical canonical helper provenance verified. Independent source review passed. Native import validation pending integration onto current main.

Assisted-by: OpenAI Codex:model version unavailable

Lib/os.py

@@ -58,6 +58,11 @@ if 'posix' in _names:
         __all__.append('_exit')
     except ImportError:
         pass
+    try:
+        from posix import _clearenv
+        __all__.append('_clearenv')
+    except ImportError:
+        pass
     import posixpath as path
 
     try:
@@ -131,6 +136,8 @@ if _exists("_have_functions"):
     _add("HAVE_UNLINKAT",   "unlink")
     _add("HAVE_UNLINKAT",   "rmdir")
     _add("HAVE_UTIMENSAT",  "utime")
+    if _exists("statx"):
+        _set.add(statx)
     supports_dir_fd = _set
 
     _set = set()
@@ -152,6 +159,8 @@ if _exists("_have_functions"):
     _add("HAVE_FPATHCONF",  "pathconf")
     if _exists("statvfs") and _exists("fstatvfs"): # mac os x10.3
         _add("HAVE_FSTATVFS", "statvfs")
+    if _exists("statx"):
+        _set.add(statx)
     supports_fd = _set
 
     _set = set()
@@ -190,6 +199,8 @@ if _exists("_have_functions"):
     _add("HAVE_FSTATAT",    "stat")
     _add("HAVE_UTIMENSAT",  "utime")
     _add("MS_WINDOWS",      "stat")
+    if _exists("statx"):
+        _set.add(statx)
     supports_follow_symlinks = _set
 
     del _set
@@ -208,14 +219,17 @@ SEEK_END = 2
 # Super directory utilities.
 # (Inspired by Eric Raymond; the doc strings are mostly his)
 
-def makedirs(name, mode=0o777, exist_ok=False):
-    """makedirs(name [, mode=0o777][, exist_ok=False])
+def makedirs(name, mode=0o777, exist_ok=False, *, parent_mode=None):
+    """makedirs(name [, mode=0o777][, exist_ok=False][, parent_mode=None])
 
     Super-mkdir; create a leaf directory and all intermediate ones.  Works
     like mkdir, except that any intermediate path segment (not just the
     rightmost) will be created if it does not exist.  If the target
     directory already exists, raise an OSError if exist_ok is False.
-    Otherwise no exception is raised.  This is recursive.
+    Otherwise no exception is raised.  If parent_mode is not None, it will
+    be used as the mode for any newly-created, intermediate-level
+    directories. Otherwise, intermediate directories are created with the
+    default permissions (respecting umask).  This is recursive.
 
     """
     head, tail = path.split(name)
@@ -223,7 +237,11 @@ def makedirs(name, mode=0o777, exist_ok=False):
         head, tail = path.split(head)
     if head and tail and not path.exists(head):
         try:
-            makedirs(head, exist_ok=exist_ok)
+            if parent_mode is not None:
+                makedirs(head, mode=parent_mode, exist_ok=exist_ok,
+                         parent_mode=parent_mode)
+            else:
+                makedirs(head, exist_ok=exist_ok)
         except FileExistsError:
             # Defeats race condition when another thread created the path
             pass
@@ -417,14 +435,16 @@ def walk(top, topdown=True, onerror=None, followlinks=False):
             # Yield before sub-directory traversal if going top down
             yield top, dirs, nondirs
             # Traverse into sub-directories
-            for dirname in reversed(dirs):
-                new_path = join(top, dirname)
-                # bpo-23605: os.path.islink() is used instead of caching
-                # entry.is_symlink() result during the loop on os.scandir() because
-                # the caller can replace the directory entry during the "yield"
-                # above.
-                if followlinks or not islink(new_path):
-                    stack.append(new_path)
+            if dirs:
+                prefix = join(top, top[:0])  # Add trailing slash
+                for dirname in reversed(dirs):
+                    new_path = prefix + dirname
+                    # bpo-23605: os.path.islink() is used instead of caching
+                    # entry.is_symlink() result during the loop on os.scandir() because
+                    # the caller can replace the directory entry during the "yield"
+                    # above.
+                    if followlinks or not islink(new_path):
+                        stack.append(new_path)
         else:
             # Yield after sub-directory traversal if going bottom up
             stack.append((top, dirs, nondirs))
@@ -766,6 +786,12 @@ class _Environ(MutableMapping):
         new.update(self)
         return new
 
+    if _exists("_clearenv"):
+        def clear(self):
+            _clearenv()
+            self._data.clear()
+
+
 def _create_environ_mapping():
     if name == 'nt':
         # Where Env Var Names Must Be UPPERCASE

Lib/test/support/__init__.py

@@ -22,6 +22,7 @@ import warnings
 
 
 __all__ = [
+    "requires_root_user", "requires_non_root_user",
     # globals
     "PIPE_MAX_SIZE", "verbose", "max_memuse", "use_resources", "failfast",
     # exceptions
@@ -3336,6 +3337,11 @@ def control_characters_c0() -> list[str]:
     return [chr(c) for c in range(0x00, 0x20)] + ["\x7F"]
 
 
+_ROOT_IN_POSIX = hasattr(os, 'geteuid') and os.geteuid() == 0
+requires_root_user = unittest.skipUnless(_ROOT_IN_POSIX, "test needs root privilege")
+requires_non_root_user = unittest.skipIf(_ROOT_IN_POSIX, "test needs non-root account")
+
+
 STATUS_DLL_INIT_FAILED = 0xC0000142
 def skip_on_low_desktop_heap_memory_subprocess(returncode):
     if sys.platform not in ('win32', 'cygwin'):

Lib/test/test_os/__init__.py

@@ -0,0 +1,6 @@
+import os.path
+from test.support import load_package_tests
+
+
+def load_tests(*args):
+    return load_package_tests(os.path.dirname(__file__), *args)

Lib/test/test_os/test_posix.py

@@ -46,7 +46,8 @@ def _supports_sched():
     try:
         posix.sched_getscheduler(0)
     except OSError as e:
-        if e.errno == errno.ENOSYS:
+        # DragonFly BSD requires privileges to use the scheduler API.
+        if e.errno in (errno.ENOSYS, errno.EPERM):
             return False
     return True
 
@@ -70,7 +71,7 @@ class PosixTester(unittest.TestCase):
         NO_ARG_FUNCTIONS = [ "ctermid", "getcwd", "getcwdb", "uname",
                              "times", "getloadavg",
                              "getegid", "geteuid", "getgid", "getgroups",
-                             "getpid", "getpgrp", "getppid", "getuid", "sync",
+                             "getpid", "getpgrp", "getppid", "getuid",
                            ]
 
         for name in NO_ARG_FUNCTIONS:
@@ -80,6 +81,13 @@ class PosixTester(unittest.TestCase):
                     posix_func()
                     self.assertRaises(TypeError, posix_func, 1)
 
+    # gh-102184: sync() can block for a long time.
+    @support.requires_resource('walltime')
+    @unittest.skipUnless(hasattr(posix, 'sync'), 'test needs posix.sync()')
+    def test_sync(self):
+        posix.sync()
+        self.assertRaises(TypeError, posix.sync, 1)
+
     @unittest.skipUnless(hasattr(posix, 'getresuid'),
                          'test needs posix.getresuid()')
     def test_getresuid(self):
@@ -415,8 +423,10 @@ class PosixTester(unittest.TestCase):
             if inst.errno == errno.EINVAL and sys.platform.startswith(
                 ('sunos', 'freebsd', 'openbsd', 'gnukfreebsd')):
                 raise unittest.SkipTest("test may fail on ZFS filesystems")
-            elif inst.errno == errno.EOPNOTSUPP and sys.platform.startswith("netbsd"):
-                raise unittest.SkipTest("test may fail on FFS filesystems")
+            elif inst.errno == errno.EOPNOTSUPP:
+                # ZFS on FreeBSD, FFS on NetBSD, etc.
+                raise unittest.SkipTest(
+                    "the file system does not support posix_fallocate()")
             else:
                 raise
         finally:
@@ -680,22 +690,65 @@ class PosixTester(unittest.TestCase):
                 'cannot use fd and follow_symlinks together'):
             posix.stat(1, follow_symlinks=False)
 
-    def test_stat(self):
-        self.assertTrue(posix.stat(os_helper.TESTFN))
-        self.assertTrue(posix.stat(os.fsencode(os_helper.TESTFN)))
+    def check_statlike_path(self, func):
+        self.assertTrue(func(os_helper.TESTFN))
+        self.assertTrue(func(os.fsencode(os_helper.TESTFN)))
+        self.assertTrue(func(os_helper.FakePath(os_helper.TESTFN)))
 
         self.assertRaisesRegex(TypeError,
                 'should be string, bytes, os.PathLike or integer, not',
-                posix.stat, bytearray(os.fsencode(os_helper.TESTFN)))
+                func, bytearray(os.fsencode(os_helper.TESTFN)))
         self.assertRaisesRegex(TypeError,
                 'should be string, bytes, os.PathLike or integer, not',
-                posix.stat, None)
+                func, None)
         self.assertRaisesRegex(TypeError,
                 'should be string, bytes, os.PathLike or integer, not',
-                posix.stat, list(os_helper.TESTFN))
+                func, list(os_helper.TESTFN))
         self.assertRaisesRegex(TypeError,
                 'should be string, bytes, os.PathLike or integer, not',
-                posix.stat, list(os.fsencode(os_helper.TESTFN)))
+                func, list(os.fsencode(os_helper.TESTFN)))
+
+    def test_stat(self):
+        self.check_statlike_path(posix.stat)
+
+    @unittest.skipUnless(hasattr(posix, 'statx'), 'test needs posix.statx()')
+    def test_statx(self):
+        def func(path, **kwargs):
+            return posix.statx(path, posix.STATX_BASIC_STATS, **kwargs)
+        self.check_statlike_path(func)
+
+    @unittest.skipUnless(hasattr(posix, 'statx'), 'test needs posix.statx()')
+    def test_statx_flags(self):
+        # glibc's fallback implementation of statx via the stat family fails
+        # with EINVAL on the (nonzero) sync flags.  If you see this failure,
+        # update your kernel and/or seccomp syscall filter.
+        valid_flag_names = ('AT_NO_AUTOMOUNT', 'AT_STATX_SYNC_AS_STAT',
+                            'AT_STATX_FORCE_SYNC', 'AT_STATX_DONT_SYNC')
+        for flag_name in valid_flag_names:
+            flag = getattr(posix, flag_name)
+            with self.subTest(msg=flag_name, flags=flag):
+                posix.statx(os_helper.TESTFN, posix.STATX_BASIC_STATS,
+                            flags=flag)
+
+        # These flags are not exposed to Python because their functionality is
+        # implemented via kwargs instead.
+        kwarg_equivalent_flags = (
+            (0x0100, 'AT_SYMLINK_NOFOLLOW', 'follow_symlinks'),
+            (0x0400, 'AT_SYMLINK_FOLLOW', 'follow_symlinks'),
+            (0x1000, 'AT_EMPTY_PATH', 'dir_fd'),
+        )
+        for flag, flag_name, kwarg_name in kwarg_equivalent_flags:
+            with self.subTest(msg=flag_name, flags=flag):
+                with self.assertRaisesRegex(ValueError, kwarg_name):
+                    posix.statx(os_helper.TESTFN, posix.STATX_BASIC_STATS,
+                                flags=flag)
+
+        with self.subTest(msg="AT_STATX_FORCE_SYNC | AT_STATX_DONT_SYNC"):
+            with self.assertRaises(OSError) as ctx:
+                flags = posix.AT_STATX_FORCE_SYNC | posix.AT_STATX_DONT_SYNC
+                posix.statx(os_helper.TESTFN, posix.STATX_BASIC_STATS,
+                            flags=flags)
+            self.assertEqual(ctx.exception.errno, errno.EINVAL)
 
     @unittest.skipUnless(hasattr(posix, 'mkfifo'), "don't have mkfifo()")
     def test_mkfifo(self):
@@ -769,12 +822,29 @@ class PosixTester(unittest.TestCase):
             self.assertRaises((ValueError, OverflowError), posix.makedev, x, minor)
             self.assertRaises((ValueError, OverflowError), posix.makedev, major, x)
 
-        if sys.platform == 'linux' and not support.linked_to_musl():
-            NODEV = -1
+        # The following tests are needed to test functions accepting or
+        # returning the special value NODEV (if it is defined). major(), minor()
+        # and makefile() are the only easily reproducible examples, but that
+        # behavior is platform specific -- on some platforms their code has
+        # a special case for NODEV, on others this is just an implementation
+        # artifact.
+        if (hasattr(posix, 'NODEV') and
+            sys.platform.startswith(('linux', 'macos', 'freebsd', 'sunos'))):
+            NODEV = posix.NODEV
             self.assertEqual(posix.major(NODEV), NODEV)
             self.assertEqual(posix.minor(NODEV), NODEV)
             self.assertEqual(posix.makedev(NODEV, NODEV), NODEV)
 
+    def test_nodev(self):
+        # NODEV is not a part of Posix, but is defined on many systems.
+        if (not hasattr(posix, 'NODEV')
+            and (not sys.platform.startswith(('linux', 'macos', 'freebsd',
+                                              'dragonfly', 'netbsd', 'openbsd',
+                                              'sunos'))
+                 or support.linked_to_musl())):
+            self.skipTest('not defined on this platform')
+        self.assertHasAttr(posix, 'NODEV')
+
     def _test_all_chown_common(self, chown_func, first_param, stat_func):
         """Common code for chown, fchown and lchown tests."""
         def check_stat(uid, gid):
@@ -838,7 +908,9 @@ class PosixTester(unittest.TestCase):
             self.assertRaises(OSError, chown_func, first_param, 0, -1)
             check_stat(uid, gid)
             if hasattr(os, 'getgroups'):
-                if 0 not in os.getgroups():
+                # Also check the effective gid, which the kernel
+                # accepts for chown even if not in getgroups().
+                if 0 not in os.getgroups() and os.getegid() != 0:
                     self.assertRaises(OSError, chown_func, first_param, -1, 0)
                     check_stat(uid, gid)
         # test illegal types
@@ -1386,6 +1458,7 @@ class PosixTester(unittest.TestCase):
         del sched_priority, param  # should not crash
         support.gc_collect()  # just to be sure
 
+    @requires_sched
     @unittest.skipUnless(hasattr(posix, "sched_rr_get_interval"), "no function")
     def test_sched_rr_get_interval(self):
         try:
@@ -1631,33 +1704,47 @@ class TestPosixDirFd(unittest.TestCase):
         with self.prepare_file() as (dir_fd, name, fullname):
             posix.chown(name, os.getuid(), os.getgid(), dir_fd=dir_fd)
 
-    @unittest.skipUnless(os.stat in os.supports_dir_fd, "test needs dir_fd support in os.stat()")
-    def test_stat_dir_fd(self):
+    def check_statlike_dir_fd(self, func, prefix):
         with self.prepare() as (dir_fd, name, fullname):
             with open(fullname, 'w') as outfile:
                 outfile.write("testline\n")
             self.addCleanup(posix.unlink, fullname)
 
-            s1 = posix.stat(fullname)
-            s2 = posix.stat(name, dir_fd=dir_fd)
-            self.assertEqual(s1, s2)
-            s2 = posix.stat(fullname, dir_fd=None)
-            self.assertEqual(s1, s2)
+            def get(result, attr):
+                return getattr(result, prefix + attr)
+
+            s1 = func(fullname)
+            s2 = func(name, dir_fd=dir_fd)
+            self.assertEqual((get(s1, "dev"), get(s1, "ino")),
+                             (get(s2, "dev"), get(s2, "ino")))
+            s2 = func(fullname, dir_fd=None)
+            self.assertEqual((get(s1, "dev"), get(s1, "ino")),
+                             (get(s2, "dev"), get(s2, "ino")))
 
             self.assertRaisesRegex(TypeError, 'should be integer or None, not',
-                    posix.stat, name, dir_fd=posix.getcwd())
+                    func, name, dir_fd=posix.getcwd())
             self.assertRaisesRegex(TypeError, 'should be integer or None, not',
-                    posix.stat, name, dir_fd=float(dir_fd))
+                    func, name, dir_fd=float(dir_fd))
             self.assertRaises(OverflowError,
-                    posix.stat, name, dir_fd=10**20)
+                    func, name, dir_fd=10**20)
 
             for fd in False, True:
                 with self.assertWarnsRegex(RuntimeWarning,
                         'bool is used as a file descriptor') as cm:
                     with self.assertRaises(OSError):
-                        posix.stat('nonexisting', dir_fd=fd)
+                        func('nonexisting', dir_fd=fd)
                 self.assertEqual(cm.filename, __file__)
 
+    @unittest.skipUnless(os.stat in os.supports_dir_fd, "test needs dir_fd support in os.stat()")
+    def test_stat_dir_fd(self):
+        self.check_statlike_dir_fd(posix.stat, prefix="st_")
+
+    @unittest.skipUnless(hasattr(posix, 'statx'), "test needs os.statx()")
+    def test_statx_dir_fd(self):
+        def func(path, **kwargs):
+            return posix.statx(path, os.STATX_INO, **kwargs)
+        self.check_statlike_dir_fd(func, prefix="stx_")
+
     @unittest.skipUnless(os.utime in os.supports_dir_fd, "test needs dir_fd support in os.utime()")
     def test_utime_dir_fd(self):
         with self.prepare_file() as (dir_fd, name, fullname):
@@ -2271,6 +2358,22 @@ class TestPosixWeaklinking(unittest.TestCase):
             self.assertNotHasAttr(os, "pwritev")
             self.assertNotHasAttr(os, "preadv")
 
+    def test_pipe2(self):
+        self._verify_available("HAVE_PIPE2")
+        if self.mac_ver >= (27, 0):
+            self.assertHasAttr(os, "pipe2")
+        else:
+            self.assertNotHasAttr(os, "pipe2")
+
+    def test_dup3(self):
+        self._verify_available("HAVE_DUP3")
+        r, w = os.pipe()
+        self.addCleanup(os.close, r)
+        self.addCleanup(os.close, w)
+        # Must not crash even when dup3 unavailable at runtime.
+        # os.dup2 returns fd2 (here w); do not double-close.
+        os.dup2(r, w, inheritable=False)
+
     def test_stat(self):
         self._verify_available("HAVE_FSTATAT")
         if self.mac_ver >= (10, 10):

Lib/test/test_os/test_windows.py

@@ -0,0 +1,605 @@
+import sys
+import unittest
+
+if sys.platform != "win32":
+    raise unittest.SkipTest("Win32 specific tests")
+
+import _winapi
+import fnmatch
+import mmap
+import os
+import shutil
+import signal
+import stat
+import subprocess
+import textwrap
+import time
+import uuid
+from test import support
+from test.support import import_helper
+from test.support import os_helper
+from .utils import create_file
+
+
+class Win32KillTests(unittest.TestCase):
+    def _kill(self, sig):
+        # Start sys.executable as a subprocess and communicate from the
+        # subprocess to the parent that the interpreter is ready. When it
+        # becomes ready, send *sig* via os.kill to the subprocess and check
+        # that the return code is equal to *sig*.
+        import ctypes
+        from ctypes import wintypes
+        import msvcrt
+
+        # Since we can't access the contents of the process' stdout until the
+        # process has exited, use PeekNamedPipe to see what's inside stdout
+        # without waiting. This is done so we can tell that the interpreter
+        # is started and running at a point where it could handle a signal.
+        PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe
+        PeekNamedPipe.restype = wintypes.BOOL
+        PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle
+                                  ctypes.POINTER(ctypes.c_char), # stdout buf
+                                  wintypes.DWORD, # Buffer size
+                                  ctypes.POINTER(wintypes.DWORD), # bytes read
+                                  ctypes.POINTER(wintypes.DWORD), # bytes avail
+                                  ctypes.POINTER(wintypes.DWORD)) # bytes left
+        msg = "running"
+        proc = subprocess.Popen([sys.executable, "-c",
+                                 "import sys;"
+                                 "sys.stdout.write('{}');"
+                                 "sys.stdout.flush();"
+                                 "input()".format(msg)],
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE,
+                                stdin=subprocess.PIPE)
+        self.addCleanup(proc.stdout.close)
+        self.addCleanup(proc.stderr.close)
+        self.addCleanup(proc.stdin.close)
+
+        count, max = 0, 100
+        while count < max and proc.poll() is None:
+            # Create a string buffer to store the result of stdout from the pipe
+            buf = ctypes.create_string_buffer(len(msg))
+            # Obtain the text currently in proc.stdout
+            # Bytes read/avail/left are left as NULL and unused
+            rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()),
+                                 buf, ctypes.sizeof(buf), None, None, None)
+            self.assertNotEqual(rslt, 0, "PeekNamedPipe failed")
+            if buf.value:
+                self.assertEqual(msg, buf.value.decode())
+                break
+            time.sleep(0.1)
+            count += 1
+        else:
+            self.fail("Did not receive communication from the subprocess")
+
+        os.kill(proc.pid, sig)
+        self.assertEqual(proc.wait(), sig)
+
+    def test_kill_sigterm(self):
+        # SIGTERM doesn't mean anything special, but make sure it works
+        self._kill(signal.SIGTERM)
+
+    def test_kill_int(self):
+        # os.kill on Windows can take an int which gets set as the exit code
+        self._kill(100)
+
+    @unittest.skipIf(mmap is None, "requires mmap")
+    def _kill_with_event(self, event, name):
+        tagname = "test_os_%s" % uuid.uuid1()
+        m = mmap.mmap(-1, 1, tagname)
+        m[0] = 0
+
+        # Run a script which has console control handling enabled.
+        script = os.path.join(os.path.dirname(__file__),
+                              "win_console_handler.py")
+        cmd = [sys.executable, script, tagname]
+        proc = subprocess.Popen(cmd,
+                                creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
+
+        with proc:
+            # Let the interpreter startup before we send signals. See #3137.
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if proc.poll() is None:
+                    break
+            else:
+                # Forcefully kill the process if we weren't able to signal it.
+                proc.kill()
+                self.fail("Subprocess didn't finish initialization")
+
+            os.kill(proc.pid, event)
+
+            try:
+                # proc.send_signal(event) could also be done here.
+                # Allow time for the signal to be passed and the process to exit.
+                proc.wait(timeout=support.SHORT_TIMEOUT)
+            except subprocess.TimeoutExpired:
+                # Forcefully kill the process if we weren't able to signal it.
+                proc.kill()
+                self.fail("subprocess did not stop on {}".format(name))
+
+    @unittest.skip("subprocesses aren't inheriting Ctrl+C property")
+    @support.requires_subprocess()
+    def test_CTRL_C_EVENT(self):
+        from ctypes import wintypes
+        import ctypes
+
+        # Make a NULL value by creating a pointer with no argument.
+        NULL = ctypes.POINTER(ctypes.c_int)()
+        SetConsoleCtrlHandler = ctypes.windll.kernel32.SetConsoleCtrlHandler
+        SetConsoleCtrlHandler.argtypes = (ctypes.POINTER(ctypes.c_int),
+                                          wintypes.BOOL)
+        SetConsoleCtrlHandler.restype = wintypes.BOOL
+
+        # Calling this with NULL and FALSE causes the calling process to
+        # handle Ctrl+C, rather than ignore it. This property is inherited
+        # by subprocesses.
+        SetConsoleCtrlHandler(NULL, 0)
+
+        self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT")
+
+    @support.requires_subprocess()
+    def test_CTRL_BREAK_EVENT(self):
+        self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
+
+
+class Win32ListdirTests(unittest.TestCase):
+    """Test listdir on Windows."""
+
+    def setUp(self):
+        self.created_paths = []
+        for i in range(2):
+            dir_name = 'SUB%d' % i
+            dir_path = os.path.join(os_helper.TESTFN, dir_name)
+            file_name = 'FILE%d' % i
+            file_path = os.path.join(os_helper.TESTFN, file_name)
+            os.makedirs(dir_path)
+            with open(file_path, 'w', encoding='utf-8') as f:
+                f.write("I'm %s and proud of it. Blame test_os.\n" % file_path)
+            self.created_paths.extend([dir_name, file_name])
+        self.created_paths.sort()
+
+    def tearDown(self):
+        shutil.rmtree(os_helper.TESTFN)
+
+    def test_listdir_no_extended_path(self):
+        """Test when the path is not an "extended" path."""
+        # unicode
+        self.assertEqual(
+                sorted(os.listdir(os_helper.TESTFN)),
+                self.created_paths)
+
+        # bytes
+        self.assertEqual(
+                sorted(os.listdir(os.fsencode(os_helper.TESTFN))),
+                [os.fsencode(path) for path in self.created_paths])
+
+    def test_listdir_extended_path(self):
+        """Test when the path starts with '\\\\?\\'."""
+        # See: http://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx#maxpath
+        # unicode
+        path = '\\\\?\\' + os.path.abspath(os_helper.TESTFN)
+        self.assertEqual(
+                sorted(os.listdir(path)),
+                self.created_paths)
+
+        # bytes
+        path = b'\\\\?\\' + os.fsencode(os.path.abspath(os_helper.TESTFN))
+        self.assertEqual(
+                sorted(os.listdir(path)),
+                [os.fsencode(path) for path in self.created_paths])
+
+
+@unittest.skipUnless(os.name == "nt", "NT specific tests")
+class Win32ListdriveTests(unittest.TestCase):
+    """Test listdrive, listmounts and listvolume on Windows."""
+
+    def setUp(self):
+        # Get drives and volumes from fsutil
+        out = subprocess.check_output(
+            ["fsutil.exe", "volume", "list"],
+            cwd=os.path.join(os.getenv("SystemRoot", "\\Windows"), "System32"),
+            encoding="mbcs",
+            errors="ignore",
+        )
+        lines = out.splitlines()
+        self.known_volumes = {l for l in lines if l.startswith('\\\\?\\')}
+        self.known_drives = {l for l in lines if l[1:] == ':\\'}
+        self.known_mounts = {l for l in lines if l[1:3] == ':\\'}
+
+    def test_listdrives(self):
+        drives = os.listdrives()
+        self.assertIsInstance(drives, list)
+        self.assertSetEqual(
+            self.known_drives,
+            self.known_drives & set(drives),
+        )
+
+    def test_listvolumes(self):
+        volumes = os.listvolumes()
+        self.assertIsInstance(volumes, list)
+        self.assertSetEqual(
+            self.known_volumes,
+            self.known_volumes & set(volumes),
+        )
+
+    def test_listmounts(self):
+        for volume in os.listvolumes():
+            try:
+                mounts = os.listmounts(volume)
+            except OSError as ex:
+                if support.verbose:
+                    print("Skipping", volume, "because of", ex)
+            else:
+                self.assertIsInstance(mounts, list)
+                self.assertSetEqual(
+                    set(mounts),
+                    self.known_mounts & set(mounts),
+                )
+
+
+@os_helper.skip_unless_symlink
+class Win32SymlinkTests(unittest.TestCase):
+    filelink = 'filelinktest'
+    filelink_target = os.path.abspath(__file__)
+    dirlink = 'dirlinktest'
+    dirlink_target = os.path.dirname(filelink_target)
+    missing_link = 'missing link'
+
+    def setUp(self):
+        assert os.path.exists(self.dirlink_target)
+        assert os.path.exists(self.filelink_target)
+        assert not os.path.exists(self.dirlink)
+        assert not os.path.exists(self.filelink)
+        assert not os.path.exists(self.missing_link)
+
+    def tearDown(self):
+        if os.path.exists(self.filelink):
+            os.remove(self.filelink)
+        if os.path.exists(self.dirlink):
+            os.rmdir(self.dirlink)
+        if os.path.lexists(self.missing_link):
+            os.remove(self.missing_link)
+
+    def test_directory_link(self):
+        os.symlink(self.dirlink_target, self.dirlink)
+        self.assertTrue(os.path.exists(self.dirlink))
+        self.assertTrue(os.path.isdir(self.dirlink))
+        self.assertTrue(os.path.islink(self.dirlink))
+        self.check_stat(self.dirlink, self.dirlink_target)
+
+    def test_file_link(self):
+        os.symlink(self.filelink_target, self.filelink)
+        self.assertTrue(os.path.exists(self.filelink))
+        self.assertTrue(os.path.isfile(self.filelink))
+        self.assertTrue(os.path.islink(self.filelink))
+        self.check_stat(self.filelink, self.filelink_target)
+
+    def _create_missing_dir_link(self):
+        'Create a "directory" link to a non-existent target'
+        linkname = self.missing_link
+        if os.path.lexists(linkname):
+            os.remove(linkname)
+        target = r'c:\\target does not exist.29r3c740'
+        assert not os.path.exists(target)
+        target_is_dir = True
+        os.symlink(target, linkname, target_is_dir)
+
+    def test_remove_directory_link_to_missing_target(self):
+        self._create_missing_dir_link()
+        # For compatibility with Unix, os.remove will check the
+        #  directory status and call RemoveDirectory if the symlink
+        #  was created with target_is_dir==True.
+        os.remove(self.missing_link)
+
+    def test_isdir_on_directory_link_to_missing_target(self):
+        self._create_missing_dir_link()
+        self.assertFalse(os.path.isdir(self.missing_link))
+
+    def test_rmdir_on_directory_link_to_missing_target(self):
+        self._create_missing_dir_link()
+        os.rmdir(self.missing_link)
+
+    def check_stat(self, link, target):
+        self.assertEqual(os.stat(link), os.stat(target))
+        self.assertNotEqual(os.lstat(link), os.stat(link))
+
+        bytes_link = os.fsencode(link)
+        self.assertEqual(os.stat(bytes_link), os.stat(target))
+        self.assertNotEqual(os.lstat(bytes_link), os.stat(bytes_link))
+
+    def test_12084(self):
+        level1 = os.path.abspath(os_helper.TESTFN)
+        level2 = os.path.join(level1, "level2")
+        level3 = os.path.join(level2, "level3")
+        self.addCleanup(os_helper.rmtree, level1)
+
+        os.mkdir(level1)
+        os.mkdir(level2)
+        os.mkdir(level3)
+
+        file1 = os.path.abspath(os.path.join(level1, "file1"))
+        create_file(file1)
+
+        orig_dir = os.getcwd()
+        try:
+            os.chdir(level2)
+            link = os.path.join(level2, "link")
+            os.symlink(os.path.relpath(file1), "link")
+            self.assertIn("link", os.listdir(os.getcwd()))
+
+            # Check os.stat calls from the same dir as the link
+            self.assertEqual(os.stat(file1), os.stat("link"))
+
+            # Check os.stat calls from a dir below the link
+            os.chdir(level1)
+            self.assertEqual(os.stat(file1),
+                             os.stat(os.path.relpath(link)))
+
+            # Check os.stat calls from a dir above the link
+            os.chdir(level3)
+            self.assertEqual(os.stat(file1),
+                             os.stat(os.path.relpath(link)))
+        finally:
+            os.chdir(orig_dir)
+
+    @unittest.skipUnless(os.path.lexists(r'C:\Users\All Users')
+                            and os.path.exists(r'C:\ProgramData'),
+                            'Test directories not found')
+    def test_29248(self):
+        # os.symlink() calls CreateSymbolicLink, which creates
+        # the reparse data buffer with the print name stored
+        # first, so the offset is always 0. CreateSymbolicLink
+        # stores the "PrintName" DOS path (e.g. "C:\") first,
+        # with an offset of 0, followed by the "SubstituteName"
+        # NT path (e.g. "\??\C:\"). The "All Users" link, on
+        # the other hand, seems to have been created manually
+        # with an inverted order.
+        target = os.readlink(r'C:\Users\All Users')
+        self.assertTrue(os.path.samefile(target, r'C:\ProgramData'))
+
+    def test_buffer_overflow(self):
+        # Older versions would have a buffer overflow when detecting
+        # whether a link source was a directory. This test ensures we
+        # no longer crash, but does not otherwise validate the behavior
+        segment = 'X' * 27
+        path = os.path.join(*[segment] * 10)
+        test_cases = [
+            # overflow with absolute src
+            ('\\' + path, segment),
+            # overflow dest with relative src
+            (segment, path),
+            # overflow when joining src
+            (path[:180], path[:180]),
+        ]
+        for src, dest in test_cases:
+            try:
+                os.symlink(src, dest)
+            except FileNotFoundError:
+                pass
+            else:
+                try:
+                    os.remove(dest)
+                except OSError:
+                    pass
+            # Also test with bytes, since that is a separate code path.
+            try:
+                os.symlink(os.fsencode(src), os.fsencode(dest))
+            except FileNotFoundError:
+                pass
+            else:
+                try:
+                    os.remove(dest)
+                except OSError:
+                    pass
+
+    def test_appexeclink(self):
+        root = os.path.expandvars(r'%LOCALAPPDATA%\Microsoft\WindowsApps')
+        if not os.path.isdir(root):
+            self.skipTest("test requires a WindowsApps directory")
+
+        aliases = [os.path.join(root, a)
+                   for a in fnmatch.filter(os.listdir(root), '*.exe')]
+
+        for alias in aliases:
+            if support.verbose:
+                print()
+                print("Testing with", alias)
+            st = os.lstat(alias)
+            self.assertEqual(st, os.stat(alias))
+            self.assertFalse(stat.S_ISLNK(st.st_mode))
+            self.assertEqual(st.st_reparse_tag, stat.IO_REPARSE_TAG_APPEXECLINK)
+            self.assertTrue(os.path.isfile(alias))
+            # testing the first one we see is sufficient
+            break
+        else:
+            self.skipTest("test requires an app execution alias")
+
+
+class Win32JunctionTests(unittest.TestCase):
+    junction = 'junctiontest'
+    junction_target = os.path.dirname(os.path.abspath(__file__))
+
+    def setUp(self):
+        assert os.path.exists(self.junction_target)
+        assert not os.path.lexists(self.junction)
+
+    def tearDown(self):
+        if os.path.lexists(self.junction):
+            os.unlink(self.junction)
+
+    def test_create_junction(self):
+        _winapi.CreateJunction(self.junction_target, self.junction)
+        self.assertTrue(os.path.lexists(self.junction))
+        self.assertTrue(os.path.exists(self.junction))
+        self.assertTrue(os.path.isdir(self.junction))
+        self.assertNotEqual(os.stat(self.junction), os.lstat(self.junction))
+        self.assertEqual(os.stat(self.junction), os.stat(self.junction_target))
+
+        # bpo-37834: Junctions are not recognized as links.
+        self.assertFalse(os.path.islink(self.junction))
+        self.assertEqual(os.path.normcase("\\\\?\\" + self.junction_target),
+                         os.path.normcase(os.readlink(self.junction)))
+
+    def test_unlink_removes_junction(self):
+        _winapi.CreateJunction(self.junction_target, self.junction)
+        self.assertTrue(os.path.exists(self.junction))
+        self.assertTrue(os.path.lexists(self.junction))
+
+        os.unlink(self.junction)
+        self.assertFalse(os.path.exists(self.junction))
+
+
+class Win32NtTests(unittest.TestCase):
+    def test_getfinalpathname_handles(self):
+        nt = import_helper.import_module('nt')
+        ctypes = import_helper.import_module('ctypes')
+        # Ruff false positive -- it thinks we're redefining `ctypes` here
+        import ctypes.wintypes  # noqa: F811
+
+        kernel = ctypes.WinDLL('Kernel32.dll', use_last_error=True)
+        kernel.GetCurrentProcess.restype = ctypes.wintypes.HANDLE
+
+        kernel.GetProcessHandleCount.restype = ctypes.wintypes.BOOL
+        kernel.GetProcessHandleCount.argtypes = (ctypes.wintypes.HANDLE,
+                                                 ctypes.wintypes.LPDWORD)
+
+        # This is a pseudo-handle that doesn't need to be closed
+        hproc = kernel.GetCurrentProcess()
+
+        handle_count = ctypes.wintypes.DWORD()
+        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
+        self.assertEqual(1, ok)
+
+        before_count = handle_count.value
+
+        # The first two test the error path, __file__ tests the success path
+        filenames = [
+            r'\\?\C:',
+            r'\\?\NUL',
+            r'\\?\CONIN',
+            __file__,
+        ]
+
+        for _ in range(10):
+            for name in filenames:
+                try:
+                    nt._getfinalpathname(name)
+                except Exception:
+                    # Failure is expected
+                    pass
+                try:
+                    os.stat(name)
+                except Exception:
+                    pass
+
+        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
+        self.assertEqual(1, ok)
+
+        handle_delta = handle_count.value - before_count
+
+        self.assertEqual(0, handle_delta)
+
+    @support.requires_subprocess()
+    def test_stat_unlink_race(self):
+        # bpo-46785: the implementation of os.stat() falls back to reading
+        # the parent directory if CreateFileW() fails with a permission
+        # error. If reading the parent directory fails because the file or
+        # directory are subsequently unlinked, or because the volume or
+        # share are no longer available, then the original permission error
+        # should not be restored.
+        filename =  os_helper.TESTFN
+        self.addCleanup(os_helper.unlink, filename)
+        deadline = time.time() + 5
+        command = textwrap.dedent("""\
+            import os
+            import sys
+            import time
+
+            filename = sys.argv[1]
+            deadline = float(sys.argv[2])
+
+            while time.time() < deadline:
+                try:
+                    with open(filename, "w") as f:
+                        pass
+                except OSError:
+                    pass
+                try:
+                    os.remove(filename)
+                except OSError:
+                    pass
+            """)
+
+        with subprocess.Popen([sys.executable, '-c', command, filename, str(deadline)]) as proc:
+            while time.time() < deadline:
+                try:
+                    os.stat(filename)
+                except FileNotFoundError as e:
+                    assert e.winerror == 2  # ERROR_FILE_NOT_FOUND
+            try:
+                proc.wait(1)
+            except subprocess.TimeoutExpired:
+                proc.terminate()
+
+    @support.requires_subprocess()
+    def test_stat_inaccessible_file(self):
+        filename = os_helper.TESTFN
+        ICACLS = os.path.expandvars(r"%SystemRoot%\System32\icacls.exe")
+
+        with open(filename, "wb") as f:
+            f.write(b'Test data')
+
+        stat1 = os.stat(filename)
+
+        try:
+            # Remove all permissions from the file
+            subprocess.check_output([ICACLS, filename, "/inheritance:r"],
+                                    stderr=subprocess.STDOUT)
+        except subprocess.CalledProcessError as ex:
+            if support.verbose:
+                print(ICACLS, filename, "/inheritance:r", "failed.")
+                print(ex.stdout.decode("oem", "replace").rstrip())
+            try:
+                os.unlink(filename)
+            except OSError:
+                pass
+            self.skipTest("Unable to create inaccessible file")
+
+        def cleanup():
+            # Give delete permission to the owner (us)
+            subprocess.check_output([ICACLS, filename, "/grant", "*WD:(D)"],
+                                    stderr=subprocess.STDOUT)
+            os.unlink(filename)
+
+        self.addCleanup(cleanup)
+
+        if support.verbose:
+            print("File:", filename)
+            print("stat with access:", stat1)
+
+        # First test - we shouldn't raise here, because we still have access to
+        # the directory and can extract enough information from its metadata.
+        stat2 = os.stat(filename)
+
+        if support.verbose:
+            print(" without access:", stat2)
+
+        # We may not get st_dev/st_ino, so ensure those are 0 or match
+        self.assertIn(stat2.st_dev, (0, stat1.st_dev))
+        self.assertIn(stat2.st_ino, (0, stat1.st_ino))
+
+        # st_mode and st_size should match (for a normal file, at least)
+        self.assertEqual(stat1.st_mode, stat2.st_mode)
+        self.assertEqual(stat1.st_size, stat2.st_size)
+
+        # st_ctime and st_mtime should be the same
+        self.assertEqual(stat1.st_ctime, stat2.st_ctime)
+        self.assertEqual(stat1.st_mtime, stat2.st_mtime)
+
+        # st_atime should be the same or later
+        self.assertGreaterEqual(stat1.st_atime, stat2.st_atime)
+
+
+if __name__ == "__main__":
+    unittest.main()

Lib/test/test_os/utils.py

@@ -0,0 +1,3 @@
+def create_file(filename, content=b'content'):
+    with open(filename, "xb", 0) as fp:
+        fp.write(content)

crates/host_env/src/crt_fd.rs

@@ -319,7 +319,10 @@ impl AsRawFd for Borrowed<'_> {
 
 #[inline]
 fn ebadf() -> io::Error {
-    io::Error::from_raw_os_error(c::EBADF)
+    cfg_select! {
+        windows => crate::os::io_error_from_errno(c::EBADF),
+        _ => io::Error::from_raw_os_error(c::EBADF),
+    }
 }
 
 pub fn open(path: &ffi::CStr, flags: i32, mode: i32) -> io::Result<Owned> {
@@ -397,7 +400,12 @@ fn close_range_slow(low: Raw, high: Raw) {
 }
 
 pub fn ftruncate(fd: Borrowed<'_>, len: Offset) -> io::Result<()> {
-    let ret = unsafe { suppress_iph!(c::ftruncate(fd.as_raw(), len)) };
+    ftruncate_raw(fd.as_raw(), len)
+}
+
+/// Truncate a caller-supplied descriptor, letting the CRT validate its value.
+pub fn ftruncate_raw(fd: Raw, len: Offset) -> io::Result<()> {
+    let ret = unsafe { suppress_iph!(c::ftruncate(fd, len)) };
     // On Windows, _chsize_s returns 0 on success, or a positive error code (errno value) on failure.
     // On other platforms, ftruncate returns 0 on success, or -1 on failure with errno set.
     cfg_select! {

crates/host_env/src/posix.rs

@@ -15,6 +15,10 @@ pub use super::posix_unix_like::*;
 
 pub use libc::{c_char, pid_t};
 
+/// `<sys/param.h>` on glibc defines NODEV as `(dev_t)-1`.
+#[cfg(all(target_os = "linux", target_env = "gnu"))]
+pub const NODEV: i32 = -1;
+
 /// `<sysexits.h>`. The `libc` crate does not bind these.
 pub const EX_OK: i32 = 0;
 pub const EX_USAGE: i32 = 64;
@@ -1148,13 +1152,22 @@ pub fn sync() {
     unsafe { libc::sync() };
 }
 
-pub fn getlogin() -> Option<CString> {
+pub fn getlogin() -> std::io::Result<CString> {
+    let old_errno = crate::os::get_errno();
+    crate::os::clear_errno();
     let ptr = unsafe { libc::getlogin() };
-    if ptr.is_null() {
-        None
+    let result = if ptr.is_null() {
+        let errno = crate::os::get_errno();
+        Err(if errno == 0 {
+            std::io::Error::other("unable to determine login name")
+        } else {
+            std::io::Error::from_raw_os_error(errno)
+        })
     } else {
-        Some(unsafe { CStr::from_ptr(ptr) }.to_owned())
-    }
+        Ok(unsafe { CStr::from_ptr(ptr) }.to_owned())
+    };
+    crate::os::set_errno(old_errno);
+    result
 }
 
 pub fn restore_signals() {
@@ -1945,26 +1958,71 @@ pub fn posix_spawn(config: PosixSpawnConfig<'_>) -> std::io::Result<libc::pid_t>
 
 #[cfg(target_os = "linux")]
 pub fn sendfile(
-    out_fd: BorrowedFd<'_>,
-    in_fd: BorrowedFd<'_>,
-    offset: &mut crate::crt_fd::Offset,
+    out_fd: i32,
+    in_fd: i32,
+    offset: Option<&mut crate::crt_fd::Offset>,
     count: usize,
 ) -> std::io::Result<usize> {
-    nix::sys::sendfile::sendfile(out_fd, in_fd, Some(offset), count).map_err(std::io::Error::from)
+    let offset = offset.map_or(core::ptr::null_mut(), |offset| offset as *mut _);
+    // The syscall validates raw descriptor numbers; no Rust fd lifetime is claimed.
+    let result = unsafe { libc::sendfile(out_fd, in_fd, offset, count) };
+    if result < 0 {
+        Err(std::io::Error::last_os_error())
+    } else {
+        Ok(result as usize)
+    }
 }
 
 #[cfg(target_os = "macos")]
 pub fn sendfile(
-    in_fd: BorrowedFd<'_>,
-    out_fd: BorrowedFd<'_>,
+    in_fd: i32,
+    out_fd: i32,
     offset: crate::crt_fd::Offset,
-    count: i64,
+    written: &mut i64,
     headers: Option<&[&[u8]]>,
     trailers: Option<&[&[u8]]>,
-) -> (std::io::Result<()>, i64) {
-    let (res, written) =
-        nix::sys::sendfile::sendfile(in_fd, out_fd, offset, Some(count), headers, trailers);
-    (res.map_err(std::io::Error::from), written)
+    flags: i32,
+) -> std::io::Result<()> {
+    let header_count = headers.map_or(0, |buffers| buffers.len());
+    let trailer_count = trailers.map_or(0, |buffers| buffers.len());
+    let (Ok(header_count), Ok(trailer_count)) =
+        (i32::try_from(header_count), i32::try_from(trailer_count))
+    else {
+        return Err(std::io::Error::from(std::io::ErrorKind::InvalidInput));
+    };
+    let iovecs = |buffers: Option<&[&[u8]]>| {
+        buffers
+            .unwrap_or_default()
+            .iter()
+            .map(|buffer| libc::iovec {
+                iov_base: buffer.as_ptr().cast_mut().cast(),
+                iov_len: buffer.len(),
+            })
+            .collect::<Vec<_>>()
+    };
+    let mut headers = iovecs(headers);
+    let mut trailers = iovecs(trailers);
+    let mut hdtr = libc::sf_hdtr {
+        headers: if headers.is_empty() {
+            core::ptr::null_mut()
+        } else {
+            headers.as_mut_ptr()
+        },
+        hdr_cnt: header_count,
+        trailers: if trailers.is_empty() {
+            core::ptr::null_mut()
+        } else {
+            trailers.as_mut_ptr()
+        },
+        trl_cnt: trailer_count,
+    };
+    // The slices and both iovec arrays remain live for this synchronous call.
+    let result = unsafe { libc::sendfile(in_fd, out_fd, offset, written, &mut hdtr, flags) };
+    if result < 0 {
+        Err(std::io::Error::last_os_error())
+    } else {
+        Ok(())
+    }
 }
 
 #[cfg(any(

crates/vm/src/stdlib/os.rs

@@ -209,16 +209,17 @@ pub(super) mod _os {
     #[cfg(any(unix, windows))]
     use crate::utils::ToCString;
     use crate::{
-        AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, TryFromObject,
+        AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult,
         builtins::{
-            PyBytesRef, PyFloat, PyGenericAlias, PyIntRef, PyStrRef, PyTuple, PyTupleRef, PyTypeRef,
+            PyBytes, PyBytesRef, PyFloat, PyGenericAlias, PyIntRef, PyStr, PyStrRef, PyTuple,
+            PyTupleRef, PyTypeRef,
         },
         class::PyClassDef,
         common::lock::{OnceCell, PyRwLock},
         convert::{IntoPyException, ToPyException, ToPyObject},
         exceptions::{OSErrorBuilder, ToOSErrorBuilder},
         function::{
-            ArgBytesLike, ArgMemoryBuffer, ArgumentError, FromArgs, FsPath, FuncArgs, Param,
+            ArgBytesLike, ArgMemoryBuffer, ArgumentError, Either, FromArgs, FsPath, FuncArgs, Param,
         },
         host_env::crt_fd,
         ospath::{OsPath, OsPathOrFd, OutputMode, PathConverter},
@@ -1911,41 +1912,35 @@ pub(super) mod _os {
     }
 
     fn file_time(obj: &PyObject, vm: &VirtualMachine) -> PyResult<FileTime> {
-        if let Some(float) = obj.downcast_ref::<PyFloat>() {
-            let value = float.to_f64();
-            if value.is_nan() {
-                return Err(vm.new_value_error("Invalid value NaN (not a number)"));
-            }
-            let mut seconds = value.trunc();
-            let mut nanoseconds = (value.fract() * 1_000_000_000.0).floor();
-            if nanoseconds < 0.0 {
-                nanoseconds += 1_000_000_000.0;
-                seconds -= 1.0;
-            }
-            let minimum = FileTimeSeconds::MIN as f64;
-            // The signed maximum rounds up when represented by f64.
-            if !(minimum..-minimum).contains(&seconds) {
-                return Err(vm.new_overflow_error("timestamp out of range for platform time_t"));
-            }
-            Ok(FileTime {
-                seconds: seconds as FileTimeSeconds,
-                nanoseconds: nanoseconds as libc::c_long,
-            })
-        } else {
-            let seconds = match file_time_seconds(obj, vm) {
-                Err(err) if err.fast_isinstance(vm.ctx.exceptions.type_error) => {
-                    return Err(vm.new_type_error(format!(
-                        "argument must be int or float, not {}",
-                        obj.class().fully_qualified_name(vm)?
-                    )));
-                }
-                result => result?,
-            };
-            Ok(FileTime {
-                seconds,
+        if obj.number().is_index() {
+            return Ok(FileTime {
+                seconds: file_time_seconds(obj, vm)?,
                 nanoseconds: 0,
-            })
+            });
         }
+        let value = if let Some(float) = obj.downcast_ref::<PyFloat>() {
+            float.to_f64()
+        } else {
+            obj.try_float(vm)?.to_f64()
+        };
+        if value.is_nan() {
+            return Err(vm.new_value_error("Invalid value NaN (not a number)"));
+        }
+        let mut seconds = value.trunc();
+        let mut nanoseconds = (value.fract() * 1_000_000_000.0).floor();
+        if nanoseconds < 0.0 {
+            nanoseconds += 1_000_000_000.0;
+            seconds -= 1.0;
+        }
+        let minimum = FileTimeSeconds::MIN as f64;
+        // The signed maximum rounds up when represented by f64.
+        if !(minimum..-minimum).contains(&seconds) {
+            return Err(vm.new_overflow_error("timestamp out of range for platform time_t"));
+        }
+        Ok(FileTime {
+            seconds: seconds as FileTimeSeconds,
+            nanoseconds: nanoseconds as libc::c_long,
+        })
     }
 
     #[pyfunction]
@@ -2183,27 +2178,66 @@ pub(super) mod _os {
         crate::host_env::time::strerror(code)
     }
 
-    #[pyfunction]
     pub(crate) fn ftruncate(fd: crt_fd::Borrowed<'_>, length: crt_fd::Offset) -> io::Result<()> {
         crt_fd::ftruncate(fd, length)
     }
 
+    #[pyfunction(name = "ftruncate")]
+    fn py_ftruncate(fd: i32, length: crt_fd::Offset, vm: &VirtualMachine) -> PyResult<()> {
+        ftruncate_impl(fd, length, vm)
+    }
+
+    fn ftruncate_impl(
+        fd: crt_fd::Raw,
+        length: crt_fd::Offset,
+        vm: &VirtualMachine,
+    ) -> PyResult<()> {
+        use crate::host_env::os::ErrorExt;
+
+        vm.audit("os.truncate", || (fd, length))?;
+        loop {
+            match vm.allow_threads(|| crt_fd::ftruncate_raw(fd, length)) {
+                Ok(()) => return Ok(()),
+                Err(err) if err.posix_errno() == libc::EINTR => vm.check_signals()?,
+                Err(err) => return Err(err.into_pyexception(vm)),
+            }
+        }
+    }
+
     #[derive(FromArgs)]
     struct TruncateArgs {
         #[pyarg(any)]
         path: PyObjectRef,
         #[pyarg(any)]
-        length: crt_fd::Offset,
+        length: PyObjectRef,
     }
 
     #[pyfunction]
     fn truncate(args: TruncateArgs, vm: &VirtualMachine) -> PyResult<()> {
         let TruncateArgs { path, length } = args;
-        match path.clone().try_into_value::<crt_fd::Borrowed<'_>>(vm) {
-            Ok(fd) => return ftruncate(fd, length).map_err(|e| e.into_pyexception(vm)),
-            Err(e) if e.fast_isinstance(vm.ctx.exceptions.warning) => return Err(e),
-            Err(_) => {}
-        }
+        let index =
+            if path.downcast_ref::<PyStr>().is_some() || path.downcast_ref::<PyBytes>().is_some() {
+                None
+            } else {
+                path.try_index_opt(vm)
+            };
+        let path = if let Some(index) = index {
+            super::warn_if_bool_fd(&path, vm)?;
+            Either::A(index?.try_to_primitive::<crt_fd::Raw>(vm)?)
+        } else {
+            Either::B(
+                PathConverter::new()
+                    .function("truncate")
+                    .try_path_inner(path, true, vm)?,
+            )
+        };
+        let length = length.try_into_value::<crt_fd::Offset>(vm)?;
+        let path = match path {
+            Either::A(fd) => {
+                return ftruncate_impl(fd, length, vm);
+            }
+            Either::B(path) => path,
+        };
 
         #[cold]
         fn error(
@@ -2214,7 +2248,8 @@ pub(super) mod _os {
             OSErrorBuilder::with_filename(&error, path, vm)
         }
 
-        let path = OsPath::try_from_object(vm, path)?;
+        vm.audit("os.truncate", || (path.filename(vm), length))?;
+
         // TODO: just call libc::truncate() on POSIX
         let f = match crate::host_env::fs::open_write(&path) {
             Ok(f) => f,
@@ -2439,6 +2474,7 @@ pub(super) mod _os {
             SupportFunc::new("rmdir", Some(false), Some(RMDIR_DIR_FD), Some(false)),
             SupportFunc::new("scandir", Some(SCANDIR_FD), Some(false), Some(false)),
             SupportFunc::new("stat", Some(true), Some(STAT_DIR_FD), Some(true)),
+            SupportFunc::new("lstat", Some(false), Some(STAT_DIR_FD), Some(false)),
             SupportFunc::new("fstat", Some(false), Some(false), Some(false)),
             SupportFunc::new("symlink", Some(false), Some(SYMLINK_DIR_FD), Some(false)),
             SupportFunc::new("truncate", Some(true), Some(false), Some(false)),

crates/vm/src/stdlib/posix.rs

@@ -49,6 +49,10 @@ pub mod module {
     #[pyattr]
     use rustpython_host_env::posix::PIDFD_NONBLOCK;
 
+    #[cfg(all(target_os = "linux", target_env = "gnu"))]
+    #[pyattr]
+    use rustpython_host_env::posix::NODEV;
+
     #[cfg(target_os = "macos")]
     #[pyattr]
     use rustpython_host_env::posix::{
@@ -881,7 +885,11 @@ pub mod module {
 
     /// Warn if forking from a multi-threaded process.
     /// `num_os_threads` should be captured before parent after-fork hooks run.
-    fn warn_if_multi_threaded(name: &str, num_os_threads: isize, vm: &VirtualMachine) {
+    fn warn_if_multi_threaded(
+        name: &str,
+        num_os_threads: isize,
+        vm: &VirtualMachine,
+    ) -> PyResult<()> {
         let num_threads = if num_os_threads > 0 {
             num_os_threads as usize
         } else {
@@ -895,7 +903,7 @@ pub mod module {
                 .and_then(|m| m.get_item("threading", vm))
             {
                 Ok(m) => m,
-                Err(_) => return,
+                Err(_) => return Ok(()),
             };
             let active = threading.get_attr("_active", vm).ok();
             let limbo = threading.get_attr("_limbo", vm).ok();
@@ -920,10 +928,9 @@ pub mod module {
             );
 
             // Match PyErr_WarnFormat(..., stacklevel=1) in CPython.
-            // Best effort: ignore failures like CPython does in this path.
-            let _ =
-                crate::stdlib::_warnings::warn(vm.ctx.exceptions.deprecation_warning, msg, 1, vm);
+            crate::stdlib::_warnings::warn(vm.ctx.exceptions.deprecation_warning, msg, 1, vm)?;
         }
+        Ok(())
     }
 
     #[pyfunction]
@@ -960,7 +967,7 @@ pub mod module {
                 let num_os_threads = get_number_of_os_threads();
                 py_os_after_fork_parent(vm);
                 // Warn only after parent callback path resumes the world.
-                warn_if_multi_threaded("fork", num_os_threads, vm);
+                warn_if_multi_threaded("fork", num_os_threads, vm)?;
                 Ok(pid)
             }
             Err(err) => {
@@ -1002,7 +1009,7 @@ pub mod module {
             Ok((pid, master)) => {
                 let num_os_threads = get_number_of_os_threads();
                 py_os_after_fork_parent(vm);
-                warn_if_multi_threaded("forkpty", num_os_threads, vm);
+                warn_if_multi_threaded("forkpty", num_os_threads, vm)?;
                 Ok((pid, master))
             }
             Err(err) => {
@@ -2347,23 +2354,10 @@ pub mod module {
     }
 
     #[pyfunction]
-    fn getlogin(vm: &VirtualMachine) -> PyResult<String> {
-        // Get a pointer to the login name string. The string is statically
-        // allocated and might be overwritten on subsequent calls to this
-        // function or to `cuserid()`. See man getlogin(3) for more information.
-        let Some(login) = rustpython_host_env::posix::getlogin() else {
-            return Err(vm.new_os_error("unable to determine login name"));
-        };
-        login.to_str().map(|s| s.to_owned()).map_err(|e| {
-            vm.new_unicode_decode_error(
-                vm.ctx.new_str("utf-8"),
-                vm.ctx.new_bytes(login.as_bytes().to_vec()),
-                e.valid_up_to(),
-                e.error_len()
-                    .map_or(login.as_bytes().len(), |n| e.valid_up_to() + n),
-                vm.ctx.new_str("unable to decode login name"),
-            )
-        })
+    fn getlogin(vm: &VirtualMachine) -> PyResult<crate::builtins::PyStrRef> {
+        let login =
+            rustpython_host_env::posix::getlogin().map_err(|err| err.into_pyexception(vm))?;
+        Ok(vm.fsdecode(std::ffi::OsString::from_vec(login.into_bytes())))
     }
 
     // cfg from nix
@@ -2868,10 +2862,16 @@ pub mod module {
 
     #[cfg(any(target_os = "linux", target_os = "macos"))]
     #[derive(FromArgs)]
-    struct SendFileArgs<'fd> {
-        out_fd: BorrowedFd<'fd>,
-        in_fd: BorrowedFd<'fd>,
+    struct SendFileArgs {
+        out_fd: i32,
+        in_fd: i32,
+        #[cfg(target_os = "linux")]
+        offset: PyObjectRef,
+        #[cfg(target_os = "macos")]
         offset: rustpython_host_env::crt_fd::Offset,
+        #[cfg(target_os = "linux")]
+        count: isize,
+        #[cfg(target_os = "macos")]
         count: i64,
         #[cfg(target_os = "macos")]
         // Missing means an empty header list.
@@ -2882,35 +2882,58 @@ pub mod module {
         #[pyarg(any, optional, py_default = "()")]
         trailers: OptionalArg<PyObjectRef>,
         #[cfg(target_os = "macos")]
-        #[allow(dead_code)]
         #[pyarg(any, default)]
-        // TODO: not implemented
         flags: i32,
     }
 
     #[cfg(target_os = "linux")]
     #[pyfunction]
-    fn sendfile(args: SendFileArgs<'_>, vm: &VirtualMachine) -> PyResult {
-        let mut file_offset = args.offset;
+    fn sendfile(args: SendFileArgs, vm: &VirtualMachine) -> PyResult {
+        if args.count < 0 {
+            return Err(vm.new_value_error("count cannot be negative"));
+        }
+        let mut file_offset = if vm.is_none(&args.offset) {
+            None
+        } else {
+            Some(args.offset.try_into_value(vm)?)
+        };
 
-        let res = rustpython_host_env::posix::sendfile(
-            args.out_fd,
-            args.in_fd,
-            &mut file_offset,
-            args.count as usize,
-        )
-        .map_err(|err| err.into_pyexception(vm))?;
-        Ok(vm.ctx.new_int(res as u64).into())
+        loop {
+            let result = vm.allow_threads(|| {
+                rustpython_host_env::posix::sendfile(
+                    args.out_fd,
+                    args.in_fd,
+                    file_offset.as_mut(),
+                    args.count as usize,
+                )
+            });
+            match result {
+                Ok(written) => return Ok(vm.ctx.new_int(written).into()),
+                Err(err) if err.raw_os_error() == Some(libc::EINTR) => vm.check_signals()?,
+                Err(err) => return Err(err.into_pyexception(vm)),
+            }
+        }
     }
 
     #[cfg(target_os = "macos")]
     fn _extract_vec_bytes(
         x: OptionalArg,
+        kind: &str,
         vm: &VirtualMachine,
     ) -> PyResult<Option<Vec<crate::function::ArgBytesLike>>> {
         x.into_option()
             .map(|x| {
-                let v: Vec<crate::function::ArgBytesLike> = x.try_to_value(vm)?;
+                let sequence = x.try_sequence(vm).map_err(|_| {
+                    vm.new_type_error(format!("sendfile() {kind}s must be a sequence"))
+                })?;
+                let count = sequence.length(vm)?;
+                if count > i32::MAX as usize {
+                    return Err(vm.new_overflow_error(format!("sendfile() {kind} is too large")));
+                }
+                let mut v = Vec::with_capacity(count);
+                for index in 0..count {
+                    v.push(sequence.get_item(index as isize, vm)?.try_into_value(vm)?);
+                }
                 Ok(if v.is_empty() { None } else { Some(v) })
             })
             .transpose()
@@ -2919,47 +2942,78 @@ pub mod module {
 
     #[cfg(target_os = "macos")]
     #[pyfunction]
-    fn sendfile(args: SendFileArgs<'_>, vm: &VirtualMachine) -> PyResult {
-        let headers = _extract_vec_bytes(args.headers, vm)?;
-        let count = headers
-            .as_ref()
-            .map_or(0, |v| v.iter().map(|s| s.len()).sum()) as i64
-            + args.count;
+    fn sendfile(args: SendFileArgs, vm: &VirtualMachine) -> PyResult {
+        if args.count < 0 {
+            return Err(vm.new_value_error("count cannot be negative"));
+        }
+        let headers = _extract_vec_bytes(args.headers, "header", vm)?;
+        let mut count = args.count;
+        if let Some(headers) = &headers {
+            for header in headers {
+                let length = i64::try_from(header.len())
+                    .map_err(|_| vm.new_overflow_error("sendfile() header is too large"))?;
+                if count >= i64::MAX - length {
+                    return Err(vm.new_overflow_error("sendfile() header is too large"));
+                }
+                count += length;
+            }
+        }
 
+        let trailers = _extract_vec_bytes(args.trailers, "trailer", vm)?;
+        // Complete all Python argument callbacks before snapshotting mutable
+        // exporters. No buffer lock may survive another callback or blocking I/O.
         let headers = headers
             .as_ref()
-            .map(|v| v.iter().map(|b| b.borrow_buf()).collect::<Vec<_>>());
+            .map(|v| {
+                v.iter()
+                    .map(|b| b.borrow_buf_unlocked(vm))
+                    .collect::<PyResult<Vec<_>>>()
+            })
+            .transpose()?;
         let headers = headers
             .as_ref()
             .map(|v| v.iter().map(|borrowed| &**borrowed).collect::<Vec<_>>());
         let headers = headers.as_deref();
 
-        let trailers = _extract_vec_bytes(args.trailers, vm)?;
         let trailers = trailers
             .as_ref()
-            .map(|v| v.iter().map(|b| b.borrow_buf()).collect::<Vec<_>>());
+            .map(|v| {
+                v.iter()
+                    .map(|b| b.borrow_buf_unlocked(vm))
+                    .collect::<PyResult<Vec<_>>>()
+            })
+            .transpose()?;
         let trailers = trailers
             .as_ref()
             .map(|v| v.iter().map(|borrowed| &**borrowed).collect::<Vec<_>>());
         let trailers = trailers.as_deref();
 
-        let (res, written) = rustpython_host_env::posix::sendfile(
-            args.in_fd,
-            args.out_fd,
-            args.offset,
-            count,
-            headers,
-            trailers,
-        );
-        // On macOS, sendfile can return EAGAIN even when some bytes were written.
-        // In that case, we should return the number of bytes written rather than
-        // raising an exception. Only raise an error if no bytes were written.
-        if let Err(err) = res
-            && written == 0
-        {
-            return Err(err.into_pyexception(vm));
+        let mut written = count;
+        loop {
+            let result = vm.allow_threads(|| {
+                rustpython_host_env::posix::sendfile(
+                    args.in_fd,
+                    args.out_fd,
+                    args.offset,
+                    &mut written,
+                    headers,
+                    trailers,
+                    args.flags,
+                )
+            });
+            match result {
+                Ok(()) => return Ok(vm.ctx.new_int(written).into()),
+                Err(err) if err.raw_os_error() == Some(libc::EINTR) => vm.check_signals()?,
+                // CPython permits partial success only for EAGAIN or EBUSY.
+                Err(err)
+                    if written != 0
+                        && matches!(err.raw_os_error(), Some(libc::EAGAIN | libc::EBUSY)) =>
+                {
+                    return Ok(vm.ctx.new_int(written).into());
+                }
+                Err(err) => return Err(err.into_pyexception(vm)),
+            }
         }
-        Ok(vm.ctx.new_int(written as u64).into())
     }
 
     #[cfg(target_os = "linux")]

crates/vm/src/stdlib/posix_compat.rs

@@ -10,6 +10,8 @@ pub(crate) use module::module_def;
     super::posix_unix_like::_posix_unix_like,
 ))]
 pub(crate) mod module {
+    #[cfg(target_os = "wasi")]
+    use crate::{AsObject, function::OptionalArg, ospath::PathConverter};
     use crate::{
         Py, PyObjectRef, PyResult, VirtualMachine,
         builtins::PyStrRef,
@@ -17,6 +19,64 @@ pub(crate) mod module {
         stdlib::os::{_os, DirFd, SupportFunc, SymlinkArgs, TargetIsDirectory},
     };
 
+    #[cfg(target_os = "wasi")]
+    #[derive(FromArgs)]
+    struct ChmodArgs {
+        #[pyarg(any)]
+        path: PyObjectRef,
+        #[pyarg(any)]
+        mode: PyObjectRef,
+        #[pyarg(named, default, py_default = "None")]
+        dir_fd: Option<PyObjectRef>,
+        #[pyarg(named, optional, py_default = "(os.name != 'nt')")]
+        follow_symlinks: OptionalArg<PyObjectRef>,
+    }
+
+    #[cfg(target_os = "wasi")]
+    #[pyfunction]
+    fn chmod(args: ChmodArgs, vm: &VirtualMachine) -> PyResult<()> {
+        // Keep the clinic conversion order even though WASI has no chmod syscall.
+        let path = PathConverter::new()
+            .function("chmod")
+            .argument("path")
+            .try_path(args.path, vm)?;
+        let mode = args.mode.try_index(vm)?;
+        let mode = i32::try_from(mode.as_bigint())
+            .map_err(|_| vm.new_overflow_error("Python int too large to convert to C int"))?;
+        if let Some(dir_fd) = args.dir_fd {
+            super::super::os::warn_if_bool_fd(&dir_fd, vm)?;
+            let fd = dir_fd.try_index_opt(vm).unwrap_or_else(|| {
+                Err(vm.new_type_error(format!(
+                    "argument should be integer or None, not {}",
+                    dir_fd.class().name()
+                )))
+            })?;
+            let fd = i32::try_from(fd.as_bigint()).map_err(|_| {
+                vm.new_overflow_error(if fd.as_bigint().sign() == malachite_bigint::Sign::Minus {
+                    "fd is less than minimum"
+                } else {
+                    "fd is greater than maximum"
+                })
+            })?;
+            if fd != rustpython_host_env::os::AT_FDCWD {
+                return Err(vm.new_not_implemented_error("dir_fd unavailable on this platform"));
+            }
+        }
+        let follow_symlinks = match args.follow_symlinks {
+            OptionalArg::Present(value) => value.try_to_bool(vm)?,
+            OptionalArg::Missing => true,
+        };
+        if !follow_symlinks {
+            return Err(
+                vm.new_not_implemented_error("chmod: follow_symlinks unavailable on this platform")
+            );
+        }
+        vm.audit("os.chmod", || (path.filename(vm), mode, -1))?;
+        // CPython 3.15 posixmodule.c explicitly ignores the absent WASI syscall
+        // after argument conversion and auditing. No supports_* capability is set.
+        Ok(())
+    }
+
     #[derive(FromArgs)]
     struct AccessArgs<'a> {
         #[pyarg(any)]

scripts/update_lib/cmd_migrate.py

@@ -20,7 +20,9 @@ This will:
 
 import argparse
 import ast
+import builtins
 import collections
+import hashlib
 import json
 import pathlib
 import shutil
@@ -186,6 +188,244 @@ def _verify_call(contents: str, expected: str) -> None:
         raise ValueError("Ambiguous environment-reset call text")
 
 
+def copy_support_assignments(source, destination, names, exports):
+    """Copy contiguous canonical assignments and export lines, failing closed.
+
+    Only additions are supported. Dependencies must already be imported with the
+    same bindings, or be builtins/earlier selected assignments. The next unchanged
+    top-level statement is the insertion anchor; unrelated bytes stay untouched.
+    """
+    if (
+        not names
+        or len(set(names)) != len(names)
+        or not all(name.isidentifier() for name in names)
+    ):
+        raise ValueError("Assignment names must be unique identifiers")
+    if len(set(exports)) != len(exports) or not set(exports) <= set(names):
+        raise ValueError("Exports must be unique selected assignment names")
+    src, dst = ast.parse(source), ast.parse(destination)
+
+    def assignment(tree, name):
+        matches = [
+            node
+            for node in ast.walk(tree)
+            if isinstance(node, ast.Name)
+            and isinstance(node.ctx, ast.Store)
+            and node.id == name
+        ]
+        nodes = [
+            node
+            for node in tree.body
+            if isinstance(node, ast.Assign)
+            and len(node.targets) == 1
+            and isinstance(node.targets[0], ast.Name)
+            and node.targets[0].id == name
+        ]
+        rebound = any(
+            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
+            and node.name == name
+            or isinstance(node, ast.alias)
+            and (node.asname or node.name.split(".")[0]) == name
+            for node in ast.walk(tree)
+        )
+        if len(nodes) != 1 or matches != [nodes[0].targets[0]] or rebound:
+            raise ValueError(f"Expected one top-level assignment to {name}")
+        return nodes[0]
+
+    selected = [assignment(src, name) for name in names]
+    start = src.body.index(selected[0])
+    if src.body[start : start + len(names)] != selected or start + len(names) == len(
+        src.body
+    ):
+        raise ValueError("Assignments must be contiguous and followed by an anchor")
+    if (start and src.body[start - 1].end_lineno >= selected[0].lineno) or (
+        selected[-1].end_lineno >= src.body[start + len(names)].lineno
+    ):
+        raise ValueError("Selected assignments must occupy complete source lines")
+    for node in ast.walk(dst):
+        if (
+            isinstance(node, ast.Name)
+            and isinstance(node.ctx, ast.Store)
+            and node.id in names
+            or isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
+            and node.name in names
+            or isinstance(node, ast.alias)
+            and (node.asname or node.name.split(".")[0]) in names
+        ):
+            raise ValueError("A selected assignment already exists in the destination")
+
+    anchor = src.body[start + len(names)]
+    matches = [node for node in dst.body if ast.dump(node) == ast.dump(anchor)]
+    if len(matches) != 1:
+        raise ValueError("Expected one unchanged destination anchor")
+    for tree, node in ((src, anchor), (dst, matches[0])):
+        index = tree.body.index(node)
+        if (
+            getattr(node, "decorator_list", ())
+            or node.col_offset
+            or index
+            and tree.body[index - 1].end_lineno >= node.lineno
+            or index + 1 < len(tree.body)
+            and tree.body[index + 1].lineno <= node.end_lineno
+        ):
+            raise ValueError(
+                "Anchor must be undecorated and occupy complete source lines"
+            )
+    available = set(vars(builtins))
+    imports = (set(), set())
+    for node in src.body:
+        matching = [
+            other
+            for other in dst.body
+            if ast.dump(node) == ast.dump(other) and other.lineno < matches[0].lineno
+        ]
+        if (
+            isinstance(node, (ast.Import, ast.ImportFrom))
+            and node.lineno < selected[0].lineno
+            and len(matching) == 1
+        ):
+            imports[0].update(node.names)
+            imports[1].update(matching[0].names)
+            available.update(
+                alias.asname or alias.name.split(".")[0] for alias in node.names
+            )
+    dependencies = set()
+    for name, node in zip(names, selected):
+        used = {
+            child.id
+            for child in ast.walk(node.value)
+            if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load)
+        }
+        missing = used - available
+        if missing:
+            raise ValueError(f"Missing dependencies for {name}: {sorted(missing)}")
+        dependencies.update(used - set(names))
+        available.add(name)
+    for tree, boundary, allowed in (
+        (src, selected[0].lineno, imports[0]),
+        (dst, matches[0].lineno, imports[1]),
+    ):
+        if dependencies and any(
+            isinstance(node, ast.ImportFrom)
+            and node.lineno < boundary
+            and any(alias.name == "*" for alias in node.names)
+            for node in ast.walk(tree)
+        ):
+            raise ValueError("Wildcard import makes dependency bindings ambiguous")
+        for dependency in dependencies:
+            bindings = []
+            for node in ast.walk(tree):
+                if getattr(node, "lineno", boundary) >= boundary:
+                    continue
+                if isinstance(node, ast.Name) and isinstance(
+                    node.ctx, (ast.Store, ast.Del)
+                ):
+                    name = node.id
+                elif isinstance(node, ast.alias):
+                    name = node.asname or node.name.split(".")[0]
+                elif isinstance(
+                    node,
+                    (
+                        ast.FunctionDef,
+                        ast.AsyncFunctionDef,
+                        ast.ClassDef,
+                        ast.ExceptHandler,
+                        ast.MatchAs,
+                        ast.MatchStar,
+                    ),
+                ):
+                    name = node.name
+                elif isinstance(node, ast.arg):
+                    name = node.arg
+                elif isinstance(node, ast.MatchMapping):
+                    name = node.rest
+                else:
+                    continue
+                if name == dependency:
+                    bindings.append(node)
+            if len(bindings) > 1 or any(node not in allowed for node in bindings):
+                raise ValueError(f"Ambiguous dependency binding: {dependency}")
+
+    source_lines, lines = source.splitlines(True), destination.splitlines(True)
+    additions = [
+        (
+            matches[0].lineno - 1,
+            source_lines[selected[0].lineno - 1 : anchor.lineno - 1],
+        )
+    ]
+    if exports:
+        src_all, dst_all = assignment(src, "__all__"), assignment(dst, "__all__")
+        if not isinstance(src_all.value, ast.List) or not isinstance(
+            dst_all.value, ast.List
+        ):
+            raise ValueError("__all__ must be a literal list")
+        if any(
+            not isinstance(node, ast.Constant) or not isinstance(node.value, str)
+            for node in src_all.value.elts + dst_all.value.elts
+        ):
+            raise ValueError("__all__ entries must be literal strings")
+        src_values, dst_values = (
+            ast.literal_eval(src_all.value),
+            ast.literal_eval(dst_all.value),
+        )
+        if any(src_values.count(name) != 1 or name in dst_values for name in exports):
+            raise ValueError(
+                "Exports must occur once in canonical __all__ and be absent locally"
+            )
+        export_nodes = [node for node in src_all.value.elts if node.value in exports]
+        export_lines = sorted({node.lineno - 1 for node in export_nodes})
+        fragment = "".join(source_lines[line] for line in export_lines)
+        if ast.literal_eval("[\n" + fragment + "]") != exports:
+            raise ValueError(
+                "Canonical export lines include unselected entries or differ in order"
+            )
+        if lines[dst_all.lineno - 1].strip() != "__all__ = [":
+            raise ValueError("Expected a multiline destination __all__ list")
+        additions.append((dst_all.lineno, [fragment]))
+    for offset, fragment in sorted(additions, reverse=True):
+        lines[offset:offset] = fragment
+    result = "".join(lines)
+    ast.parse(result)
+    return result
+
+
+def migrate_support_assignments(path, names, exports, source_hash, destination_hash):
+    target = parse_lib_path(path)
+    if (
+        target.parts[:3] != ("Lib", "test", "support")
+        or ".." in target.parts
+        or target.suffix != ".py"
+    ):
+        raise ValueError(
+            "Selective assignments are restricted to Lib/test/support Python files"
+        )
+    source, destination = path.read_bytes(), target.read_bytes()
+    for label, contents, expected in (
+        ("source", source, source_hash),
+        ("destination", destination, destination_hash),
+    ):
+        if hashlib.sha256(contents).hexdigest() != expected:
+            raise ValueError(f"{label} SHA256 does not match the pinned input")
+    result = copy_support_assignments(
+        source.decode("utf-8"), destination.decode("utf-8"), names, exports
+    ).encode("utf-8")
+    if target.read_bytes() != destination:
+        raise ValueError("Destination changed during assignment preparation")
+    target.write_bytes(result)
+    print(
+        json.dumps(
+            {
+                "assignments": names,
+                "exports": exports,
+                "source_sha256": source_hash,
+                "destination_before_sha256": destination_hash,
+                "destination_after_sha256": hashlib.sha256(result).hexdigest(),
+            },
+            sort_keys=True,
+        )
+    )
+
+
 def patch_single_content(
     src_path: pathlib.Path,
     lib_path: pathlib.Path,
@@ -485,9 +725,35 @@ def main(argv: list[str] | None = None) -> int:
         type=pathlib.Path,
         help="Single-file target, remaining decorator patches, and optional doctests JSON",
     )
+    parser.add_argument(
+        "--support-assignments",
+        nargs="+",
+        help="Add selected canonical test.support assignments",
+    )
+    parser.add_argument("--support-exports", nargs="+", default=[])
+    parser.add_argument("--source-sha256")
+    parser.add_argument("--destination-sha256")
     args = parser.parse_args(argv)
 
     try:
+        if args.support_assignments:
+            if args.legacy_map is not None or args.patches is not None:
+                raise ValueError(
+                    "Selective support imports cannot combine with --legacy-map or --patches"
+                )
+            try:
+                migrate_support_assignments(
+                    args.path,
+                    args.support_assignments,
+                    args.support_exports,
+                    args.source_sha256,
+                    args.destination_sha256,
+                )
+            except (SyntaxError, UnicodeError) as exc:
+                raise ValueError(str(exc)) from exc
+            return 0
+        if args.support_exports or args.source_sha256 or args.destination_sha256:
+            raise ValueError("Selective options require --support-assignments")
         if args.patches is not None and (
             args.path.is_dir() or args.legacy_map is not None
         ):

scripts/update_lib/tests/test_migrate.py

@@ -1,6 +1,7 @@
 """Tests for migrate.py - file migration operations."""
 
 import contextlib
+import hashlib
 import io
 import json
 import pathlib
@@ -16,6 +17,7 @@ from update_lib.cmd_migrate import (
     LOCAL_CALL,
     SYSCONFIG_LOCAL,
     SYSCONFIG_UPSTREAM,
+    copy_support_assignments,
     main,
     migrate_package,
     patch_directory,
@@ -35,6 +37,235 @@ from update_lib.patch_spec import (
 )
 
 
+class TestSupportAssignments(unittest.TestCase):
+    names = ["_available", "requires_available", "requires_unavailable"]
+    exports = names[1:]
+    imports = "import os\nimport unittest\n\n"
+    export_line = '    "requires_available", "requires_unavailable",\n'
+    block = (
+        "_available = hasattr(os, 'geteuid') and os.geteuid() == 0\n"
+        'requires_available = unittest.skipUnless(_available, "needs privilege")\n'
+        'requires_unavailable = unittest.skipIf(_available, "needs no privilege")\n\n\n'
+    )
+    source = (
+        imports
+        + '__all__ = [\n    "existing",\n'
+        + export_line
+        + "]\n\n\n"
+        + block
+        + "STATUS = 1\n"
+    )
+    destination = (
+        imports
+        + '__all__ = [\n    "existing",\n]\n\n\ndef legacy():\n    return "keep π"\n\n\nSTATUS = 1\n'
+    )
+
+    def test_canonical_bytes_and_unrelated_helpers_are_preserved(self):
+        result = copy_support_assignments(
+            self.source, self.destination, self.names, self.exports
+        )
+        self.assertEqual(
+            result,
+            self.destination.replace(
+                "__all__ = [\n", "__all__ = [\n" + self.export_line
+            ).replace("STATUS = 1\n", self.block + "STATUS = 1\n"),
+        )
+
+    def test_missing_ambiguous_and_noncontiguous_assignments_are_rejected(self):
+        for source in (
+            self.source.replace("_available =", "_other ="),
+            self.source + "_available = False\n",
+            self.source + "def requires_available(): pass\n",
+            self.source + "import other as requires_unavailable\n",
+            self.source.replace(
+                "requires_available =", "unselected = 0\nrequires_available ="
+            ),
+            self.source.replace("_available =", "unselected = 0; _available =", 1),
+            self.source.replace("STATUS = 1", ""),
+        ):
+            with self.subTest(source=source), self.assertRaises(ValueError):
+                copy_support_assignments(
+                    source, self.destination, self.names, self.exports
+                )
+
+    def test_existing_bindings_dependencies_and_anchor_fail_closed(self):
+        for destination in (
+            self.destination + "_available = False\n",
+            self.destination + "def requires_available(): pass\n",
+            self.destination + "import other as requires_unavailable\n",
+            self.destination.replace("import os\n", ""),
+            self.destination.replace("import os\n", "import os as other\n"),
+            self.destination.replace("import os\n", "") + "import os\n",
+            self.destination.replace("STATUS = 1", "STATUS = 2"),
+            self.destination + "STATUS = 1\n",
+        ):
+            with self.subTest(destination=destination), self.assertRaises(ValueError):
+                copy_support_assignments(
+                    self.source, destination, self.names, self.exports
+                )
+
+    def test_invalid_exports_are_rejected(self):
+        for source, destination, exports in (
+            (self.source, self.destination, self.exports + self.exports),
+            (self.source, self.destination, ["unselected"]),
+            (self.source, self.destination, list(reversed(self.exports))),
+            (self.source, self.destination, self.exports[:1]),
+            (
+                self.source,
+                self.destination.replace('"existing"', '"requires_available"'),
+                self.exports,
+            ),
+            (self.source.replace(self.export_line, ""), self.destination, self.exports),
+            (
+                self.source,
+                self.destination.replace('[\n    "existing",\n]', '("existing",)'),
+                self.exports,
+            ),
+            (self.source.replace('"existing"', "3"), self.destination, self.exports),
+            (self.source, self.destination.replace('"existing"', "3"), self.exports),
+        ):
+            with self.subTest(exports=exports), self.assertRaises(ValueError):
+                copy_support_assignments(source, destination, self.names, exports)
+
+    def test_dependency_rebinding_deletion_and_builtin_shadowing_are_rejected(self):
+        for binding in (
+            "os = replacement",
+            "del os",
+            "import other as os",
+            "hasattr = replacement",
+            "del hasattr",
+            "if condition:\n    unittest = replacement",
+        ):
+            for source_side in (False, True):
+                source, destination = self.source, self.destination
+                if source_side:
+                    source = source.replace("__all__ = [", binding + "\n__all__ = [")
+                else:
+                    destination = destination.replace(
+                        "__all__ = [", binding + "\n__all__ = ["
+                    )
+                with (
+                    self.subTest(binding=binding, source_side=source_side),
+                    self.assertRaisesRegex(ValueError, "Ambiguous dependency binding"),
+                ):
+                    copy_support_assignments(
+                        source, destination, self.names, self.exports
+                    )
+
+    def test_decorated_and_partial_line_anchors_are_rejected(self):
+        for replacement in (
+            "@decorator\ndef anchor(): pass",
+            "STATUS = 1; other = 2",
+            "other = 2; STATUS = 1",
+        ):
+            with self.subTest(replacement=replacement), self.assertRaises(ValueError):
+                copy_support_assignments(
+                    self.source.replace("STATUS = 1", replacement),
+                    self.destination.replace("STATUS = 1", replacement),
+                    self.names,
+                    self.exports,
+                )
+
+    def test_wildcard_import_dependencies_are_rejected(self):
+        for source, destination in (
+            ("from other import *\n" + self.source, self.destination),
+            (self.source, "from other import *\n" + self.destination),
+        ):
+            with self.assertRaisesRegex(ValueError, "Wildcard import"):
+                copy_support_assignments(source, destination, self.names, self.exports)
+
+    def test_cli_hash_pins_scope_and_receipt(self):
+        with tempfile.TemporaryDirectory() as temp, contextlib.chdir(temp):
+            source = pathlib.Path("canonical/Lib/test/support/__init__.py")
+            target = pathlib.Path("Lib/test/support/__init__.py")
+            source.parent.mkdir(parents=True)
+            target.parent.mkdir(parents=True)
+            source.write_text(self.source)
+            target.write_text(self.destination)
+            args = [
+                str(source),
+                "--support-assignments",
+                *self.names,
+                "--support-exports",
+                *self.exports,
+                "--source-sha256",
+                hashlib.sha256(source.read_bytes()).hexdigest(),
+                "--destination-sha256",
+                "0" * 64,
+            ]
+            with contextlib.redirect_stderr(io.StringIO()):
+                self.assertEqual(main(args), 1)
+                self.assertEqual(
+                    main([str(source), "--support-assignments", *self.names]), 1
+                )
+            self.assertEqual(target.read_text(), self.destination)
+            args[-1] = hashlib.sha256(target.read_bytes()).hexdigest()
+            with contextlib.redirect_stdout(io.StringIO()) as output:
+                self.assertEqual(main(args), 0)
+            self.assertIn('"destination_after_sha256"', output.getvalue())
+            self.assertIn(self.block, target.read_text())
+            for path in (
+                "canonical/Lib/os.py",
+                "canonical/Lib/test/support/../../os.py",
+            ):
+                with contextlib.redirect_stderr(io.StringIO()):
+                    self.assertEqual(main([path, *args[1:]]), 1)
+
+    def test_selective_mode_rejects_existing_modes_before_reading_metadata(self):
+        for mode in ("--patches", "--legacy-map"):
+            with contextlib.redirect_stderr(io.StringIO()) as errors:
+                self.assertEqual(
+                    main(
+                        [
+                            "missing/Lib/test/support/example.py",
+                            "--support-assignments",
+                            *self.names,
+                            mode,
+                            "missing-metadata.json",
+                        ]
+                    ),
+                    1,
+                )
+            self.assertIn("cannot combine", errors.getvalue())
+            with contextlib.redirect_stderr(io.StringIO()) as errors:
+                self.assertEqual(
+                    main(
+                        [
+                            "missing/Lib/test/support/example.py",
+                            "--support-exports",
+                            *self.exports,
+                            mode,
+                            "missing-metadata.json",
+                        ]
+                    ),
+                    1,
+                )
+            self.assertIn("require --support-assignments", errors.getvalue())
+
+    def test_default_syntax_error_behavior_is_unchanged(self):
+        with patch(
+            "update_lib.cmd_migrate.patch_file", side_effect=SyntaxError("invalid")
+        ):
+            with self.assertRaises(SyntaxError):
+                main(["missing/Lib/test/support/example.py"])
+        with patch(
+            "update_lib.cmd_migrate.migrate_support_assignments",
+            side_effect=SyntaxError("invalid"),
+        ):
+            with contextlib.redirect_stderr(io.StringIO()) as errors:
+                self.assertEqual(
+                    main(
+                        [
+                            "missing/Lib/test/support/example.py",
+                            "--support-assignments",
+                            *self.names,
+                        ]
+                    ),
+                    1,
+                )
+            self.assertIn("invalid", errors.getvalue())
+
+
 class TestPatchSingleContent(unittest.TestCase):
     """Tests for patch_single_content function."""