Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/index.py: 31%
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# index.py -- File parser/writer for the git index file
2# Copyright (C) 2008-2013 Jelmer Vernooij <jelmer@jelmer.uk>
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"""Parser for the git index file format."""
24__all__ = [
25 "DEFAULT_VERSION",
26 "EOIE_EXTENSION",
27 "EXTENDED_FLAG_INTEND_TO_ADD",
28 "EXTENDED_FLAG_SKIP_WORKTREE",
29 "FLAG_EXTENDED",
30 "FLAG_NAMEMASK",
31 "FLAG_STAGEMASK",
32 "FLAG_STAGESHIFT",
33 "FLAG_VALID",
34 "HFS_IGNORABLE_CHARS",
35 "IEOT_EXTENSION",
36 "INVALID_DOTNAMES",
37 "RESERVED_WINDOWS_DEVICE_NAMES",
38 "REUC_EXTENSION",
39 "SDIR_EXTENSION",
40 "TREE_EXTENSION",
41 "UNTR_EXTENSION",
42 "ConflictedIndexEntry",
43 "Index",
44 "IndexChecksumReader",
45 "IndexChecksumWriter",
46 "IndexEntry",
47 "IndexExtension",
48 "InvalidPathError",
49 "ResolveUndoExtension",
50 "SerializedIndexEntry",
51 "SparseDirExtension",
52 "Stage",
53 "TreeDict",
54 "TreeExtension",
55 "UnmergedEntries",
56 "UnsupportedIndexFormat",
57 "UntrackedExtension",
58 "apply_stat_refresh",
59 "blob_from_path_and_mode",
60 "blob_from_path_and_stat",
61 "build_file_from_blob",
62 "build_index_from_tree",
63 "changes_from_tree",
64 "cleanup_mode",
65 "commit_index",
66 "commit_tree",
67 "detect_case_only_renames",
68 "get_path_element_normalizer",
69 "get_path_element_validator",
70 "get_symlink_fn",
71 "get_unstaged_changes",
72 "index_entry_from_directory",
73 "index_entry_from_path",
74 "index_entry_from_stat",
75 "index_entry_from_tree_entry",
76 "iter_fresh_entries",
77 "iter_fresh_objects",
78 "locked_index",
79 "make_path_normalizer",
80 "os_sep_bytes",
81 "pathjoin",
82 "pathsplit",
83 "read_cache_entry",
84 "read_cache_time",
85 "read_index",
86 "read_index_dict",
87 "read_index_dict_with_version",
88 "read_index_header",
89 "read_submodule_head",
90 "refresh_index",
91 "symlink",
92 "update_working_tree",
93 "validate_path",
94 "validate_path_element_default",
95 "validate_path_element_hfs",
96 "validate_path_element_ntfs",
97 "verify_leading_dirs",
98 "verify_tree_path",
99 "write_cache_entry",
100 "write_cache_time",
101 "write_index",
102 "write_index_dict",
103 "write_index_extension",
104]
106import errno
107import logging
108import os
109import shutil
110import stat
111import struct
112import sys
113import types
114from collections.abc import (
115 Callable,
116 Generator,
117 Iterable,
118 Iterator,
119 Mapping,
120 Sequence,
121 Set,
122)
123from dataclasses import dataclass
124from enum import Enum
125from typing import (
126 TYPE_CHECKING,
127 Any,
128 Protocol,
129)
131if TYPE_CHECKING:
132 from .config import Config
133 from .diff_tree import TreeChange
134 from .file import _GitFile
135 from .filters import FilterBlobNormalizer
136 from .object_store import BaseObjectStore
137 from .repo import Repo
139from . import _deprecated_aliases
140from .errors import ChecksumMismatch
141from .file import GitFile, SharedPerm
142from .object_format import DEFAULT_OBJECT_FORMAT, ObjectFormat
143from .object_store import iter_tree_contents
144from .objects import (
145 S_IFGITLINK,
146 S_ISGITLINK,
147 Blob,
148 ObjectID,
149 RawObjectID,
150 Tree,
151 TreeEntry,
152 hex_to_sha,
153 sha_to_hex,
154)
155from .pack import ObjectContainer
157logger = logging.getLogger(__name__)
160__getattr__ = _deprecated_aliases(
161 __name__,
162 {
163 "SHA1Reader": "dulwich.pack.SHA1Reader",
164 "SHA1Writer": "dulwich.pack.SHA1Writer",
165 },
166)
169# Type alias for recursive tree structure used in commit_tree
170TreeDict = dict[bytes, "TreeDict | tuple[int, ObjectID]"]
172# 2-bit stage (during merge)
173FLAG_STAGEMASK = 0x3000
174FLAG_STAGESHIFT = 12
175FLAG_NAMEMASK = 0x0FFF
177# assume-valid
178FLAG_VALID = 0x8000
180# extended flag (must be zero in version 2)
181FLAG_EXTENDED = 0x4000
183# used by sparse checkout
184EXTENDED_FLAG_SKIP_WORKTREE = 0x4000
186# used by "git add -N"
187EXTENDED_FLAG_INTEND_TO_ADD = 0x2000
189DEFAULT_VERSION = 2
191# Index extension signatures
192TREE_EXTENSION = b"TREE"
193REUC_EXTENSION = b"REUC"
194UNTR_EXTENSION = b"UNTR"
195EOIE_EXTENSION = b"EOIE"
196IEOT_EXTENSION = b"IEOT"
197SDIR_EXTENSION = b"sdir" # Sparse directory extension
200def _encode_varint(value: int) -> bytes:
201 """Encode an integer using variable-width encoding.
203 Same format as used for OFS_DELTA pack entries and index v4 path compression.
204 Uses 7 bits per byte, with the high bit indicating continuation.
206 Args:
207 value: Integer to encode
208 Returns:
209 Encoded bytes
210 """
211 if value == 0:
212 return b"\x00"
214 result = []
215 while value > 0:
216 byte = value & 0x7F # Take lower 7 bits
217 value >>= 7
218 if value > 0:
219 byte |= 0x80 # Set continuation bit
220 result.append(byte)
222 return bytes(result)
225def _decode_varint(data: bytes, offset: int = 0) -> tuple[int, int]:
226 """Decode a variable-width encoded integer.
228 Args:
229 data: Bytes to decode from
230 offset: Starting offset in data
231 Returns:
232 tuple of (decoded_value, new_offset)
233 """
234 value = 0
235 shift = 0
236 pos = offset
238 while pos < len(data):
239 byte = data[pos]
240 pos += 1
241 value |= (byte & 0x7F) << shift
242 shift += 7
243 if not (byte & 0x80): # No continuation bit
244 break
246 return value, pos
249def _compress_path(path: bytes, previous_path: bytes) -> bytes:
250 """Compress a path relative to the previous path for index version 4.
252 Args:
253 path: Path to compress
254 previous_path: Previous path for comparison
255 Returns:
256 Compressed path data (varint prefix_len + suffix)
257 """
258 # Find the common prefix length
259 common_len = 0
260 min_len = min(len(path), len(previous_path))
262 for i in range(min_len):
263 if path[i] == previous_path[i]:
264 common_len += 1
265 else:
266 break
268 # The number of bytes to remove from the end of previous_path
269 # to get the common prefix
270 remove_len = len(previous_path) - common_len
272 # The suffix to append
273 suffix = path[common_len:]
275 # Encode: varint(remove_len) + suffix + NUL
276 return _encode_varint(remove_len) + suffix + b"\x00"
279def _decompress_path(
280 data: bytes, offset: int, previous_path: bytes
281) -> tuple[bytes, int]:
282 """Decompress a path from index version 4 compressed format.
284 Args:
285 data: Raw data containing compressed path
286 offset: Starting offset in data
287 previous_path: Previous path for decompression
288 Returns:
289 tuple of (decompressed_path, new_offset)
290 """
291 # Decode the number of bytes to remove from previous path
292 remove_len, new_offset = _decode_varint(data, offset)
294 # Find the NUL terminator for the suffix
295 suffix_start = new_offset
296 suffix_end = suffix_start
297 while suffix_end < len(data) and data[suffix_end] != 0:
298 suffix_end += 1
300 if suffix_end >= len(data):
301 raise ValueError("Unterminated path suffix in compressed entry")
303 suffix = data[suffix_start:suffix_end]
304 new_offset = suffix_end + 1 # Skip the NUL terminator
306 # Reconstruct the path
307 if remove_len > len(previous_path):
308 raise ValueError(
309 f"Invalid path compression: trying to remove {remove_len} bytes from {len(previous_path)}-byte path"
310 )
312 prefix = previous_path[:-remove_len] if remove_len > 0 else previous_path
313 path = prefix + suffix
315 return path, new_offset
318class _IndexWritable(Protocol):
319 """The subset of a binary file that the index serializer writes to."""
321 def write(self, data: bytes, /) -> int:
322 """Write data, returning the number of bytes written."""
324 def tell(self) -> int:
325 """Return the current stream position."""
327 def close(self) -> None:
328 """Close the stream."""
331class _IndexReadable(Protocol):
332 """The subset of a binary file that the index parser reads from."""
334 def read(self, size: int = -1, /) -> bytes:
335 """Read up to size bytes."""
337 def tell(self) -> int:
338 """Return the current stream position."""
340 def seek(self, offset: int, whence: int = 0, /) -> int:
341 """Change the stream position."""
344def _decompress_path_from_stream(
345 f: _IndexReadable, previous_path: bytes
346) -> tuple[bytes, int]:
347 """Decompress a path from index version 4 compressed format, reading from stream.
349 Args:
350 f: File-like object to read from
351 previous_path: Previous path for decompression
352 Returns:
353 tuple of (decompressed_path, bytes_consumed)
354 """
355 # Decode the varint for remove_len by reading byte by byte
356 remove_len = 0
357 shift = 0
358 bytes_consumed = 0
360 while True:
361 byte_data = f.read(1)
362 if not byte_data:
363 raise ValueError("Unexpected end of file while reading varint")
364 byte = byte_data[0]
365 bytes_consumed += 1
366 remove_len |= (byte & 0x7F) << shift
367 shift += 7
368 if not (byte & 0x80): # No continuation bit
369 break
371 # Read the suffix until NUL terminator
372 suffix = b""
373 while True:
374 byte_data = f.read(1)
375 if not byte_data:
376 raise ValueError("Unexpected end of file while reading path suffix")
377 byte = byte_data[0]
378 bytes_consumed += 1
379 if byte == 0: # NUL terminator
380 break
381 suffix += bytes([byte])
383 # Reconstruct the path
384 if remove_len > len(previous_path):
385 raise ValueError(
386 f"Invalid path compression: trying to remove {remove_len} bytes from {len(previous_path)}-byte path"
387 )
389 prefix = previous_path[:-remove_len] if remove_len > 0 else previous_path
390 path = prefix + suffix
392 return path, bytes_consumed
395class Stage(Enum):
396 """Represents the stage of an index entry during merge conflicts."""
398 NORMAL = 0
399 MERGE_CONFLICT_ANCESTOR = 1
400 MERGE_CONFLICT_THIS = 2
401 MERGE_CONFLICT_OTHER = 3
404@dataclass
405class SerializedIndexEntry:
406 """Represents a serialized index entry as stored in the index file.
408 This dataclass holds the raw data for an index entry before it's
409 parsed into the more user-friendly IndexEntry format.
410 """
412 name: bytes
413 ctime: int | float | tuple[int, int]
414 mtime: int | float | tuple[int, int]
415 dev: int
416 ino: int
417 mode: int
418 uid: int
419 gid: int
420 size: int
421 sha: ObjectID
422 flags: int
423 extended_flags: int
425 def stage(self) -> Stage:
426 """Extract the stage from the flags field.
428 Returns:
429 Stage enum value indicating merge conflict state
430 """
431 return Stage((self.flags & FLAG_STAGEMASK) >> FLAG_STAGESHIFT)
433 def is_sparse_dir(self) -> bool:
434 """Check if this entry represents a sparse directory.
436 A sparse directory entry is a collapsed representation of an entire
437 directory tree in a sparse index. It has:
438 - Directory mode (0o040000)
439 - SKIP_WORKTREE flag set
440 - Path ending with '/'
441 - SHA pointing to a tree object
443 Returns:
444 True if entry is a sparse directory entry
445 """
446 return (
447 stat.S_ISDIR(self.mode)
448 and bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
449 and self.name.endswith(b"/")
450 )
453@dataclass
454class IndexExtension:
455 """Base class for index extensions."""
457 signature: bytes
458 data: bytes
460 @classmethod
461 def from_raw(cls, signature: bytes, data: bytes) -> "IndexExtension":
462 """Create an extension from raw data.
464 Args:
465 signature: 4-byte extension signature
466 data: Extension data
467 Returns:
468 Parsed extension object
469 """
470 if signature == TREE_EXTENSION:
471 return TreeExtension.from_bytes(data)
472 elif signature == REUC_EXTENSION:
473 return ResolveUndoExtension.from_bytes(data)
474 elif signature == UNTR_EXTENSION:
475 return UntrackedExtension.from_bytes(data)
476 elif signature == SDIR_EXTENSION:
477 return SparseDirExtension.from_bytes(data)
478 else:
479 # Unknown extension - just store raw data
480 return cls(signature, data)
482 def to_bytes(self) -> bytes:
483 """Serialize extension to bytes."""
484 return self.data
487class TreeExtension(IndexExtension):
488 """Tree cache extension."""
490 def __init__(self, entries: list[tuple[bytes, bytes, int]]) -> None:
491 """Initialize TreeExtension.
493 Args:
494 entries: List of tree cache entries (path, sha, flags)
495 """
496 self.entries = entries
497 super().__init__(TREE_EXTENSION, b"")
499 @classmethod
500 def from_bytes(cls, data: bytes) -> "TreeExtension":
501 """Parse TreeExtension from bytes.
503 Args:
504 data: Raw bytes to parse
506 Returns:
507 TreeExtension instance
508 """
509 # TODO: Implement tree cache parsing
510 return cls([])
512 def to_bytes(self) -> bytes:
513 """Serialize TreeExtension to bytes.
515 Returns:
516 Serialized extension data
517 """
518 # TODO: Implement tree cache serialization
519 return b""
522class ResolveUndoExtension(IndexExtension):
523 """Resolve undo extension for recording merge conflicts."""
525 def __init__(self, entries: list[tuple[bytes, list[tuple[int, bytes]]]]) -> None:
526 """Initialize ResolveUndoExtension.
528 Args:
529 entries: List of (path, stages) where stages is a list of (stage, sha) tuples
530 """
531 self.entries = entries
532 super().__init__(REUC_EXTENSION, b"")
534 @classmethod
535 def from_bytes(cls, data: bytes) -> "ResolveUndoExtension":
536 """Parse ResolveUndoExtension from bytes.
538 Args:
539 data: Raw bytes to parse
541 Returns:
542 ResolveUndoExtension instance
543 """
544 # TODO: Implement resolve undo parsing
545 return cls([])
547 def to_bytes(self) -> bytes:
548 """Serialize ResolveUndoExtension to bytes.
550 Returns:
551 Serialized extension data
552 """
553 # TODO: Implement resolve undo serialization
554 return b""
557class UntrackedExtension(IndexExtension):
558 """Untracked cache extension."""
560 def __init__(self, data: bytes) -> None:
561 """Initialize UntrackedExtension.
563 Args:
564 data: Raw untracked cache data
565 """
566 super().__init__(UNTR_EXTENSION, data)
568 @classmethod
569 def from_bytes(cls, data: bytes) -> "UntrackedExtension":
570 """Parse UntrackedExtension from bytes.
572 Args:
573 data: Raw bytes to parse
575 Returns:
576 UntrackedExtension instance
577 """
578 return cls(data)
581class SparseDirExtension(IndexExtension):
582 """Sparse directory extension.
584 This extension indicates that the index contains sparse directory entries.
585 Tools that don't understand sparse index should avoid interacting with
586 the index when this extension is present.
588 The extension data is empty - its presence is the signal.
589 """
591 def __init__(self) -> None:
592 """Initialize SparseDirExtension."""
593 super().__init__(SDIR_EXTENSION, b"")
595 @classmethod
596 def from_bytes(cls, data: bytes) -> "SparseDirExtension":
597 """Parse SparseDirExtension from bytes.
599 Args:
600 data: Raw bytes to parse (should be empty)
602 Returns:
603 SparseDirExtension instance
604 """
605 return cls()
607 def to_bytes(self) -> bytes:
608 """Serialize SparseDirExtension to bytes.
610 Returns:
611 Empty bytes (extension presence is the signal)
612 """
613 return b""
616@dataclass
617class IndexEntry:
618 """Represents an entry in the Git index.
620 This is a higher-level representation of an index entry that includes
621 parsed data and convenience methods.
622 """
624 ctime: int | float | tuple[int, int]
625 mtime: int | float | tuple[int, int]
626 dev: int
627 ino: int
628 mode: int
629 uid: int
630 gid: int
631 size: int
632 sha: ObjectID
633 flags: int = 0
634 extended_flags: int = 0
636 @classmethod
637 def from_serialized(cls, serialized: SerializedIndexEntry) -> "IndexEntry":
638 """Create an IndexEntry from a SerializedIndexEntry.
640 Args:
641 serialized: SerializedIndexEntry to convert
643 Returns:
644 New IndexEntry instance
645 """
646 return cls(
647 ctime=serialized.ctime,
648 mtime=serialized.mtime,
649 dev=serialized.dev,
650 ino=serialized.ino,
651 mode=serialized.mode,
652 uid=serialized.uid,
653 gid=serialized.gid,
654 size=serialized.size,
655 sha=serialized.sha,
656 flags=serialized.flags,
657 extended_flags=serialized.extended_flags,
658 )
660 def serialize(self, name: bytes, stage: Stage) -> SerializedIndexEntry:
661 """Serialize this entry with a given name and stage.
663 Args:
664 name: Path name for the entry
665 stage: Merge conflict stage
667 Returns:
668 SerializedIndexEntry ready for writing to disk
669 """
670 # Clear out any existing stage bits, then set them from the Stage.
671 new_flags = self.flags & ~FLAG_STAGEMASK
672 new_flags |= stage.value << FLAG_STAGESHIFT
673 return SerializedIndexEntry(
674 name=name,
675 ctime=self.ctime,
676 mtime=self.mtime,
677 dev=self.dev,
678 ino=self.ino,
679 mode=self.mode,
680 uid=self.uid,
681 gid=self.gid,
682 size=self.size,
683 sha=self.sha,
684 flags=new_flags,
685 extended_flags=self.extended_flags,
686 )
688 def stage(self) -> Stage:
689 """Get the merge conflict stage of this entry.
691 Returns:
692 Stage enum value
693 """
694 return Stage((self.flags & FLAG_STAGEMASK) >> FLAG_STAGESHIFT)
696 @property
697 def skip_worktree(self) -> bool:
698 """Return True if the skip-worktree bit is set in extended_flags."""
699 return bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
701 def set_skip_worktree(self, skip: bool = True) -> None:
702 """Helper method to set or clear the skip-worktree bit in extended_flags.
704 Also sets FLAG_EXTENDED in self.flags if needed.
705 """
706 if skip:
707 # Turn on the skip-worktree bit
708 self.extended_flags |= EXTENDED_FLAG_SKIP_WORKTREE
709 # Also ensure the main 'extended' bit is set in flags
710 self.flags |= FLAG_EXTENDED
711 else:
712 # Turn off the skip-worktree bit
713 self.extended_flags &= ~EXTENDED_FLAG_SKIP_WORKTREE
714 # Optionally unset the main extended bit if no extended flags remain
715 if self.extended_flags == 0:
716 self.flags &= ~FLAG_EXTENDED
718 def is_sparse_dir(self, name: bytes) -> bool:
719 """Check if this entry represents a sparse directory.
721 A sparse directory entry is a collapsed representation of an entire
722 directory tree in a sparse index. It has:
723 - Directory mode (0o040000)
724 - SKIP_WORKTREE flag set
725 - Path ending with '/'
726 - SHA pointing to a tree object
728 Args:
729 name: The path name for this entry (IndexEntry doesn't store name)
731 Returns:
732 True if entry is a sparse directory entry
733 """
734 return (
735 stat.S_ISDIR(self.mode)
736 and bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
737 and name.endswith(b"/")
738 )
741class ConflictedIndexEntry:
742 """Index entry that represents a conflict."""
744 ancestor: IndexEntry | None
745 this: IndexEntry | None
746 other: IndexEntry | None
748 def __init__(
749 self,
750 ancestor: IndexEntry | None = None,
751 this: IndexEntry | None = None,
752 other: IndexEntry | None = None,
753 ) -> None:
754 """Initialize ConflictedIndexEntry.
756 Args:
757 ancestor: The common ancestor entry
758 this: The current branch entry
759 other: The other branch entry
760 """
761 self.ancestor = ancestor
762 self.this = this
763 self.other = other
766class UnmergedEntries(Exception):
767 """Unmerged entries exist in the index."""
770def pathsplit(path: bytes) -> tuple[bytes, bytes]:
771 """Split a /-delimited path into a directory part and a basename.
773 Args:
774 path: The path to split.
776 Returns:
777 Tuple with directory name and basename
778 """
779 try:
780 (dirname, basename) = path.rsplit(b"/", 1)
781 except ValueError:
782 return (b"", path)
783 else:
784 return (dirname, basename)
787def pathjoin(*args: bytes) -> bytes:
788 """Join a /-delimited path."""
789 return b"/".join([p for p in args if p])
792def read_cache_time(f: _IndexReadable) -> tuple[int, int]:
793 """Read a cache time.
795 Args:
796 f: File-like object to read from
797 Returns:
798 Tuple with seconds and nanoseconds
799 """
800 return struct.unpack(">LL", f.read(8))
803def write_cache_time(f: _IndexWritable, t: int | float | tuple[int, int]) -> None:
804 """Write a cache time.
806 Args:
807 f: File-like object to write to
808 t: Time to write (as int, float or tuple with secs and nsecs)
809 """
810 if isinstance(t, int):
811 t = (t, 0)
812 elif isinstance(t, float):
813 (secs, nsecs) = divmod(t, 1.0)
814 t = (int(secs), int(nsecs * 1000000000))
815 elif not isinstance(t, tuple):
816 raise TypeError(t)
817 f.write(struct.pack(">LL", *t))
820class IndexChecksumWriter:
821 """Wrap a file, tracking the checksum of everything written to it.
823 The index trailer uses the repository's object format, so unlike
824 :class:`dulwich.pack.SHA1Writer` this is not tied to SHA-1.
825 """
827 def __init__(
828 self, f: _IndexWritable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
829 ) -> None:
830 """Initialize the writer.
832 Args:
833 f: File-like object to wrap
834 object_format: Object format determining the checksum algorithm
835 """
836 self.f = f
837 self.object_format = object_format
838 self._hash = object_format.new_hash()
840 def write(self, data: bytes) -> int:
841 """Write data and update the running checksum."""
842 self._hash.update(data)
843 return self.f.write(data)
845 def tell(self) -> int:
846 """Return the position of the underlying file."""
847 return self.f.tell()
849 def write_checksum(self) -> bytes:
850 """Write the trailing checksum and return it."""
851 digest = self._hash.digest()
852 self.f.write(digest)
853 return digest
855 def close(self) -> None:
856 """Write the trailing checksum and close the underlying file."""
857 self.write_checksum()
858 self.f.close()
861class IndexChecksumReader:
862 """Wrap a file, tracking the checksum of everything read from it."""
864 def __init__(
865 self, f: _IndexReadable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
866 ) -> None:
867 """Initialize the reader.
869 Args:
870 f: File-like object to wrap
871 object_format: Object format determining the checksum algorithm
872 """
873 self.f = f
874 self.object_format = object_format
875 self._hash = object_format.new_hash()
877 def read(self, size: int = -1) -> bytes:
878 """Read data and update the running checksum."""
879 data = self.f.read(size)
880 self._hash.update(data)
881 return data
883 def tell(self) -> int:
884 """Return the position of the underlying file."""
885 return self.f.tell()
887 def seek(self, offset: int, whence: int = 0) -> int:
888 """Seek the underlying file.
890 Bytes skipped by seeking are not folded into the checksum, so this
891 is only safe for the trailer probing done while reading extensions.
892 """
893 return self.f.seek(offset, whence)
895 def check_checksum(self, allow_empty: bool = False) -> None:
896 """Verify the trailing checksum.
898 Args:
899 allow_empty: Accept an all-zero checksum, as written when the
900 index.skipHash config option is set.
902 Raises:
903 ChecksumMismatch: If the stored checksum does not match.
904 """
905 length = self.object_format.oid_length
906 stored = self.f.read(length)
907 expected = self._hash.digest()
908 if stored == expected:
909 return
910 if allow_empty and (len(stored) != length or stored == b"\x00" * length):
911 return
912 raise ChecksumMismatch(
913 self._hash.hexdigest(), sha_to_hex(RawObjectID(stored)).decode("ascii")
914 )
917def read_cache_entry(
918 f: _IndexReadable,
919 version: int,
920 previous_path: bytes = b"",
921 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
922) -> SerializedIndexEntry:
923 """Read an entry from a cache file.
925 Args:
926 f: File-like object to read from
927 version: Index version
928 previous_path: Previous entry's path (for version 4 compression)
929 object_format: Object format determining the object ID width
930 """
931 beginoffset = f.tell()
932 ctime = read_cache_time(f)
933 mtime = read_cache_time(f)
934 oid_length = object_format.oid_length
935 (
936 dev,
937 ino,
938 mode,
939 uid,
940 gid,
941 size,
942 sha,
943 flags,
944 ) = struct.unpack(f">LLLLLL{oid_length}sH", f.read(oid_length + 4 * 6 + 2))
945 if flags & FLAG_EXTENDED:
946 if version < 3:
947 raise AssertionError("extended flag set in index with version < 3")
948 (extended_flags,) = struct.unpack(">H", f.read(2))
949 else:
950 extended_flags = 0
952 if version >= 4:
953 # Version 4: paths are always compressed (name_len should be 0)
954 name, _consumed = _decompress_path_from_stream(f, previous_path)
955 else:
956 # Versions < 4: regular name reading
957 name = f.read(flags & FLAG_NAMEMASK)
959 # Padding:
960 if version < 4:
961 real_size = (f.tell() - beginoffset + 8) & ~7
962 f.read((beginoffset + real_size) - f.tell())
964 return SerializedIndexEntry(
965 name,
966 ctime,
967 mtime,
968 dev,
969 ino,
970 mode,
971 uid,
972 gid,
973 size,
974 sha_to_hex(sha),
975 flags & ~FLAG_NAMEMASK,
976 extended_flags,
977 )
980def write_cache_entry(
981 f: _IndexWritable,
982 entry: SerializedIndexEntry,
983 version: int,
984 previous_path: bytes = b"",
985 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
986) -> None:
987 """Write an index entry to a file.
989 Args:
990 f: File object
991 entry: IndexEntry to write
992 version: Index format version
993 previous_path: Previous entry's path (for version 4 compression)
994 object_format: Object format determining the object ID width
995 """
996 if len(entry.sha) != object_format.hex_length:
997 raise ValueError(
998 f"Object ID {entry.sha!r} for {entry.name!r} is not valid for "
999 f"a {object_format.name} index"
1000 )
1002 beginoffset = f.tell()
1003 write_cache_time(f, entry.ctime)
1004 write_cache_time(f, entry.mtime)
1006 compressed_path = b""
1007 if version >= 4:
1008 # Version 4: use compression but set name_len to actual filename length
1009 # This matches how C Git implements index v4 flags
1010 compressed_path = _compress_path(entry.name, previous_path)
1011 flags = len(entry.name) | (entry.flags & ~FLAG_NAMEMASK)
1013 if entry.extended_flags:
1014 flags |= FLAG_EXTENDED
1015 if flags & FLAG_EXTENDED and version is not None and version < 3:
1016 raise AssertionError("unable to use extended flags in version < 3")
1018 f.write(
1019 struct.pack(
1020 f">LLLLLL{object_format.oid_length}sH",
1021 entry.dev & 0xFFFFFFFF,
1022 entry.ino & 0xFFFFFFFF,
1023 entry.mode,
1024 entry.uid,
1025 entry.gid,
1026 entry.size,
1027 hex_to_sha(entry.sha),
1028 flags,
1029 )
1030 )
1031 if flags & FLAG_EXTENDED:
1032 f.write(struct.pack(b">H", entry.extended_flags))
1034 if version >= 4:
1035 # Version 4: always write compressed path
1036 f.write(compressed_path)
1037 else:
1038 # Versions < 4: write regular path and padding
1039 f.write(entry.name)
1040 real_size = (f.tell() - beginoffset + 8) & ~7
1041 f.write(b"\0" * ((beginoffset + real_size) - f.tell()))
1044class UnsupportedIndexFormat(Exception):
1045 """An unsupported index format was encountered."""
1047 def __init__(self, version: int) -> None:
1048 """Initialize UnsupportedIndexFormat exception.
1050 Args:
1051 version: The unsupported index format version
1052 """
1053 self.index_format_version = version
1056def read_index_header(f: _IndexReadable) -> tuple[int, int]:
1057 """Read an index header from a file.
1059 Returns:
1060 tuple of (version, num_entries)
1061 """
1062 header = f.read(4)
1063 if header != b"DIRC":
1064 raise AssertionError(f"Invalid index file header: {header!r}")
1065 (version, num_entries) = struct.unpack(b">LL", f.read(4 * 2))
1066 if version not in (1, 2, 3, 4):
1067 raise UnsupportedIndexFormat(version)
1068 return version, num_entries
1071def write_index_extension(f: _IndexWritable, extension: IndexExtension) -> None:
1072 """Write an index extension.
1074 Args:
1075 f: File-like object to write to
1076 extension: Extension to write
1077 """
1078 data = extension.to_bytes()
1079 f.write(extension.signature)
1080 f.write(struct.pack(">I", len(data)))
1081 f.write(data)
1084def read_index(
1085 f: _IndexReadable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
1086) -> Iterator[SerializedIndexEntry]:
1087 """Read an index file, yielding the individual entries."""
1088 version, num_entries = read_index_header(f)
1089 previous_path = b""
1090 for i in range(num_entries):
1091 entry = read_cache_entry(f, version, previous_path, object_format)
1092 previous_path = entry.name
1093 yield entry
1096def read_index_dict_with_version(
1097 f: _IndexReadable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
1098) -> tuple[dict[bytes, IndexEntry | ConflictedIndexEntry], int, list[IndexExtension]]:
1099 """Read an index file and return it as a dictionary along with the version.
1101 Returns:
1102 tuple of (entries_dict, version, extensions)
1103 """
1104 version, num_entries = read_index_header(f)
1106 ret: dict[bytes, IndexEntry | ConflictedIndexEntry] = {}
1107 previous_path = b""
1108 for i in range(num_entries):
1109 entry = read_cache_entry(f, version, previous_path, object_format)
1110 previous_path = entry.name
1111 stage = entry.stage()
1112 if stage == Stage.NORMAL:
1113 ret[entry.name] = IndexEntry.from_serialized(entry)
1114 else:
1115 existing = ret.setdefault(entry.name, ConflictedIndexEntry())
1116 if isinstance(existing, IndexEntry):
1117 raise AssertionError(f"Non-conflicted entry for {entry.name!r} exists")
1118 if stage == Stage.MERGE_CONFLICT_ANCESTOR:
1119 existing.ancestor = IndexEntry.from_serialized(entry)
1120 elif stage == Stage.MERGE_CONFLICT_THIS:
1121 existing.this = IndexEntry.from_serialized(entry)
1122 elif stage == Stage.MERGE_CONFLICT_OTHER:
1123 existing.other = IndexEntry.from_serialized(entry)
1125 # Read extensions
1126 extensions = []
1127 while True:
1128 # Check if we're at the end (trailing checksum before EOF)
1129 current_pos = f.tell()
1130 f.seek(0, 2) # EOF
1131 eof_pos = f.tell()
1132 f.seek(current_pos)
1134 if current_pos >= eof_pos - object_format.oid_length:
1135 break
1137 # Try to read extension signature
1138 signature = f.read(4)
1139 if len(signature) < 4:
1140 break
1142 # Check if it's a valid extension signature (4 uppercase letters)
1143 if not all(65 <= b <= 90 for b in signature):
1144 # Not an extension, seek back
1145 f.seek(-4, 1)
1146 break
1148 # Read extension size
1149 size_data = f.read(4)
1150 if len(size_data) < 4:
1151 break
1152 size = struct.unpack(">I", size_data)[0]
1154 # Read extension data
1155 data = f.read(size)
1156 if len(data) < size:
1157 break
1159 extension = IndexExtension.from_raw(signature, data)
1160 extensions.append(extension)
1162 return ret, version, extensions
1165def read_index_dict(
1166 f: _IndexReadable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
1167) -> dict[bytes, IndexEntry | ConflictedIndexEntry]:
1168 """Read an index file and return it as a dictionary.
1170 Dict Key is tuple of path and stage number, as
1171 path alone is not unique
1172 Args:
1173 f: File object to read fromls.
1174 object_format: Object format determining the object ID width
1175 """
1176 ret: dict[bytes, IndexEntry | ConflictedIndexEntry] = {}
1177 for entry in read_index(f, object_format):
1178 stage = entry.stage()
1179 if stage == Stage.NORMAL:
1180 ret[entry.name] = IndexEntry.from_serialized(entry)
1181 else:
1182 existing = ret.setdefault(entry.name, ConflictedIndexEntry())
1183 if isinstance(existing, IndexEntry):
1184 raise AssertionError(f"Non-conflicted entry for {entry.name!r} exists")
1185 if stage == Stage.MERGE_CONFLICT_ANCESTOR:
1186 existing.ancestor = IndexEntry.from_serialized(entry)
1187 elif stage == Stage.MERGE_CONFLICT_THIS:
1188 existing.this = IndexEntry.from_serialized(entry)
1189 elif stage == Stage.MERGE_CONFLICT_OTHER:
1190 existing.other = IndexEntry.from_serialized(entry)
1191 return ret
1194def write_index(
1195 f: _IndexWritable,
1196 entries: Sequence[SerializedIndexEntry],
1197 version: int | None = None,
1198 extensions: Sequence[IndexExtension] | None = None,
1199 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
1200) -> None:
1201 """Write an index file.
1203 Args:
1204 f: File-like object to write to
1205 version: Version number to write
1206 entries: Iterable over the entries to write
1207 extensions: Optional list of extensions to write
1208 object_format: Object format determining the object ID width
1209 """
1210 if version is None:
1211 version = DEFAULT_VERSION
1212 # STEP 1: check if any extended_flags are set
1213 uses_extended_flags = any(e.extended_flags != 0 for e in entries)
1214 if uses_extended_flags and version < 3:
1215 # Force or bump the version to 3
1216 version = 3
1217 # The rest is unchanged, but you might insert a final check:
1218 if version < 3:
1219 # Double-check no extended flags appear
1220 for e in entries:
1221 if e.extended_flags != 0:
1222 raise AssertionError("Attempt to use extended flags in index < v3")
1223 # Proceed with the existing code to write the header and entries.
1224 f.write(b"DIRC")
1225 f.write(struct.pack(b">LL", version, len(entries)))
1226 previous_path = b""
1227 for entry in entries:
1228 write_cache_entry(
1229 f,
1230 entry,
1231 version=version,
1232 previous_path=previous_path,
1233 object_format=object_format,
1234 )
1235 previous_path = entry.name
1237 # Write extensions
1238 if extensions:
1239 for extension in extensions:
1240 write_index_extension(f, extension)
1243def write_index_dict(
1244 f: _IndexWritable,
1245 entries: Mapping[bytes, IndexEntry | ConflictedIndexEntry],
1246 version: int | None = None,
1247 extensions: Sequence[IndexExtension] | None = None,
1248 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
1249) -> None:
1250 """Write an index file based on the contents of a dictionary.
1252 being careful to sort by path and then by stage.
1253 """
1254 entries_list = []
1255 for key in sorted(entries):
1256 value = entries[key]
1257 if isinstance(value, ConflictedIndexEntry):
1258 if value.ancestor is not None:
1259 entries_list.append(
1260 value.ancestor.serialize(key, Stage.MERGE_CONFLICT_ANCESTOR)
1261 )
1262 if value.this is not None:
1263 entries_list.append(
1264 value.this.serialize(key, Stage.MERGE_CONFLICT_THIS)
1265 )
1266 if value.other is not None:
1267 entries_list.append(
1268 value.other.serialize(key, Stage.MERGE_CONFLICT_OTHER)
1269 )
1270 else:
1271 entries_list.append(value.serialize(key, Stage.NORMAL))
1273 write_index(
1274 f,
1275 entries_list,
1276 version=version,
1277 extensions=extensions,
1278 object_format=object_format,
1279 )
1282def cleanup_mode(mode: int) -> int:
1283 """Cleanup a mode value.
1285 This will return a mode that can be stored in a tree object.
1287 Args:
1288 mode: Mode to clean up.
1290 Returns:
1291 mode
1292 """
1293 if stat.S_ISLNK(mode):
1294 return stat.S_IFLNK
1295 elif stat.S_ISDIR(mode):
1296 return stat.S_IFDIR
1297 elif S_ISGITLINK(mode):
1298 return S_IFGITLINK
1299 ret = stat.S_IFREG | 0o644
1300 if mode & 0o100:
1301 ret |= 0o111
1302 return ret
1305class Index:
1306 """A Git Index file."""
1308 _byname: dict[bytes, IndexEntry | ConflictedIndexEntry]
1310 def __init__(
1311 self,
1312 filename: bytes | str | os.PathLike[str],
1313 read: bool = True,
1314 skip_hash: bool = False,
1315 version: int | None = None,
1316 *,
1317 shared_perm: "SharedPerm | None" = None,
1318 path_normalizer: Callable[[bytes], bytes] | None = None,
1319 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
1320 ) -> None:
1321 """Create an index object associated with the given filename.
1323 Args:
1324 filename: Path to the index file
1325 read: Whether to initialize the index from the given file, should it exist.
1326 skip_hash: Whether to skip SHA1 hash when writing (for manyfiles feature)
1327 version: Index format version to use (None = auto-detect from file or use default)
1328 shared_perm: Optional shared repository permission setting
1329 object_format: Object format of the repository this index belongs to.
1330 Determines the width of stored object IDs and of the trailing
1331 checksum.
1332 path_normalizer: Optional function mapping a filesystem path to a
1333 canonical form (e.g. case-folded, NFC-normalized). When provided,
1334 lookups (``index[path]``, ``path in index``, ``del index[path]``)
1335 transparently match paths that normalize to the same form as an
1336 existing entry.
1337 """
1338 self._filename = os.fspath(filename)
1339 # TODO(jelmer): Store the version returned by read_index
1340 self._version = version
1341 self._skip_hash = skip_hash
1342 self._shared_perm = shared_perm
1343 self._object_format = object_format
1344 self._extensions: list[IndexExtension] = []
1345 self._path_normalizer = path_normalizer
1346 self._normalized: dict[bytes, bytes] | None = (
1347 {} if path_normalizer is not None else None
1348 )
1349 self.clear()
1350 if read:
1351 self.read()
1353 def canonical_path(self, name: bytes) -> bytes:
1354 """Resolve ``name`` to the canonical key stored in the index.
1356 If an entry already exists under ``name`` (or no normalizer is
1357 configured), ``name`` is returned unchanged. Otherwise the
1358 normalizer is applied and the key of any entry with the same
1359 normalized form is returned. Falls back to ``name`` if none.
1361 Normally callers do not need this because ``index[name]``,
1362 ``name in index``, and ``del index[name]`` already apply
1363 normalization transparently. Use this when the path is also
1364 being used outside the index (for example to look up the same
1365 entry in a commit tree), so that both sides agree on the key.
1366 """
1367 if self._normalized is None or name in self._byname:
1368 return name
1369 assert self._path_normalizer is not None
1370 return self._normalized.get(self._path_normalizer(name), name)
1372 @property
1373 def path(self) -> bytes | str:
1374 """Get the path to the index file.
1376 Returns:
1377 Path to the index file
1378 """
1379 return self._filename
1381 @property
1382 def object_format(self) -> ObjectFormat:
1383 """Object format used for object IDs in this index."""
1384 return self._object_format
1386 def __repr__(self) -> str:
1387 """Return string representation of Index."""
1388 return f"{self.__class__.__name__}({self._filename!r})"
1390 def write(self) -> None:
1391 """Write current contents of index to disk."""
1392 f = GitFile(self._filename, "wb", shared_perm=self._shared_perm)
1393 try:
1394 # Filter out extensions with no meaningful data
1395 meaningful_extensions = []
1396 for ext in self._extensions:
1397 # Skip extensions that have empty data
1398 ext_data = ext.to_bytes()
1399 if ext_data:
1400 meaningful_extensions.append(ext)
1402 if self._skip_hash:
1403 # When skipHash is enabled, write the index without computing
1404 # the trailing checksum
1405 write_index_dict(
1406 f,
1407 self._byname,
1408 version=self._version,
1409 extensions=meaningful_extensions,
1410 object_format=self._object_format,
1411 )
1412 # Write zero bytes instead of the checksum
1413 f.write(b"\x00" * self._object_format.oid_length)
1414 f.close()
1415 else:
1416 checksum_writer = IndexChecksumWriter(f, self._object_format)
1417 write_index_dict(
1418 checksum_writer,
1419 self._byname,
1420 version=self._version,
1421 extensions=meaningful_extensions,
1422 object_format=self._object_format,
1423 )
1424 checksum_writer.close()
1425 except:
1426 f.close()
1427 raise
1429 def read(self) -> None:
1430 """Read current contents of index from disk."""
1431 if not os.path.exists(self._filename):
1432 return
1433 f = GitFile(self._filename, "rb")
1434 try:
1435 checksum_reader = IndexChecksumReader(f, self._object_format)
1436 entries, version, extensions = read_index_dict_with_version(
1437 checksum_reader, self._object_format
1438 )
1439 self._version = version
1440 self._extensions = extensions
1441 for name, value in entries.items():
1442 self.set_verbatim(name, value)
1443 # Extensions have already been read by read_index_dict_with_version
1444 checksum_reader.check_checksum(allow_empty=True)
1445 finally:
1446 f.close()
1448 def __len__(self) -> int:
1449 """Number of entries in this index file."""
1450 return len(self._byname)
1452 def __getitem__(self, key: bytes) -> IndexEntry | ConflictedIndexEntry:
1453 """Retrieve entry by relative path and stage.
1455 Returns: Either a IndexEntry or a ConflictedIndexEntry
1456 Raises KeyError: if the entry does not exist
1457 """
1458 return self._byname[self.canonical_path(key)]
1460 def __iter__(self) -> Iterator[bytes]:
1461 """Iterate over the paths and stages in this index."""
1462 return iter(self._byname)
1464 def __contains__(self, key: bytes) -> bool:
1465 """Check if a path exists in the index."""
1466 return self.canonical_path(key) in self._byname
1468 def get_sha1(self, path: bytes) -> ObjectID:
1469 """Return the (git object) SHA1 for the object at a path."""
1470 value = self[path]
1471 if isinstance(value, ConflictedIndexEntry):
1472 raise UnmergedEntries
1473 return value.sha
1475 def get_mode(self, path: bytes) -> int:
1476 """Return the POSIX file mode for the object at a path."""
1477 value = self[path]
1478 if isinstance(value, ConflictedIndexEntry):
1479 raise UnmergedEntries
1480 return value.mode
1482 def iterobjects(self) -> Iterable[tuple[bytes, ObjectID, int]]:
1483 """Iterate over path, sha, mode tuples for use with commit_tree."""
1484 for path in self:
1485 entry = self[path]
1486 if isinstance(entry, ConflictedIndexEntry):
1487 raise UnmergedEntries
1488 yield path, entry.sha, cleanup_mode(entry.mode)
1490 def has_conflicts(self) -> bool:
1491 """Check if the index contains any conflicted entries.
1493 Returns:
1494 True if any entries are conflicted, False otherwise
1495 """
1496 for value in self._byname.values():
1497 if isinstance(value, ConflictedIndexEntry):
1498 return True
1499 return False
1501 def clear(self) -> None:
1502 """Remove all contents from this index."""
1503 self._byname = {}
1504 if self._normalized is not None:
1505 self._normalized = {}
1507 def __setitem__(
1508 self, name: bytes, value: IndexEntry | ConflictedIndexEntry
1509 ) -> None:
1510 """Set an entry in the index."""
1511 assert isinstance(name, bytes)
1512 name = self.canonical_path(name)
1513 is_new = name not in self._byname
1514 self._byname[name] = value
1515 if is_new and self._normalized is not None:
1516 assert self._path_normalizer is not None
1517 self._normalized.setdefault(self._path_normalizer(name), name)
1519 def set_verbatim(
1520 self, name: bytes, value: IndexEntry | ConflictedIndexEntry
1521 ) -> None:
1522 """Set an entry without folding onto an existing normalized key.
1524 Use this when populating the index from an authoritative source
1525 that may legitimately hold two entries whose paths only differ by
1526 case (or NFD/NFC form), such as an on-disk index or a tree from
1527 the object store. ``__setitem__`` would collapse the second entry
1528 onto the first via the path normalizer, losing content.
1529 """
1530 assert isinstance(name, bytes)
1531 self._byname[name] = value
1532 if self._normalized is not None:
1533 assert self._path_normalizer is not None
1534 self._normalized.setdefault(self._path_normalizer(name), name)
1536 def __delitem__(self, name: bytes) -> None:
1537 """Delete an entry from the index."""
1538 name = self.canonical_path(name)
1539 del self._byname[name]
1540 if self._normalized is not None:
1541 assert self._path_normalizer is not None
1542 normalized_key = self._path_normalizer(name)
1543 if self._normalized.get(normalized_key) == name:
1544 del self._normalized[normalized_key]
1546 def iteritems(
1547 self,
1548 ) -> Iterator[tuple[bytes, IndexEntry | ConflictedIndexEntry]]:
1549 """Iterate over (path, entry) pairs in the index.
1551 Returns:
1552 Iterator of (path, entry) tuples
1553 """
1554 return iter(self._byname.items())
1556 def items(self) -> Iterator[tuple[bytes, IndexEntry | ConflictedIndexEntry]]:
1557 """Get an iterator over (path, entry) pairs.
1559 Returns:
1560 Iterator of (path, entry) tuples
1561 """
1562 return iter(self._byname.items())
1564 def update(self, entries: dict[bytes, IndexEntry | ConflictedIndexEntry]) -> None:
1565 """Update the index with multiple entries.
1567 Args:
1568 entries: Dictionary mapping paths to index entries
1569 """
1570 for key, value in entries.items():
1571 self[key] = value
1573 def paths(self) -> Generator[bytes, None, None]:
1574 """Generate all paths in the index.
1576 Yields:
1577 Path names as bytes
1578 """
1579 yield from self._byname.keys()
1581 def changes_from_tree(
1582 self,
1583 object_store: ObjectContainer,
1584 tree: ObjectID | None,
1585 want_unchanged: bool = False,
1586 ) -> Generator[
1587 tuple[
1588 tuple[bytes | None, bytes | None],
1589 tuple[int | None, int | None],
1590 tuple[bytes | None, bytes | None],
1591 ],
1592 None,
1593 None,
1594 ]:
1595 """Find the differences between the contents of this index and a tree.
1597 Args:
1598 object_store: Object store to use for retrieving tree contents
1599 tree: SHA1 of the root tree
1600 want_unchanged: Whether unchanged files should be reported
1601 Returns: Iterator over tuples with (oldpath, newpath), (oldmode,
1602 newmode), (oldsha, newsha)
1603 """
1605 def lookup_entry(path: bytes) -> tuple[bytes, int]:
1606 entry = self[path]
1607 if isinstance(entry, IndexEntry):
1608 return entry.sha, cleanup_mode(entry.mode)
1609 else:
1610 # Handle ConflictedIndexEntry case
1611 return b"", 0
1613 yield from changes_from_tree(
1614 self.paths(),
1615 lookup_entry,
1616 object_store,
1617 tree,
1618 want_unchanged=want_unchanged,
1619 )
1621 def commit(self, object_store: ObjectContainer) -> ObjectID:
1622 """Create a new tree from an index.
1624 Args:
1625 object_store: Object store to save the tree in
1626 Returns:
1627 Root tree SHA
1628 """
1629 return commit_tree(
1630 object_store, self.iterobjects(), object_format=self._object_format
1631 )
1633 def is_sparse(self) -> bool:
1634 """Check if this index contains sparse directory entries.
1636 Returns:
1637 True if any sparse directory extension is present
1638 """
1639 return any(isinstance(ext, SparseDirExtension) for ext in self._extensions)
1641 def ensure_full_index(self, object_store: "BaseObjectStore") -> None:
1642 """Expand all sparse directory entries into full file entries.
1644 This converts a sparse index into a full index by recursively
1645 expanding any sparse directory entries into their constituent files.
1647 Args:
1648 object_store: Object store to read tree objects from
1650 Raises:
1651 KeyError: If a tree object referenced by a sparse dir entry doesn't exist
1652 """
1653 if not self.is_sparse():
1654 return
1656 # Find all sparse directory entries
1657 sparse_dirs = []
1658 for path, entry in list(self._byname.items()):
1659 if isinstance(entry, IndexEntry) and entry.is_sparse_dir(path):
1660 sparse_dirs.append((path, entry))
1662 # Expand each sparse directory
1663 for path, entry in sparse_dirs:
1664 # Remove the sparse directory entry
1665 del self[path]
1667 # Get the tree object
1668 tree = object_store[entry.sha]
1669 if not isinstance(tree, Tree):
1670 raise ValueError(f"Sparse directory {path!r} points to non-tree object")
1672 # Recursively add all entries from the tree
1673 self._expand_tree(path.rstrip(b"/"), tree, object_store, entry)
1675 # Remove the sparse directory extension
1676 self._extensions = [
1677 ext for ext in self._extensions if not isinstance(ext, SparseDirExtension)
1678 ]
1680 def _expand_tree(
1681 self,
1682 prefix: bytes,
1683 tree: Tree,
1684 object_store: "BaseObjectStore",
1685 template_entry: IndexEntry,
1686 ) -> None:
1687 """Recursively expand a tree into index entries.
1689 Args:
1690 prefix: Path prefix for entries (without trailing slash)
1691 tree: Tree object to expand
1692 object_store: Object store to read nested trees from
1693 template_entry: Template entry to copy metadata from
1694 """
1695 for name, mode, sha in tree.items():
1696 if prefix:
1697 full_path = prefix + b"/" + name
1698 else:
1699 full_path = name
1701 if stat.S_ISDIR(mode):
1702 # Recursively expand subdirectories
1703 subtree = object_store[sha]
1704 if not isinstance(subtree, Tree):
1705 raise ValueError(
1706 f"Directory entry {full_path!r} points to non-tree object"
1707 )
1708 self._expand_tree(full_path, subtree, object_store, template_entry)
1709 else:
1710 # Create an index entry for this file
1711 # Use the template entry for metadata but with the file's sha and mode
1712 new_entry = IndexEntry(
1713 ctime=template_entry.ctime,
1714 mtime=template_entry.mtime,
1715 dev=template_entry.dev,
1716 ino=template_entry.ino,
1717 mode=mode,
1718 uid=template_entry.uid,
1719 gid=template_entry.gid,
1720 size=0, # Size is unknown from tree
1721 sha=sha,
1722 flags=0,
1723 extended_flags=0, # Don't copy skip-worktree flag
1724 )
1725 self[full_path] = new_entry
1727 def convert_to_sparse(
1728 self,
1729 object_store: "BaseObjectStore",
1730 tree_sha: ObjectID,
1731 sparse_dirs: Set[bytes],
1732 ) -> None:
1733 """Convert full index entries to sparse directory entries.
1735 This collapses directories that are entirely outside the sparse
1736 checkout cone into single sparse directory entries.
1738 Args:
1739 object_store: Object store to read tree objects
1740 tree_sha: SHA of the tree (usually HEAD) to base sparse dirs on
1741 sparse_dirs: Set of directory paths (with trailing /) to collapse
1743 Raises:
1744 KeyError: If tree_sha or a subdirectory doesn't exist
1745 """
1746 if not sparse_dirs:
1747 return
1749 # Get the base tree
1750 tree = object_store[tree_sha]
1751 if not isinstance(tree, Tree):
1752 raise ValueError(f"tree_sha {tree_sha!r} is not a tree object")
1754 # For each sparse directory, find its tree SHA and create sparse entry
1755 for dir_path in sparse_dirs:
1756 dir_path_stripped = dir_path.rstrip(b"/")
1758 # Find the tree SHA for this directory
1759 subtree_sha = self._find_subtree_sha(tree, dir_path_stripped, object_store)
1760 if subtree_sha is None:
1761 # Directory doesn't exist in tree, skip it
1762 continue
1764 # Remove all entries under this directory
1765 entries_to_remove = [
1766 path
1767 for path in self._byname
1768 if path.startswith(dir_path) or path == dir_path_stripped
1769 ]
1770 for path in entries_to_remove:
1771 del self[path]
1773 # Create a sparse directory entry
1774 # Use minimal metadata since it's not a real file
1775 from dulwich.objects import ObjectID
1777 sparse_entry = IndexEntry(
1778 ctime=0,
1779 mtime=0,
1780 dev=0,
1781 ino=0,
1782 mode=stat.S_IFDIR,
1783 uid=0,
1784 gid=0,
1785 size=0,
1786 sha=ObjectID(subtree_sha),
1787 flags=0,
1788 extended_flags=EXTENDED_FLAG_SKIP_WORKTREE,
1789 )
1790 self[dir_path] = sparse_entry
1792 # Add sparse directory extension if not present
1793 if not self.is_sparse():
1794 self._extensions.append(SparseDirExtension())
1796 def _find_subtree_sha(
1797 self,
1798 tree: Tree,
1799 path: bytes,
1800 object_store: "BaseObjectStore",
1801 ) -> bytes | None:
1802 """Find the SHA of a subtree at a given path.
1804 Args:
1805 tree: Root tree object to search in
1806 path: Path to the subdirectory (no trailing slash)
1807 object_store: Object store to read nested trees from
1809 Returns:
1810 SHA of the subtree, or None if path doesn't exist
1811 """
1812 if not path:
1813 return tree.get_id(self._object_format)
1815 parts = path.split(b"/")
1816 current_tree = tree
1818 for part in parts:
1819 # Look for this part in the current tree
1820 try:
1821 mode, sha = current_tree[part]
1822 except KeyError:
1823 return None
1825 if not stat.S_ISDIR(mode):
1826 # Path component is a file, not a directory
1827 return None
1829 # Load the next tree
1830 obj = object_store[sha]
1831 if not isinstance(obj, Tree):
1832 return None
1833 current_tree = obj
1835 return current_tree.get_id(self._object_format)
1838def commit_tree(
1839 object_store: ObjectContainer,
1840 blobs: Iterable[tuple[bytes, ObjectID, int]],
1841 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
1842) -> ObjectID:
1843 """Commit a new tree.
1845 Args:
1846 object_store: Object store to add trees to
1847 blobs: Iterable over blob path, sha, mode entries
1848 object_format: Object format to name the created trees with
1849 Returns:
1850 ID of the created tree.
1851 """
1852 trees: dict[bytes, TreeDict] = {b"": {}}
1854 def add_tree(path: bytes) -> TreeDict:
1855 if path in trees:
1856 return trees[path]
1857 dirname, basename = pathsplit(path)
1858 t = add_tree(dirname)
1859 assert isinstance(basename, bytes)
1860 newtree: TreeDict = {}
1861 t[basename] = newtree
1862 trees[path] = newtree
1863 return newtree
1865 for path, sha, mode in blobs:
1866 tree_path, basename = pathsplit(path)
1867 tree = add_tree(tree_path)
1868 tree[basename] = (mode, sha)
1870 def build_tree(path: bytes) -> ObjectID:
1871 tree = Tree()
1872 for basename, entry in trees[path].items():
1873 if isinstance(entry, dict):
1874 mode = stat.S_IFDIR
1875 sha = build_tree(pathjoin(path, basename))
1876 else:
1877 (mode, sha) = entry
1878 tree.add(basename, mode, sha)
1879 object_store.add_object(tree)
1880 return ObjectID(tree.get_id(object_format))
1882 return build_tree(b"")
1885def commit_index(object_store: ObjectContainer, index: Index) -> ObjectID:
1886 """Create a new tree from an index.
1888 Args:
1889 object_store: Object store to save the tree in
1890 index: Index file
1891 Note: This function is deprecated, use index.commit() instead.
1892 Returns: Root tree sha.
1893 """
1894 return commit_tree(
1895 object_store, index.iterobjects(), object_format=index.object_format
1896 )
1899def changes_from_tree(
1900 names: Iterable[bytes],
1901 lookup_entry: Callable[[bytes], tuple[bytes, int]],
1902 object_store: ObjectContainer,
1903 tree: ObjectID | None,
1904 want_unchanged: bool = False,
1905) -> Iterable[
1906 tuple[
1907 tuple[bytes | None, bytes | None],
1908 tuple[int | None, int | None],
1909 tuple[bytes | None, bytes | None],
1910 ]
1911]:
1912 """Find the differences between the contents of a tree and a working copy.
1914 Args:
1915 names: Iterable of names in the working copy
1916 lookup_entry: Function to lookup an entry in the working copy
1917 object_store: Object store to use for retrieving tree contents
1918 tree: SHA1 of the root tree, or None for an empty tree
1919 want_unchanged: Whether unchanged files should be reported
1920 Returns: Iterator over tuples with (oldpath, newpath), (oldmode, newmode),
1921 (oldsha, newsha)
1922 """
1923 # TODO(jelmer): Support a include_trees option
1924 other_names = set(names)
1926 if tree is not None:
1927 for name, mode, sha in iter_tree_contents(object_store, tree):
1928 assert name is not None and mode is not None and sha is not None
1929 try:
1930 (other_sha, other_mode) = lookup_entry(name)
1931 except KeyError:
1932 # Was removed
1933 yield ((name, None), (mode, None), (sha, None))
1934 else:
1935 other_names.remove(name)
1936 if want_unchanged or other_sha != sha or other_mode != mode:
1937 yield ((name, name), (mode, other_mode), (sha, other_sha))
1939 # Mention added files
1940 for name in other_names:
1941 try:
1942 (other_sha, other_mode) = lookup_entry(name)
1943 except KeyError:
1944 pass
1945 else:
1946 yield ((None, name), (None, other_mode), (None, other_sha))
1949def index_entry_from_stat(
1950 stat_val: os.stat_result,
1951 hex_sha: bytes,
1952 mode: int | None = None,
1953) -> IndexEntry:
1954 """Create a new index entry from a stat value.
1956 Args:
1957 stat_val: POSIX stat_result instance
1958 hex_sha: Hex sha of the object
1959 mode: Optional file mode, will be derived from stat if not provided
1960 """
1961 if mode is None:
1962 mode = cleanup_mode(stat_val.st_mode)
1964 from dulwich.objects import ObjectID
1966 # Use nanosecond precision when available to avoid precision loss
1967 # through float representation
1968 ctime: int | float | tuple[int, int]
1969 mtime: int | float | tuple[int, int]
1970 st_ctime_ns = getattr(stat_val, "st_ctime_ns", None)
1971 if st_ctime_ns is not None:
1972 ctime = (
1973 st_ctime_ns // 1_000_000_000,
1974 st_ctime_ns % 1_000_000_000,
1975 )
1976 else:
1977 ctime = stat_val.st_ctime
1979 st_mtime_ns = getattr(stat_val, "st_mtime_ns", None)
1980 if st_mtime_ns is not None:
1981 mtime = (
1982 st_mtime_ns // 1_000_000_000,
1983 st_mtime_ns % 1_000_000_000,
1984 )
1985 else:
1986 mtime = stat_val.st_mtime
1988 return IndexEntry(
1989 ctime=ctime,
1990 mtime=mtime,
1991 dev=stat_val.st_dev,
1992 ino=stat_val.st_ino,
1993 mode=mode,
1994 uid=stat_val.st_uid,
1995 gid=stat_val.st_gid,
1996 size=stat_val.st_size,
1997 sha=ObjectID(hex_sha),
1998 flags=0,
1999 extended_flags=0,
2000 )
2003def index_entry_from_tree_entry(
2004 mode: int,
2005 hex_sha: bytes,
2006 size: int = 0,
2007) -> IndexEntry:
2008 """Create an index entry from a tree entry, with zeroed stat fields.
2010 Use this when populating an index directly from a tree without touching
2011 the filesystem, matching ``git read-tree``. The stat-derived fields
2012 (ctime/mtime/dev/ino/uid/gid) are zeroed; git treats such entries as
2013 stat-unmerged and refreshes them on the next stat.
2015 Args:
2016 mode: File mode from the tree entry
2017 hex_sha: Hex sha of the object
2018 size: Size of the object (0 for gitlinks or when unknown)
2019 """
2020 from dulwich.objects import ObjectID
2022 return IndexEntry(
2023 ctime=0,
2024 mtime=0,
2025 dev=0,
2026 ino=0,
2027 mode=mode,
2028 uid=0,
2029 gid=0,
2030 size=size,
2031 sha=ObjectID(hex_sha),
2032 flags=0,
2033 extended_flags=0,
2034 )
2037if sys.platform == "win32":
2038 # On Windows, creating symlinks either requires administrator privileges
2039 # or developer mode. Raise a more helpful error when we're unable to
2040 # create symlinks
2042 # https://github.com/jelmer/dulwich/issues/1005
2044 class WindowsSymlinkPermissionError(PermissionError):
2045 """Windows-specific error for symlink creation failures.
2047 This error is raised when symlink creation fails on Windows,
2048 typically due to lack of developer mode or administrator privileges.
2049 """
2051 def __init__(self, errno: int, msg: str, filename: str | None) -> None:
2052 """Initialize WindowsSymlinkPermissionError."""
2053 super().__init__(
2054 errno,
2055 f"Unable to create symlink; do you have developer mode enabled? {msg}",
2056 filename,
2057 )
2059 def symlink(
2060 src: str | bytes,
2061 dst: str | bytes,
2062 target_is_directory: bool = False,
2063 *,
2064 dir_fd: int | None = None,
2065 ) -> None:
2066 """Create a symbolic link on Windows with better error handling.
2068 Args:
2069 src: Source path for the symlink
2070 dst: Destination path where symlink will be created
2071 target_is_directory: Whether the target is a directory
2072 dir_fd: Optional directory file descriptor
2074 Raises:
2075 WindowsSymlinkPermissionError: If symlink creation fails due to permissions
2076 """
2077 try:
2078 return os.symlink(
2079 src, dst, target_is_directory=target_is_directory, dir_fd=dir_fd
2080 )
2081 except PermissionError as e:
2082 raise WindowsSymlinkPermissionError(
2083 e.errno or 0, e.strerror or "", e.filename
2084 ) from e
2085else:
2086 symlink = os.symlink
2089def get_symlink_fn(
2090 config: "Config",
2091) -> Callable[[str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None]:
2092 """Get the function to use for creating symlinks in the working tree.
2094 With ``core.symlinks`` set to false, symlinks are checked out as plain
2095 files containing the link target, like git does.
2097 Args:
2098 config: Repository configuration
2100 Returns:
2101 Function taking the link target and the path to create
2102 """
2103 if config.get_boolean(b"core", b"symlinks", True):
2105 def create_symlink(
2106 source: str | bytes | os.PathLike[str],
2107 target: str | bytes | os.PathLike[str],
2108 ) -> None:
2109 symlink(source, target) # type: ignore[arg-type,unused-ignore]
2111 return create_symlink
2113 def write_link_as_file(
2114 source: str | bytes | os.PathLike[str],
2115 target: str | bytes | os.PathLike[str],
2116 ) -> None:
2117 mode = "w" + ("b" if isinstance(source, bytes) else "")
2118 with open(target, mode) as f:
2119 f.write(source)
2121 return write_link_as_file
2124def build_file_from_blob(
2125 blob: Blob,
2126 mode: int,
2127 target_path: bytes,
2128 *,
2129 honor_filemode: bool = True,
2130 tree_encoding: str = "utf-8",
2131 symlink_fn: Callable[
2132 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2133 ]
2134 | None = None,
2135) -> os.stat_result:
2136 """Build a file or symlink on disk based on a Git object.
2138 Args:
2139 blob: The git object
2140 mode: File mode
2141 target_path: Path to write to
2142 honor_filemode: An optional flag to honor core.filemode setting in
2143 config file, default is core.filemode=True, change executable bit
2144 tree_encoding: Encoding to use for tree contents
2145 symlink_fn: Function to use for creating symlinks
2146 Returns: stat object for the file
2147 """
2148 try:
2149 oldstat = os.lstat(target_path)
2150 except FileNotFoundError:
2151 oldstat = None
2152 contents = blob.as_raw_string()
2153 if stat.S_ISLNK(mode):
2154 if oldstat:
2155 _remove_file_with_readonly_handling(target_path)
2156 if sys.platform == "win32":
2157 # os.readlink on Python3 on Windows requires a unicode string.
2158 contents_str = contents.decode(tree_encoding)
2159 target_path_str = target_path.decode(tree_encoding)
2160 (symlink_fn or symlink)(contents_str, target_path_str)
2161 else:
2162 (symlink_fn or symlink)(contents, target_path)
2163 else:
2164 if oldstat is not None and stat.S_ISLNK(oldstat.st_mode):
2165 # A symlink left at the target path must not be written through.
2166 # open(..., "wb") would follow it and clobber whatever it points
2167 # at, possibly outside the work tree. Replace it with a fresh
2168 # regular file, like git does on checkout.
2169 _remove_file_with_readonly_handling(target_path)
2170 oldstat = None
2171 if oldstat is not None and oldstat.st_size == len(contents):
2172 with open(target_path, "rb") as f:
2173 unchanged = f.read() == contents
2174 if unchanged:
2175 if not honor_filemode or bool(oldstat.st_mode & stat.S_IXUSR) == bool(
2176 mode & stat.S_IXUSR
2177 ):
2178 return oldstat
2179 os.chmod(target_path, cleanup_mode(mode))
2180 return os.lstat(target_path)
2182 with open(target_path, "wb") as f:
2183 # Write out file
2184 f.write(contents)
2186 if honor_filemode:
2187 # Canonicalize to the permission bits git honors (0o644/0o755).
2188 # The mode comes from a tree/index entry, which for an untrusted
2189 # repository can carry setuid/setgid/sticky or world-writable bits;
2190 # git reduces every regular-file mode to 0o644 or 0o755 on checkout.
2191 os.chmod(target_path, cleanup_mode(mode))
2193 return os.lstat(target_path)
2196INVALID_DOTNAMES = (b".git", b".", b"..", b"")
2199def _normalize_path_element_default(element: bytes) -> bytes:
2200 """Normalize path element for default case-insensitive comparison."""
2201 return element.lower()
2204def _normalize_path_element_ntfs(element: bytes) -> bytes:
2205 """Normalize path element for NTFS filesystem."""
2206 return element.rstrip(b". ").lower()
2209def _normalize_path_element_hfs(element: bytes) -> bytes:
2210 """Normalize path element for HFS+ filesystem."""
2211 import unicodedata
2213 # Decode to Unicode (let UnicodeDecodeError bubble up)
2214 element_str = element.decode("utf-8", errors="strict")
2216 # Remove HFS+ ignorable characters
2217 filtered = "".join(c for c in element_str if ord(c) not in HFS_IGNORABLE_CHARS)
2218 # Normalize to NFD
2219 normalized = unicodedata.normalize("NFD", filtered)
2220 return normalized.lower().encode("utf-8", errors="strict")
2223def get_path_element_normalizer(config: "Config") -> Callable[[bytes], bytes]:
2224 """Get the appropriate path element normalization function based on config.
2226 Args:
2227 config: Repository configuration object
2229 Returns:
2230 Function that normalizes path elements for the configured filesystem
2231 """
2232 import os
2233 import sys
2235 if config.get_boolean(b"core", b"protectNTFS", os.name == "nt"):
2236 return _normalize_path_element_ntfs
2237 elif config.get_boolean(b"core", b"protectHFS", sys.platform == "darwin"):
2238 return _normalize_path_element_hfs
2239 else:
2240 return _normalize_path_element_default
2243def make_path_normalizer(
2244 config: "Config",
2245) -> Callable[[bytes], bytes] | None:
2246 """Build a path normalizer honoring ``core.ignorecase`` and ``core.precomposeunicode``.
2248 The returned callable maps a filesystem-form path to a canonical form
2249 used to match equivalent paths (e.g. ``Foo.txt`` ↔ ``foo.txt`` when
2250 ``core.ignorecase=true``, NFD ↔ NFC when ``core.precomposeunicode=true``).
2251 Returns ``None`` when neither option is active so callers can skip the
2252 comparison entirely.
2253 """
2254 ignorecase = config.get_boolean(b"core", b"ignorecase", False)
2255 precompose = config.get_boolean(b"core", b"precomposeunicode", False)
2256 if not ignorecase and not precompose:
2257 return None
2259 def normalize(path: bytes) -> bytes:
2260 if precompose:
2261 import unicodedata
2263 try:
2264 path = unicodedata.normalize("NFC", path.decode("utf-8")).encode(
2265 "utf-8"
2266 )
2267 except UnicodeDecodeError:
2268 pass
2269 if ignorecase:
2270 path = path.lower()
2271 return path
2273 return normalize
2276class InvalidPathError(Exception):
2277 """Raised when a tree entry's path is unsafe to write to the work tree."""
2279 def __init__(self, path: bytes) -> None:
2280 """Initialize the exception with the offending path."""
2281 self.path = path
2282 super().__init__(f"invalid path {path.decode('utf-8', 'replace')!r}")
2285def validate_path_element_default(element: bytes) -> bool:
2286 """Validate a path element using default rules.
2288 Args:
2289 element: Path element to validate
2291 Returns:
2292 True if path element is valid, False otherwise
2293 """
2294 return _normalize_path_element_default(element) not in INVALID_DOTNAMES
2297def _is_ntfs_dotgit(name: bytes) -> bool:
2298 """Match NTFS-dangerous spellings of ``.git`` at the start of ``name``.
2300 Matches ``.git`` or the 8.3 short name ``git~1`` followed only by
2301 dots/spaces and then the end of the element, a separator, or a ``:``
2302 (an alternate-data-stream marker, as in ``.git::$INDEX_ALLOCATION``).
2303 """
2304 if name[:1] == b".":
2305 if name[1:4].lower() != b"git":
2306 return False
2307 i = 4
2308 elif name[:1].lower() == b"g": # ``git~1`` 8.3 short name
2309 if name[1:3].lower() != b"it" or name[3:5] != b"~1":
2310 return False
2311 i = 5
2312 else:
2313 return False
2315 while i < len(name):
2316 c = name[i : i + 1]
2317 if c == b":":
2318 return True
2319 if c != b"." and c != b" ":
2320 return False
2321 i += 1
2322 return True
2325# Reserved Windows device names. Opening any of these on Windows
2326# resolves to a device rather than a file, regardless of any
2327# extension or trailing dots/spaces (``NUL``, ``NUL.txt``,
2328# ``aux.foo.bar`` all hit the device). They are ordinary filenames
2329# everywhere else, so this is only enforced when running on Windows.
2330RESERVED_WINDOWS_DEVICE_NAMES = frozenset(
2331 [b"con", b"prn", b"aux", b"nul"]
2332 + [b"com%d" % i for i in range(1, 10)]
2333 + [b"lpt%d" % i for i in range(1, 10)]
2334)
2337def _is_reserved_windows_device_name(normalized: bytes) -> bool:
2338 """Match Windows reserved device names regardless of extension."""
2339 # The "stem" is the portion before the first ``.``; Windows
2340 # also strips trailing spaces from that stem when resolving.
2341 stem = normalized.split(b".", 1)[0].rstrip(b" ")
2342 return stem in RESERVED_WINDOWS_DEVICE_NAMES
2345def _has_dos_drive_prefix(name: bytes) -> bool:
2346 """Return True if name begins with a DOS drive-letter prefix like 'C:'.
2348 Any ASCII byte followed by a colon counts, matching the ASCII branch of
2349 C git's win32_has_dos_drive_prefix: subst allows non-alphabetic drives.
2350 """
2351 if len(name) < 2 or name[1:2] != b":":
2352 return False
2353 return name[0] < 0x80
2356def validate_path_element_ntfs(element: bytes) -> bool:
2357 """Validate a path element using NTFS filesystem rules.
2359 Args:
2360 element: Path element to validate
2362 Returns:
2363 True if path element is valid for NTFS, False otherwise
2364 """
2365 # A backslash is a separator on Windows but an ordinary filename
2366 # character on POSIX, so only reject it on Windows.
2367 if os.name == "nt" and b"\\" in element:
2368 return False
2369 # Backslash also separates ``.git`` spellings, so check each segment.
2370 for segment in element.split(b"\\"):
2371 if _is_ntfs_dotgit(segment):
2372 return False
2373 normalized = _normalize_path_element_ntfs(element)
2374 if normalized in INVALID_DOTNAMES:
2375 return False
2376 # Like the backslash check above, a reserved device name is only a
2377 # hazard on Windows; C git confines this to is_valid_win32_path in
2378 # compat/mingw.c, and accepts e.g. ``aux`` on POSIX (issue #2351).
2379 if os.name == "nt" and _is_reserved_windows_device_name(normalized):
2380 return False
2381 return True
2384# HFS+ ignorable Unicode codepoints (from Git's utf8.c)
2385HFS_IGNORABLE_CHARS = {
2386 0x200C, # ZERO WIDTH NON-JOINER
2387 0x200D, # ZERO WIDTH JOINER
2388 0x200E, # LEFT-TO-RIGHT MARK
2389 0x200F, # RIGHT-TO-LEFT MARK
2390 0x202A, # LEFT-TO-RIGHT EMBEDDING
2391 0x202B, # RIGHT-TO-LEFT EMBEDDING
2392 0x202C, # POP DIRECTIONAL FORMATTING
2393 0x202D, # LEFT-TO-RIGHT OVERRIDE
2394 0x202E, # RIGHT-TO-LEFT OVERRIDE
2395 0x206A, # INHIBIT SYMMETRIC SWAPPING
2396 0x206B, # ACTIVATE SYMMETRIC SWAPPING
2397 0x206C, # INHIBIT ARABIC FORM SHAPING
2398 0x206D, # ACTIVATE ARABIC FORM SHAPING
2399 0x206E, # NATIONAL DIGIT SHAPES
2400 0x206F, # NOMINAL DIGIT SHAPES
2401 0xFEFF, # ZERO WIDTH NO-BREAK SPACE
2402}
2405def validate_path_element_hfs(element: bytes) -> bool:
2406 """Validate path element for HFS+ filesystem.
2408 Equivalent to Git's is_hfs_dotgit and related checks.
2409 Uses NFD normalization and ignores HFS+ ignorable characters.
2410 """
2411 try:
2412 normalized = _normalize_path_element_hfs(element)
2413 except UnicodeDecodeError:
2414 # Malformed UTF-8 - be conservative and reject
2415 return False
2417 # Check against invalid names
2418 if normalized in INVALID_DOTNAMES:
2419 return False
2421 # Also check for 8.3 short name
2422 if normalized == b"git~1":
2423 return False
2425 return True
2428def get_path_element_validator(config: "Config") -> Callable[[bytes], bool]:
2429 """Get the path-element validator to use when checking out a tree.
2431 ``core.protectNTFS`` defaults to true on every platform (matching Git's
2432 ``PROTECT_NTFS_DEFAULT=1``) because a repository authored on POSIX can
2433 still be cloned on Windows later; ``core.protectHFS`` defaults to true on
2434 macOS. Both protections are independent and apply together, so on macOS
2435 (where both default on) a path element must satisfy the NTFS and HFS+
2436 checks. With both disabled this falls back to the default validator, which
2437 only refuses ``.git``, ``.`` and ``..``.
2439 Args:
2440 config: Repository configuration object
2442 Returns:
2443 Function that validates a single path element for the configured
2444 filesystem protections.
2445 """
2446 validators: list[Callable[[bytes], bool]] = []
2447 if config.get_boolean(b"core", b"protectNTFS", True):
2448 validators.append(validate_path_element_ntfs)
2449 if config.get_boolean(b"core", b"protectHFS", sys.platform == "darwin"):
2450 validators.append(validate_path_element_hfs)
2451 if not validators:
2452 return validate_path_element_default
2453 if len(validators) == 1:
2454 return validators[0]
2456 def validate_all(element: bytes) -> bool:
2457 return all(validator(element) for validator in validators)
2459 return validate_all
2462def validate_path(
2463 path: bytes,
2464 element_validator: Callable[[bytes], bool] = validate_path_element_default,
2465) -> bool:
2466 """Default path validator that just checks for .git/."""
2467 # A leading drive-letter prefix lets os.path.join discard the work-tree
2468 # root on Windows. Matches C git's verify_path, whose has_dos_drive_prefix
2469 # is a no-op stub on non-Windows.
2470 if os.name == "nt" and _has_dos_drive_prefix(path):
2471 return False
2472 parts = path.split(b"/")
2473 for p in parts:
2474 if not element_validator(p):
2475 return False
2476 else:
2477 return True
2480def verify_leading_dirs(
2481 tree_path: bytes,
2482 safe_prefix: list[bytes],
2483 repo_path: bytes,
2484) -> None:
2485 """Reject writes whose leading path resolves through a symlink.
2487 Callers that materialize many paths in sorted order can pass a shared
2488 ``safe_prefix`` list to cache the deepest chain of directory components
2489 already verified to be real directories (or absent); each call only
2490 ``lstat``s the components that differ from that chain. Callers that only
2491 verify a single path can pass an empty list. Mirrors git's
2492 ``lstat_cache_matchlen`` (see CVE-2021-21300).
2494 Args:
2495 tree_path: Tree-form path (``/``-separated) about to be written.
2496 safe_prefix: Mutable cache of directory components already verified
2497 under ``repo_path``. Updated in place.
2498 repo_path: Filesystem path to the work-tree root.
2500 Raises:
2501 InvalidPathError: If any leading component is a symlink.
2502 """
2503 slash = tree_path.rfind(b"/")
2504 if slash <= 0:
2505 return
2506 components = tree_path[:slash].split(b"/")
2508 common = 0
2509 while (
2510 common < len(safe_prefix)
2511 and common < len(components)
2512 and safe_prefix[common] == components[common]
2513 ):
2514 common += 1
2515 del safe_prefix[common:]
2517 current = repo_path
2518 for part in components[:common]:
2519 current = os.path.join(current, part)
2520 for part in components[common:]:
2521 current = os.path.join(current, part)
2522 try:
2523 st = os.lstat(current)
2524 except FileNotFoundError:
2525 # Anything below here doesn't exist yet; makedirs will create
2526 # it under a verified-real-directory prefix.
2527 break
2528 if stat.S_ISLNK(st.st_mode):
2529 raise InvalidPathError(tree_path)
2530 safe_prefix.append(part)
2533def verify_tree_path(
2534 tree_path: bytes,
2535 validate_path_element: Callable[[bytes], bool],
2536 safe_prefix: list[bytes],
2537 root_path: bytes,
2538) -> None:
2539 """Reject a tree path that is unsafe to write below ``root_path``.
2541 Combines ``validate_path`` and ``verify_leading_dirs``.
2543 Args:
2544 tree_path: Tree-form path (``/``-separated) about to be written.
2545 validate_path_element: Function to validate a single path element.
2546 safe_prefix: Mutable cache shared across calls, see
2547 ``verify_leading_dirs``.
2548 root_path: Filesystem path to the work-tree root.
2550 Raises:
2551 InvalidPathError: If the path has a component rejected by
2552 ``validate_path_element`` or a leading component is a symlink.
2553 """
2554 if not validate_path(tree_path, validate_path_element):
2555 raise InvalidPathError(tree_path)
2556 verify_leading_dirs(tree_path, safe_prefix, root_path)
2559def build_index_from_tree(
2560 root_path: str | bytes,
2561 index_path: str | bytes,
2562 object_store: ObjectContainer,
2563 tree_id: ObjectID,
2564 honor_filemode: bool = True,
2565 validate_path_element: Callable[[bytes], bool] = validate_path_element_default,
2566 symlink_fn: Callable[
2567 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2568 ]
2569 | None = None,
2570 blob_normalizer: "FilterBlobNormalizer | None" = None,
2571 tree_encoding: str = "utf-8",
2572 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
2573) -> None:
2574 """Generate and materialize index from a tree.
2576 Args:
2577 tree_id: Tree to materialize
2578 root_path: Target dir for materialized index files
2579 index_path: Target path for generated index
2580 object_store: Non-empty object store holding tree contents
2581 honor_filemode: An optional flag to honor core.filemode setting in
2582 config file, default is core.filemode=True, change executable bit
2583 validate_path_element: Function to validate path elements to check
2584 out; default just refuses .git and .. directories.
2585 symlink_fn: Function to use for creating symlinks
2586 blob_normalizer: An optional BlobNormalizer to use for converting line
2587 endings when writing blobs to the working directory.
2588 tree_encoding: Encoding used for tree paths (default: utf-8)
2589 object_format: Object format of the repository the index belongs to
2591 Note: existing index is wiped and contents are not merged
2592 in a working dir. Suitable only for fresh clones.
2593 """
2594 index = Index(index_path, read=False, object_format=object_format)
2595 if not isinstance(root_path, bytes):
2596 root_path = os.fsencode(root_path)
2598 # Cache of leading directory components already verified to be real
2599 # directories, shared across the sorted iteration. See verify_leading_dirs.
2600 safe_prefix: list[bytes] = []
2602 for entry in iter_tree_contents(object_store, tree_id):
2603 assert (
2604 entry.path is not None and entry.mode is not None and entry.sha is not None
2605 )
2606 # Validate as we go and abort on the first invalid path, leaving any
2607 # files already written in place. This also refuses a leading path
2608 # that resolves through a symlink materialized by an earlier entry.
2609 verify_tree_path(entry.path, validate_path_element, safe_prefix, root_path)
2610 full_path = _tree_to_fs_path(root_path, entry.path, tree_encoding)
2612 if not os.path.exists(os.path.dirname(full_path)):
2613 os.makedirs(os.path.dirname(full_path))
2615 # TODO(jelmer): Merge new index into working tree
2616 if S_ISGITLINK(entry.mode):
2617 if not os.path.isdir(full_path):
2618 os.mkdir(full_path)
2619 st = os.lstat(full_path)
2620 # TODO(jelmer): record and return submodule paths
2621 else:
2622 obj = object_store[entry.sha]
2623 assert isinstance(obj, Blob)
2624 # Apply blob normalization for checkout if normalizer is provided
2625 if blob_normalizer is not None:
2626 obj = blob_normalizer.checkout_normalize(obj, entry.path)
2627 st = build_file_from_blob(
2628 obj,
2629 entry.mode,
2630 full_path,
2631 honor_filemode=honor_filemode,
2632 tree_encoding=tree_encoding,
2633 symlink_fn=symlink_fn,
2634 )
2636 # Add file to index
2637 if not honor_filemode or S_ISGITLINK(entry.mode):
2638 # we can not use tuple slicing to build a new tuple,
2639 # because on windows that will convert the times to
2640 # longs, which causes errors further along
2641 st_tuple = (
2642 entry.mode,
2643 st.st_ino,
2644 st.st_dev,
2645 st.st_nlink,
2646 st.st_uid,
2647 st.st_gid,
2648 st.st_size,
2649 st.st_atime,
2650 st.st_mtime,
2651 st.st_ctime,
2652 )
2653 st = st.__class__(st_tuple)
2654 # default to a stage 0 index entry (normal)
2655 # when reading from the filesystem
2656 index[entry.path] = index_entry_from_stat(st, entry.sha)
2658 index.write()
2661def blob_from_path_and_mode(
2662 fs_path: bytes, mode: int, tree_encoding: str = "utf-8"
2663) -> Blob:
2664 """Create a blob from a path and a stat object.
2666 Args:
2667 fs_path: Full file system path to file
2668 mode: File mode
2669 tree_encoding: Encoding to use for tree contents
2670 Returns: A `Blob` object
2671 """
2672 assert isinstance(fs_path, bytes)
2673 blob = Blob()
2674 if stat.S_ISLNK(mode):
2675 if sys.platform == "win32":
2676 # os.readlink on Python3 on Windows requires a unicode string.
2677 blob.data = os.readlink(os.fsdecode(fs_path)).encode(tree_encoding)
2678 else:
2679 blob.data = os.readlink(fs_path)
2680 else:
2681 with open(fs_path, "rb") as f:
2682 blob.data = f.read()
2683 return blob
2686def blob_from_path_and_stat(
2687 fs_path: bytes, st: os.stat_result, tree_encoding: str = "utf-8"
2688) -> Blob:
2689 """Create a blob from a path and a stat object.
2691 Args:
2692 fs_path: Full file system path to file
2693 st: A stat object
2694 tree_encoding: Encoding to use for tree contents
2695 Returns: A `Blob` object
2696 """
2697 return blob_from_path_and_mode(fs_path, st.st_mode, tree_encoding)
2700def read_submodule_head(path: str | bytes) -> bytes | None:
2701 """Read the head commit of a submodule.
2703 Args:
2704 path: path to the submodule
2705 Returns: HEAD sha, None if not a valid head/repository
2706 """
2707 from .errors import NotGitRepository
2708 from .repo import Repo
2710 # Repo currently expects a "str", so decode if necessary.
2711 # TODO(jelmer): Perhaps move this into Repo() ?
2712 if not isinstance(path, str):
2713 path = os.fsdecode(path)
2714 try:
2715 repo = Repo(path)
2716 except NotGitRepository:
2717 return None
2718 try:
2719 return repo.head()
2720 except KeyError:
2721 return None
2724def _has_directory_changed(tree_path: bytes, entry: IndexEntry) -> bool:
2725 """Check if a directory has changed after getting an error.
2727 When handling an error trying to create a blob from a path, call this
2728 function. It will check if the path is a directory. If it's a directory
2729 and a submodule, check the submodule head to see if it's has changed. If
2730 not, consider the file as changed as Git tracked a file and not a
2731 directory.
2733 Return true if the given path should be considered as changed and False
2734 otherwise or if the path is not a directory.
2735 """
2736 # This is actually a directory
2737 if os.path.exists(os.path.join(tree_path, b".git")):
2738 # Submodule
2739 head = read_submodule_head(tree_path)
2740 if entry.sha != head:
2741 return True
2742 else:
2743 # The file was changed to a directory, so consider it removed.
2744 return True
2746 return False
2749os_sep_bytes = os.sep.encode("ascii")
2752def _ensure_parent_dir_exists(full_path: bytes) -> None:
2753 """Ensure parent directory exists, checking no parent is a file."""
2754 parent_dir = os.path.dirname(full_path)
2755 if parent_dir and not os.path.exists(parent_dir):
2756 # Walk up the directory tree to find the first existing parent
2757 current = parent_dir
2758 parents_to_check: list[bytes] = []
2760 while current and not os.path.exists(current):
2761 parents_to_check.insert(0, current)
2762 new_parent = os.path.dirname(current)
2763 if new_parent == current:
2764 # Reached the root or can't go up further
2765 break
2766 current = new_parent
2768 # Check if the existing parent (if any) is a directory
2769 if current and os.path.exists(current) and not os.path.isdir(current):
2770 raise OSError(
2771 f"Cannot create directory, parent path is a file: {current!r}"
2772 )
2774 # Now check each parent we need to create isn't blocked by an existing file
2775 for parent_path in parents_to_check:
2776 if os.path.exists(parent_path) and not os.path.isdir(parent_path):
2777 raise OSError(
2778 f"Cannot create directory, parent path is a file: {parent_path!r}"
2779 )
2781 os.makedirs(parent_dir)
2784def _remove_file_with_readonly_handling(path: bytes) -> None:
2785 """Remove a file, handling read-only files on Windows.
2787 Args:
2788 path: Path to the file to remove
2789 """
2790 try:
2791 os.unlink(path)
2792 except PermissionError:
2793 # On Windows, remove read-only attribute and retry
2794 if sys.platform == "win32":
2795 os.chmod(path, stat.S_IWRITE | stat.S_IREAD)
2796 os.unlink(path)
2797 else:
2798 raise
2801def _remove_empty_parents(path: bytes, stop_at: bytes) -> None:
2802 """Remove empty parent directories up to stop_at."""
2803 parent = os.path.dirname(path)
2804 while parent and parent != stop_at:
2805 try:
2806 os.rmdir(parent)
2807 parent = os.path.dirname(parent)
2808 except FileNotFoundError:
2809 # Directory doesn't exist - stop trying
2810 break
2811 except OSError as e:
2812 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
2813 # Directory not empty - stop trying
2814 break
2815 raise
2818def _check_symlink_matches(
2819 full_path: bytes, repo_object_store: "BaseObjectStore", entry_sha: ObjectID
2820) -> bool:
2821 """Check if symlink target matches expected target.
2823 Returns True if symlink matches, False if it doesn't match.
2824 """
2825 try:
2826 current_target = os.readlink(full_path)
2827 blob_obj = repo_object_store[entry_sha]
2828 expected_target = blob_obj.as_raw_string()
2829 if isinstance(current_target, str):
2830 current_target = current_target.encode()
2831 return current_target == expected_target
2832 except FileNotFoundError:
2833 # Symlink doesn't exist
2834 return False
2835 except OSError as e:
2836 if e.errno == errno.EINVAL:
2837 # Not a symlink
2838 return False
2839 raise
2842def _check_file_matches(
2843 repo_object_store: "BaseObjectStore",
2844 full_path: bytes,
2845 entry_sha: ObjectID,
2846 entry_mode: int,
2847 current_stat: os.stat_result,
2848 honor_filemode: bool,
2849 blob_normalizer: "FilterBlobNormalizer | None" = None,
2850 tree_path: bytes | None = None,
2851) -> bool:
2852 """Check if a file on disk matches the expected git object.
2854 Returns True if file matches, False if it doesn't match.
2855 """
2856 # Check mode first (if honor_filemode is True)
2857 if honor_filemode:
2858 current_mode = stat.S_IMODE(current_stat.st_mode)
2859 expected_mode = stat.S_IMODE(entry_mode)
2861 # For regular files, only check the user executable bit, not group/other permissions
2862 # This matches Git's behavior where umask differences don't count as modifications
2863 if stat.S_ISREG(current_stat.st_mode):
2864 # Normalize regular file modes to ignore group/other write permissions
2865 current_mode_normalized = (
2866 current_mode & 0o755
2867 ) # Keep only user rwx and all read+execute
2868 expected_mode_normalized = expected_mode & 0o755
2870 # For Git compatibility, regular files should be either 644 or 755
2871 if expected_mode_normalized not in (0o644, 0o755):
2872 expected_mode_normalized = 0o644 # Default for regular files
2873 if current_mode_normalized not in (0o644, 0o755):
2874 # Determine if it should be executable based on user execute bit
2875 if current_mode & 0o100: # User execute bit is set
2876 current_mode_normalized = 0o755
2877 else:
2878 current_mode_normalized = 0o644
2880 if current_mode_normalized != expected_mode_normalized:
2881 return False
2882 else:
2883 # For non-regular files (symlinks, etc.), check mode exactly
2884 if current_mode != expected_mode:
2885 return False
2887 # If mode matches (or we don't care), check content via size first
2888 blob_obj = repo_object_store[entry_sha]
2889 if current_stat.st_size != blob_obj.raw_length():
2890 return False
2892 # Size matches, check actual content
2893 try:
2894 with open(full_path, "rb") as f:
2895 current_content = f.read()
2896 expected_content = blob_obj.as_raw_string()
2897 if blob_normalizer and tree_path is not None:
2898 assert isinstance(blob_obj, Blob)
2899 normalized_blob = blob_normalizer.checkout_normalize(
2900 blob_obj, tree_path
2901 )
2902 expected_content = normalized_blob.as_raw_string()
2903 return current_content == expected_content
2904 except (FileNotFoundError, PermissionError, IsADirectoryError):
2905 return False
2908def _transition_to_submodule(
2909 repo: "Repo",
2910 path: bytes,
2911 full_path: bytes,
2912 current_stat: os.stat_result | None,
2913 entry: IndexEntry | TreeEntry,
2914 index: Index,
2915) -> None:
2916 """Transition any type to submodule."""
2917 from .submodule import ensure_submodule_placeholder
2919 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2920 # Already a directory, just ensure .git file exists
2921 ensure_submodule_placeholder(repo, path)
2922 else:
2923 # Remove whatever is there and create submodule
2924 if current_stat is not None:
2925 _remove_file_with_readonly_handling(full_path)
2926 ensure_submodule_placeholder(repo, path)
2928 st = os.lstat(full_path)
2929 assert entry.sha is not None
2930 index[path] = index_entry_from_stat(st, entry.sha)
2933def _transition_to_file(
2934 object_store: "BaseObjectStore",
2935 path: bytes,
2936 full_path: bytes,
2937 current_stat: os.stat_result | None,
2938 entry: IndexEntry | TreeEntry,
2939 index: Index,
2940 honor_filemode: bool,
2941 symlink_fn: Callable[
2942 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2943 ]
2944 | None,
2945 blob_normalizer: "FilterBlobNormalizer | None",
2946 tree_encoding: str = "utf-8",
2947) -> None:
2948 """Transition any type to regular file or symlink."""
2949 assert entry.sha is not None and entry.mode is not None
2950 # Check if we need to update
2951 if (
2952 current_stat is not None
2953 and stat.S_ISREG(current_stat.st_mode)
2954 and not stat.S_ISLNK(entry.mode)
2955 ):
2956 # File to file - check if update needed
2957 file_matches = _check_file_matches(
2958 object_store,
2959 full_path,
2960 entry.sha,
2961 entry.mode,
2962 current_stat,
2963 honor_filemode,
2964 blob_normalizer,
2965 path,
2966 )
2967 needs_update = not file_matches
2968 elif (
2969 current_stat is not None
2970 and stat.S_ISLNK(current_stat.st_mode)
2971 and stat.S_ISLNK(entry.mode)
2972 ):
2973 # Symlink to symlink - check if update needed
2974 symlink_matches = _check_symlink_matches(full_path, object_store, entry.sha)
2975 needs_update = not symlink_matches
2976 else:
2977 needs_update = True
2979 if not needs_update:
2980 # Just update index - current_stat should always be valid here since we're not updating
2981 assert current_stat is not None
2982 index[path] = index_entry_from_stat(current_stat, entry.sha)
2983 return
2985 # Remove existing entry if needed
2986 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2987 # Remove directory
2988 dir_contents = set(os.listdir(full_path))
2989 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
2991 if git_file_name in dir_contents:
2992 if dir_contents != {git_file_name}:
2993 raise IsADirectoryError(
2994 f"Cannot replace submodule with untracked files: {full_path!r}"
2995 )
2996 shutil.rmtree(full_path)
2997 else:
2998 try:
2999 os.rmdir(full_path)
3000 except OSError as e:
3001 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
3002 raise IsADirectoryError(
3003 f"Cannot replace non-empty directory with file: {full_path!r}"
3004 )
3005 raise
3006 elif current_stat is not None:
3007 _remove_file_with_readonly_handling(full_path)
3009 # Ensure parent directory exists
3010 _ensure_parent_dir_exists(full_path)
3012 # Write the file
3013 blob_obj = object_store[entry.sha]
3014 assert isinstance(blob_obj, Blob)
3015 if blob_normalizer:
3016 blob_obj = blob_normalizer.checkout_normalize(blob_obj, path)
3017 st = build_file_from_blob(
3018 blob_obj,
3019 entry.mode,
3020 full_path,
3021 honor_filemode=honor_filemode,
3022 tree_encoding=tree_encoding,
3023 symlink_fn=symlink_fn,
3024 )
3025 index[path] = index_entry_from_stat(st, entry.sha)
3028def _transition_to_absent(
3029 repo: "Repo",
3030 path: bytes,
3031 full_path: bytes,
3032 current_stat: os.stat_result | None,
3033 index: Index,
3034) -> None:
3035 """Remove any type of entry."""
3036 if current_stat is None:
3037 return
3039 if stat.S_ISDIR(current_stat.st_mode):
3040 # Check if it's a submodule directory
3041 dir_contents = set(os.listdir(full_path))
3042 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
3044 if git_file_name in dir_contents and dir_contents == {git_file_name}:
3045 shutil.rmtree(full_path)
3046 else:
3047 try:
3048 os.rmdir(full_path)
3049 except OSError as e:
3050 if e.errno not in (errno.ENOTEMPTY, errno.EEXIST):
3051 raise
3052 else:
3053 _remove_file_with_readonly_handling(full_path)
3055 try:
3056 del index[path]
3057 except KeyError:
3058 pass
3060 # Try to remove empty parent directories
3061 _remove_empty_parents(
3062 full_path, repo.path if isinstance(repo.path, bytes) else repo.path.encode()
3063 )
3066def detect_case_only_renames(
3067 changes: Sequence["TreeChange"],
3068 config: "Config",
3069) -> list["TreeChange"]:
3070 """Detect and transform case-only renames in a list of tree changes.
3072 This function identifies file renames that only differ in case (e.g.,
3073 README.txt -> readme.txt) and transforms matching ADD/DELETE pairs into
3074 CHANGE_RENAME operations. It uses filesystem-appropriate path normalization
3075 based on the repository configuration.
3077 Args:
3078 changes: List of TreeChange objects representing file changes
3079 config: Repository configuration object
3081 Returns:
3082 New list of TreeChange objects with case-only renames converted to CHANGE_RENAME
3083 """
3084 from .diff_tree import (
3085 CHANGE_ADD,
3086 CHANGE_COPY,
3087 CHANGE_DELETE,
3088 CHANGE_MODIFY,
3089 CHANGE_RENAME,
3090 TreeChange,
3091 )
3093 # Build dictionaries of old and new paths with their normalized forms
3094 old_paths_normalized = {}
3095 new_paths_normalized = {}
3096 old_changes = {} # Map from old path to change object
3097 new_changes = {} # Map from new path to change object
3099 # Get the appropriate normalizer based on config
3100 normalize_func = get_path_element_normalizer(config)
3102 def normalize_path(path: bytes) -> bytes:
3103 """Normalize entire path using element normalization."""
3104 return b"/".join(normalize_func(part) for part in path.split(b"/"))
3106 # Pre-normalize all paths once to avoid repeated normalization
3107 for change in changes:
3108 if change.type == CHANGE_DELETE and change.old:
3109 assert change.old.path is not None
3110 try:
3111 normalized = normalize_path(change.old.path)
3112 except UnicodeDecodeError:
3113 logger.warning(
3114 "Skipping case-only rename detection for path with invalid UTF-8: %r",
3115 change.old.path,
3116 )
3117 else:
3118 old_paths_normalized[normalized] = change.old.path
3119 old_changes[change.old.path] = change
3120 elif change.type == CHANGE_RENAME and change.old:
3121 assert change.old.path is not None
3122 # Treat RENAME as DELETE + ADD for case-only detection
3123 try:
3124 normalized = normalize_path(change.old.path)
3125 except UnicodeDecodeError:
3126 logger.warning(
3127 "Skipping case-only rename detection for path with invalid UTF-8: %r",
3128 change.old.path,
3129 )
3130 else:
3131 old_paths_normalized[normalized] = change.old.path
3132 old_changes[change.old.path] = change
3134 if (
3135 change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY)
3136 and change.new
3137 ):
3138 assert change.new.path is not None
3139 try:
3140 normalized = normalize_path(change.new.path)
3141 except UnicodeDecodeError:
3142 logger.warning(
3143 "Skipping case-only rename detection for path with invalid UTF-8: %r",
3144 change.new.path,
3145 )
3146 else:
3147 new_paths_normalized[normalized] = change.new.path
3148 new_changes[change.new.path] = change
3150 # Find case-only renames and transform changes
3151 case_only_renames = set()
3152 new_rename_changes = []
3154 for norm_path, old_path in old_paths_normalized.items():
3155 if norm_path in new_paths_normalized:
3156 new_path = new_paths_normalized[norm_path]
3157 if old_path != new_path:
3158 # Found a case-only rename
3159 old_change = old_changes[old_path]
3160 new_change = new_changes[new_path]
3162 # Create a CHANGE_RENAME to replace the DELETE and ADD/MODIFY pair
3163 if new_change.type == CHANGE_ADD:
3164 # Simple case: DELETE + ADD becomes RENAME
3165 rename_change = TreeChange(
3166 CHANGE_RENAME, old_change.old, new_change.new
3167 )
3168 else:
3169 # Complex case: DELETE + MODIFY becomes RENAME
3170 # Use the old file from DELETE and new file from MODIFY
3171 rename_change = TreeChange(
3172 CHANGE_RENAME, old_change.old, new_change.new
3173 )
3175 new_rename_changes.append(rename_change)
3177 # Mark the old changes for removal
3178 case_only_renames.add(old_change)
3179 case_only_renames.add(new_change)
3181 # Return new list with original ADD/DELETE changes replaced by renames
3182 result = [change for change in changes if change not in case_only_renames]
3183 result.extend(new_rename_changes)
3184 return result
3187def update_working_tree(
3188 repo: "Repo",
3189 old_tree_id: bytes | None,
3190 new_tree_id: bytes,
3191 change_iterator: Iterator["TreeChange"],
3192 honor_filemode: bool = True,
3193 validate_path_element: Callable[[bytes], bool] | None = None,
3194 symlink_fn: Callable[
3195 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
3196 ]
3197 | None = None,
3198 force_remove_untracked: bool = False,
3199 blob_normalizer: "FilterBlobNormalizer | None" = None,
3200 tree_encoding: str = "utf-8",
3201 allow_overwrite_modified: bool = False,
3202 *,
3203 config: "Config | None" = None,
3204) -> None:
3205 """Update the working tree and index to match a new tree.
3207 This function handles:
3208 - Adding new files
3209 - Updating modified files
3210 - Removing deleted files
3211 - Cleaning up empty directories
3213 Args:
3214 repo: Repository object
3215 old_tree_id: SHA of the tree before the update
3216 new_tree_id: SHA of the tree to update to
3217 change_iterator: Iterator of TreeChange objects to apply
3218 honor_filemode: An optional flag to honor core.filemode setting
3219 validate_path_element: Function to validate path elements to check out.
3220 If None, derived from ``config`` so that ``core.protectNTFS`` and
3221 ``core.protectHFS`` are honored by default.
3222 symlink_fn: Function to use for creating symlinks
3223 force_remove_untracked: If True, remove files that exist in working
3224 directory but not in target tree, even if old_tree_id is None
3225 blob_normalizer: An optional BlobNormalizer to use for converting line
3226 endings when writing blobs to the working directory.
3227 tree_encoding: Encoding used for tree paths (default: utf-8)
3228 allow_overwrite_modified: If False, raise an error when attempting to
3229 overwrite files that have been modified compared to old_tree_id
3230 config: Repository configuration. If None, falls back to
3231 ``repo.get_config_stack()``.
3232 """
3233 from .diff_tree import (
3234 CHANGE_ADD,
3235 CHANGE_COPY,
3236 CHANGE_DELETE,
3237 CHANGE_MODIFY,
3238 CHANGE_RENAME,
3239 CHANGE_UNCHANGED,
3240 )
3242 if force_remove_untracked:
3243 import warnings
3245 warnings.warn(
3246 "force_remove_untracked is a no-op and will be removed in a future release",
3247 DeprecationWarning,
3248 )
3249 repo_path = repo.path if isinstance(repo.path, bytes) else repo.path.encode()
3250 if config is None:
3251 config = repo.get_config_stack()
3253 if validate_path_element is None:
3254 # Derive the validator from config so callers cannot accidentally
3255 # skip ``core.protectNTFS``/``core.protectHFS`` enforcement by
3256 # omitting this argument.
3257 validate_path_element = get_path_element_validator(config)
3259 index = repo.open_index(config=config)
3261 # Convert iterator to list since we need multiple passes
3262 changes = list(change_iterator)
3264 # Transform case-only renames on case-insensitive filesystems
3265 import platform
3267 default_ignore_case = platform.system() in ("Windows", "Darwin")
3268 config = repo.get_config()
3269 ignore_case = config.get_boolean((b"core",), b"ignorecase", default_ignore_case)
3271 if ignore_case:
3272 config = repo.get_config()
3273 changes = detect_case_only_renames(changes, config)
3275 # Check for path conflicts where files need to become directories
3276 paths_becoming_dirs = set()
3277 for change in changes:
3278 if change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY):
3279 assert change.new is not None
3280 path = change.new.path
3281 assert path is not None
3282 if b"/" in path: # This is a file inside a directory
3283 # Check if any parent path exists as a file in the old tree or changes
3284 parts = path.split(b"/")
3285 for i in range(1, len(parts)):
3286 parent = b"/".join(parts[:i])
3287 # See if this parent path is being deleted (was a file, becoming a dir)
3288 for other_change in changes:
3289 if (
3290 other_change.type == CHANGE_DELETE
3291 and other_change.old
3292 and other_change.old.path == parent
3293 ):
3294 paths_becoming_dirs.add(parent)
3296 # Check if any path that needs to become a directory has been modified
3297 for path in paths_becoming_dirs:
3298 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3299 try:
3300 current_stat = os.lstat(full_path)
3301 except FileNotFoundError:
3302 continue # File doesn't exist, nothing to check
3303 except OSError as e:
3304 raise OSError(
3305 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3306 ) from e
3308 if stat.S_ISREG(current_stat.st_mode):
3309 # Find the old entry for this path
3310 old_change = None
3311 for change in changes:
3312 if (
3313 change.type == CHANGE_DELETE
3314 and change.old
3315 and change.old.path == path
3316 ):
3317 old_change = change
3318 break
3320 if old_change:
3321 # Check if file has been modified
3322 assert old_change.old is not None
3323 assert (
3324 old_change.old.sha is not None and old_change.old.mode is not None
3325 )
3326 file_matches = _check_file_matches(
3327 repo.object_store,
3328 full_path,
3329 old_change.old.sha,
3330 old_change.old.mode,
3331 current_stat,
3332 honor_filemode,
3333 blob_normalizer,
3334 path,
3335 )
3336 if not file_matches:
3337 raise OSError(
3338 f"Cannot replace modified file with directory: {path!r}"
3339 )
3341 # Check for uncommitted modifications before making any changes
3342 if not allow_overwrite_modified and old_tree_id:
3343 for change in changes:
3344 # Only check files that are being modified or deleted
3345 if change.type in (CHANGE_MODIFY, CHANGE_DELETE) and change.old:
3346 path = change.old.path
3347 assert path is not None
3348 if not validate_path(path, validate_path_element):
3349 continue
3351 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3352 try:
3353 current_stat = os.lstat(full_path)
3354 except FileNotFoundError:
3355 continue # File doesn't exist, nothing to check
3356 except OSError as e:
3357 raise OSError(
3358 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3359 ) from e
3361 if stat.S_ISREG(current_stat.st_mode):
3362 # Check if working tree file differs from old tree
3363 assert change.old.sha is not None and change.old.mode is not None
3364 file_matches = _check_file_matches(
3365 repo.object_store,
3366 full_path,
3367 change.old.sha,
3368 change.old.mode,
3369 current_stat,
3370 honor_filemode,
3371 blob_normalizer,
3372 path,
3373 )
3374 if not file_matches:
3375 from .errors import WorkingTreeModifiedError
3377 raise WorkingTreeModifiedError(
3378 f"Your local changes to '{path.decode('utf-8', errors='replace')}' "
3379 f"would be overwritten by checkout. "
3380 f"Please commit your changes or stash them before you switch branches."
3381 )
3383 # Apply the changes
3384 for change in changes:
3385 if change.type in (CHANGE_DELETE, CHANGE_RENAME):
3386 # Remove file/directory
3387 assert change.old is not None and change.old.path is not None
3388 path = change.old.path
3389 if not validate_path(path, validate_path_element):
3390 continue
3392 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3393 try:
3394 delete_stat: os.stat_result | None = os.lstat(full_path)
3395 except FileNotFoundError:
3396 delete_stat = None
3397 except OSError as e:
3398 raise OSError(
3399 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3400 ) from e
3402 _transition_to_absent(repo, path, full_path, delete_stat, index)
3404 if change.type in (
3405 CHANGE_ADD,
3406 CHANGE_MODIFY,
3407 CHANGE_UNCHANGED,
3408 CHANGE_COPY,
3409 CHANGE_RENAME,
3410 ):
3411 # Add or modify file
3412 assert (
3413 change.new is not None
3414 and change.new.path is not None
3415 and change.new.mode is not None
3416 )
3417 path = change.new.path
3418 # Validate as we go and abort on the first invalid path,
3419 # leaving any changes already applied in place.
3420 verify_tree_path(path, validate_path_element, [], repo_path)
3421 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3422 try:
3423 modify_stat: os.stat_result | None = os.lstat(full_path)
3424 except FileNotFoundError:
3425 modify_stat = None
3426 except OSError as e:
3427 raise OSError(
3428 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3429 ) from e
3431 if S_ISGITLINK(change.new.mode):
3432 _transition_to_submodule(
3433 repo, path, full_path, modify_stat, change.new, index
3434 )
3435 else:
3436 _transition_to_file(
3437 repo.object_store,
3438 path,
3439 full_path,
3440 modify_stat,
3441 change.new,
3442 index,
3443 honor_filemode,
3444 symlink_fn,
3445 blob_normalizer,
3446 tree_encoding,
3447 )
3449 index.write()
3452def _stat_matches_entry(
3453 st: os.stat_result, entry: IndexEntry, trust_ctime: bool = True
3454) -> bool:
3455 """Check if filesystem stat matches index entry stat.
3457 This is used to determine if a file might have changed without reading its content.
3458 Git uses this optimization to avoid expensive filter operations on unchanged files.
3460 Args:
3461 st: Filesystem stat result
3462 entry: Index entry to compare against
3463 trust_ctime: If True, also check ctime (default: True, matching Git behavior)
3464 Returns: True if stat matches and file is likely unchanged
3465 """
3466 # Compare change time (ctime) if trust_ctime is enabled
3467 if trust_ctime:
3468 # Get entry ctime with nanosecond precision if available
3469 if isinstance(entry.ctime, tuple):
3470 entry_ctime_sec = entry.ctime[0]
3471 entry_ctime_nsec = entry.ctime[1]
3472 else:
3473 entry_ctime_sec = int(entry.ctime)
3474 entry_ctime_nsec = 0
3476 if hasattr(st, "st_ctime_ns"):
3477 # Use nanosecond precision when available
3478 st_ctime_nsec = st.st_ctime_ns
3479 entry_ctime_nsec_total = entry_ctime_sec * 1_000_000_000 + entry_ctime_nsec
3480 if st_ctime_nsec != entry_ctime_nsec_total:
3481 return False
3482 else:
3483 # Fall back to second precision
3484 if int(st.st_ctime) != entry_ctime_sec:
3485 return False
3487 # Get entry mtime with nanosecond precision if available
3488 if isinstance(entry.mtime, tuple):
3489 entry_mtime_sec = entry.mtime[0]
3490 entry_mtime_nsec = entry.mtime[1]
3491 else:
3492 entry_mtime_sec = int(entry.mtime)
3493 entry_mtime_nsec = 0
3495 # Compare modification time with nanosecond precision if available
3496 # This is important for fast workflows (e.g., stash) where files can be
3497 # modified multiple times within the same second
3498 if hasattr(st, "st_mtime_ns"):
3499 # Use nanosecond precision when available
3500 st_mtime_nsec = st.st_mtime_ns
3501 entry_mtime_nsec_total = entry_mtime_sec * 1_000_000_000 + entry_mtime_nsec
3502 if st_mtime_nsec != entry_mtime_nsec_total:
3503 return False
3504 else:
3505 # Fall back to second precision
3506 if int(st.st_mtime) != entry_mtime_sec:
3507 return False
3509 # Compare file size
3510 if st.st_size != entry.size:
3511 return False
3513 # If all checks pass, file is likely unchanged
3514 return True
3517def _check_entry_for_changes(
3518 tree_path: bytes,
3519 entry: IndexEntry | ConflictedIndexEntry,
3520 root_path: bytes,
3521 filter_blob_callback: Callable[[Blob, bytes], Blob] | None = None,
3522 trust_ctime: bool = True,
3523 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
3524 refresh_stat: list[tuple[bytes, os.stat_result]] | None = None,
3525) -> bytes | None:
3526 """Check a single index entry for changes.
3528 Args:
3529 tree_path: Path in the tree
3530 entry: Index entry to check
3531 root_path: Root filesystem path
3532 filter_blob_callback: Optional callback to filter blobs
3533 trust_ctime: If True, use ctime for change detection (default: True)
3534 object_format: Object format to hash the working tree blob with
3535 refresh_stat: If provided, entries with drifted stat but matching
3536 content have their (tree_path, stat_result) appended so the
3537 caller can update the index stat cache in a single pass.
3538 Returns: tree_path if changed, None otherwise
3539 """
3540 if isinstance(entry, ConflictedIndexEntry):
3541 # Conflicted files are always unstaged
3542 return tree_path
3544 full_path = _tree_to_fs_path(root_path, tree_path)
3545 try:
3546 st = os.lstat(full_path)
3547 if stat.S_ISDIR(st.st_mode):
3548 if _has_directory_changed(tree_path, entry):
3549 return tree_path
3550 return None
3552 if not stat.S_ISREG(st.st_mode) and not stat.S_ISLNK(st.st_mode):
3553 return None
3555 # Optimization: If stat matches index entry (mtime and size unchanged),
3556 # we can skip reading and filtering the file entirely. This is a significant
3557 # performance improvement for repositories with many unchanged files.
3558 # Even with filters (e.g., LFS), if the file hasn't been modified (stat unchanged),
3559 # the filter output would be the same, so we can safely skip the expensive
3560 # filter operation. This addresses performance issues with LFS repositories
3561 # where filter operations can be very slow.
3562 if _stat_matches_entry(st, entry, trust_ctime):
3563 return None
3565 blob = blob_from_path_and_stat(full_path, st)
3567 if filter_blob_callback is not None:
3568 blob = filter_blob_callback(blob, tree_path)
3569 except FileNotFoundError:
3570 # The file was removed, so we assume that counts as
3571 # different from whatever file used to exist.
3572 return tree_path
3573 else:
3574 if blob.get_id(object_format) != entry.sha:
3575 return tree_path
3576 if refresh_stat is not None:
3577 refresh_stat.append((tree_path, st))
3578 return None
3581def get_unstaged_changes(
3582 index: Index,
3583 root_path: str | bytes,
3584 filter_blob_callback: Callable[..., Any] | None = None,
3585 preload_index: bool = False,
3586 trust_ctime: bool = True,
3587 max_stat: int | None = None,
3588 refresh_stat: list[tuple[bytes, os.stat_result]] | None = None,
3589) -> Generator[bytes, None, None]:
3590 """Walk through an index and check for differences against working tree.
3592 Args:
3593 index: index to check
3594 root_path: path in which to find files
3595 filter_blob_callback: Optional callback to filter blobs
3596 preload_index: If True, use parallel threads to check files (requires threading support)
3597 trust_ctime: If True, use ctime for change detection (default: True)
3598 max_stat: If set, limit the number of stat operations performed.
3599 When the limit is reached, remaining files are assumed unchanged.
3600 refresh_stat: If provided, entries whose stat has drifted but whose
3601 content is unchanged are appended as ``(tree_path, stat_result)``
3602 pairs so the caller can update the index stat cache without a
3603 second pass.
3604 Returns: iterator over paths with unstaged changes
3605 """
3606 # For each entry in the index check the sha1 & ensure not staged
3607 if not isinstance(root_path, bytes):
3608 root_path = os.fsencode(root_path)
3610 stat_count = 0
3612 if preload_index:
3613 # Use parallel processing for better performance on slow filesystems
3614 try:
3615 import multiprocessing
3616 from concurrent.futures import ThreadPoolExecutor
3617 except ImportError:
3618 # If threading is not available, fall back to serial processing
3619 preload_index = False
3620 else:
3621 # Collect all entries first
3622 entries = list(index.iteritems())
3624 if max_stat is not None:
3625 # When max_stat is set, limit the entries we process
3626 entries = entries[:max_stat]
3628 # Use number of CPUs but cap at 8 threads to avoid overhead
3629 num_workers = min(multiprocessing.cpu_count(), 8)
3631 # Per-thread refresh accumulators so appends stay lock-free,
3632 # then merged into the caller's list once workers finish.
3633 worker_refresh: list[list[tuple[bytes, os.stat_result]] | None] = [
3634 [] if refresh_stat is not None else None for _ in entries
3635 ]
3637 # Process entries in parallel
3638 with ThreadPoolExecutor(max_workers=num_workers) as executor:
3639 # Submit all tasks
3640 futures = [
3641 executor.submit(
3642 _check_entry_for_changes,
3643 tree_path,
3644 entry,
3645 root_path,
3646 filter_blob_callback,
3647 trust_ctime,
3648 index.object_format,
3649 worker_refresh[i],
3650 )
3651 for i, (tree_path, entry) in enumerate(entries)
3652 ]
3654 # Yield results as they complete
3655 for future in futures:
3656 result = future.result()
3657 if result is not None:
3658 yield result
3660 if refresh_stat is not None:
3661 for bucket in worker_refresh:
3662 if bucket:
3663 refresh_stat.extend(bucket)
3665 if not preload_index:
3666 # Serial processing
3667 for tree_path, entry in index.iteritems():
3668 if max_stat is not None and stat_count >= max_stat:
3669 return
3670 result = _check_entry_for_changes(
3671 tree_path,
3672 entry,
3673 root_path,
3674 filter_blob_callback,
3675 trust_ctime,
3676 index.object_format,
3677 refresh_stat,
3678 )
3679 stat_count += 1
3680 if result is not None:
3681 yield result
3684def _decode_utf8_with_fallback(data: bytes) -> str:
3685 """Decode bytes as UTF-8, with lossy fallbacks for invalid sequences.
3687 Mirrors the behaviour of git-for-windows's ``xutftowcsn`` (in
3688 ``compat/mingw.c``) so that tree paths containing legacy-encoded or
3689 otherwise invalid UTF-8 produce the same on-disk filename as C git.
3691 Rules:
3692 * Valid UTF-8 (1-4 byte sequences, excluding overlongs and codepoints
3693 > U+10FFFF) is decoded normally.
3694 * Invalid bytes in 0xa0-0xff map 1:1 to U+00A0-U+00FF.
3695 * Invalid bytes in 0x80-0x9f are expanded to two lowercase ASCII hex
3696 digits (e.g. byte 0x80 -> "80").
3697 * Truncated multi-byte sequences and overlong/out-of-range encodings
3698 cause the lead byte to fall through to the above invalid-byte rules
3699 (the trail bytes are re-evaluated on the next iteration).
3700 """
3701 out: list[str] = []
3702 i = 0
3703 n = len(data)
3704 while i < n:
3705 c = data[i]
3706 if c < 0x80:
3707 out.append(chr(c))
3708 i += 1
3709 elif 0xC2 <= c < 0xE0 and i + 1 < n and (data[i + 1] & 0xC0) == 0x80:
3710 cp = ((c & 0x1F) << 6) | (data[i + 1] & 0x3F)
3711 out.append(chr(cp))
3712 i += 2
3713 elif (
3714 0xE0 <= c < 0xF0
3715 and i + 2 < n
3716 and not (c == 0xE0 and data[i + 1] < 0xA0)
3717 and (data[i + 1] & 0xC0) == 0x80
3718 and (data[i + 2] & 0xC0) == 0x80
3719 ):
3720 cp = ((c & 0x0F) << 12) | ((data[i + 1] & 0x3F) << 6) | (data[i + 2] & 0x3F)
3721 out.append(chr(cp))
3722 i += 3
3723 elif (
3724 0xF0 <= c < 0xF5
3725 and i + 3 < n
3726 and not (c == 0xF0 and data[i + 1] < 0x90)
3727 and not (c == 0xF4 and data[i + 1] >= 0x90)
3728 and (data[i + 1] & 0xC0) == 0x80
3729 and (data[i + 2] & 0xC0) == 0x80
3730 and (data[i + 3] & 0xC0) == 0x80
3731 ):
3732 cp = (
3733 ((c & 0x07) << 18)
3734 | ((data[i + 1] & 0x3F) << 12)
3735 | ((data[i + 2] & 0x3F) << 6)
3736 | (data[i + 3] & 0x3F)
3737 )
3738 out.append(chr(cp))
3739 i += 4
3740 elif c >= 0xA0:
3741 out.append(chr(c))
3742 i += 1
3743 else:
3744 out.append(f"{c:02x}")
3745 i += 1
3746 return "".join(out)
3749def _tree_to_fs_path(
3750 root_path: bytes, tree_path: bytes, tree_encoding: str = "utf-8"
3751) -> bytes:
3752 """Convert a git tree path to a file system path.
3754 Args:
3755 root_path: Root filesystem path
3756 tree_path: Git tree path as bytes (encoded with tree_encoding)
3757 tree_encoding: Encoding used for tree paths (default: utf-8)
3759 Returns: File system path.
3760 """
3761 assert isinstance(tree_path, bytes)
3762 if os_sep_bytes != b"/":
3763 sep_corrected_path = tree_path.replace(b"/", os_sep_bytes)
3764 else:
3765 sep_corrected_path = tree_path
3767 # On Windows, decode tree-encoded bytes to a str so they can flow into
3768 # the wide-char Win32 APIs via Python's filesystem layer. For UTF-8
3769 # (the default tree encoding) we use a lossy decoder that matches C
3770 # git's xutftowcsn fallbacks; for other encodings we let UnicodeDecodeError
3771 # propagate rather than silently producing a corrupt path.
3772 if sys.platform == "win32":
3773 if tree_encoding == "utf-8":
3774 tree_path_str = _decode_utf8_with_fallback(sep_corrected_path)
3775 else:
3776 tree_path_str = sep_corrected_path.decode(tree_encoding)
3777 sep_corrected_path = os.fsencode(tree_path_str)
3779 return os.path.join(root_path, sep_corrected_path)
3782def _fs_to_tree_path(fs_path: str | bytes, tree_encoding: str = "utf-8") -> bytes:
3783 """Convert a file system path to a git tree path.
3785 Args:
3786 fs_path: File system path.
3787 tree_encoding: Encoding to use for tree paths (default: utf-8)
3789 Returns: Git tree path as bytes (encoded with tree_encoding)
3790 """
3791 if not isinstance(fs_path, bytes):
3792 fs_path_bytes = os.fsencode(fs_path)
3793 else:
3794 fs_path_bytes = fs_path
3796 # On Windows the on-disk filename is a UTF-16 wide string; Python gives
3797 # us either str (already decoded) or bytes encoded via the filesystem
3798 # codec. Normalise to str, then encode under the tree encoding so the
3799 # resulting tree path is plain UTF-8. This matches C git's xwcstoutf,
3800 # which is just WideCharToMultiByte(CP_UTF8); it makes no attempt to
3801 # reverse the xutftowcsn fallbacks, so a file that was checked out from
3802 # a tree path with invalid UTF-8 will read back as the lossy form (the
3803 # same divergence C git exhibits, documented as a one-way mapping).
3804 if sys.platform == "win32":
3805 fs_path_str = os.fsdecode(fs_path_bytes)
3806 fs_path_bytes = fs_path_str.encode(tree_encoding)
3808 if os_sep_bytes != b"/":
3809 tree_path = fs_path_bytes.replace(os_sep_bytes, b"/")
3810 else:
3811 tree_path = fs_path_bytes
3812 return tree_path
3815def index_entry_from_directory(st: os.stat_result, path: bytes) -> IndexEntry | None:
3816 """Create an index entry for a directory.
3818 This is only used for submodules (directories containing .git).
3820 Args:
3821 st: Stat result for the directory
3822 path: Path to the directory
3824 Returns:
3825 IndexEntry for a submodule, or None if not a submodule
3826 """
3827 if os.path.exists(os.path.join(path, b".git")):
3828 head = read_submodule_head(path)
3829 if head is None:
3830 return None
3831 return index_entry_from_stat(st, head, mode=S_IFGITLINK)
3832 return None
3835def index_entry_from_path(
3836 path: bytes,
3837 object_store: ObjectContainer | None = None,
3838 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
3839) -> IndexEntry | None:
3840 """Create an index from a filesystem path.
3842 This returns an index value for files, symlinks
3843 and tree references. for directories and
3844 non-existent files it returns None
3846 Args:
3847 path: Path to create an index entry for
3848 object_store: Optional object store to
3849 save new blobs in
3850 object_format: Object format to name new blobs with
3851 Returns: An index entry; None for directories
3852 """
3853 assert isinstance(path, bytes)
3854 st = os.lstat(path)
3855 if stat.S_ISDIR(st.st_mode):
3856 return index_entry_from_directory(st, path)
3858 if stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode):
3859 blob = blob_from_path_and_stat(path, st)
3860 if object_store is not None:
3861 object_store.add_object(blob)
3862 return index_entry_from_stat(st, ObjectID(blob.get_id(object_format)))
3864 return None
3867def iter_fresh_entries(
3868 paths: Iterable[bytes],
3869 root_path: bytes,
3870 object_store: ObjectContainer | None = None,
3871 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
3872) -> Iterator[tuple[bytes, IndexEntry | None]]:
3873 """Iterate over current versions of index entries on disk.
3875 Args:
3876 paths: Paths to iterate over
3877 root_path: Root path to access from
3878 object_store: Optional store to save new blobs in
3879 object_format: Object format to name new blobs with
3880 Returns: Iterator over path, index_entry
3881 """
3882 for path in paths:
3883 p = _tree_to_fs_path(root_path, path)
3884 try:
3885 entry = index_entry_from_path(
3886 p, object_store=object_store, object_format=object_format
3887 )
3888 except (FileNotFoundError, IsADirectoryError):
3889 entry = None
3890 yield path, entry
3893def iter_fresh_objects(
3894 paths: Iterable[bytes],
3895 root_path: bytes,
3896 include_deleted: bool = False,
3897 object_store: ObjectContainer | None = None,
3898 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
3899) -> Iterator[tuple[bytes, ObjectID | None, int | None]]:
3900 """Iterate over versions of objects on disk referenced by index.
3902 Args:
3903 paths: Paths to check
3904 root_path: Root path to access from
3905 include_deleted: Include deleted entries with sha and
3906 mode set to None
3907 object_store: Optional object store to report new items to
3908 object_format: Object format to name new blobs with
3909 Returns: Iterator over path, sha, mode
3910 """
3911 for path, entry in iter_fresh_entries(
3912 paths, root_path, object_store=object_store, object_format=object_format
3913 ):
3914 if entry is None:
3915 if include_deleted:
3916 yield path, None, None
3917 else:
3918 yield path, entry.sha, cleanup_mode(entry.mode)
3921def refresh_index(index: Index, root_path: bytes) -> None:
3922 """Refresh the contents of an index.
3924 This is the equivalent to running 'git commit -a'.
3926 Args:
3927 index: Index to update
3928 root_path: Root filesystem path
3929 """
3930 for path, entry in iter_fresh_entries(
3931 index, root_path, object_format=index.object_format
3932 ):
3933 if entry:
3934 index[path] = entry
3937def apply_stat_refresh(
3938 index: Index, refreshed: Iterable[tuple[bytes, os.stat_result]]
3939) -> bool:
3940 """Apply refreshed stat info collected during an index walk.
3942 ``refreshed`` should be the list populated by passing ``refresh_stat``
3943 into `get_unstaged_changes`: it contains (tree_path, stat_result)
3944 pairs for entries whose on-disk stat drifted but whose content still
3945 matches the index entry sha.
3947 Returns True if any entries were updated (caller may want to write
3948 the index).
3949 """
3950 updated = False
3951 for tree_path, st in refreshed:
3952 entry = index[tree_path]
3953 if isinstance(entry, ConflictedIndexEntry):
3954 continue
3955 index[tree_path] = index_entry_from_stat(st, entry.sha, mode=entry.mode)
3956 updated = True
3957 return updated
3960class locked_index:
3961 """Lock the index while making modifications.
3963 Works as a context manager.
3964 """
3966 _file: "_GitFile"
3968 def __init__(
3969 self,
3970 path: bytes | str,
3971 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
3972 ) -> None:
3973 """Initialize locked_index.
3975 Args:
3976 path: Path to the index file
3977 object_format: Object format of the repository owning the index
3978 """
3979 self._path = path
3980 self._object_format = object_format
3982 def __enter__(self) -> Index:
3983 """Enter context manager and lock index."""
3984 f = GitFile(self._path, "wb")
3985 self._file = f
3986 self._index = Index(self._path, object_format=self._object_format)
3987 return self._index
3989 def __exit__(
3990 self,
3991 exc_type: type | None,
3992 exc_value: BaseException | None,
3993 traceback: types.TracebackType | None,
3994 ) -> None:
3995 """Exit context manager and unlock index."""
3996 if exc_type is not None:
3997 self._file.abort()
3998 return
3999 try:
4000 f = IndexChecksumWriter(self._file, self._object_format)
4001 write_index_dict(f, self._index._byname, object_format=self._object_format)
4002 except BaseException:
4003 self._file.abort()
4004 else:
4005 f.close()