diff --git a/‎git/util.py‎
Lines changed: 47 additions & 6 deletions b/‎git/util.py‎
Lines changed: 47 additions & 6 deletions
diff --git a/‎test/test_util.py‎
Lines changed: 68 additions & 2 deletions b/‎test/test_util.py‎
Lines changed: 68 additions & 2 deletions
@@ -1166,15 +1166,56 @@ def _obtain_lock_or_raise(self) -> None:
         if self._has_lock():
             return
         lock_file = self._lock_file_path()
-        if osp.isfile(lock_file):
+        # Create the lock in one step. Checking for it first would allow another
+        # holder to create it between the check and the open.
+        try:
+            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)
-            )
-
-        try:
-            with open(lock_file, mode="w"):
-                pass
+            ) from e
         except OSError as e:
             raise OSError(str(e)) from e
 
 
@@ -12,6 +12,7 @@
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 from unittest import SkipTest, mock
 
@@ -418,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.
@@ -429,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)
@@ -437,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)
@@ -446,6 +451,67 @@ 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, 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()
+            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:
+            my_file = os.path.join(tdir, "my-lock-file")
+            racers = 8
+            at_the_line = threading.Barrier(racers)
+            holders = []
+            guard = threading.Lock()
+
+            def obtain():
+                lock_file = LockFile(my_file)
+                at_the_line.wait()
+                try:
+                    lock_file._obtain_lock_or_raise()
+                except OSError:
+                    return
+                with guard:
+                    holders.append(lock_file)
+
+            threads = [threading.Thread(target=obtain) for _ in range(racers)]
+            for thread in threads:
+                thread.start()
+            for thread in threads:
+                thread.join()
+
+            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:
             my_file = os.path.join(tdir, "my-lock-file")