bpo-40138: Fix Windows os.waitpid() for large exit code

Lib/test/test_os.py

@@ -2789,40 +2789,66 @@ def test_getppid(self):
         # We are the parent of our subprocess
         self.assertEqual(int(stdout), os.getpid())
 
+    def check_waitpid(self, code, exitcode, callback=None):
+        if sys.platform == 'win32':
+            # On Windows, os.spawnv() simply joins arguments with spaces:
+            # arguments need to be quoted
+            args = [f'"{sys.executable}"', '-c', f'"{code}"']
+        else:
+            args = [sys.executable, '-c', code]
+        pid = os.spawnv(os.P_NOWAIT, sys.executable, args)
+
+        if callback is not None:
+            callback(pid)
+
+        # don't use support.wait_process() to test directly os.waitpid()
+        # and os.waitstatus_to_exitcode()
+        pid2, status = os.waitpid(pid, 0)
+        self.assertEqual(os.waitstatus_to_exitcode(status), exitcode)
+        self.assertEqual(pid2, pid)
+
     def test_waitpid(self):
-        args = [sys.executable, '-c', 'pass']
-        # Add an implicit test for PyUnicode_FSConverter().
-        pid = os.spawnv(os.P_NOWAIT, FakePath(args[0]), args)
-        support.wait_process(pid, exitcode=0)
+        self.check_waitpid(code='pass', exitcode=0)
 
     def test_waitstatus_to_exitcode(self):
         exitcode = 23
-        filename = support.TESTFN
-        self.addCleanup(support.unlink, filename)
+        code = f'import sys; sys.exit({exitcode})'
+        self.check_waitpid(code, exitcode=exitcode)
 
-        with open(filename, "w") as fp:
-            print(f'import sys; sys.exit({exitcode})', file=fp)
-            fp.flush()
-        args = [sys.executable, filename]
-        pid = os.spawnv(os.P_NOWAIT, args[0], args)
+        with self.assertRaises(TypeError):
+            os.waitstatus_to_exitcode(0.0)
 
-        pid2, status = os.waitpid(pid, 0)
-        self.assertEqual(os.waitstatus_to_exitcode(status), exitcode)
-        self.assertEqual(pid2, pid)
+    @unittest.skipUnless(sys.platform == 'win32', 'win32-specific test')
+    def test_waitpid_windows(self):
+        # bpo-40138: test os.waitpid() and os.waitstatus_to_exitcode()
+        # with exit code larger than INT_MAX.
+        STATUS_CONTROL_C_EXIT = 0xC000013A
+        code = f'import _winapi; _winapi.ExitProcess({STATUS_CONTROL_C_EXIT})'
+        self.check_waitpid(code, exitcode=STATUS_CONTROL_C_EXIT)
+
+    @unittest.skipUnless(sys.platform == 'win32', 'win32-specific test')
+    def test_waitstatus_to_exitcode_windows(self):
+        max_exitcode = 2 ** 32 - 1
+        for exitcode in (0, 1, 5, max_exitcode):
+            self.assertEqual(os.waitstatus_to_exitcode(exitcode << 8),
+                             exitcode)
+
+        # invalid values
+        with self.assertRaises(ValueError):
+            os.waitstatus_to_exitcode((max_exitcode + 1) << 8)
+        with self.assertRaises(OverflowError):
+            os.waitstatus_to_exitcode(-1)
 
     # Skip the test on Windows
     @unittest.skipUnless(hasattr(signal, 'SIGKILL'), 'need signal.SIGKILL')
     def test_waitstatus_to_exitcode_kill(self):
+        code = f'import time; time.sleep({support.LONG_TIMEOUT})'
         signum = signal.SIGKILL
-        args = [sys.executable, '-c',
-                f'import time; time.sleep({support.LONG_TIMEOUT})']
-        pid = os.spawnv(os.P_NOWAIT, args[0], args)
 
-        os.kill(pid, signum)
+        def kill_process(pid):
+            os.kill(pid, signum)
 
