Vulnerable code sample

import abc
import errno
import os
import stat
import typing as t
from pathlib import Path

# This code represents a minimal, self-contained version of the filelock
# library's SoftFileLock class prior to version 3.20.3, demonstrating
# the structure that leads to the TOCTOU vulnerability.

_AcquireRetType = t.Union["BaseFileLock", t.Literal[False]]
_P = t.TypeVar("_P", bound="BaseFileLock")


class LockError(Exception):
    """Base class for all lock-related errors."""


class BaseFileLock(abc.ABC):
    """
    Abstract base class for all file lock implementations.
    """

    def __init__(self, lock_file: t.Union[str, "os.PathLike[str]"], timeout: float = -1) -> None:
        self._lock_file: str = os.fspath(lock_file)
        self.timeout: float = timeout
        self.is_locked: bool = False
        self._lock_file_fd: t.Optional[int] = None

    @abc.abstractmethod
    def _acquire(self) -> None:
        raise NotImplementedError

    @abc.abstractmethod
    def _release(self) -> None:
        raise NotImplementedError

    def acquire(self, timeout: t.Optional[float] = None, poll_interval: float = 0.05) -> _AcquireRetType:
        # Simplified acquire logic for demonstration purposes.
        if timeout is None:
            timeout = self.timeout

        try:
            self._acquire()
        except LockError:
            return False

        self.is_locked = True
        return self

    def release(self) -> None:
        if self.is_locked:
            self._release()
            self.is_locked = False

    def __enter__(self: _P) -> _P:
        self.acquire()
        return self

    def __exit__(self, exc_type: t.Any, exc_value: t.Any, traceback: t.Any) -> None:
        self.release()

    @property
    def lock_file(self) -> str:
        return self._lock_file


class SoftFileLock(BaseFileLock):
    """
    A platform-independent file lock that uses a file to indicate the lock.
    This is the vulnerable implementation.
    """

    def __init__(self, lock_file: t.Union[str, "os.PathLike[str]"], timeout: float = -1) -> None:
        super().__init__(lock_file, timeout)
        self._lock_counter: int = 0

    def _acquire(self) -> None:
        if self._lock_counter > 0:
            self._lock_counter += 1
            return

        # VULNERABILITY: A race condition (TOCTOU) exists between the check
        # for writability and the actual file creation.

        # TIME OF CHECK (TOC): Check if the directory is writable and the
        # path is not a directory.
        self.raise_on_not_writable_file()

        # RACE WINDOW: An attacker can replace `self.lock_file` with a
        # symlink to a different file after the check above but before
        # the os.open() call below.

        try:
            # TIME OF USE (TOU): Attempt to open/create the lock file.
            # If a symlink was created in the race window, this operation
            # will act on the symlink's target.
            mode = stat.S_IWUSR | stat.S_IRUSR
            fd = os.open(self.lock_file, os.O_WRONLY | os.O_CREAT, mode)
        except OSError as exc:
            if exc.errno == errno.EISDIR:
                raise LockError(f"Lock file '{self.lock_file}' is a directory.") from exc
            # Other OSErrors could be caused by the attacker's symlink,
            # for example, pointing to a file they don't have permission to write.
            raise LockError(f"Failed to create lock file '{self.lock_file}'.") from exc
        else:
            self._lock_file_fd = fd
            self._lock_counter = 1

    def _release(self) -> None:
        if self._lock_counter > 1:
            self._lock_counter -= 1
            return

        if self._lock_file_fd is not None:
            os.close(self._lock_file_fd)
            self._lock_file_fd = None

        try:
            os.remove(self.lock_file)
        except OSError:
            # The file might have been deleted by another process.
            pass
        finally:
            self._lock_counter = 0

    def raise_on_not_writable_file(self) -> None:
        """
        Check if the lock file can be created. This method is part of the
        vulnerable TOCTOU sequence.
        """
        lock_path = Path(self.lock_file)
        if lock_path.is_dir():
            raise LockError(f"The lock file path '{lock_path}' is a directory.")

        # Check if the parent directory is writable.
        parent_dir = lock_path.parent
        if not parent_dir.is_dir():
            raise LockError(f"The parent directory '{parent_dir}' of the lock file does not exist.")
        if not os.access(parent_dir, os.W_OK):
            raise LockError(f"The parent directory '{parent_dir}' is not writable.")