fix(util): acquire Windows locks without following symlinks

Windows follows dangling symlinks even when `os.open` uses `O_CREAT | O_EXCL`,
so acquiring a lock could create its symlink target and incorrectly report
ownership. The `_winapi.CreateFile` workaround also uses the ANSI API on
Python 3.8 through 3.10, causing failures in Unicode directories or creating
locks under mangled filenames.

Use `CreateFileW` with `CREATE_NEW` and `FILE_FLAG_OPEN_REPARSE_POINT` to
create the lock atomically while rejecting existing links. Declare the
`ctypes` argument and return types explicitly so Unicode paths and native
handle sizes are preserved. Close the handle before recording ownership,
propagate Windows errors, and reject embedded NULs before the native API
can truncate a path. Preserve exclusive `os.open` creation on POSIX.

Expand the lock tests to cover Unicode filenames and directories, including
non-BMP characters, and verify that the requested lock path is actually
created and removed. Check NUL rejection, preserve both existing and missing
symlink targets, and explicitly release the concurrent test's acquired locks.

Reproduced the CI failure in `test_clone_from_with_path_contains_unicode`
on Windows/Python 3.8.10 before the fix. The affected utility, clone, and
configuration modules pass on Python 3.8.10 and 3.13.14: 169 passed,
41 skipped, and 2 expected failures on each version. The new Unicode and
NUL regressions also failed before their respective fixes. `ruff check`,
`ruff format --check`, `mypy --python-version=3.13`, and
`basedpyright --warnings` pass.

Assisted-by: GPT 6.0 Astra
Co-authored-by: GPT 6 <codex@openai.com>

Author: Byron

git/util.py

@@ -1166,20 +1166,58 @@ def _obtain_lock_or_raise(self) -> None:
         if self._has_lock():
             return
         lock_file = self._lock_file_path()
-        # Create the lock in one step, the way Git and gitdb's LockedFD do. Testing
-        # for the file first leaves a window in which another holder creates it and
-        # both proceed, and O_CREAT|O_EXCL additionally refuses to follow a symbolic
-        # link planted at the lock path instead of writing through it.
+        # Create the lock in one step. Checking for it first would allow another
+        # holder to create it between the check and the open.
         try:
-            fd = os.open(lock_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            if sys.platform == "win32":
+                if "\0" in lock_file:
+                    raise ValueError("embedded null character")
+
+                import ctypes
+                from ctypes import wintypes
+
+                # Unlike POSIX, Windows follows dangling symlinks even with O_EXCL.
+                # Open the reparse point itself so an existing link is rejected.
+                # Call the Unicode API directly: older _winapi.CreateFile wrappers
+                # use the ANSI API and can create a lock under the wrong filename.
+                kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+                create_file = kernel32.CreateFileW
+                create_file.argtypes = (
+                    wintypes.LPCWSTR,
+                    wintypes.DWORD,
+                    wintypes.DWORD,
+                    wintypes.LPVOID,
+                    wintypes.DWORD,
+                    wintypes.DWORD,
+                    wintypes.HANDLE,
+                )
+                create_file.restype = wintypes.HANDLE
+                close_handle = kernel32.CloseHandle
+                close_handle.argtypes = (wintypes.HANDLE,)
+                close_handle.restype = wintypes.BOOL
+                handle = create_file(
+                    lock_file,
+                    0x40000000,  # GENERIC_WRITE
+                    0,
+                    None,
+                    1,  # CREATE_NEW
+                    0x00200000,  # FILE_FLAG_OPEN_REPARSE_POINT
+                    None,
+                )
+                if handle == wintypes.HANDLE(-1).value:
+                    raise ctypes.WinError(ctypes.get_last_error())
+                if not close_handle(handle):
+                    raise ctypes.WinError(ctypes.get_last_error())
+            else:
+                fd = os.open(lock_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+                os.close(fd)
         except FileExistsError as e:
             raise OSError(
                 "Lock for file %r did already exist, delete %r in case the lock is illegal"
                 % (self._file_path, lock_file)
             ) from e
         except OSError as e:
             raise OSError(str(e)) from e
-        os.close(fd)
 
         self._owns_lock = True
 

test/test_util.py

@@ -419,9 +419,11 @@ def test_it_should_dashify(self):
         self.assertEqual("this-is-my-argument", dashify("this_is_my_argument"))
         self.assertEqual("foo", dashify("foo"))
 
-    def test_lock_file(self):
+    @ddt.data("my-lock-file", "my-lock-file-\u0394", "\u0394/my-lock-file", "\U0001f680/my-lock-file")
+    def test_lock_file(self, filename):
         with tempfile.TemporaryDirectory() as tdir:
-            my_file = os.path.join(tdir, "my-lock-file")
+            my_file = os.path.join(tdir, filename)
+            os.makedirs(os.path.dirname(my_file), exist_ok=True)
             lock_file = LockFile(my_file)
             assert not lock_file._has_lock()
             # Release lock we don't have - fine.
@@ -430,6 +432,7 @@ def test_lock_file(self):
             # Get lock.
             lock_file._obtain_lock_or_raise()
             assert lock_file._has_lock()
+            assert os.path.isfile(my_file + ".lock")
 
             # Concurrent access.
             other_lock_file = LockFile(my_file)
@@ -438,6 +441,7 @@ def test_lock_file(self):
 
             lock_file._release_lock()
             assert not lock_file._has_lock()
+            assert not os.path.exists(my_file + ".lock")
 
             other_lock_file._obtain_lock_or_raise()
             self.assertRaises(IOError, lock_file._obtain_lock_or_raise)
@@ -447,17 +451,36 @@ def test_lock_file(self):
             lock_file._obtain_lock_or_raise()
             lock_file._release_lock()
 
+    def test_lock_file_rejects_embedded_nul(self):
+        with tempfile.TemporaryDirectory() as tdir:
+            my_file = os.path.join(tdir, "my-lock-file")
+            lock_file = LockFile(my_file + "\0suffix")
+            self.assertRaises(ValueError, lock_file._obtain_lock_or_raise)
+            assert not lock_file._has_lock()
+            assert not os.path.exists(my_file)
+
+    @ddt.data(False, True)
     @requires_symlinks
-    def test_lock_file_does_not_follow_a_symlink(self):
+    def test_lock_file_does_not_follow_a_symlink(self, target_exists):
         with tempfile.TemporaryDirectory() as tdir:
             my_file = os.path.join(tdir, "my-lock-file")
             outside = os.path.join(tdir, "outside-the-lock")
+            content = b"Do not modify the symlink target."
+            if target_exists:
+                with open(outside, "wb") as stream:
+                    stream.write(content)
             os.symlink(outside, my_file + ".lock")
 
             lock_file = LockFile(my_file)
             self.assertRaises(IOError, lock_file._obtain_lock_or_raise)
             assert not lock_file._has_lock()
-            assert not os.path.exists(outside)
+            lock_file._release_lock()
+            assert os.path.islink(my_file + ".lock")
+            if target_exists:
+                with open(outside, "rb") as stream:
+                    self.assertEqual(stream.read(), content)
+            else:
+                assert not os.path.exists(outside)
 
     def test_lock_file_is_obtained_by_a_single_holder(self):
         with tempfile.TemporaryDirectory() as tdir:
@@ -483,7 +506,11 @@ def obtain():
             for thread in threads:
                 thread.join()
 
-            self.assertEqual(1, len(holders))
+            try:
+                self.assertEqual(1, len(holders))
+            finally:
+                for lock_file in holders:
+                    lock_file._release_lock()
 
     def test_blocking_lock_file(self):
         with tempfile.TemporaryDirectory() as tdir: