Windows: grammar-cache creation can hang when cache writes are denied

third_party/yapf_third_party/_ylib2to3/pgen2/grammar.py

@@ -14,11 +14,26 @@
 # Python imports
 import os
 import pickle
-import tempfile
+import uuid
 
 # Local imports
 from . import token
 
+# Flags for opening the temporary cache file written by ``Grammar.dump``.
+# We avoid ``tempfile``/``mkstemp`` here: on Windows, Python's tempfile
+# module treats a ``PermissionError`` from ``os.open`` as a possible name
+# collision and retries up to ``tempfile.TMP_MAX`` (over two billion) times
+# whenever the destination directory exists and ``os.access`` reports it
+# writable. That can make this optional cache write hang for a very long
+# time instead of failing, e.g. in a restricted-token sandbox where the
+# directory exists but file creation is denied. Doing a single, explicit
+# ``os.open`` attempt lets a denied write fail immediately.
+_TEMPFILE_OPEN_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+if hasattr(os, 'O_BINARY'):
+  _TEMPFILE_OPEN_FLAGS |= os.O_BINARY
+if hasattr(os, 'O_NOINHERIT'):
+  _TEMPFILE_OPEN_FLAGS |= os.O_NOINHERIT
+
 
 class Grammar(object):
   """Pgen parsing tables conversion class.
@@ -102,25 +117,27 @@ def dump(self, filename):
     #   ever have to leave a tempfile around for failure of deletion,
     #   it will have a reasonable filename extension and its name will help
     #   explain is nature.
-    tempfile_dir = os.path.dirname(filename)
-    tempfile_prefix, tempfile_suffix = os.path.splitext(filename)
-    with tempfile.NamedTemporaryFile(
-        mode='wb',
-        suffix=tempfile_suffix,
-        prefix=tempfile_prefix,
-        dir=tempfile_dir,
-        delete=False) as f:
-      pickle.dump(self.__dict__, f.file, pickle.HIGHEST_PROTOCOL)
+    # - We close the tempfile before calling ``os.rename``, since a rename
+    #   of a still-open file can fail on Windows.
+    tempfile_dir = os.path.dirname(filename) or '.'
+    tempfile_prefix, tempfile_suffix = os.path.splitext(
+        os.path.basename(filename))
+    temp_filename = os.path.join(
+        tempfile_dir,
+        '{}.{}{}'.format(tempfile_prefix, uuid.uuid4().hex, tempfile_suffix))
+    try:
+      fd = os.open(temp_filename, _TEMPFILE_OPEN_FLAGS, 0o600)
+      with os.fdopen(fd, 'wb') as f:
+        pickle.dump(self.__dict__, f, pickle.HIGHEST_PROTOCOL)
+      os.rename(temp_filename, filename)
+    except OSError:
+      # This makes sure that we do not leave the tempfile around
+      # unless we have to...
       try:
-        os.rename(f.name, filename)
+        os.remove(temp_filename)
       except OSError:
-        # This makes sure that we do not leave the tempfile around
-        # unless we have to...
-        try:
-          os.remove(f.name)
-        except OSError:
-          pass
-        raise
+        pass
+      raise
 
   def load(self, filename):
     """Load the grammar tables from a pickle file."""

yapftests/pgen2_grammar_test.py

@@ -0,0 +1,65 @@
+# Copyright 2026 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for yapf_third_party._ylib2to3.pgen2.grammar."""
+
+import os
+import shutil
+import stat
+import tempfile
+import unittest
+
+from yapf_third_party._ylib2to3.pgen2 import grammar
+
+from yapftests import yapf_test_helper
+
+
+class GrammarDumpTest(yapf_test_helper.YAPFTest):
+
+  def setUp(self):
+    self.test_tmpdir = tempfile.mkdtemp()
+
+  def tearDown(self):
+    shutil.rmtree(self.test_tmpdir, ignore_errors=True)
+
+  def testDumpAndLoadRoundTrip(self):
+    g = grammar.Grammar()
+    target = os.path.join(self.test_tmpdir, 'test.pickle')
+    g.dump(target)
+    self.assertTrue(os.path.exists(target))
+
+    loaded = grammar.Grammar()
+    loaded.load(target)
+    self.assertEqual(g.__dict__, loaded.__dict__)
+
+  @unittest.skipIf(
+      os.name == 'nt' or os.geteuid() == 0,
+      'permission bits are not enforced for root or on Windows')
+  def testDumpFailsImmediatelyWhenDirectoryIsUnwritable(self):
+    # Regression test for #1311: a denied cache write must raise promptly
+    # instead of retrying (potentially for a very long time, as observed on
+    # Windows with the stdlib tempfile module).
+    unwritable_dir = os.path.join(self.test_tmpdir, 'unwritable')
+    os.mkdir(unwritable_dir)
+    os.chmod(unwritable_dir, stat.S_IRUSR | stat.S_IXUSR)
+
+    g = grammar.Grammar()
+    target = os.path.join(unwritable_dir, 'test.pickle')
+    with self.assertRaises(OSError):
+      g.dump(target)
+
+    self.assertEqual([], os.listdir(unwritable_dir))
+
+
+if __name__ == '__main__':
+  unittest.main()