Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/file.py: 60%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# file.py -- Safe access to git files
2# Copyright (C) 2010 Google, Inc.
3#
4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
6# General Public License as published by the Free Software Foundation; version 2.0
7# or (at your option) any later version. You can redistribute it and/or
8# modify it under the terms of either of these two licenses.
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16# You should have received a copy of the licenses; if not, see
17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
19# License, Version 2.0.
20#
22"""Safe access to git files."""
24__all__ = [
25 "PERM_EVERYBODY",
26 "PERM_GROUP",
27 "FileLocked",
28 "GitFile",
29 "SharedPerm",
30 "adjust_shared_perm",
31 "calc_shared_perm",
32 "ensure_dir_exists",
33 "open_nofollow",
34 "open_nofollow_read",
35]
37import errno
38import os
39import stat
40import sys
41import warnings
42from collections.abc import Iterable, Iterator
43from dataclasses import dataclass
44from types import TracebackType
45from typing import IO, Any, BinaryIO, ClassVar, Literal, overload
47from ._typing import Buffer
49if sys.version_info >= (3, 11):
50 from typing import Self
51else:
52 from typing_extensions import Self
55@dataclass(frozen=True)
56class SharedPerm:
57 """A parsed core.sharedRepository setting.
59 Attributes:
60 tweak: Permission bits the setting asks for
61 replace: Whether those bits state the mode outright, as an explicit
62 octal setting does, rather than only loosening the existing mode
63 """
65 tweak: int
66 replace: bool = False
69# git's PERM_GROUP and PERM_EVERYBODY.
70PERM_GROUP = SharedPerm(tweak=0o660)
71PERM_EVERYBODY = SharedPerm(tweak=0o664)
74def calc_shared_perm(mode: int, perm: "SharedPerm") -> int:
75 """Apply a shared permission setting to an existing mode.
77 Mirrors git's calc_shared_perm(). The umask never appears here: it has
78 already been applied by the kernel when the file was created, and named
79 settings only ever loosen the mode that resulted.
81 Args:
82 mode: Current permission bits of the file or directory
83 perm: Shared permission setting to apply
85 Returns:
86 The adjusted permission bits
87 """
88 tweak = perm.tweak
89 if not mode & stat.S_IWUSR:
90 tweak &= ~0o222
91 if mode & stat.S_IXUSR:
92 # Copy read bits to execute bits
93 tweak |= (tweak & 0o444) >> 2
94 if perm.replace:
95 return (mode & ~0o777) | tweak
96 return mode | tweak
99def adjust_shared_perm(
100 path: str | bytes | os.PathLike[str] | os.PathLike[bytes],
101 perm: "SharedPerm | None",
102) -> None:
103 """Widen the permissions of a path per core.sharedRepository.
105 Mirrors git's adjust_shared_perm(). The mode is read back from the path,
106 so whatever the umask already removed at creation time stays removed for
107 named settings.
109 Args:
110 path: File or directory to adjust
111 perm: Shared permission setting, or None to leave the path alone
112 """
113 if perm is None:
114 return
116 st = os.stat(path)
117 old_mode = stat.S_IMODE(st.st_mode)
118 new_mode = calc_shared_perm(old_mode, perm)
120 if stat.S_ISDIR(st.st_mode):
121 # Copy read bits to execute bits
122 new_mode |= (new_mode & 0o444) >> 2
123 # g+s matters only if group membership grants extra access
124 if new_mode & 0o060:
125 new_mode |= stat.S_ISGID
127 if new_mode != old_mode:
128 os.chmod(path, new_mode)
131def ensure_dir_exists(
132 dirname: str | bytes | os.PathLike[str] | os.PathLike[bytes],
133) -> None:
134 """Ensure a directory exists, creating if necessary."""
135 try:
136 os.makedirs(dirname)
137 except FileExistsError:
138 pass
141def open_nofollow(
142 path: str | bytes | os.PathLike[str] | os.PathLike[bytes],
143 mode: int = 0o666,
144) -> IO[bytes]:
145 """Open a path for writing, refusing to follow a symlink at the final name.
147 For output written into a directory the caller named but does not
148 necessarily control, following a symlink pre-planted at the target name
149 would write outside that directory.
151 Args:
152 path: File to create or truncate
153 mode: Permission bits for a newly created file, before the umask
155 Returns:
156 A binary file object open for writing
158 Raises:
159 OSError: If the path is a symlink, with errno ELOOP
160 """
161 flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
162 if hasattr(os, "O_NOFOLLOW"):
163 flags |= os.O_NOFOLLOW
164 elif os.path.islink(path):
165 # Windows has no O_NOFOLLOW; this check races, but creating symlinks
166 # there requires privileges that make the attack far less reachable.
167 raise OSError(errno.ELOOP, os.strerror(errno.ELOOP), os.fspath(path))
169 return os.fdopen(os.open(path, flags, mode), "wb")
172def open_nofollow_read(
173 path: str | bytes | os.PathLike[str] | os.PathLike[bytes],
174) -> BinaryIO:
175 """Open a path for reading, refusing to follow a symlink at the final name.
177 ``.gitignore`` and ``.gitattributes`` are read from the work tree at names
178 the tree supplies; git opens both with ``open_nofollow()`` and skips them
179 with a warning if they are symlinks.
181 Args:
182 path: File to read
184 Returns:
185 A binary file object open for reading
187 Raises:
188 OSError: If the path is a symlink, with errno ELOOP (EMLINK on some BSDs)
189 """
190 flags = os.O_RDONLY | getattr(os, "O_BINARY", 0)
191 if hasattr(os, "O_NOFOLLOW"):
192 flags |= os.O_NOFOLLOW
193 elif os.path.islink(path):
194 # Windows has no O_NOFOLLOW; this check races, but creating symlinks
195 # there requires privileges.
196 raise OSError(errno.ELOOP, os.strerror(errno.ELOOP), os.fspath(path))
198 return os.fdopen(os.open(path, flags), "rb")
201def _fancy_rename(oldname: str | bytes, newname: str | bytes) -> None:
202 """Rename file with temporary backup file to rollback if rename fails."""
203 if not os.path.exists(newname):
204 os.rename(oldname, newname)
205 return
207 # Defer the tempfile import since it pulls in a lot of other things.
208 import tempfile
210 # destination file exists
211 (fd, tmpfile) = tempfile.mkstemp(".tmp", prefix=str(oldname), dir=".")
212 os.close(fd)
213 os.remove(tmpfile)
214 os.rename(newname, tmpfile)
215 try:
216 os.rename(oldname, newname)
217 except OSError:
218 os.rename(tmpfile, newname)
219 raise
220 os.remove(tmpfile)
223@overload
224def GitFile(
225 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
226 mode: Literal["wb"],
227 bufsize: int = -1,
228 mask: int = 0o644,
229 fsync: bool = True,
230 shared_perm: "SharedPerm | None" = None,
231) -> "_GitFile": ...
234@overload
235def GitFile(
236 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
237 mode: Literal["rb"] = "rb",
238 bufsize: int = -1,
239 mask: int = 0o644,
240 fsync: bool = True,
241 shared_perm: "SharedPerm | None" = None,
242) -> IO[bytes]: ...
245@overload
246def GitFile(
247 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
248 mode: str = "rb",
249 bufsize: int = -1,
250 mask: int = 0o644,
251 fsync: bool = True,
252 shared_perm: "SharedPerm | None" = None,
253) -> "IO[bytes] | _GitFile": ...
256def GitFile(
257 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
258 mode: str = "rb",
259 bufsize: int = -1,
260 mask: int = 0o644,
261 fsync: bool = True,
262 shared_perm: "SharedPerm | None" = None,
263) -> "IO[bytes] | _GitFile":
264 """Create a file object that obeys the git file locking protocol.
266 Returns: a builtin file object or a _GitFile object
268 Note: See _GitFile for a description of the file locking protocol.
270 Only read-only and write-only (binary) modes are supported; r+, w+, and a
271 are not. To read and write from the same file, you can take advantage of
272 the fact that opening a file for write does not actually open the file you
273 request.
275 The default file mask makes any created files user-writable and
276 world-readable.
278 Args:
279 filename: Path to the file
280 mode: File mode (only 'rb' and 'wb' are supported)
281 bufsize: Buffer size for file operations
282 mask: File mask for created files
283 fsync: Whether to call fsync() before closing (default: True)
284 shared_perm: core.sharedRepository setting to widen the mode with
286 """
287 if "a" in mode:
288 raise OSError("append mode not supported for Git files")
289 if "+" in mode:
290 raise OSError("read/write mode not supported for Git files")
291 if "b" not in mode:
292 raise OSError("text mode not supported for Git files")
293 if "w" in mode:
294 return _GitFile(filename, mode, bufsize, mask, fsync, shared_perm)
295 else:
296 return open(filename, mode, bufsize)
299class FileLocked(Exception):
300 """File is already locked."""
302 def __init__(
303 self,
304 filename: str | bytes,
305 lockfilename: str | bytes,
306 ) -> None:
307 """Initialize FileLocked.
309 Args:
310 filename: Name of the file that is locked
311 lockfilename: Name of the lock file
312 """
313 self.filename = filename
314 self.lockfilename = lockfilename
315 super().__init__(filename, lockfilename)
318class _GitFile(IO[bytes]):
319 """File that follows the git locking protocol for writes.
321 All writes to a file foo will be written into foo.lock in the same
322 directory, and the lockfile will be renamed to overwrite the original file
323 on close.
325 Note: You *must* call close() or abort() on a _GitFile for the lock to be
326 released. Typically this will happen in a finally block.
327 """
329 _file: IO[bytes]
330 _filename: str | bytes
331 _lockfilename: str | bytes
332 _closed: bool
334 PROXY_PROPERTIES: ClassVar[set[str]] = {
335 "encoding",
336 "errors",
337 "mode",
338 "name",
339 "newlines",
340 "softspace",
341 }
342 PROXY_METHODS: ClassVar[set[str]] = {
343 "__iter__",
344 "__next__",
345 "flush",
346 "fileno",
347 "isatty",
348 "read",
349 "readable",
350 "readline",
351 "readlines",
352 "seek",
353 "seekable",
354 "tell",
355 "truncate",
356 "writable",
357 "write",
358 "writelines",
359 }
361 def __init__(
362 self,
363 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
364 mode: str,
365 bufsize: int,
366 mask: int,
367 fsync: bool = True,
368 shared_perm: "SharedPerm | None" = None,
369 ) -> None:
370 # Convert PathLike to str/bytes for our internal use
371 self._filename: str | bytes = os.fspath(filename)
372 self._fsync = fsync
373 self._shared_perm = shared_perm
374 if isinstance(self._filename, bytes):
375 self._lockfilename: str | bytes = self._filename + b".lock"
376 else:
377 self._lockfilename = self._filename + ".lock"
378 try:
379 fd = os.open(
380 self._lockfilename,
381 os.O_RDWR | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0),
382 mask,
383 )
384 except FileExistsError as exc:
385 raise FileLocked(self._filename, self._lockfilename) from exc
386 self._file = os.fdopen(fd, mode, bufsize)
387 self._closed = False
389 def __iter__(self) -> Iterator[bytes]:
390 """Iterate over lines in the file."""
391 return iter(self._file)
393 def abort(self) -> None:
394 """Close and discard the lockfile without overwriting the target.
396 If the file is already closed, this is a no-op.
397 """
398 if self._closed:
399 return
400 self._file.close()
401 try:
402 os.remove(self._lockfilename)
403 self._closed = True
404 except FileNotFoundError:
405 # The file may have been removed already, which is ok.
406 self._closed = True
408 def close(self) -> None:
409 """Close this file, saving the lockfile over the original.
411 Note: If this method fails, it will attempt to delete the lockfile.
412 However, it is not guaranteed to do so (e.g. if a filesystem
413 becomes suddenly read-only), which will prevent future writes to
414 this file until the lockfile is removed manually.
416 Raises:
417 OSError: if the original file could not be overwritten. The
418 lock file is still closed, so further attempts to write to the same
419 file object will raise ValueError.
420 """
421 if self._closed:
422 return
423 self._file.flush()
424 if self._fsync:
425 os.fsync(self._file.fileno())
426 self._file.close()
427 # Adjust before the rename, so the file is never visible at the
428 # final path with the wrong permissions.
429 adjust_shared_perm(self._lockfilename, self._shared_perm)
430 try:
431 if getattr(os, "replace", None) is not None:
432 os.replace(self._lockfilename, self._filename)
433 else:
434 if sys.platform != "win32":
435 os.rename(self._lockfilename, self._filename)
436 else:
437 # Windows versions prior to Vista don't support atomic
438 # renames
439 _fancy_rename(self._lockfilename, self._filename)
440 finally:
441 self.abort()
443 def __del__(self) -> None:
444 if not getattr(self, "_closed", True):
445 warnings.warn(f"unclosed {self!r}", ResourceWarning, stacklevel=2)
446 self.abort()
448 def __enter__(self) -> Self:
449 return self
451 def __exit__(
452 self,
453 exc_type: type[BaseException] | None,
454 exc_val: BaseException | None,
455 exc_tb: TracebackType | None,
456 ) -> None:
457 if exc_type is not None:
458 self.abort()
459 else:
460 self.close()
462 def __fspath__(self) -> str | bytes:
463 """Return the file path for os.fspath() compatibility."""
464 return self._filename
466 @property
467 def closed(self) -> bool:
468 """Return whether the file is closed."""
469 return self._closed
471 def __getattr__(self, name: str) -> Any: # noqa: ANN401
472 """Proxy property calls to the underlying file."""
473 if name in self.PROXY_PROPERTIES:
474 return getattr(self._file, name)
475 raise AttributeError(name)
477 # Implement IO[bytes] methods by delegating to the underlying file
478 def read(self, size: int = -1) -> bytes:
479 return self._file.read(size)
481 # TODO: Remove type: ignore when Python 3.10 support is dropped (Oct 2026)
482 # Python 3.10 has issues with IO[bytes] overload signatures
483 def write(self, data: Buffer, /) -> int: # type: ignore[override,unused-ignore]
484 return self._file.write(data)
486 def readline(self, size: int = -1) -> bytes:
487 return self._file.readline(size)
489 def readlines(self, hint: int = -1) -> list[bytes]:
490 return self._file.readlines(hint)
492 # TODO: Remove type: ignore when Python 3.10 support is dropped (Oct 2026)
493 # Python 3.10 has issues with IO[bytes] overload signatures
494 def writelines(self, lines: Iterable[Buffer], /) -> None: # type: ignore[override,unused-ignore]
495 return self._file.writelines(lines)
497 def seek(self, offset: int, whence: int = 0) -> int:
498 return self._file.seek(offset, whence)
500 def tell(self) -> int:
501 return self._file.tell()
503 def flush(self) -> None:
504 return self._file.flush()
506 def truncate(self, size: int | None = None) -> int:
507 return self._file.truncate(size)
509 def fileno(self) -> int:
510 return self._file.fileno()
512 def isatty(self) -> bool:
513 return self._file.isatty()
515 def readable(self) -> bool:
516 return self._file.readable()
518 def writable(self) -> bool:
519 return self._file.writable()
521 def seekable(self) -> bool:
522 return self._file.seekable()
524 def __next__(self) -> bytes:
525 return next(iter(self._file))