gh-120932: Check the pyc file mtime at import

Lib/importlib/_bootstrap_external.py

@@ -26,6 +26,7 @@
 import _imp
 import _io
 import sys
+import time
 import _warnings
 import marshal
 
@@ -458,8 +459,8 @@ def _classify_pyc(data, name, exc_details):
     return flags
 
 
-def _validate_timestamp_pyc(data, source_mtime, source_size, name,
-                            exc_details):
+def _validate_timestamp_pyc(self, data, source_mtime, source_size, name,
+                            bytecode_path, exc_details):
     """Validate a pyc against the source last-modified time.
 
     *data* is the contents of the pyc file. (Only the first 16 bytes are
@@ -471,19 +472,32 @@ def _validate_timestamp_pyc(data, source_mtime, source_size, name,
 
     *name* is the name of the module being imported. It is used for logging.
 
+    *bytecode_path* is the path of the pyc file.
+
     *exc_details* is a dictionary passed to ImportError if it raised for
     improved debugging.
 
     An ImportError is raised if the bytecode is stale.
 
     """
-    if _unpack_uint32(data[8:12]) != (source_mtime & 0xFFFFFFFF):
+    timestamp = _unpack_uint32(data[8:12])
+    if timestamp != (int(source_mtime) & 0xFFFFFFFF):
         message = f'bytecode is stale for {name!r}'
         _bootstrap._verbose_message('{}', message)
         raise ImportError(message, **exc_details)
     if (source_size is not None and
         _unpack_uint32(data[12:16]) != (source_size & 0xFFFFFFFF)):
         raise ImportError(f'bytecode is stale for {name!r}', **exc_details)
+    if time.time() - source_mtime < 2:
+        try:
+            bytecode_mtime = self.path_stats(bytecode_path)['mtime']
+        except OSError:
+            pass
+        else:
+            if bytecode_mtime < source_mtime:
+                message = f'bytecode is stale for {name!r}'
+                _bootstrap._verbose_message('{}', message)
+                raise ImportError(message, **exc_details)
 
 
 def _validate_hash_pyc(data, source_hash, name, exc_details):
@@ -527,7 +541,7 @@ def _code_to_timestamp_pyc(code, mtime=0, source_size=0):
     "Produce the data for a timestamp-based pyc."
     data = bytearray(MAGIC_NUMBER)
     data.extend(_pack_uint32(0))
-    data.extend(_pack_uint32(mtime))
+    data.extend(_pack_uint32(int(mtime)))
     data.extend(_pack_uint32(source_size))
     data.extend(marshal.dumps(code))
     return data
@@ -850,7 +864,7 @@ def get_code(self, fullname):
             except OSError:
                 pass
             else:
-                source_mtime = int(st['mtime'])
+                source_mtime = st['mtime']
                 try:
                     data = self.get_data(bytecode_path)
                 except OSError:
@@ -878,10 +892,12 @@ def get_code(self, fullname):
                                                    exc_details)
                         else:
                             _validate_timestamp_pyc(
+                                self,
                                 data,
                                 source_mtime,
                                 st['size'],
                                 fullname,
+                                bytecode_path,
                                 exc_details,
                             )
                     except (ImportError, EOFError):

Lib/test/test_import/__init__.py

@@ -1761,6 +1761,34 @@ def test_recompute_pyc_same_second(self):
         m = __import__(TESTFN)
         self.assertEqual(m.x, 5)
 
+    def test_recompute_pyc_same_second_same_size(self):
+        # Even when the source file doesn't change timestamp (truncated
+        # to seconds) and size, the difference between the source and
+        # the bytecode timestamps is enough to trigger recomputation of
+        # the pyc file.
+        pyc_file = importlib.util.cache_from_source(self.source)
+        for i in range(10, 100):
+            try:
+                bytecode_mtime = os.stat(pyc_file).st_mtime
+            except FileNotFoundError:
+                # The pyc file has not yet be created.
+                bytecode_mtime = 0
+            delay = 1e-3
+            for j in range(10):
+                time.sleep(delay)
+                with open(self.source, 'w', encoding='utf-8') as fp:
+                    print(f"x = {i}", file=fp)
+                if os.stat(self.source).st_mtime > bytecode_mtime:
+                    break
+                delay *= 2
+
+            m = __import__(TESTFN)
+            self.assertEqual(m.x, i)
+            unload(TESTFN)
+            m = __import__(TESTFN)
+            self.assertEqual(m.x, i)
+            unload(TESTFN)
+
 
 class TestSymbolicallyLinkedPackage(unittest.TestCase):
     package_name = 'sample'

Misc/NEWS.d/next/Core_and_Builtins/2024-07-12-19-57-12.gh-issue-121620.AgNaIS.rst

@@ -0,0 +1,2 @@
+Import checks now also the modification time of the ``.pyc`` file for
+recently modified source file. This reduces chance of using stale bytecode.