-        pid2, status = os.waitpid(pid, 0)
-        self.assertEqual(os.waitstatus_to_exitcode(status), -signum)
-        self.assertEqual(pid2, pid)
+        self.check_waitpid(code, exitcode=-signum, callback=kill_process)
 
 
 class SpawnTests(unittest.TestCase):
@@ -2884,6 +2910,10 @@ def test_spawnv(self):
         exitcode = os.spawnv(os.P_WAIT, args[0], args)
         self.assertEqual(exitcode, self.exitcode)
 
+        # Test for PyUnicode_FSConverter()
+        exitcode = os.spawnv(os.P_WAIT, FakePath(args[0]), args)
+        self.assertEqual(exitcode, self.exitcode)
+
     @requires_os_func('spawnve')
     def test_spawnve(self):
         args = self.create_args(with_env=True)

Misc/NEWS.d/next/Library/2020-04-22-00-05-10.bpo-40138.i_oGqa.rst

@@ -0,0 +1,2 @@
+Fix the Windows implementation of :func:`os.waitpid` for exit code larger than
+``INT_MAX >> 8``. The exit status is now interpreted as an unsigned number.

Modules/posixmodule.c

@@ -7972,8 +7972,10 @@ os_waitpid_impl(PyObject *module, intptr_t pid, int options)
     if (res < 0)
         return (!async_err) ? posix_error() : NULL;
 
+    unsigned long long ustatus = (unsigned int)status;
+
     /* shift the status left a byte so this is more like the POSIX waitpid */
-    return Py_BuildValue(_Py_PARSE_INTPTR "i", res, status << 8);
+    return Py_BuildValue(_Py_PARSE_INTPTR "K", res, ustatus << 8);
 }
 #endif
 
@@ -13829,7 +13831,7 @@ os__remove_dll_directory_impl(PyObject *module, PyObject *cookie)
 /*[clinic input]
 os.waitstatus_to_exitcode
 
-    status: int
+    status as status_obj: object
 
 Convert a wait status to an exit code.
 
@@ -13847,10 +13849,20 @@ This function must not be called if WIFSTOPPED(status) is true.
 [clinic start generated code]*/
 
 static PyObject *
-os_waitstatus_to_exitcode_impl(PyObject *module, int status)
-/*[clinic end generated code: output=c7c2265731f79b7a input=edfa5ca5006276fb]*/
+os_waitstatus_to_exitcode_impl(PyObject *module, PyObject *status_obj)
+/*[clinic end generated code: output=db50b1b0ba3c7153 input=7fe2d7fdaea3db42]*/
 {
+    if (PyFloat_Check(status_obj)) {
+        PyErr_SetString(PyExc_TypeError,
+                        "integer argument expected, got float" );
+        return NULL;
+    }
 #ifndef MS_WINDOWS
+    int status = _PyLong_AsInt(status_obj);
+    if (status == -1 && PyErr_Occurred()) {
+        return NULL;
+    }
+
     WAIT_TYPE wait_status;
     WAIT_STATUS_INT(wait_status) = status;
     int exitcode;
@@ -13889,8 +13901,19 @@ os_waitstatus_to_exitcode_impl(PyObject *module, int status)
 #else
     /* Windows implementation: see os.waitpid() implementation
        which uses _cwait(). */
-    int exitcode = (status >> 8);
-    return PyLong_FromLong(exitcode);
+    unsigned long long status = PyLong_AsUnsignedLongLong(status_obj);
+    if (status == (unsigned long long)-1 && PyErr_Occurred()) {
+        return NULL;
+    }
+
+    unsigned long long exitcode = (status >> 8);
+    /* ExitProcess() accepts an UINT type:
+       reject exit code which doesn't fit in an UINT */
+    if (exitcode > UINT_MAX) {
+        PyErr_Format(PyExc_ValueError, "invalid exit code: %llu", exitcode);
+        return NULL;
+    }
+    return PyLong_FromUnsignedLong((unsigned long)exitcode);
 #endif
 }
 #endif