gh-104690 Disallow thread creation and fork at interpreter finalization (#104826)

Disallow thread creation and fork at interpreter finalization.

in the following functions, check if interpreter is finalizing and raise `RuntimeError` with appropriate message:
* `_thread.start_new_thread` and thus `threading`
* `posix.fork`
* `posix.fork1`
* `posix.forkpty`
* `_posixsubprocess.fork_exec` when a `preexec_fn=` is supplied.

---------

Co-authored-by: blurb-it[bot] <43283697+blurb-it[bot]@users.noreply.github.com>
Co-authored-by: Gregory P. Smith <greg@krypto.org>

3 files changed, 51 insertions, 28 deletions

diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py
index c6810c0..8de4ef7 100644
--- a/Lib/test/test_os.py
+++ b/Lib/test/test_os.py
@@ -4706,6 +4706,22 @@ class ForkTests(unittest.TestCase):
         self.assertEqual(err.decode("utf-8"), "")
         self.assertEqual(out.decode("utf-8"), "")
 
+    def test_fork_at_exit(self):
+        code = """if 1:
+            import atexit
+            import os
+
+            def exit_handler():
+                pid = os.fork()
+                if pid != 0:
+                    print("shouldn't be printed")
+
+            atexit.register(exit_handler)
+        """
+        _, out, err = assert_python_ok("-c", code)
+        self.assertEqual(b"", out)
+        self.assertIn(b"can't fork at interpreter shutdown", err)
+
 
 # Only test if the C version is provided, otherwise TestPEP519 already tested
 # the pure Python implementation.
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index 92f81ea..51ba423 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -5,6 +5,7 @@ from test.support import check_sanitizer
 from test.support import import_helper
 from test.support import os_helper
 from test.support import warnings_helper
+from test.support.script_helper import assert_python_ok
 import subprocess
 import sys
 import signal
@@ -3329,6 +3330,24 @@ class POSIXProcessTestCase(BaseTestCase):
             except subprocess.TimeoutExpired:
                 pass
 
+    def test_preexec_at_exit(self):
+        code = f"""if 1:
+        import atexit
+        import subprocess
+
+        def dummy():
+            pass
+
+        def exit_handler():
+            subprocess.Popen({ZERO_RETURN_CMD}, preexec_fn=dummy)
+            print("shouldn't be printed")
+
+        atexit.register(exit_handler)
+        """
+        _, out, err = assert_python_ok("-c", code)
+        self.assertEqual(out, b'')
+        self.assertIn(b"preexec_fn not supported at interpreter shutdown", err)
+
 
 @unittest.skipUnless(mswindows, "Windows specific tests")
 class Win32ProcessTestCase(BaseTestCase):
diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py
index 9716526..9e4972e 100644
--- a/Lib/test/test_threading.py
+++ b/Lib/test/test_threading.py
@@ -532,34 +532,6 @@ class ThreadTests(BaseTestCase):
         self.assertTrue(t.daemon)
 
     @support.requires_fork()
-    def test_fork_at_exit(self):
-        # bpo-42350: Calling os.fork() after threading._shutdown() must
-        # not log an error.
-        code = textwrap.dedent("""
-            import atexit
-            import os
-            import sys
-            from test.support import wait_process
-
-            # Import the threading module to register its "at fork" callback
-            import threading
-
-            def exit_handler():
-                pid = os.fork()
-                if not pid:
-                    print("child process ok", file=sys.stderr, flush=True)
-                    # child process
-                else:
-                    wait_process(pid, exitcode=0)
-
-            # exit_handler() will be called after threading._shutdown()
-            atexit.register(exit_handler)
-        """)
-        _, out, err = assert_python_ok("-c", code)
-        self.assertEqual(out, b'')
-        self.assertEqual(err.rstrip(), b'child process ok')
-
-    @support.requires_fork()
     def test_dummy_thread_after_fork(self):
         # Issue #14308: a dummy thread in the active list doesn't mess up
         # the after-fork mechanism.
@@ -1048,6 +1020,22 @@ class ThreadTests(BaseTestCase):
         self.assertEqual(out, b'')
         self.assertEqual(err, b'')
 
+    def test_start_new_thread_at_exit(self):
+        code = """if 1:
+            import atexit
+            import _thread
+
+            def f():
+                print("shouldn't be printed")
+
+            def exit_handler():
+                _thread.start_new_thread(f, ())
+
+            atexit.register(exit_handler)
+        """
+        _, out, err = assert_python_ok("-c", code)
+        self.assertEqual(out, b'')
+        self.assertIn(b"can't create new thread at interpreter shutdown", err)
 
 class ThreadJoinOnShutdown(BaseTestCase):