File size: 6,356 Bytes
8ddc930 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | from __future__ import annotations
import os
import secrets
import stat
import sys
from errno import EACCES, EIO, EISDIR
from pathlib import Path
from typing import Final
def write_all(fd: int, data: bytes) -> None:
"""
Write the whole buffer to *fd*, looping over the short writes ``os.write`` is allowed to make.
A marker written with a bare ``os.write`` can land partially: a peer reading it mid-write parses a truncated record
as malformed or as a foreign holder. Looping until the buffer drains keeps the record atomic in the process and
kernel view. No ``fsync``: filelock needs a complete record, not crash-durable storage.
:param fd: file descriptor open for writing.
:param data: bytes to write in full.
:raises OSError: if a write reports zero progress before the record is complete.
"""
remaining = memoryview(data)
while remaining:
if (written := os.write(fd, remaining)) == 0:
raise OSError(EIO, "os.write wrote 0 bytes before the record was complete")
remaining = remaining[written:]
def raise_on_not_writable_file(filename: str) -> None:
"""
Raise an exception if attempting to open the file for writing would fail.
Separates files that can never be written from files that are writable but currently locked.
:param filename: file to check
:raises OSError: as if the file was opened for writing.
"""
try:
# lstat, not stat: settles exists-and-writable in one syscall, and a hostile symlink at the lock path would
# make stat inspect the link target, letting an attacker turn a contended acquire into a misleading
# PermissionError / IsADirectoryError and probe that target's attributes. The real open passes O_NOFOLLOW and
# refuses the symlink anyway.
file_stat = os.lstat(filename)
except OSError:
return # does not exist, or an error the caller cannot act on
# No mtime guard: the old `if st_mtime != 0` skip covered NFS/Linux quirks where os.lstat returned an all-zero
# struct, which it no longer does. Skipping on mtime 0 let a read-only file or a directory at the lock path pass
# as missing, so acquire() blocked forever on an open that cannot succeed.
if not (file_stat.st_mode & stat.S_IWUSR):
raise PermissionError(EACCES, "Permission denied", filename)
if stat.S_ISDIR(file_stat.st_mode):
if sys.platform == "win32": # pragma: win32 cover
raise PermissionError(EACCES, "Permission denied", filename)
raise IsADirectoryError(EISDIR, "Is a directory", filename) # pragma: win32 no cover
def ensure_directory_exists(filename: Path | str) -> None:
"""
Ensure the directory containing the file exists (create it if necessary).
:param filename: file.
"""
Path(filename).parent.mkdir(parents=True, exist_ok=True)
def break_lock_file(lock_file: str, mtime_before: float, ino_before: int) -> None:
"""
Atomically break a stale lock file judged stale at modification time *mtime_before*.
Rename the file to a process-private name before unlinking it, so two processes breaking the same lock cannot
delete each other's work: only one rename of a given inode wins, the loser gets ``OSError``. After the rename,
re-check the file. A newer modification time, or a different inode than *ino_before*, means a peer recreated the
lock between the stale decision and the rename, so we grabbed a live file and abort, leaving the renamed file in
place. A rollback rename is itself racy, the same trade-off as the soft read/write marker break. The inode check
matters because filesystems with coarse modification-time granularity (NFS, FAT) can give a same-second recreation
the old mtime, so mtime alone would miss it and unlink a live lock; the inode is the reliable identity, mirroring
the token re-check in the soft read/write marker break. ``lstat`` avoids following a hostile symlink swapped in
after the decision.
The break name carries a random token so it is unguessable and unique per attempt. Without it two breakers in the
same process share ``<lock>.break.<pid>``, and a second break can rename a recreated live lock onto that path in
the window between the re-verify ``lstat`` above and the ``unlink`` below, deleting a live lock the inode check
just approved. A private name keeps anyone else from targeting our break path, matching the soft read/write marker
break.
:param lock_file: path to the lock file to break.
:param mtime_before: modification time observed when the lock was judged stale.
:param ino_before: inode number observed when the lock was judged stale.
:raises OSError: if the rename fails (e.g. the file vanished or is not owned in a sticky directory).
"""
break_path = f"{lock_file}.break.{os.getpid()}.{secrets.token_hex(16)}"
Path(lock_file).rename(break_path)
try:
st_after = os.lstat(break_path)
except OSError:
return
if st_after.st_mtime > mtime_before or st_after.st_ino != ino_before:
return
Path(break_path).unlink()
def touch(name: str, *, fd: int | None = None) -> None:
# Prefer the already-open, already-verified fd so a peer that swaps a symlink or a different file in at the
# path after our O_NOFOLLOW read cannot redirect the touch: utime then targets the inode behind the fd.
# Where the platform cannot utime an fd, fall back to a path-based touch that still refuses to follow a
# symlink where supported, matching the O_NOFOLLOW reads used elsewhere here.
if fd is not None and _SUPPORTS_UTIME_FD: # pragma: needs utime-fd
os.utime(fd, None)
return
os.utime(name, None, follow_symlinks=not _SUPPORTS_UTIME_NOFOLLOW)
# Retargeting os.utime to an open fd lets a heartbeat refresh the exact inode it verified instead of whatever the
# pathname now names.
_SUPPORTS_UTIME_FD: Final[bool] = sys.platform != "win32" and os.utime in os.supports_fd
# os.utime follows symlinks unless told not to; not every platform can refuse the follow, so probe support.
_SUPPORTS_UTIME_NOFOLLOW: Final[bool] = os.utime in os.supports_follow_symlinks
__all__ = [
"break_lock_file",
"ensure_directory_exists",
"raise_on_not_writable_file",
"touch",
"write_all",
]
|