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]
36import errno
37import os
38import stat
39import sys
40import warnings
41from collections.abc import Iterable, Iterator
42from dataclasses import dataclass
43from types import TracebackType
44from typing import IO, Any, ClassVar, Literal, overload
46from ._typing import Buffer
48if sys.version_info >= (3, 11):
49 from typing import Self
50else:
51 from typing_extensions import Self
54@dataclass(frozen=True)
55class SharedPerm:
56 """A parsed core.sharedRepository setting.
58 Attributes:
59 tweak: Permission bits the setting asks for
60 replace: Whether those bits state the mode outright, as an explicit
61 octal setting does, rather than only loosening the existing mode
62 """
64 tweak: int
65 replace: bool = False
68# git's PERM_GROUP and PERM_EVERYBODY.
69PERM_GROUP = SharedPerm(tweak=0o660)
70PERM_EVERYBODY = SharedPerm(tweak=0o664)
73def calc_shared_perm(mode: int, perm: "SharedPerm") -> int:
74 """Apply a shared permission setting to an existing mode.
76 Mirrors git's calc_shared_perm(). The umask never appears here: it has
77 already been applied by the kernel when the file was created, and named
78 settings only ever loosen the mode that resulted.
80 Args:
81 mode: Current permission bits of the file or directory
82 perm: Shared permission setting to apply
84 Returns:
85 The adjusted permission bits
86 """
87 tweak = perm.tweak
88 if not mode & stat.S_IWUSR:
89 tweak &= ~0o222
90 if mode & stat.S_IXUSR:
91 # Copy read bits to execute bits
92 tweak |= (tweak & 0o444) >> 2
93 if perm.replace:
94 return (mode & ~0o777) | tweak
95 return mode | tweak
98def adjust_shared_perm(
99 path: str | bytes | os.PathLike[str] | os.PathLike[bytes],
100 perm: "SharedPerm | None",
101) -> None:
102 """Widen the permissions of a path per core.sharedRepository.
104 Mirrors git's adjust_shared_perm(). The mode is read back from the path,
105 so whatever the umask already removed at creation time stays removed for
106 named settings.
108 Args:
109 path: File or directory to adjust
110 perm: Shared permission setting, or None to leave the path alone
111 """
112 if perm is None:
113 return
115 st = os.stat(path)
116 old_mode = stat.S_IMODE(st.st_mode)
117 new_mode = calc_shared_perm(old_mode, perm)
119 if stat.S_ISDIR(st.st_mode):
120 # Copy read bits to execute bits
121 new_mode |= (new_mode & 0o444) >> 2
122 # g+s matters only if group membership grants extra access
123 if new_mode & 0o060:
124 new_mode |= stat.S_ISGID
126 if new_mode != old_mode:
127 os.chmod(path, new_mode)
130def ensure_dir_exists(
131 dirname: str | bytes | os.PathLike[str] | os.PathLike[bytes],
132) -> None:
133 """Ensure a directory exists, creating if necessary."""
134 try:
135 os.makedirs(dirname)
136 except FileExistsError:
137 pass
140def open_nofollow(
141 path: str | bytes | os.PathLike[str] | os.PathLike[bytes],
142 mode: int = 0o666,
143) -> IO[bytes]:
144 """Open a path for writing, refusing to follow a symlink at the final name.
146 For output written into a directory the caller named but does not
147 necessarily control, following a symlink pre-planted at the target name
148 would write outside that directory.
150 Args:
151 path: File to create or truncate
152 mode: Permission bits for a newly created file, before the umask
154 Returns:
155 A binary file object open for writing
157 Raises:
158 OSError: If the path is a symlink, with errno ELOOP
159 """
160 flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
161 if hasattr(os, "O_NOFOLLOW"):
162 flags |= os.O_NOFOLLOW
163 elif os.path.islink(path):
164 # Windows has no O_NOFOLLOW; this check races, but creating symlinks
165 # there requires privileges that make the attack far less reachable.
166 raise OSError(errno.ELOOP, os.strerror(errno.ELOOP), os.fspath(path))
168 return os.fdopen(os.open(path, flags, mode), "wb")
171def _fancy_rename(oldname: str | bytes, newname: str | bytes) -> None:
172 """Rename file with temporary backup file to rollback if rename fails."""
173 if not os.path.exists(newname):
174 os.rename(oldname, newname)
175 return
177 # Defer the tempfile import since it pulls in a lot of other things.
178 import tempfile
180 # destination file exists
181 (fd, tmpfile) = tempfile.mkstemp(".tmp", prefix=str(oldname), dir=".")
182 os.close(fd)
183 os.remove(tmpfile)
184 os.rename(newname, tmpfile)
185 try:
186 os.rename(oldname, newname)
187 except OSError:
188 os.rename(tmpfile, newname)
189 raise
190 os.remove(tmpfile)
193@overload
194def GitFile(
195 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
196 mode: Literal["wb"],
197 bufsize: int = -1,
198 mask: int = 0o644,
199 fsync: bool = True,
200 shared_perm: "SharedPerm | None" = None,
201) -> "_GitFile": ...
204@overload
205def GitFile(
206 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
207 mode: Literal["rb"] = "rb",
208 bufsize: int = -1,
209 mask: int = 0o644,
210 fsync: bool = True,
211 shared_perm: "SharedPerm | None" = None,
212) -> IO[bytes]: ...
215@overload
216def GitFile(
217 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
218 mode: str = "rb",
219 bufsize: int = -1,
220 mask: int = 0o644,
221 fsync: bool = True,
222 shared_perm: "SharedPerm | None" = None,
223) -> "IO[bytes] | _GitFile": ...
226def GitFile(
227 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
228 mode: str = "rb",
229 bufsize: int = -1,
230 mask: int = 0o644,
231 fsync: bool = True,
232 shared_perm: "SharedPerm | None" = None,
233) -> "IO[bytes] | _GitFile":
234 """Create a file object that obeys the git file locking protocol.
236 Returns: a builtin file object or a _GitFile object
238 Note: See _GitFile for a description of the file locking protocol.
240 Only read-only and write-only (binary) modes are supported; r+, w+, and a
241 are not. To read and write from the same file, you can take advantage of
242 the fact that opening a file for write does not actually open the file you
243 request.
245 The default file mask makes any created files user-writable and
246 world-readable.
248 Args:
249 filename: Path to the file
250 mode: File mode (only 'rb' and 'wb' are supported)
251 bufsize: Buffer size for file operations
252 mask: File mask for created files
253 fsync: Whether to call fsync() before closing (default: True)
254 shared_perm: core.sharedRepository setting to widen the mode with
256 """
257 if "a" in mode:
258 raise OSError("append mode not supported for Git files")
259 if "+" in mode:
260 raise OSError("read/write mode not supported for Git files")
261 if "b" not in mode:
262 raise OSError("text mode not supported for Git files")
263 if "w" in mode:
264 return _GitFile(filename, mode, bufsize, mask, fsync, shared_perm)
265 else:
266 return open(filename, mode, bufsize)
269class FileLocked(Exception):
270 """File is already locked."""
272 def __init__(
273 self,
274 filename: str | bytes,
275 lockfilename: str | bytes,
276 ) -> None:
277 """Initialize FileLocked.
279 Args:
280 filename: Name of the file that is locked
281 lockfilename: Name of the lock file
282 """
283 self.filename = filename
284 self.lockfilename = lockfilename
285 super().__init__(filename, lockfilename)
288class _GitFile(IO[bytes]):
289 """File that follows the git locking protocol for writes.
291 All writes to a file foo will be written into foo.lock in the same
292 directory, and the lockfile will be renamed to overwrite the original file
293 on close.
295 Note: You *must* call close() or abort() on a _GitFile for the lock to be
296 released. Typically this will happen in a finally block.
297 """
299 _file: IO[bytes]
300 _filename: str | bytes
301 _lockfilename: str | bytes
302 _closed: bool
304 PROXY_PROPERTIES: ClassVar[set[str]] = {
305 "encoding",
306 "errors",
307 "mode",
308 "name",
309 "newlines",
310 "softspace",
311 }
312 PROXY_METHODS: ClassVar[set[str]] = {
313 "__iter__",
314 "__next__",
315 "flush",
316 "fileno",
317 "isatty",
318 "read",
319 "readable",
320 "readline",
321 "readlines",
322 "seek",
323 "seekable",
324 "tell",
325 "truncate",
326 "writable",
327 "write",
328 "writelines",
329 }
331 def __init__(
332 self,
333 filename: str | bytes | os.PathLike[str] | os.PathLike[bytes],
334 mode: str,
335 bufsize: int,
336 mask: int,
337 fsync: bool = True,
338 shared_perm: "SharedPerm | None" = None,
339 ) -> None:
340 # Convert PathLike to str/bytes for our internal use
341 self._filename: str | bytes = os.fspath(filename)
342 self._fsync = fsync
343 self._shared_perm = shared_perm
344 if isinstance(self._filename, bytes):
345 self._lockfilename: str | bytes = self._filename + b".lock"
346 else:
347 self._lockfilename = self._filename + ".lock"
348 try:
349 fd = os.open(
350 self._lockfilename,
351 os.O_RDWR | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0),
352 mask,
353 )
354 except FileExistsError as exc:
355 raise FileLocked(self._filename, self._lockfilename) from exc
356 self._file = os.fdopen(fd, mode, bufsize)
357 self._closed = False
359 def __iter__(self) -> Iterator[bytes]:
360 """Iterate over lines in the file."""
361 return iter(self._file)
363 def abort(self) -> None:
364 """Close and discard the lockfile without overwriting the target.
366 If the file is already closed, this is a no-op.
367 """
368 if self._closed:
369 return
370 self._file.close()
371 try:
372 os.remove(self._lockfilename)
373 self._closed = True
374 except FileNotFoundError:
375 # The file may have been removed already, which is ok.
376 self._closed = True
378 def close(self) -> None:
379 """Close this file, saving the lockfile over the original.
381 Note: If this method fails, it will attempt to delete the lockfile.
382 However, it is not guaranteed to do so (e.g. if a filesystem
383 becomes suddenly read-only), which will prevent future writes to
384 this file until the lockfile is removed manually.
386 Raises:
387 OSError: if the original file could not be overwritten. The
388 lock file is still closed, so further attempts to write to the same
389 file object will raise ValueError.
390 """
391 if self._closed:
392 return
393 self._file.flush()
394 if self._fsync:
395 os.fsync(self._file.fileno())
396 self._file.close()
397 # Adjust before the rename, so the file is never visible at the
398 # final path with the wrong permissions.
399 adjust_shared_perm(self._lockfilename, self._shared_perm)
400 try:
401 if getattr(os, "replace", None) is not None:
402 os.replace(self._lockfilename, self._filename)
403 else:
404 if sys.platform != "win32":
405 os.rename(self._lockfilename, self._filename)
406 else:
407 # Windows versions prior to Vista don't support atomic
408 # renames
409 _fancy_rename(self._lockfilename, self._filename)
410 finally:
411 self.abort()
413 def __del__(self) -> None:
414 if not getattr(self, "_closed", True):
415 warnings.warn(f"unclosed {self!r}", ResourceWarning, stacklevel=2)
416 self.abort()
418 def __enter__(self) -> Self:
419 return self
421 def __exit__(
422 self,
423 exc_type: type[BaseException] | None,
424 exc_val: BaseException | None,
425 exc_tb: TracebackType | None,
426 ) -> None:
427 if exc_type is not None:
428 self.abort()
429 else:
430 self.close()
432 def __fspath__(self) -> str | bytes:
433 """Return the file path for os.fspath() compatibility."""
434 return self._filename
436 @property
437 def closed(self) -> bool:
438 """Return whether the file is closed."""
439 return self._closed
441 def __getattr__(self, name: str) -> Any: # noqa: ANN401
442 """Proxy property calls to the underlying file."""
443 if name in self.PROXY_PROPERTIES:
444 return getattr(self._file, name)
445 raise AttributeError(name)
447 # Implement IO[bytes] methods by delegating to the underlying file
448 def read(self, size: int = -1) -> bytes:
449 return self._file.read(size)
451 # TODO: Remove type: ignore when Python 3.10 support is dropped (Oct 2026)
452 # Python 3.10 has issues with IO[bytes] overload signatures
453 def write(self, data: Buffer, /) -> int: # type: ignore[override,unused-ignore]
454 return self._file.write(data)
456 def readline(self, size: int = -1) -> bytes:
457 return self._file.readline(size)
459 def readlines(self, hint: int = -1) -> list[bytes]:
460 return self._file.readlines(hint)
462 # TODO: Remove type: ignore when Python 3.10 support is dropped (Oct 2026)
463 # Python 3.10 has issues with IO[bytes] overload signatures
464 def writelines(self, lines: Iterable[Buffer], /) -> None: # type: ignore[override,unused-ignore]
465 return self._file.writelines(lines)
467 def seek(self, offset: int, whence: int = 0) -> int:
468 return self._file.seek(offset, whence)
470 def tell(self) -> int:
471 return self._file.tell()
473 def flush(self) -> None:
474 return self._file.flush()
476 def truncate(self, size: int | None = None) -> int:
477 return self._file.truncate(size)
479 def fileno(self) -> int:
480 return self._file.fileno()
482 def isatty(self) -> bool:
483 return self._file.isatty()
485 def readable(self) -> bool:
486 return self._file.readable()
488 def writable(self) -> bool:
489 return self._file.writable()
491 def seekable(self) -> bool:
492 return self._file.seekable()
494 def __next__(self) -> bytes:
495 return next(iter(self._file))