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 "unchanged_entry_flags",
93 "update_working_tree",
94 "validate_path",
95 "validate_path_element_default",
96 "validate_path_element_hfs",
97 "validate_path_element_ntfs",
98 "verify_leading_dirs",
99 "verify_tree_path",
100 "write_cache_entry",
101 "write_cache_time",
102 "write_index",
103 "write_index_dict",
104 "write_index_extension",
105]
107import errno
108import logging
109import os
110import shutil
111import stat
112import struct
113import sys
114import types
115from collections.abc import (
116 Callable,
117 Generator,
118 Iterable,
119 Iterator,
120 Mapping,
121 Sequence,
122 Set,
123)
124from dataclasses import dataclass
125from enum import Enum
126from typing import (
127 TYPE_CHECKING,
128 Any,
129 Protocol,
130)
132if TYPE_CHECKING:
133 from .config import Config
134 from .diff_tree import TreeChange
135 from .file import _GitFile
136 from .filters import FilterBlobNormalizer
137 from .object_store import BaseObjectStore
138 from .repo import Repo
140from . import _deprecated_aliases
141from .errors import ChecksumMismatch
142from .file import GitFile, SharedPerm
143from .object_format import DEFAULT_OBJECT_FORMAT, ObjectFormat
144from .object_store import iter_tree_contents
145from .objects import (
146 S_IFGITLINK,
147 S_ISGITLINK,
148 Blob,
149 ObjectID,
150 RawObjectID,
151 Tree,
152 TreeEntry,
153 hex_to_sha,
154 sha_to_hex,
155)
156from .pack import ObjectContainer
158logger = logging.getLogger(__name__)
161__getattr__ = _deprecated_aliases(
162 __name__,
163 {
164 "SHA1Reader": "dulwich.pack.SHA1Reader",
165 "SHA1Writer": "dulwich.pack.SHA1Writer",
166 },
167)
170# Type alias for recursive tree structure used in commit_tree
171TreeDict = dict[bytes, "TreeDict | tuple[int, ObjectID]"]
173# 2-bit stage (during merge)
174FLAG_STAGEMASK = 0x3000
175FLAG_STAGESHIFT = 12
176FLAG_NAMEMASK = 0x0FFF
178# assume-valid
179FLAG_VALID = 0x8000
181# extended flag (must be zero in version 2)
182FLAG_EXTENDED = 0x4000
184# used by sparse checkout
185EXTENDED_FLAG_SKIP_WORKTREE = 0x4000
187# used by "git add -N"
188EXTENDED_FLAG_INTEND_TO_ADD = 0x2000
190DEFAULT_VERSION = 2
192# Index extension signatures
193TREE_EXTENSION = b"TREE"
194REUC_EXTENSION = b"REUC"
195UNTR_EXTENSION = b"UNTR"
196EOIE_EXTENSION = b"EOIE"
197IEOT_EXTENSION = b"IEOT"
198SDIR_EXTENSION = b"sdir" # Sparse directory extension
201def _encode_varint(value: int) -> bytes:
202 """Encode an integer using variable-width encoding.
204 Same format as used for OFS_DELTA pack entries and index v4 path compression.
205 Uses 7 bits per byte, with the high bit indicating continuation.
207 Args:
208 value: Integer to encode
209 Returns:
210 Encoded bytes
211 """
212 if value == 0:
213 return b"\x00"
215 result = []
216 while value > 0:
217 byte = value & 0x7F # Take lower 7 bits
218 value >>= 7
219 if value > 0:
220 byte |= 0x80 # Set continuation bit
221 result.append(byte)
223 return bytes(result)
226def _decode_varint(data: bytes, offset: int = 0) -> tuple[int, int]:
227 """Decode a variable-width encoded integer.
229 Args:
230 data: Bytes to decode from
231 offset: Starting offset in data
232 Returns:
233 tuple of (decoded_value, new_offset)
234 """
235 value = 0
236 shift = 0
237 pos = offset
239 while pos < len(data):
240 byte = data[pos]
241 pos += 1
242 value |= (byte & 0x7F) << shift
243 shift += 7
244 if not (byte & 0x80): # No continuation bit
245 break
247 return value, pos
250def _compress_path(path: bytes, previous_path: bytes) -> bytes:
251 """Compress a path relative to the previous path for index version 4.
253 Args:
254 path: Path to compress
255 previous_path: Previous path for comparison
256 Returns:
257 Compressed path data (varint prefix_len + suffix)
258 """
259 # Find the common prefix length
260 common_len = 0
261 min_len = min(len(path), len(previous_path))
263 for i in range(min_len):
264 if path[i] == previous_path[i]:
265 common_len += 1
266 else:
267 break
269 # The number of bytes to remove from the end of previous_path
270 # to get the common prefix
271 remove_len = len(previous_path) - common_len
273 # The suffix to append
274 suffix = path[common_len:]
276 # Encode: varint(remove_len) + suffix + NUL
277 return _encode_varint(remove_len) + suffix + b"\x00"
280def _decompress_path(
281 data: bytes, offset: int, previous_path: bytes
282) -> tuple[bytes, int]:
283 """Decompress a path from index version 4 compressed format.
285 Args:
286 data: Raw data containing compressed path
287 offset: Starting offset in data
288 previous_path: Previous path for decompression
289 Returns:
290 tuple of (decompressed_path, new_offset)
291 """
292 # Decode the number of bytes to remove from previous path
293 remove_len, new_offset = _decode_varint(data, offset)
295 # Find the NUL terminator for the suffix
296 suffix_start = new_offset
297 suffix_end = suffix_start
298 while suffix_end < len(data) and data[suffix_end] != 0:
299 suffix_end += 1
301 if suffix_end >= len(data):
302 raise ValueError("Unterminated path suffix in compressed entry")
304 suffix = data[suffix_start:suffix_end]
305 new_offset = suffix_end + 1 # Skip the NUL terminator
307 # Reconstruct the path
308 if remove_len > len(previous_path):
309 raise ValueError(
310 f"Invalid path compression: trying to remove {remove_len} bytes from {len(previous_path)}-byte path"
311 )
313 prefix = previous_path[:-remove_len] if remove_len > 0 else previous_path
314 path = prefix + suffix
316 return path, new_offset
319class _IndexWritable(Protocol):
320 """The subset of a binary file that the index serializer writes to."""
322 def write(self, data: bytes, /) -> int:
323 """Write data, returning the number of bytes written."""
325 def tell(self) -> int:
326 """Return the current stream position."""
328 def close(self) -> None:
329 """Close the stream."""
332class _IndexReadable(Protocol):
333 """The subset of a binary file that the index parser reads from."""
335 def read(self, size: int = -1, /) -> bytes:
336 """Read up to size bytes."""
338 def tell(self) -> int:
339 """Return the current stream position."""
341 def seek(self, offset: int, whence: int = 0, /) -> int:
342 """Change the stream position."""
345def _decompress_path_from_stream(
346 f: _IndexReadable, previous_path: bytes
347) -> tuple[bytes, int]:
348 """Decompress a path from index version 4 compressed format, reading from stream.
350 Args:
351 f: File-like object to read from
352 previous_path: Previous path for decompression
353 Returns:
354 tuple of (decompressed_path, bytes_consumed)
355 """
356 # Decode the varint for remove_len by reading byte by byte
357 remove_len = 0
358 shift = 0
359 bytes_consumed = 0
361 while True:
362 byte_data = f.read(1)
363 if not byte_data:
364 raise ValueError("Unexpected end of file while reading varint")
365 byte = byte_data[0]
366 bytes_consumed += 1
367 remove_len |= (byte & 0x7F) << shift
368 shift += 7
369 if not (byte & 0x80): # No continuation bit
370 break
372 # Read the suffix until NUL terminator
373 suffix = b""
374 while True:
375 byte_data = f.read(1)
376 if not byte_data:
377 raise ValueError("Unexpected end of file while reading path suffix")
378 byte = byte_data[0]
379 bytes_consumed += 1
380 if byte == 0: # NUL terminator
381 break
382 suffix += bytes([byte])
384 # Reconstruct the path
385 if remove_len > len(previous_path):
386 raise ValueError(
387 f"Invalid path compression: trying to remove {remove_len} bytes from {len(previous_path)}-byte path"
388 )
390 prefix = previous_path[:-remove_len] if remove_len > 0 else previous_path
391 path = prefix + suffix
393 return path, bytes_consumed
396class Stage(Enum):
397 """Represents the stage of an index entry during merge conflicts."""
399 NORMAL = 0
400 MERGE_CONFLICT_ANCESTOR = 1
401 MERGE_CONFLICT_THIS = 2
402 MERGE_CONFLICT_OTHER = 3
405@dataclass
406class SerializedIndexEntry:
407 """Represents a serialized index entry as stored in the index file.
409 This dataclass holds the raw data for an index entry before it's
410 parsed into the more user-friendly IndexEntry format.
411 """
413 name: bytes
414 ctime: int | float | tuple[int, int]
415 mtime: int | float | tuple[int, int]
416 dev: int
417 ino: int
418 mode: int
419 uid: int
420 gid: int
421 size: int
422 sha: ObjectID
423 flags: int
424 extended_flags: int
426 def stage(self) -> Stage:
427 """Extract the stage from the flags field.
429 Returns:
430 Stage enum value indicating merge conflict state
431 """
432 return Stage((self.flags & FLAG_STAGEMASK) >> FLAG_STAGESHIFT)
434 def is_sparse_dir(self) -> bool:
435 """Check if this entry represents a sparse directory.
437 A sparse directory entry is a collapsed representation of an entire
438 directory tree in a sparse index. It has:
439 - Directory mode (0o040000)
440 - SKIP_WORKTREE flag set
441 - Path ending with '/'
442 - SHA pointing to a tree object
444 Returns:
445 True if entry is a sparse directory entry
446 """
447 return (
448 stat.S_ISDIR(self.mode)
449 and bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
450 and self.name.endswith(b"/")
451 )
454@dataclass
455class IndexExtension:
456 """Base class for index extensions."""
458 signature: bytes
459 data: bytes
461 @classmethod
462 def from_raw(cls, signature: bytes, data: bytes) -> "IndexExtension":
463 """Create an extension from raw data.
465 Args:
466 signature: 4-byte extension signature
467 data: Extension data
468 Returns:
469 Parsed extension object
470 """
471 if signature == TREE_EXTENSION:
472 return TreeExtension.from_bytes(data)
473 elif signature == REUC_EXTENSION:
474 return ResolveUndoExtension.from_bytes(data)
475 elif signature == UNTR_EXTENSION:
476 return UntrackedExtension.from_bytes(data)
477 elif signature == SDIR_EXTENSION:
478 return SparseDirExtension.from_bytes(data)
479 else:
480 # Unknown extension - just store raw data
481 return cls(signature, data)
483 def to_bytes(self) -> bytes:
484 """Serialize extension to bytes."""
485 return self.data
488class TreeExtension(IndexExtension):
489 """Tree cache extension."""
491 def __init__(self, entries: list[tuple[bytes, bytes, int]]) -> None:
492 """Initialize TreeExtension.
494 Args:
495 entries: List of tree cache entries (path, sha, flags)
496 """
497 self.entries = entries
498 super().__init__(TREE_EXTENSION, b"")
500 @classmethod
501 def from_bytes(cls, data: bytes) -> "TreeExtension":
502 """Parse TreeExtension from bytes.
504 Args:
505 data: Raw bytes to parse
507 Returns:
508 TreeExtension instance
509 """
510 # TODO: Implement tree cache parsing
511 return cls([])
513 def to_bytes(self) -> bytes:
514 """Serialize TreeExtension to bytes.
516 Returns:
517 Serialized extension data
518 """
519 # TODO: Implement tree cache serialization
520 return b""
523class ResolveUndoExtension(IndexExtension):
524 """Resolve undo extension for recording merge conflicts."""
526 def __init__(self, entries: list[tuple[bytes, list[tuple[int, bytes]]]]) -> None:
527 """Initialize ResolveUndoExtension.
529 Args:
530 entries: List of (path, stages) where stages is a list of (stage, sha) tuples
531 """
532 self.entries = entries
533 super().__init__(REUC_EXTENSION, b"")
535 @classmethod
536 def from_bytes(cls, data: bytes) -> "ResolveUndoExtension":
537 """Parse ResolveUndoExtension from bytes.
539 Args:
540 data: Raw bytes to parse
542 Returns:
543 ResolveUndoExtension instance
544 """
545 # TODO: Implement resolve undo parsing
546 return cls([])
548 def to_bytes(self) -> bytes:
549 """Serialize ResolveUndoExtension to bytes.
551 Returns:
552 Serialized extension data
553 """
554 # TODO: Implement resolve undo serialization
555 return b""
558class UntrackedExtension(IndexExtension):
559 """Untracked cache extension."""
561 def __init__(self, data: bytes) -> None:
562 """Initialize UntrackedExtension.
564 Args:
565 data: Raw untracked cache data
566 """
567 super().__init__(UNTR_EXTENSION, data)
569 @classmethod
570 def from_bytes(cls, data: bytes) -> "UntrackedExtension":
571 """Parse UntrackedExtension from bytes.
573 Args:
574 data: Raw bytes to parse
576 Returns:
577 UntrackedExtension instance
578 """
579 return cls(data)
582class SparseDirExtension(IndexExtension):
583 """Sparse directory extension.
585 This extension indicates that the index contains sparse directory entries.
586 Tools that don't understand sparse index should avoid interacting with
587 the index when this extension is present.
589 The extension data is empty - its presence is the signal.
590 """
592 def __init__(self) -> None:
593 """Initialize SparseDirExtension."""
594 super().__init__(SDIR_EXTENSION, b"")
596 @classmethod
597 def from_bytes(cls, data: bytes) -> "SparseDirExtension":
598 """Parse SparseDirExtension from bytes.
600 Args:
601 data: Raw bytes to parse (should be empty)
603 Returns:
604 SparseDirExtension instance
605 """
606 return cls()
608 def to_bytes(self) -> bytes:
609 """Serialize SparseDirExtension to bytes.
611 Returns:
612 Empty bytes (extension presence is the signal)
613 """
614 return b""
617@dataclass
618class IndexEntry:
619 """Represents an entry in the Git index.
621 This is a higher-level representation of an index entry that includes
622 parsed data and convenience methods.
623 """
625 ctime: int | float | tuple[int, int]
626 mtime: int | float | tuple[int, int]
627 dev: int
628 ino: int
629 mode: int
630 uid: int
631 gid: int
632 size: int
633 sha: ObjectID
634 flags: int = 0
635 extended_flags: int = 0
637 @classmethod
638 def from_serialized(cls, serialized: SerializedIndexEntry) -> "IndexEntry":
639 """Create an IndexEntry from a SerializedIndexEntry.
641 Args:
642 serialized: SerializedIndexEntry to convert
644 Returns:
645 New IndexEntry instance
646 """
647 return cls(
648 ctime=serialized.ctime,
649 mtime=serialized.mtime,
650 dev=serialized.dev,
651 ino=serialized.ino,
652 mode=serialized.mode,
653 uid=serialized.uid,
654 gid=serialized.gid,
655 size=serialized.size,
656 sha=serialized.sha,
657 flags=serialized.flags,
658 extended_flags=serialized.extended_flags,
659 )
661 def serialize(self, name: bytes, stage: Stage) -> SerializedIndexEntry:
662 """Serialize this entry with a given name and stage.
664 Args:
665 name: Path name for the entry
666 stage: Merge conflict stage
668 Returns:
669 SerializedIndexEntry ready for writing to disk
670 """
671 # Clear out any existing stage bits, then set them from the Stage.
672 new_flags = self.flags & ~FLAG_STAGEMASK
673 new_flags |= stage.value << FLAG_STAGESHIFT
674 return SerializedIndexEntry(
675 name=name,
676 ctime=self.ctime,
677 mtime=self.mtime,
678 dev=self.dev,
679 ino=self.ino,
680 mode=self.mode,
681 uid=self.uid,
682 gid=self.gid,
683 size=self.size,
684 sha=self.sha,
685 flags=new_flags,
686 extended_flags=self.extended_flags,
687 )
689 def stage(self) -> Stage:
690 """Get the merge conflict stage of this entry.
692 Returns:
693 Stage enum value
694 """
695 return Stage((self.flags & FLAG_STAGEMASK) >> FLAG_STAGESHIFT)
697 @property
698 def skip_worktree(self) -> bool:
699 """Return True if the skip-worktree bit is set in extended_flags."""
700 return bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
702 def set_skip_worktree(self, skip: bool = True) -> None:
703 """Helper method to set or clear the skip-worktree bit in extended_flags.
705 Also sets FLAG_EXTENDED in self.flags if needed.
706 """
707 if skip:
708 # Turn on the skip-worktree bit
709 self.extended_flags |= EXTENDED_FLAG_SKIP_WORKTREE
710 # Also ensure the main 'extended' bit is set in flags
711 self.flags |= FLAG_EXTENDED
712 else:
713 # Turn off the skip-worktree bit
714 self.extended_flags &= ~EXTENDED_FLAG_SKIP_WORKTREE
715 # Optionally unset the main extended bit if no extended flags remain
716 if self.extended_flags == 0:
717 self.flags &= ~FLAG_EXTENDED
719 def is_sparse_dir(self, name: bytes) -> bool:
720 """Check if this entry represents a sparse directory.
722 A sparse directory entry is a collapsed representation of an entire
723 directory tree in a sparse index. It has:
724 - Directory mode (0o040000)
725 - SKIP_WORKTREE flag set
726 - Path ending with '/'
727 - SHA pointing to a tree object
729 Args:
730 name: The path name for this entry (IndexEntry doesn't store name)
732 Returns:
733 True if entry is a sparse directory entry
734 """
735 return (
736 stat.S_ISDIR(self.mode)
737 and bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
738 and name.endswith(b"/")
739 )
742class ConflictedIndexEntry:
743 """Index entry that represents a conflict."""
745 ancestor: IndexEntry | None
746 this: IndexEntry | None
747 other: IndexEntry | None
749 def __init__(
750 self,
751 ancestor: IndexEntry | None = None,
752 this: IndexEntry | None = None,
753 other: IndexEntry | None = None,
754 ) -> None:
755 """Initialize ConflictedIndexEntry.
757 Args:
758 ancestor: The common ancestor entry
759 this: The current branch entry
760 other: The other branch entry
761 """
762 self.ancestor = ancestor
763 self.this = this
764 self.other = other
767class UnmergedEntries(Exception):
768 """Unmerged entries exist in the index."""
771def pathsplit(path: bytes) -> tuple[bytes, bytes]:
772 """Split a /-delimited path into a directory part and a basename.
774 Args:
775 path: The path to split.
777 Returns:
778 Tuple with directory name and basename
779 """
780 try:
781 (dirname, basename) = path.rsplit(b"/", 1)
782 except ValueError:
783 return (b"", path)
784 else:
785 return (dirname, basename)
788def pathjoin(*args: bytes) -> bytes:
789 """Join a /-delimited path."""
790 return b"/".join([p for p in args if p])
793def read_cache_time(f: _IndexReadable) -> tuple[int, int]:
794 """Read a cache time.
796 Args:
797 f: File-like object to read from
798 Returns:
799 Tuple with seconds and nanoseconds
800 """
801 return struct.unpack(">LL", f.read(8))
804def write_cache_time(f: _IndexWritable, t: int | float | tuple[int, int]) -> None:
805 """Write a cache time.
807 Args:
808 f: File-like object to write to
809 t: Time to write (as int, float or tuple with secs and nsecs)
810 """
811 if isinstance(t, int):
812 t = (t, 0)
813 elif isinstance(t, float):
814 (secs, nsecs) = divmod(t, 1.0)
815 t = (int(secs), int(nsecs * 1000000000))
816 elif not isinstance(t, tuple):
817 raise TypeError(t)
818 f.write(struct.pack(">LL", *t))
821class IndexChecksumWriter:
822 """Wrap a file, tracking the checksum of everything written to it.
824 The index trailer uses the repository's object format, so unlike
825 :class:`dulwich.pack.SHA1Writer` this is not tied to SHA-1.
826 """
828 def __init__(
829 self, f: _IndexWritable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
830 ) -> None:
831 """Initialize the writer.
833 Args:
834 f: File-like object to wrap
835 object_format: Object format determining the checksum algorithm
836 """
837 self.f = f
838 self.object_format = object_format
839 self._hash = object_format.new_hash()
841 def write(self, data: bytes) -> int:
842 """Write data and update the running checksum."""
843 self._hash.update(data)
844 return self.f.write(data)
846 def tell(self) -> int:
847 """Return the position of the underlying file."""
848 return self.f.tell()
850 def write_checksum(self) -> bytes:
851 """Write the trailing checksum and return it."""
852 digest = self._hash.digest()
853 self.f.write(digest)
854 return digest
856 def close(self) -> None:
857 """Write the trailing checksum and close the underlying file."""
858 self.write_checksum()
859 self.f.close()
862class IndexChecksumReader:
863 """Wrap a file, tracking the checksum of everything read from it."""
865 def __init__(
866 self, f: _IndexReadable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
867 ) -> None:
868 """Initialize the reader.
870 Args:
871 f: File-like object to wrap
872 object_format: Object format determining the checksum algorithm
873 """
874 self.f = f
875 self.object_format = object_format
876 self._hash = object_format.new_hash()
878 def read(self, size: int = -1) -> bytes:
879 """Read data and update the running checksum."""
880 data = self.f.read(size)
881 self._hash.update(data)
882 return data
884 def tell(self) -> int:
885 """Return the position of the underlying file."""
886 return self.f.tell()
888 def seek(self, offset: int, whence: int = 0) -> int:
889 """Seek the underlying file.
891 Bytes skipped by seeking are not folded into the checksum, so this
892 is only safe for the trailer probing done while reading extensions.
893 """
894 return self.f.seek(offset, whence)
896 def check_checksum(self, allow_empty: bool = False) -> None:
897 """Verify the trailing checksum.
899 Args:
900 allow_empty: Accept an all-zero checksum, as written when the
901 index.skipHash config option is set.
903 Raises:
904 ChecksumMismatch: If the stored checksum does not match.
905 """
906 length = self.object_format.oid_length
907 stored = self.f.read(length)
908 expected = self._hash.digest()
909 if stored == expected:
910 return
911 if allow_empty and (len(stored) != length or stored == b"\x00" * length):
912 return
913 raise ChecksumMismatch(
914 self._hash.hexdigest(), sha_to_hex(RawObjectID(stored)).decode("ascii")
915 )
918def read_cache_entry(
919 f: _IndexReadable,
920 version: int,
921 previous_path: bytes = b"",
922 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
923) -> SerializedIndexEntry:
924 """Read an entry from a cache file.
926 Args:
927 f: File-like object to read from
928 version: Index version
929 previous_path: Previous entry's path (for version 4 compression)
930 object_format: Object format determining the object ID width
931 """
932 beginoffset = f.tell()
933 ctime = read_cache_time(f)
934 mtime = read_cache_time(f)
935 oid_length = object_format.oid_length
936 (
937 dev,
938 ino,
939 mode,
940 uid,
941 gid,
942 size,
943 sha,
944 flags,
945 ) = struct.unpack(f">LLLLLL{oid_length}sH", f.read(oid_length + 4 * 6 + 2))
946 if flags & FLAG_EXTENDED:
947 if version < 3:
948 raise AssertionError("extended flag set in index with version < 3")
949 (extended_flags,) = struct.unpack(">H", f.read(2))
950 else:
951 extended_flags = 0
953 if version >= 4:
954 # Version 4: paths are always compressed (name_len should be 0)
955 name, _consumed = _decompress_path_from_stream(f, previous_path)
956 else:
957 # Versions < 4: regular name reading
958 name = f.read(flags & FLAG_NAMEMASK)
960 # Padding:
961 if version < 4:
962 real_size = (f.tell() - beginoffset + 8) & ~7
963 f.read((beginoffset + real_size) - f.tell())
965 return SerializedIndexEntry(
966 name,
967 ctime,
968 mtime,
969 dev,
970 ino,
971 mode,
972 uid,
973 gid,
974 size,
975 sha_to_hex(sha),
976 flags & ~FLAG_NAMEMASK,
977 extended_flags,
978 )
981def write_cache_entry(
982 f: _IndexWritable,
983 entry: SerializedIndexEntry,
984 version: int,
985 previous_path: bytes = b"",
986 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
987) -> None:
988 """Write an index entry to a file.
990 Args:
991 f: File object
992 entry: IndexEntry to write
993 version: Index format version
994 previous_path: Previous entry's path (for version 4 compression)
995 object_format: Object format determining the object ID width
996 """
997 if len(entry.sha) != object_format.hex_length:
998 raise ValueError(
999 f"Object ID {entry.sha!r} for {entry.name!r} is not valid for "
1000 f"a {object_format.name} index"
1001 )
1003 beginoffset = f.tell()
1004 write_cache_time(f, entry.ctime)
1005 write_cache_time(f, entry.mtime)
1007 compressed_path = b""
1008 if version >= 4:
1009 # Version 4: use compression but set name_len to actual filename length
1010 # This matches how C Git implements index v4 flags
1011 compressed_path = _compress_path(entry.name, previous_path)
1012 flags = len(entry.name) | (entry.flags & ~FLAG_NAMEMASK)
1014 if entry.extended_flags:
1015 flags |= FLAG_EXTENDED
1016 if flags & FLAG_EXTENDED and version is not None and version < 3:
1017 raise AssertionError("unable to use extended flags in version < 3")
1019 f.write(
1020 struct.pack(
1021 f">LLLLLL{object_format.oid_length}sH",
1022 entry.dev & 0xFFFFFFFF,
1023 entry.ino & 0xFFFFFFFF,
1024 entry.mode,
1025 entry.uid,
1026 entry.gid,
1027 entry.size,
1028 hex_to_sha(entry.sha),
1029 flags,
1030 )
1031 )
1032 if flags & FLAG_EXTENDED:
1033 f.write(struct.pack(b">H", entry.extended_flags))
1035 if version >= 4:
1036 # Version 4: always write compressed path
1037 f.write(compressed_path)
1038 else:
1039 # Versions < 4: write regular path and padding
1040 f.write(entry.name)
1041 real_size = (f.tell() - beginoffset + 8) & ~7
1042 f.write(b"\0" * ((beginoffset + real_size) - f.tell()))
1045class UnsupportedIndexFormat(Exception):
1046 """An unsupported index format was encountered."""
1048 def __init__(self, version: int) -> None:
1049 """Initialize UnsupportedIndexFormat exception.
1051 Args:
1052 version: The unsupported index format version
1053 """
1054 self.index_format_version = version
1057def read_index_header(f: _IndexReadable) -> tuple[int, int]:
1058 """Read an index header from a file.
1060 Returns:
1061 tuple of (version, num_entries)
1062 """
1063 header = f.read(4)
1064 if header != b"DIRC":
1065 raise AssertionError(f"Invalid index file header: {header!r}")
1066 (version, num_entries) = struct.unpack(b">LL", f.read(4 * 2))
1067 if version not in (1, 2, 3, 4):
1068 raise UnsupportedIndexFormat(version)
1069 return version, num_entries
1072def write_index_extension(f: _IndexWritable, extension: IndexExtension) -> None:
1073 """Write an index extension.
1075 Args:
1076 f: File-like object to write to
1077 extension: Extension to write
1078 """
1079 data = extension.to_bytes()
1080 f.write(extension.signature)
1081 f.write(struct.pack(">I", len(data)))
1082 f.write(data)
1085def read_index(
1086 f: _IndexReadable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
1087) -> Iterator[SerializedIndexEntry]:
1088 """Read an index file, yielding the individual entries."""
1089 version, num_entries = read_index_header(f)
1090 previous_path = b""
1091 for i in range(num_entries):
1092 entry = read_cache_entry(f, version, previous_path, object_format)
1093 previous_path = entry.name
1094 yield entry
1097def read_index_dict_with_version(
1098 f: _IndexReadable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
1099) -> tuple[dict[bytes, IndexEntry | ConflictedIndexEntry], int, list[IndexExtension]]:
1100 """Read an index file and return it as a dictionary along with the version.
1102 Returns:
1103 tuple of (entries_dict, version, extensions)
1104 """
1105 version, num_entries = read_index_header(f)
1107 ret: dict[bytes, IndexEntry | ConflictedIndexEntry] = {}
1108 previous_path = b""
1109 for i in range(num_entries):
1110 entry = read_cache_entry(f, version, previous_path, object_format)
1111 previous_path = entry.name
1112 stage = entry.stage()
1113 if stage == Stage.NORMAL:
1114 ret[entry.name] = IndexEntry.from_serialized(entry)
1115 else:
1116 existing = ret.setdefault(entry.name, ConflictedIndexEntry())
1117 if isinstance(existing, IndexEntry):
1118 raise AssertionError(f"Non-conflicted entry for {entry.name!r} exists")
1119 match stage:
1120 case Stage.MERGE_CONFLICT_ANCESTOR:
1121 existing.ancestor = IndexEntry.from_serialized(entry)
1122 case Stage.MERGE_CONFLICT_THIS:
1123 existing.this = IndexEntry.from_serialized(entry)
1124 case Stage.MERGE_CONFLICT_OTHER:
1125 existing.other = IndexEntry.from_serialized(entry)
1127 # Read extensions
1128 extensions = []
1129 while True:
1130 # Check if we're at the end (trailing checksum before EOF)
1131 current_pos = f.tell()
1132 f.seek(0, 2) # EOF
1133 eof_pos = f.tell()
1134 f.seek(current_pos)
1136 if current_pos >= eof_pos - object_format.oid_length:
1137 break
1139 # Try to read extension signature
1140 signature = f.read(4)
1141 if len(signature) < 4:
1142 break
1144 # Check if it's a valid extension signature (4 uppercase letters)
1145 if not all(65 <= b <= 90 for b in signature):
1146 # Not an extension, seek back
1147 f.seek(-4, 1)
1148 break
1150 # Read extension size
1151 size_data = f.read(4)
1152 if len(size_data) < 4:
1153 break
1154 size = struct.unpack(">I", size_data)[0]
1156 # Read extension data
1157 data = f.read(size)
1158 if len(data) < size:
1159 break
1161 extension = IndexExtension.from_raw(signature, data)
1162 extensions.append(extension)
1164 return ret, version, extensions
1167def read_index_dict(
1168 f: _IndexReadable, object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT
1169) -> dict[bytes, IndexEntry | ConflictedIndexEntry]:
1170 """Read an index file and return it as a dictionary.
1172 Dict Key is tuple of path and stage number, as
1173 path alone is not unique
1174 Args:
1175 f: File object to read fromls.
1176 object_format: Object format determining the object ID width
1177 """
1178 ret: dict[bytes, IndexEntry | ConflictedIndexEntry] = {}
1179 for entry in read_index(f, object_format):
1180 stage = entry.stage()
1181 if stage == Stage.NORMAL:
1182 ret[entry.name] = IndexEntry.from_serialized(entry)
1183 else:
1184 existing = ret.setdefault(entry.name, ConflictedIndexEntry())
1185 if isinstance(existing, IndexEntry):
1186 raise AssertionError(f"Non-conflicted entry for {entry.name!r} exists")
1187 match stage:
1188 case Stage.MERGE_CONFLICT_ANCESTOR:
1189 existing.ancestor = IndexEntry.from_serialized(entry)
1190 case Stage.MERGE_CONFLICT_THIS:
1191 existing.this = IndexEntry.from_serialized(entry)
1192 case Stage.MERGE_CONFLICT_OTHER:
1193 existing.other = IndexEntry.from_serialized(entry)
1194 return ret
1197def write_index(
1198 f: _IndexWritable,
1199 entries: Sequence[SerializedIndexEntry],
1200 version: int | None = None,
1201 extensions: Sequence[IndexExtension] | None = None,
1202 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
1203) -> None:
1204 """Write an index file.
1206 Args:
1207 f: File-like object to write to
1208 version: Version number to write
1209 entries: Iterable over the entries to write
1210 extensions: Optional list of extensions to write
1211 object_format: Object format determining the object ID width
1212 """
1213 if version is None:
1214 version = DEFAULT_VERSION
1215 # STEP 1: check if any extended_flags are set
1216 uses_extended_flags = any(e.extended_flags != 0 for e in entries)
1217 if uses_extended_flags and version < 3:
1218 # Force or bump the version to 3
1219 version = 3
1220 # The rest is unchanged, but you might insert a final check:
1221 if version < 3:
1222 # Double-check no extended flags appear
1223 for e in entries:
1224 if e.extended_flags != 0:
1225 raise AssertionError("Attempt to use extended flags in index < v3")
1226 # Proceed with the existing code to write the header and entries.
1227 f.write(b"DIRC")
1228 f.write(struct.pack(b">LL", version, len(entries)))
1229 previous_path = b""
1230 for entry in entries:
1231 write_cache_entry(
1232 f,
1233 entry,
1234 version=version,
1235 previous_path=previous_path,
1236 object_format=object_format,
1237 )
1238 previous_path = entry.name
1240 # Write extensions
1241 if extensions:
1242 for extension in extensions:
1243 write_index_extension(f, extension)
1246def write_index_dict(
1247 f: _IndexWritable,
1248 entries: Mapping[bytes, IndexEntry | ConflictedIndexEntry],
1249 version: int | None = None,
1250 extensions: Sequence[IndexExtension] | None = None,
1251 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
1252) -> None:
1253 """Write an index file based on the contents of a dictionary.
1255 being careful to sort by path and then by stage.
1256 """
1257 entries_list = []
1258 for key in sorted(entries):
1259 value = entries[key]
1260 if isinstance(value, ConflictedIndexEntry):
1261 if value.ancestor is not None:
1262 entries_list.append(
1263 value.ancestor.serialize(key, Stage.MERGE_CONFLICT_ANCESTOR)
1264 )
1265 if value.this is not None:
1266 entries_list.append(
1267 value.this.serialize(key, Stage.MERGE_CONFLICT_THIS)
1268 )
1269 if value.other is not None:
1270 entries_list.append(
1271 value.other.serialize(key, Stage.MERGE_CONFLICT_OTHER)
1272 )
1273 else:
1274 entries_list.append(value.serialize(key, Stage.NORMAL))
1276 write_index(
1277 f,
1278 entries_list,
1279 version=version,
1280 extensions=extensions,
1281 object_format=object_format,
1282 )
1285def cleanup_mode(mode: int) -> int:
1286 """Cleanup a mode value.
1288 This will return a mode that can be stored in a tree object.
1290 Args:
1291 mode: Mode to clean up.
1293 Returns:
1294 mode
1295 """
1296 if stat.S_ISLNK(mode):
1297 return stat.S_IFLNK
1298 elif stat.S_ISDIR(mode):
1299 return stat.S_IFDIR
1300 elif S_ISGITLINK(mode):
1301 return S_IFGITLINK
1302 ret = stat.S_IFREG | 0o644
1303 if mode & 0o100:
1304 ret |= 0o111
1305 return ret
1308class Index:
1309 """A Git Index file."""
1311 _byname: dict[bytes, IndexEntry | ConflictedIndexEntry]
1313 def __init__(
1314 self,
1315 filename: bytes | str | os.PathLike[str],
1316 read: bool = True,
1317 skip_hash: bool = False,
1318 version: int | None = None,
1319 *,
1320 shared_perm: "SharedPerm | None" = None,
1321 path_normalizer: Callable[[bytes], bytes] | None = None,
1322 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
1323 ) -> None:
1324 """Create an index object associated with the given filename.
1326 Args:
1327 filename: Path to the index file
1328 read: Whether to initialize the index from the given file, should it exist.
1329 skip_hash: Whether to skip SHA1 hash when writing (for manyfiles feature)
1330 version: Index format version to use (None = auto-detect from file or use default)
1331 shared_perm: Optional shared repository permission setting
1332 object_format: Object format of the repository this index belongs to.
1333 Determines the width of stored object IDs and of the trailing
1334 checksum.
1335 path_normalizer: Optional function mapping a filesystem path to a
1336 canonical form (e.g. case-folded, NFC-normalized). When provided,
1337 lookups (``index[path]``, ``path in index``, ``del index[path]``)
1338 transparently match paths that normalize to the same form as an
1339 existing entry.
1340 """
1341 self._filename = os.fspath(filename)
1342 # TODO(jelmer): Store the version returned by read_index
1343 self._version = version
1344 self._skip_hash = skip_hash
1345 self._shared_perm = shared_perm
1346 self._object_format = object_format
1347 self._extensions: list[IndexExtension] = []
1348 self._path_normalizer = path_normalizer
1349 self._normalized: dict[bytes, bytes] | None = (
1350 {} if path_normalizer is not None else None
1351 )
1352 self.clear()
1353 if read:
1354 self.read()
1356 def canonical_path(self, name: bytes) -> bytes:
1357 """Resolve ``name`` to the canonical key stored in the index.
1359 If an entry already exists under ``name`` (or no normalizer is
1360 configured), ``name`` is returned unchanged. Otherwise the
1361 normalizer is applied and the key of any entry with the same
1362 normalized form is returned. Falls back to ``name`` if none.
1364 Normally callers do not need this because ``index[name]``,
1365 ``name in index``, and ``del index[name]`` already apply
1366 normalization transparently. Use this when the path is also
1367 being used outside the index (for example to look up the same
1368 entry in a commit tree), so that both sides agree on the key.
1369 """
1370 if self._normalized is None or name in self._byname:
1371 return name
1372 assert self._path_normalizer is not None
1373 return self._normalized.get(self._path_normalizer(name), name)
1375 @property
1376 def path(self) -> bytes | str:
1377 """Get the path to the index file.
1379 Returns:
1380 Path to the index file
1381 """
1382 return self._filename
1384 @property
1385 def object_format(self) -> ObjectFormat:
1386 """Object format used for object IDs in this index."""
1387 return self._object_format
1389 def __repr__(self) -> str:
1390 """Return string representation of Index."""
1391 return f"{self.__class__.__name__}({self._filename!r})"
1393 def write(self) -> None:
1394 """Write current contents of index to disk."""
1395 f = GitFile(self._filename, "wb", shared_perm=self._shared_perm)
1396 try:
1397 # Filter out extensions with no meaningful data
1398 meaningful_extensions = []
1399 for ext in self._extensions:
1400 # Skip extensions that have empty data
1401 ext_data = ext.to_bytes()
1402 if ext_data:
1403 meaningful_extensions.append(ext)
1405 if self._skip_hash:
1406 # When skipHash is enabled, write the index without computing
1407 # the trailing checksum
1408 write_index_dict(
1409 f,
1410 self._byname,
1411 version=self._version,
1412 extensions=meaningful_extensions,
1413 object_format=self._object_format,
1414 )
1415 # Write zero bytes instead of the checksum
1416 f.write(b"\x00" * self._object_format.oid_length)
1417 f.close()
1418 else:
1419 checksum_writer = IndexChecksumWriter(f, self._object_format)
1420 write_index_dict(
1421 checksum_writer,
1422 self._byname,
1423 version=self._version,
1424 extensions=meaningful_extensions,
1425 object_format=self._object_format,
1426 )
1427 checksum_writer.close()
1428 except:
1429 f.close()
1430 raise
1432 def read(self) -> None:
1433 """Read current contents of index from disk."""
1434 if not os.path.exists(self._filename):
1435 return
1436 f = GitFile(self._filename, "rb")
1437 try:
1438 checksum_reader = IndexChecksumReader(f, self._object_format)
1439 entries, version, extensions = read_index_dict_with_version(
1440 checksum_reader, self._object_format
1441 )
1442 self._version = version
1443 self._extensions = extensions
1444 for name, value in entries.items():
1445 self.set_verbatim(name, value)
1446 # Extensions have already been read by read_index_dict_with_version
1447 checksum_reader.check_checksum(allow_empty=True)
1448 finally:
1449 f.close()
1451 def __len__(self) -> int:
1452 """Number of entries in this index file."""
1453 return len(self._byname)
1455 def __getitem__(self, key: bytes) -> IndexEntry | ConflictedIndexEntry:
1456 """Retrieve entry by relative path and stage.
1458 Returns: Either a IndexEntry or a ConflictedIndexEntry
1459 Raises KeyError: if the entry does not exist
1460 """
1461 return self._byname[self.canonical_path(key)]
1463 def __iter__(self) -> Iterator[bytes]:
1464 """Iterate over the paths and stages in this index."""
1465 return iter(self._byname)
1467 def __contains__(self, key: bytes) -> bool:
1468 """Check if a path exists in the index."""
1469 return self.canonical_path(key) in self._byname
1471 def get_sha1(self, path: bytes) -> ObjectID:
1472 """Return the (git object) SHA1 for the object at a path."""
1473 value = self[path]
1474 if isinstance(value, ConflictedIndexEntry):
1475 raise UnmergedEntries
1476 return value.sha
1478 def get_mode(self, path: bytes) -> int:
1479 """Return the POSIX file mode for the object at a path."""
1480 value = self[path]
1481 if isinstance(value, ConflictedIndexEntry):
1482 raise UnmergedEntries
1483 return value.mode
1485 def iterobjects(self) -> Iterable[tuple[bytes, ObjectID, int]]:
1486 """Iterate over path, sha, mode tuples for use with commit_tree."""
1487 for path in self:
1488 entry = self[path]
1489 if isinstance(entry, ConflictedIndexEntry):
1490 raise UnmergedEntries
1491 yield path, entry.sha, cleanup_mode(entry.mode)
1493 def has_conflicts(self) -> bool:
1494 """Check if the index contains any conflicted entries.
1496 Returns:
1497 True if any entries are conflicted, False otherwise
1498 """
1499 for value in self._byname.values():
1500 if isinstance(value, ConflictedIndexEntry):
1501 return True
1502 return False
1504 def clear(self) -> None:
1505 """Remove all contents from this index."""
1506 self._byname = {}
1507 if self._normalized is not None:
1508 self._normalized = {}
1510 def __setitem__(
1511 self, name: bytes, value: IndexEntry | ConflictedIndexEntry
1512 ) -> None:
1513 """Set an entry in the index."""
1514 assert isinstance(name, bytes)
1515 name = self.canonical_path(name)
1516 is_new = name not in self._byname
1517 self._byname[name] = value
1518 if is_new and self._normalized is not None:
1519 assert self._path_normalizer is not None
1520 self._normalized.setdefault(self._path_normalizer(name), name)
1522 def set_verbatim(
1523 self, name: bytes, value: IndexEntry | ConflictedIndexEntry
1524 ) -> None:
1525 """Set an entry without folding onto an existing normalized key.
1527 Use this when populating the index from an authoritative source
1528 that may legitimately hold two entries whose paths only differ by
1529 case (or NFD/NFC form), such as an on-disk index or a tree from
1530 the object store. ``__setitem__`` would collapse the second entry
1531 onto the first via the path normalizer, losing content.
1532 """
1533 assert isinstance(name, bytes)
1534 self._byname[name] = value
1535 if self._normalized is not None:
1536 assert self._path_normalizer is not None
1537 self._normalized.setdefault(self._path_normalizer(name), name)
1539 def __delitem__(self, name: bytes) -> None:
1540 """Delete an entry from the index."""
1541 name = self.canonical_path(name)
1542 del self._byname[name]
1543 if self._normalized is not None:
1544 assert self._path_normalizer is not None
1545 normalized_key = self._path_normalizer(name)
1546 if self._normalized.get(normalized_key) == name:
1547 del self._normalized[normalized_key]
1549 def iteritems(
1550 self,
1551 ) -> Iterator[tuple[bytes, IndexEntry | ConflictedIndexEntry]]:
1552 """Iterate over (path, entry) pairs in the index.
1554 Returns:
1555 Iterator of (path, entry) tuples
1556 """
1557 return iter(self._byname.items())
1559 def items(self) -> Iterator[tuple[bytes, IndexEntry | ConflictedIndexEntry]]:
1560 """Get an iterator over (path, entry) pairs.
1562 Returns:
1563 Iterator of (path, entry) tuples
1564 """
1565 return iter(self._byname.items())
1567 def update(self, entries: dict[bytes, IndexEntry | ConflictedIndexEntry]) -> None:
1568 """Update the index with multiple entries.
1570 Args:
1571 entries: Dictionary mapping paths to index entries
1572 """
1573 for key, value in entries.items():
1574 self[key] = value
1576 def paths(self) -> Generator[bytes, None, None]:
1577 """Generate all paths in the index.
1579 Yields:
1580 Path names as bytes
1581 """
1582 yield from self._byname.keys()
1584 def changes_from_tree(
1585 self,
1586 object_store: ObjectContainer,
1587 tree: ObjectID | None,
1588 want_unchanged: bool = False,
1589 ) -> Generator[
1590 tuple[
1591 tuple[bytes | None, bytes | None],
1592 tuple[int | None, int | None],
1593 tuple[bytes | None, bytes | None],
1594 ],
1595 None,
1596 None,
1597 ]:
1598 """Find the differences between the contents of this index and a tree.
1600 Args:
1601 object_store: Object store to use for retrieving tree contents
1602 tree: SHA1 of the root tree
1603 want_unchanged: Whether unchanged files should be reported
1604 Returns: Iterator over tuples with (oldpath, newpath), (oldmode,
1605 newmode), (oldsha, newsha)
1606 """
1608 def lookup_entry(path: bytes) -> tuple[bytes, int]:
1609 # Tree paths are authoritative, so match them exactly rather
1610 # than folding case through canonical_path.
1611 entry = self._byname[path]
1612 if isinstance(entry, IndexEntry):
1613 return entry.sha, cleanup_mode(entry.mode)
1614 else:
1615 # Handle ConflictedIndexEntry case
1616 return b"", 0
1618 yield from changes_from_tree(
1619 self.paths(),
1620 lookup_entry,
1621 object_store,
1622 tree,
1623 want_unchanged=want_unchanged,
1624 )
1626 def commit(self, object_store: ObjectContainer) -> ObjectID:
1627 """Create a new tree from an index.
1629 Args:
1630 object_store: Object store to save the tree in
1631 Returns:
1632 Root tree SHA
1633 """
1634 return commit_tree(
1635 object_store, self.iterobjects(), object_format=self._object_format
1636 )
1638 def is_sparse(self) -> bool:
1639 """Check if this index contains sparse directory entries.
1641 Returns:
1642 True if any sparse directory extension is present
1643 """
1644 return any(isinstance(ext, SparseDirExtension) for ext in self._extensions)
1646 def ensure_full_index(self, object_store: "BaseObjectStore") -> None:
1647 """Expand all sparse directory entries into full file entries.
1649 This converts a sparse index into a full index by recursively
1650 expanding any sparse directory entries into their constituent files.
1652 Args:
1653 object_store: Object store to read tree objects from
1655 Raises:
1656 KeyError: If a tree object referenced by a sparse dir entry doesn't exist
1657 """
1658 if not self.is_sparse():
1659 return
1661 # Find all sparse directory entries
1662 sparse_dirs = []
1663 for path, entry in list(self._byname.items()):
1664 if isinstance(entry, IndexEntry) and entry.is_sparse_dir(path):
1665 sparse_dirs.append((path, entry))
1667 # Expand each sparse directory
1668 for path, entry in sparse_dirs:
1669 # Remove the sparse directory entry
1670 del self[path]
1672 # Get the tree object
1673 tree = object_store[entry.sha]
1674 if not isinstance(tree, Tree):
1675 raise ValueError(f"Sparse directory {path!r} points to non-tree object")
1677 # Recursively add all entries from the tree
1678 self._expand_tree(path.rstrip(b"/"), tree, object_store, entry)
1680 # Remove the sparse directory extension
1681 self._extensions = [
1682 ext for ext in self._extensions if not isinstance(ext, SparseDirExtension)
1683 ]
1685 def _expand_tree(
1686 self,
1687 prefix: bytes,
1688 tree: Tree,
1689 object_store: "BaseObjectStore",
1690 template_entry: IndexEntry,
1691 ) -> None:
1692 """Recursively expand a tree into index entries.
1694 Args:
1695 prefix: Path prefix for entries (without trailing slash)
1696 tree: Tree object to expand
1697 object_store: Object store to read nested trees from
1698 template_entry: Template entry to copy metadata from
1699 """
1700 for name, mode, sha in tree.items():
1701 if prefix:
1702 full_path = prefix + b"/" + name
1703 else:
1704 full_path = name
1706 if stat.S_ISDIR(mode):
1707 # Recursively expand subdirectories
1708 subtree = object_store[sha]
1709 if not isinstance(subtree, Tree):
1710 raise ValueError(
1711 f"Directory entry {full_path!r} points to non-tree object"
1712 )
1713 self._expand_tree(full_path, subtree, object_store, template_entry)
1714 else:
1715 # Create an index entry for this file
1716 # Use the template entry for metadata but with the file's sha and mode
1717 new_entry = IndexEntry(
1718 ctime=template_entry.ctime,
1719 mtime=template_entry.mtime,
1720 dev=template_entry.dev,
1721 ino=template_entry.ino,
1722 mode=mode,
1723 uid=template_entry.uid,
1724 gid=template_entry.gid,
1725 size=0, # Size is unknown from tree
1726 sha=sha,
1727 flags=0,
1728 extended_flags=0, # Don't copy skip-worktree flag
1729 )
1730 self[full_path] = new_entry
1732 def convert_to_sparse(
1733 self,
1734 object_store: "BaseObjectStore",
1735 tree_sha: ObjectID,
1736 sparse_dirs: Set[bytes],
1737 ) -> None:
1738 """Convert full index entries to sparse directory entries.
1740 This collapses directories that are entirely outside the sparse
1741 checkout cone into single sparse directory entries.
1743 Args:
1744 object_store: Object store to read tree objects
1745 tree_sha: SHA of the tree (usually HEAD) to base sparse dirs on
1746 sparse_dirs: Set of directory paths (with trailing /) to collapse
1748 Raises:
1749 KeyError: If tree_sha or a subdirectory doesn't exist
1750 """
1751 if not sparse_dirs:
1752 return
1754 # Get the base tree
1755 tree = object_store[tree_sha]
1756 if not isinstance(tree, Tree):
1757 raise ValueError(f"tree_sha {tree_sha!r} is not a tree object")
1759 # For each sparse directory, find its tree SHA and create sparse entry
1760 for dir_path in sparse_dirs:
1761 dir_path_stripped = dir_path.rstrip(b"/")
1763 # Find the tree SHA for this directory
1764 subtree_sha = self._find_subtree_sha(tree, dir_path_stripped, object_store)
1765 if subtree_sha is None:
1766 # Directory doesn't exist in tree, skip it
1767 continue
1769 # Remove all entries under this directory
1770 entries_to_remove = [
1771 path
1772 for path in self._byname
1773 if path.startswith(dir_path) or path == dir_path_stripped
1774 ]
1775 for path in entries_to_remove:
1776 del self[path]
1778 # Create a sparse directory entry
1779 # Use minimal metadata since it's not a real file
1780 from dulwich.objects import ObjectID
1782 sparse_entry = IndexEntry(
1783 ctime=0,
1784 mtime=0,
1785 dev=0,
1786 ino=0,
1787 mode=stat.S_IFDIR,
1788 uid=0,
1789 gid=0,
1790 size=0,
1791 sha=ObjectID(subtree_sha),
1792 flags=0,
1793 extended_flags=EXTENDED_FLAG_SKIP_WORKTREE,
1794 )
1795 self[dir_path] = sparse_entry
1797 # Add sparse directory extension if not present
1798 if not self.is_sparse():
1799 self._extensions.append(SparseDirExtension())
1801 def _find_subtree_sha(
1802 self,
1803 tree: Tree,
1804 path: bytes,
1805 object_store: "BaseObjectStore",
1806 ) -> bytes | None:
1807 """Find the SHA of a subtree at a given path.
1809 Args:
1810 tree: Root tree object to search in
1811 path: Path to the subdirectory (no trailing slash)
1812 object_store: Object store to read nested trees from
1814 Returns:
1815 SHA of the subtree, or None if path doesn't exist
1816 """
1817 if not path:
1818 return tree.get_id(self._object_format)
1820 parts = path.split(b"/")
1821 current_tree = tree
1823 for part in parts:
1824 # Look for this part in the current tree
1825 try:
1826 mode, sha = current_tree[part]
1827 except KeyError:
1828 return None
1830 if not stat.S_ISDIR(mode):
1831 # Path component is a file, not a directory
1832 return None
1834 # Load the next tree
1835 obj = object_store[sha]
1836 if not isinstance(obj, Tree):
1837 return None
1838 current_tree = obj
1840 return current_tree.get_id(self._object_format)
1843def commit_tree(
1844 object_store: ObjectContainer,
1845 blobs: Iterable[tuple[bytes, ObjectID, int]],
1846 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
1847) -> ObjectID:
1848 """Commit a new tree.
1850 Args:
1851 object_store: Object store to add trees to
1852 blobs: Iterable over blob path, sha, mode entries
1853 object_format: Object format to name the created trees with
1854 Returns:
1855 ID of the created tree.
1856 """
1857 trees: dict[bytes, TreeDict] = {b"": {}}
1859 def add_tree(path: bytes) -> TreeDict:
1860 if path in trees:
1861 return trees[path]
1862 dirname, basename = pathsplit(path)
1863 t = add_tree(dirname)
1864 assert isinstance(basename, bytes)
1865 newtree: TreeDict = {}
1866 t[basename] = newtree
1867 trees[path] = newtree
1868 return newtree
1870 for path, sha, mode in blobs:
1871 tree_path, basename = pathsplit(path)
1872 tree = add_tree(tree_path)
1873 tree[basename] = (mode, sha)
1875 def build_tree(path: bytes) -> ObjectID:
1876 tree = Tree()
1877 for basename, entry in trees[path].items():
1878 if isinstance(entry, dict):
1879 mode = stat.S_IFDIR
1880 sha = build_tree(pathjoin(path, basename))
1881 else:
1882 (mode, sha) = entry
1883 tree.add(basename, mode, sha)
1884 object_store.add_object(tree)
1885 return ObjectID(tree.get_id(object_format))
1887 return build_tree(b"")
1890def commit_index(object_store: ObjectContainer, index: Index) -> ObjectID:
1891 """Create a new tree from an index.
1893 Args:
1894 object_store: Object store to save the tree in
1895 index: Index file
1896 Note: This function is deprecated, use index.commit() instead.
1897 Returns: Root tree sha.
1898 """
1899 return commit_tree(
1900 object_store, index.iterobjects(), object_format=index.object_format
1901 )
1904def changes_from_tree(
1905 names: Iterable[bytes],
1906 lookup_entry: Callable[[bytes], tuple[bytes, int]],
1907 object_store: ObjectContainer,
1908 tree: ObjectID | None,
1909 want_unchanged: bool = False,
1910) -> Iterable[
1911 tuple[
1912 tuple[bytes | None, bytes | None],
1913 tuple[int | None, int | None],
1914 tuple[bytes | None, bytes | None],
1915 ]
1916]:
1917 """Find the differences between the contents of a tree and a working copy.
1919 Args:
1920 names: Iterable of names in the working copy
1921 lookup_entry: Function to lookup an entry in the working copy
1922 object_store: Object store to use for retrieving tree contents
1923 tree: SHA1 of the root tree, or None for an empty tree
1924 want_unchanged: Whether unchanged files should be reported
1925 Returns: Iterator over tuples with (oldpath, newpath), (oldmode, newmode),
1926 (oldsha, newsha)
1927 """
1928 # TODO(jelmer): Support a include_trees option
1929 other_names = set(names)
1931 if tree is not None:
1932 for name, mode, sha in iter_tree_contents(object_store, tree):
1933 assert name is not None and mode is not None and sha is not None
1934 try:
1935 (other_sha, other_mode) = lookup_entry(name)
1936 except KeyError:
1937 # Was removed
1938 yield ((name, None), (mode, None), (sha, None))
1939 else:
1940 other_names.remove(name)
1941 if want_unchanged or other_sha != sha or other_mode != mode:
1942 yield ((name, name), (mode, other_mode), (sha, other_sha))
1944 # Mention added files
1945 for name in other_names:
1946 try:
1947 (other_sha, other_mode) = lookup_entry(name)
1948 except KeyError:
1949 pass
1950 else:
1951 yield ((None, name), (None, other_mode), (None, other_sha))
1954def index_entry_from_stat(
1955 stat_val: os.stat_result,
1956 hex_sha: bytes,
1957 mode: int | None = None,
1958 flags: int = 0,
1959 extended_flags: int = 0,
1960) -> IndexEntry:
1961 """Create a new index entry from a stat value.
1963 Args:
1964 stat_val: POSIX stat_result instance
1965 hex_sha: Hex sha of the object
1966 mode: Optional file mode, will be derived from stat if not provided
1967 flags: Index entry flags
1968 extended_flags: Index entry extended flags
1969 """
1970 if mode is None:
1971 mode = cleanup_mode(stat_val.st_mode)
1973 from dulwich.objects import ObjectID
1975 # Use nanosecond precision when available to avoid precision loss
1976 # through float representation
1977 ctime: int | float | tuple[int, int]
1978 mtime: int | float | tuple[int, int]
1979 st_ctime_ns = getattr(stat_val, "st_ctime_ns", None)
1980 if st_ctime_ns is not None:
1981 ctime = (
1982 st_ctime_ns // 1_000_000_000,
1983 st_ctime_ns % 1_000_000_000,
1984 )
1985 else:
1986 ctime = stat_val.st_ctime
1988 st_mtime_ns = getattr(stat_val, "st_mtime_ns", None)
1989 if st_mtime_ns is not None:
1990 mtime = (
1991 st_mtime_ns // 1_000_000_000,
1992 st_mtime_ns % 1_000_000_000,
1993 )
1994 else:
1995 mtime = stat_val.st_mtime
1997 return IndexEntry(
1998 ctime=ctime,
1999 mtime=mtime,
2000 dev=stat_val.st_dev,
2001 ino=stat_val.st_ino,
2002 mode=mode,
2003 uid=stat_val.st_uid,
2004 gid=stat_val.st_gid,
2005 size=stat_val.st_size,
2006 sha=ObjectID(hex_sha),
2007 flags=flags,
2008 extended_flags=extended_flags,
2009 )
2012def unchanged_entry_flags(
2013 index: "Index | dict[bytes, IndexEntry | ConflictedIndexEntry]",
2014 path: bytes,
2015 hex_sha: bytes,
2016 mode: int,
2017) -> tuple[int, int]:
2018 """Return the flags to carry over when replacing the index entry for path.
2020 An entry keeps its flags (such as assume-unchanged or skip-worktree)
2021 as long as its sha and mode stay the same.
2023 Returns:
2024 Tuple of (flags, extended_flags); zeros if there is no matching entry
2025 """
2026 try:
2027 entry = index[path]
2028 except KeyError:
2029 return 0, 0
2030 if isinstance(entry, IndexEntry) and entry.sha == hex_sha and entry.mode == mode:
2031 return entry.flags, entry.extended_flags
2032 return 0, 0
2035def index_entry_from_tree_entry(
2036 mode: int,
2037 hex_sha: bytes,
2038 size: int = 0,
2039 flags: int = 0,
2040 extended_flags: int = 0,
2041) -> IndexEntry:
2042 """Create an index entry from a tree entry, with zeroed stat fields.
2044 Use this when populating an index directly from a tree without touching
2045 the filesystem, matching ``git read-tree``. The stat-derived fields
2046 (ctime/mtime/dev/ino/uid/gid) are zeroed; git treats such entries as
2047 stat-unmerged and refreshes them on the next stat.
2049 Args:
2050 mode: File mode from the tree entry
2051 hex_sha: Hex sha of the object
2052 size: Size of the object (0 for gitlinks or when unknown)
2053 flags: Index entry flags
2054 extended_flags: Index entry extended flags
2055 """
2056 from dulwich.objects import ObjectID
2058 return IndexEntry(
2059 ctime=0,
2060 mtime=0,
2061 dev=0,
2062 ino=0,
2063 mode=mode,
2064 uid=0,
2065 gid=0,
2066 size=size,
2067 sha=ObjectID(hex_sha),
2068 flags=flags,
2069 extended_flags=extended_flags,
2070 )
2073if sys.platform == "win32":
2074 # On Windows, creating symlinks either requires administrator privileges
2075 # or developer mode. Raise a more helpful error when we're unable to
2076 # create symlinks
2078 # https://github.com/jelmer/dulwich/issues/1005
2080 class WindowsSymlinkPermissionError(PermissionError):
2081 """Windows-specific error for symlink creation failures.
2083 This error is raised when symlink creation fails on Windows,
2084 typically due to lack of developer mode or administrator privileges.
2085 """
2087 def __init__(self, errno: int, msg: str, filename: str | None) -> None:
2088 """Initialize WindowsSymlinkPermissionError."""
2089 super().__init__(
2090 errno,
2091 f"Unable to create symlink; do you have developer mode enabled? {msg}",
2092 filename,
2093 )
2095 def symlink(
2096 src: str | bytes,
2097 dst: str | bytes,
2098 target_is_directory: bool = False,
2099 *,
2100 dir_fd: int | None = None,
2101 ) -> None:
2102 """Create a symbolic link on Windows with better error handling.
2104 Args:
2105 src: Source path for the symlink
2106 dst: Destination path where symlink will be created
2107 target_is_directory: Whether the target is a directory
2108 dir_fd: Optional directory file descriptor
2110 Raises:
2111 WindowsSymlinkPermissionError: If symlink creation fails due to permissions
2112 """
2113 try:
2114 return os.symlink(
2115 src, dst, target_is_directory=target_is_directory, dir_fd=dir_fd
2116 )
2117 except PermissionError as e:
2118 raise WindowsSymlinkPermissionError(
2119 e.errno or 0, e.strerror or "", e.filename
2120 ) from e
2121else:
2122 symlink = os.symlink
2125def get_symlink_fn(
2126 config: "Config",
2127) -> Callable[[str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None]:
2128 """Get the function to use for creating symlinks in the working tree.
2130 With ``core.symlinks`` set to false, symlinks are checked out as plain
2131 files containing the link target, like git does.
2133 Args:
2134 config: Repository configuration
2136 Returns:
2137 Function taking the link target and the path to create
2138 """
2139 if config.get_boolean(b"core", b"symlinks", True):
2141 def create_symlink(
2142 source: str | bytes | os.PathLike[str],
2143 target: str | bytes | os.PathLike[str],
2144 ) -> None:
2145 symlink(source, target) # type: ignore[arg-type,unused-ignore]
2147 return create_symlink
2149 def write_link_as_file(
2150 source: str | bytes | os.PathLike[str],
2151 target: str | bytes | os.PathLike[str],
2152 ) -> None:
2153 mode = "w" + ("b" if isinstance(source, bytes) else "")
2154 with open(target, mode) as f:
2155 f.write(source)
2157 return write_link_as_file
2160def build_file_from_blob(
2161 blob: Blob,
2162 mode: int,
2163 target_path: bytes,
2164 *,
2165 honor_filemode: bool = True,
2166 tree_encoding: str = "utf-8",
2167 symlink_fn: Callable[
2168 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2169 ]
2170 | None = None,
2171) -> os.stat_result:
2172 """Build a file or symlink on disk based on a Git object.
2174 Args:
2175 blob: The git object
2176 mode: File mode
2177 target_path: Path to write to
2178 honor_filemode: An optional flag to honor core.filemode setting in
2179 config file, default is core.filemode=True, change executable bit
2180 tree_encoding: Encoding to use for tree contents
2181 symlink_fn: Function to use for creating symlinks
2182 Returns: stat object for the file
2184 Raises:
2185 InvalidPathError: If asked to write ``.gitmodules`` as a symlink
2186 """
2187 if stat.S_ISLNK(mode) and os.path.basename(target_path).lower() == b".gitmodules":
2188 # .gitmodules is read from the work tree, so a symlink here would
2189 # source submodule configuration from a file outside the tree.
2190 raise InvalidPathError(target_path)
2191 try:
2192 oldstat = os.lstat(target_path)
2193 except FileNotFoundError:
2194 oldstat = None
2195 contents = blob.as_raw_string()
2196 if stat.S_ISLNK(mode):
2197 if oldstat:
2198 _remove_file_with_readonly_handling(target_path)
2199 if sys.platform == "win32":
2200 # os.readlink on Python3 on Windows requires a unicode string.
2201 contents_str = contents.decode(tree_encoding)
2202 target_path_str = target_path.decode(tree_encoding)
2203 (symlink_fn or symlink)(contents_str, target_path_str)
2204 else:
2205 (symlink_fn or symlink)(contents, target_path)
2206 else:
2207 if oldstat is not None and stat.S_ISLNK(oldstat.st_mode):
2208 # A symlink left at the target path must not be written through.
2209 # open(..., "wb") would follow it and clobber whatever it points
2210 # at, possibly outside the work tree. Replace it with a fresh
2211 # regular file, like git does on checkout.
2212 _remove_file_with_readonly_handling(target_path)
2213 oldstat = None
2214 if oldstat is not None and oldstat.st_size == len(contents):
2215 with open(target_path, "rb") as f:
2216 unchanged = f.read() == contents
2217 if unchanged:
2218 if not honor_filemode or bool(oldstat.st_mode & stat.S_IXUSR) == bool(
2219 mode & stat.S_IXUSR
2220 ):
2221 return oldstat
2222 os.chmod(target_path, cleanup_mode(mode))
2223 return os.lstat(target_path)
2225 with open(target_path, "wb") as f:
2226 # Write out file
2227 f.write(contents)
2229 if honor_filemode:
2230 # Canonicalize to the permission bits git honors (0o644/0o755).
2231 # The mode comes from a tree/index entry, which for an untrusted
2232 # repository can carry setuid/setgid/sticky or world-writable bits;
2233 # git reduces every regular-file mode to 0o644 or 0o755 on checkout.
2234 os.chmod(target_path, cleanup_mode(mode))
2236 return os.lstat(target_path)
2239INVALID_DOTNAMES = (b".git", b".", b"..", b"")
2242def _normalize_path_element_default(element: bytes) -> bytes:
2243 """Normalize path element for default case-insensitive comparison."""
2244 return element.lower()
2247def _normalize_path_element_ntfs(element: bytes) -> bytes:
2248 """Normalize path element for NTFS filesystem."""
2249 return element.rstrip(b". ").lower()
2252def _normalize_path_element_hfs(element: bytes) -> bytes:
2253 """Normalize path element for HFS+ filesystem."""
2254 import unicodedata
2256 # Decode to Unicode (let UnicodeDecodeError bubble up)
2257 element_str = element.decode("utf-8", errors="strict")
2259 # Remove HFS+ ignorable characters
2260 filtered = "".join(c for c in element_str if ord(c) not in HFS_IGNORABLE_CHARS)
2261 # Normalize to NFD
2262 normalized = unicodedata.normalize("NFD", filtered)
2263 return normalized.lower().encode("utf-8", errors="strict")
2266def get_path_element_normalizer(config: "Config") -> Callable[[bytes], bytes]:
2267 """Get the appropriate path element normalization function based on config.
2269 Args:
2270 config: Repository configuration object
2272 Returns:
2273 Function that normalizes path elements for the configured filesystem
2274 """
2275 import os
2276 import sys
2278 if config.get_boolean(b"core", b"protectNTFS", os.name == "nt"):
2279 return _normalize_path_element_ntfs
2280 elif config.get_boolean(b"core", b"protectHFS", sys.platform == "darwin"):
2281 return _normalize_path_element_hfs
2282 else:
2283 return _normalize_path_element_default
2286def make_path_normalizer(
2287 config: "Config",
2288) -> Callable[[bytes], bytes] | None:
2289 """Build a path normalizer honoring ``core.ignorecase`` and ``core.precomposeunicode``.
2291 The returned callable maps a filesystem-form path to a canonical form
2292 used to match equivalent paths (e.g. ``Foo.txt`` ↔ ``foo.txt`` when
2293 ``core.ignorecase=true``, NFD ↔ NFC when ``core.precomposeunicode=true``).
2294 Returns ``None`` when neither option is active so callers can skip the
2295 comparison entirely.
2296 """
2297 ignorecase = config.get_boolean(b"core", b"ignorecase", False)
2298 precompose = config.get_boolean(b"core", b"precomposeunicode", False)
2299 if not ignorecase and not precompose:
2300 return None
2302 def normalize(path: bytes) -> bytes:
2303 if precompose:
2304 import unicodedata
2306 try:
2307 path = unicodedata.normalize("NFC", path.decode("utf-8")).encode(
2308 "utf-8"
2309 )
2310 except UnicodeDecodeError:
2311 pass
2312 if ignorecase:
2313 path = path.lower()
2314 return path
2316 return normalize
2319class InvalidPathError(Exception):
2320 """Raised when a tree entry's path is unsafe to write to the work tree."""
2322 def __init__(self, path: bytes) -> None:
2323 """Initialize the exception with the offending path."""
2324 self.path = path
2325 super().__init__(f"invalid path {path.decode('utf-8', 'replace')!r}")
2328def validate_path_element_default(element: bytes) -> bool:
2329 """Validate a path element using default rules.
2331 Args:
2332 element: Path element to validate
2334 Returns:
2335 True if path element is valid, False otherwise
2336 """
2337 return _normalize_path_element_default(element) not in INVALID_DOTNAMES
2340def _is_ntfs_dotgit(name: bytes) -> bool:
2341 """Match NTFS-dangerous spellings of ``.git`` at the start of ``name``.
2343 Matches ``.git`` or the 8.3 short name ``git~1`` followed only by
2344 dots/spaces and then the end of the element, a separator, or a ``:``
2345 (an alternate-data-stream marker, as in ``.git::$INDEX_ALLOCATION``).
2346 """
2347 if name[:1] == b".":
2348 if name[1:4].lower() != b"git":
2349 return False
2350 i = 4
2351 elif name[:1].lower() == b"g": # ``git~1`` 8.3 short name
2352 if name[1:3].lower() != b"it" or name[3:5] != b"~1":
2353 return False
2354 i = 5
2355 else:
2356 return False
2358 while i < len(name):
2359 c = name[i : i + 1]
2360 if c == b":":
2361 return True
2362 if c != b"." and c != b" ":
2363 return False
2364 i += 1
2365 return True
2368# Reserved Windows device names. Opening any of these on Windows
2369# resolves to a device rather than a file, regardless of any
2370# extension or trailing dots/spaces (``NUL``, ``NUL.txt``,
2371# ``aux.foo.bar`` all hit the device). They are ordinary filenames
2372# everywhere else, so this is only enforced when running on Windows.
2373RESERVED_WINDOWS_DEVICE_NAMES = frozenset(
2374 [b"con", b"prn", b"aux", b"nul"]
2375 + [b"com%d" % i for i in range(1, 10)]
2376 + [b"lpt%d" % i for i in range(1, 10)]
2377)
2380def _is_reserved_windows_device_name(normalized: bytes) -> bool:
2381 """Match Windows reserved device names regardless of extension."""
2382 # The "stem" is the portion before the first ``.``; Windows
2383 # also strips trailing spaces from that stem when resolving.
2384 stem = normalized.split(b".", 1)[0].rstrip(b" ")
2385 return stem in RESERVED_WINDOWS_DEVICE_NAMES
2388def _has_dos_drive_prefix(name: bytes) -> bool:
2389 """Return True if name begins with a DOS drive-letter prefix like 'C:'.
2391 Any ASCII byte followed by a colon counts, matching the ASCII branch of
2392 C git's win32_has_dos_drive_prefix: subst allows non-alphabetic drives.
2393 """
2394 if len(name) < 2 or name[1:2] != b":":
2395 return False
2396 return name[0] < 0x80
2399def validate_path_element_ntfs(element: bytes) -> bool:
2400 """Validate a path element using NTFS filesystem rules.
2402 Args:
2403 element: Path element to validate
2405 Returns:
2406 True if path element is valid for NTFS, False otherwise
2407 """
2408 # A backslash is a separator on Windows but an ordinary filename
2409 # character on POSIX, so only reject it on Windows.
2410 if os.name == "nt" and b"\\" in element:
2411 return False
2412 # Backslash also separates ``.git`` spellings, so check each segment.
2413 for segment in element.split(b"\\"):
2414 if _is_ntfs_dotgit(segment):
2415 return False
2416 normalized = _normalize_path_element_ntfs(element)
2417 if normalized in INVALID_DOTNAMES:
2418 return False
2419 # Like the backslash check above, a reserved device name is only a
2420 # hazard on Windows; C git confines this to is_valid_win32_path in
2421 # compat/mingw.c, and accepts e.g. ``aux`` on POSIX (issue #2351).
2422 if os.name == "nt" and _is_reserved_windows_device_name(normalized):
2423 return False
2424 return True
2427# HFS+ ignorable Unicode codepoints (from Git's utf8.c)
2428HFS_IGNORABLE_CHARS = {
2429 0x200C, # ZERO WIDTH NON-JOINER
2430 0x200D, # ZERO WIDTH JOINER
2431 0x200E, # LEFT-TO-RIGHT MARK
2432 0x200F, # RIGHT-TO-LEFT MARK
2433 0x202A, # LEFT-TO-RIGHT EMBEDDING
2434 0x202B, # RIGHT-TO-LEFT EMBEDDING
2435 0x202C, # POP DIRECTIONAL FORMATTING
2436 0x202D, # LEFT-TO-RIGHT OVERRIDE
2437 0x202E, # RIGHT-TO-LEFT OVERRIDE
2438 0x206A, # INHIBIT SYMMETRIC SWAPPING
2439 0x206B, # ACTIVATE SYMMETRIC SWAPPING
2440 0x206C, # INHIBIT ARABIC FORM SHAPING
2441 0x206D, # ACTIVATE ARABIC FORM SHAPING
2442 0x206E, # NATIONAL DIGIT SHAPES
2443 0x206F, # NOMINAL DIGIT SHAPES
2444 0xFEFF, # ZERO WIDTH NO-BREAK SPACE
2445}
2448def validate_path_element_hfs(element: bytes) -> bool:
2449 """Validate path element for HFS+ filesystem.
2451 Equivalent to Git's is_hfs_dotgit and related checks.
2452 Uses NFD normalization and ignores HFS+ ignorable characters.
2453 """
2454 try:
2455 normalized = _normalize_path_element_hfs(element)
2456 except UnicodeDecodeError:
2457 # Malformed UTF-8 - be conservative and reject
2458 return False
2460 # Check against invalid names
2461 if normalized in INVALID_DOTNAMES:
2462 return False
2464 # Also check for 8.3 short name
2465 if normalized == b"git~1":
2466 return False
2468 return True
2471def get_path_element_validator(config: "Config") -> Callable[[bytes], bool]:
2472 """Get the path-element validator to use when checking out a tree.
2474 ``core.protectNTFS`` defaults to true on every platform (matching Git's
2475 ``PROTECT_NTFS_DEFAULT=1``) because a repository authored on POSIX can
2476 still be cloned on Windows later; ``core.protectHFS`` defaults to true on
2477 macOS. Both protections are independent and apply together, so on macOS
2478 (where both default on) a path element must satisfy the NTFS and HFS+
2479 checks. With both disabled this falls back to the default validator, which
2480 only refuses ``.git``, ``.`` and ``..``.
2482 Args:
2483 config: Repository configuration object
2485 Returns:
2486 Function that validates a single path element for the configured
2487 filesystem protections.
2488 """
2489 validators: list[Callable[[bytes], bool]] = []
2490 if config.get_boolean(b"core", b"protectNTFS", True):
2491 validators.append(validate_path_element_ntfs)
2492 if config.get_boolean(b"core", b"protectHFS", sys.platform == "darwin"):
2493 validators.append(validate_path_element_hfs)
2494 if not validators:
2495 return validate_path_element_default
2496 if len(validators) == 1:
2497 return validators[0]
2499 def validate_all(element: bytes) -> bool:
2500 return all(validator(element) for validator in validators)
2502 return validate_all
2505def validate_path(
2506 path: bytes,
2507 element_validator: Callable[[bytes], bool] = validate_path_element_default,
2508) -> bool:
2509 """Default path validator that just checks for .git/."""
2510 # A leading drive-letter prefix lets os.path.join discard the work-tree
2511 # root on Windows. Matches C git's verify_path, whose has_dos_drive_prefix
2512 # is a no-op stub on non-Windows.
2513 if os.name == "nt" and _has_dos_drive_prefix(path):
2514 return False
2515 parts = path.split(b"/")
2516 for p in parts:
2517 if not element_validator(p):
2518 return False
2519 else:
2520 return True
2523def verify_leading_dirs(
2524 tree_path: bytes,
2525 safe_prefix: list[bytes],
2526 repo_path: bytes,
2527) -> None:
2528 """Reject writes whose leading path resolves through a symlink.
2530 Callers that materialize many paths in sorted order can pass a shared
2531 ``safe_prefix`` list to cache the deepest chain of directory components
2532 already verified to be real directories (or absent); each call only
2533 ``lstat``s the components that differ from that chain. Callers that only
2534 verify a single path can pass an empty list. Mirrors git's
2535 ``lstat_cache_matchlen`` (see CVE-2021-21300).
2537 Args:
2538 tree_path: Tree-form path (``/``-separated) about to be written.
2539 safe_prefix: Mutable cache of directory components already verified
2540 under ``repo_path``. Updated in place.
2541 repo_path: Filesystem path to the work-tree root.
2543 Raises:
2544 InvalidPathError: If any leading component is a symlink.
2545 """
2546 slash = tree_path.rfind(b"/")
2547 if slash <= 0:
2548 return
2549 components = tree_path[:slash].split(b"/")
2551 common = 0
2552 while (
2553 common < len(safe_prefix)
2554 and common < len(components)
2555 and safe_prefix[common] == components[common]
2556 ):
2557 common += 1
2558 del safe_prefix[common:]
2560 current = repo_path
2561 for part in components[:common]:
2562 current = os.path.join(current, part)
2563 for part in components[common:]:
2564 current = os.path.join(current, part)
2565 try:
2566 st = os.lstat(current)
2567 except FileNotFoundError:
2568 # Anything below here doesn't exist yet; makedirs will create
2569 # it under a verified-real-directory prefix.
2570 break
2571 if stat.S_ISLNK(st.st_mode):
2572 raise InvalidPathError(tree_path)
2573 safe_prefix.append(part)
2576def verify_tree_path(
2577 tree_path: bytes,
2578 validate_path_element: Callable[[bytes], bool],
2579 safe_prefix: list[bytes],
2580 root_path: bytes,
2581) -> None:
2582 """Reject a tree path that is unsafe to write below ``root_path``.
2584 Combines ``validate_path`` and ``verify_leading_dirs``.
2586 Args:
2587 tree_path: Tree-form path (``/``-separated) about to be written.
2588 validate_path_element: Function to validate a single path element.
2589 safe_prefix: Mutable cache shared across calls, see
2590 ``verify_leading_dirs``.
2591 root_path: Filesystem path to the work-tree root.
2593 Raises:
2594 InvalidPathError: If the path has a component rejected by
2595 ``validate_path_element`` or a leading component is a symlink.
2596 """
2597 if not validate_path(tree_path, validate_path_element):
2598 raise InvalidPathError(tree_path)
2599 verify_leading_dirs(tree_path, safe_prefix, root_path)
2602def build_index_from_tree(
2603 root_path: str | bytes,
2604 index_path: str | bytes,
2605 object_store: ObjectContainer,
2606 tree_id: ObjectID,
2607 honor_filemode: bool = True,
2608 validate_path_element: Callable[[bytes], bool] = validate_path_element_default,
2609 symlink_fn: Callable[
2610 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2611 ]
2612 | None = None,
2613 blob_normalizer: "FilterBlobNormalizer | None" = None,
2614 tree_encoding: str = "utf-8",
2615 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
2616) -> None:
2617 """Generate and materialize index from a tree.
2619 Args:
2620 tree_id: Tree to materialize
2621 root_path: Target dir for materialized index files
2622 index_path: Target path for generated index
2623 object_store: Non-empty object store holding tree contents
2624 honor_filemode: An optional flag to honor core.filemode setting in
2625 config file, default is core.filemode=True, change executable bit
2626 validate_path_element: Function to validate path elements to check
2627 out; default just refuses .git and .. directories.
2628 symlink_fn: Function to use for creating symlinks
2629 blob_normalizer: An optional BlobNormalizer to use for converting line
2630 endings when writing blobs to the working directory.
2631 tree_encoding: Encoding used for tree paths (default: utf-8)
2632 object_format: Object format of the repository the index belongs to
2634 Note: existing index is wiped and contents are not merged
2635 in a working dir. Suitable only for fresh clones.
2636 """
2637 index = Index(index_path, read=False, object_format=object_format)
2638 if not isinstance(root_path, bytes):
2639 root_path = os.fsencode(root_path)
2641 # Cache of leading directory components already verified to be real
2642 # directories, shared across the sorted iteration. See verify_leading_dirs.
2643 safe_prefix: list[bytes] = []
2645 for entry in iter_tree_contents(object_store, tree_id):
2646 assert (
2647 entry.path is not None and entry.mode is not None and entry.sha is not None
2648 )
2649 # Validate as we go and abort on the first invalid path, leaving any
2650 # files already written in place. This also refuses a leading path
2651 # that resolves through a symlink materialized by an earlier entry.
2652 verify_tree_path(entry.path, validate_path_element, safe_prefix, root_path)
2653 full_path = _tree_to_fs_path(root_path, entry.path, tree_encoding)
2655 if not os.path.exists(os.path.dirname(full_path)):
2656 os.makedirs(os.path.dirname(full_path))
2658 # TODO(jelmer): Merge new index into working tree
2659 if S_ISGITLINK(entry.mode):
2660 if not os.path.isdir(full_path):
2661 os.mkdir(full_path)
2662 st = os.lstat(full_path)
2663 # TODO(jelmer): record and return submodule paths
2664 else:
2665 obj = object_store[entry.sha]
2666 assert isinstance(obj, Blob)
2667 # Apply blob normalization for checkout if normalizer is provided
2668 if blob_normalizer is not None:
2669 obj = blob_normalizer.checkout_normalize(obj, entry.path)
2670 st = build_file_from_blob(
2671 obj,
2672 entry.mode,
2673 full_path,
2674 honor_filemode=honor_filemode,
2675 tree_encoding=tree_encoding,
2676 symlink_fn=symlink_fn,
2677 )
2679 # Add file to index
2680 if not honor_filemode or S_ISGITLINK(entry.mode):
2681 # we can not use tuple slicing to build a new tuple,
2682 # because on windows that will convert the times to
2683 # longs, which causes errors further along
2684 st_tuple = (
2685 entry.mode,
2686 st.st_ino,
2687 st.st_dev,
2688 st.st_nlink,
2689 st.st_uid,
2690 st.st_gid,
2691 st.st_size,
2692 st.st_atime,
2693 st.st_mtime,
2694 st.st_ctime,
2695 )
2696 st = st.__class__(st_tuple)
2697 # default to a stage 0 index entry (normal)
2698 # when reading from the filesystem
2699 index[entry.path] = index_entry_from_stat(st, entry.sha)
2701 index.write()
2704def blob_from_path_and_mode(
2705 fs_path: bytes, mode: int, tree_encoding: str = "utf-8"
2706) -> Blob:
2707 """Create a blob from a path and a stat object.
2709 Args:
2710 fs_path: Full file system path to file
2711 mode: File mode
2712 tree_encoding: Encoding to use for tree contents
2713 Returns: A `Blob` object
2714 """
2715 assert isinstance(fs_path, bytes)
2716 blob = Blob()
2717 if stat.S_ISLNK(mode):
2718 if sys.platform == "win32":
2719 # os.readlink on Python3 on Windows requires a unicode string.
2720 blob.data = os.readlink(os.fsdecode(fs_path)).encode(tree_encoding)
2721 else:
2722 blob.data = os.readlink(fs_path)
2723 else:
2724 with open(fs_path, "rb") as f:
2725 blob.data = f.read()
2726 return blob
2729def blob_from_path_and_stat(
2730 fs_path: bytes, st: os.stat_result, tree_encoding: str = "utf-8"
2731) -> Blob:
2732 """Create a blob from a path and a stat object.
2734 Args:
2735 fs_path: Full file system path to file
2736 st: A stat object
2737 tree_encoding: Encoding to use for tree contents
2738 Returns: A `Blob` object
2739 """
2740 return blob_from_path_and_mode(fs_path, st.st_mode, tree_encoding)
2743def read_submodule_head(path: str | bytes) -> bytes | None:
2744 """Read the head commit of a submodule.
2746 Args:
2747 path: path to the submodule
2748 Returns: HEAD sha, None if not a valid head/repository
2749 """
2750 from .errors import NotGitRepository
2751 from .repo import Repo
2753 # Repo currently expects a "str", so decode if necessary.
2754 # TODO(jelmer): Perhaps move this into Repo() ?
2755 if not isinstance(path, str):
2756 path = os.fsdecode(path)
2757 try:
2758 repo = Repo(path)
2759 except NotGitRepository:
2760 return None
2761 try:
2762 return repo.head()
2763 except KeyError:
2764 return None
2767def _has_directory_changed(tree_path: bytes, entry: IndexEntry) -> bool:
2768 """Check if a directory has changed after getting an error.
2770 When handling an error trying to create a blob from a path, call this
2771 function. It will check if the path is a directory. If it's a directory
2772 and a submodule, check the submodule head to see if it's has changed. If
2773 not, consider the file as changed as Git tracked a file and not a
2774 directory.
2776 Return true if the given path should be considered as changed and False
2777 otherwise or if the path is not a directory.
2778 """
2779 # This is actually a directory
2780 if os.path.exists(os.path.join(tree_path, b".git")):
2781 # Submodule
2782 head = read_submodule_head(tree_path)
2783 if entry.sha != head:
2784 return True
2785 else:
2786 # The file was changed to a directory, so consider it removed.
2787 return True
2789 return False
2792os_sep_bytes = os.sep.encode("ascii")
2795def _ensure_parent_dir_exists(full_path: bytes) -> None:
2796 """Ensure parent directory exists, checking no parent is a file."""
2797 parent_dir = os.path.dirname(full_path)
2798 if parent_dir and not os.path.exists(parent_dir):
2799 # Walk up the directory tree to find the first existing parent
2800 current = parent_dir
2801 parents_to_check: list[bytes] = []
2803 while current and not os.path.exists(current):
2804 parents_to_check.insert(0, current)
2805 new_parent = os.path.dirname(current)
2806 if new_parent == current:
2807 # Reached the root or can't go up further
2808 break
2809 current = new_parent
2811 # Check if the existing parent (if any) is a directory
2812 if current and os.path.exists(current) and not os.path.isdir(current):
2813 raise OSError(
2814 f"Cannot create directory, parent path is a file: {current!r}"
2815 )
2817 # Now check each parent we need to create isn't blocked by an existing file
2818 for parent_path in parents_to_check:
2819 if os.path.exists(parent_path) and not os.path.isdir(parent_path):
2820 raise OSError(
2821 f"Cannot create directory, parent path is a file: {parent_path!r}"
2822 )
2824 os.makedirs(parent_dir)
2827def _remove_file_with_readonly_handling(path: bytes) -> None:
2828 """Remove a file, handling read-only files on Windows.
2830 Args:
2831 path: Path to the file to remove
2832 """
2833 try:
2834 os.unlink(path)
2835 except PermissionError:
2836 # On Windows, remove read-only attribute and retry
2837 if sys.platform == "win32":
2838 os.chmod(path, stat.S_IWRITE | stat.S_IREAD)
2839 os.unlink(path)
2840 else:
2841 raise
2844def _remove_empty_parents(path: bytes, stop_at: bytes) -> None:
2845 """Remove empty parent directories up to stop_at."""
2846 parent = os.path.dirname(path)
2847 while parent and parent != stop_at:
2848 try:
2849 os.rmdir(parent)
2850 parent = os.path.dirname(parent)
2851 except FileNotFoundError:
2852 # Directory doesn't exist - stop trying
2853 break
2854 except OSError as e:
2855 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
2856 # Directory not empty - stop trying
2857 break
2858 raise
2861def _check_symlink_matches(
2862 full_path: bytes, repo_object_store: "BaseObjectStore", entry_sha: ObjectID
2863) -> bool:
2864 """Check if symlink target matches expected target.
2866 Returns True if symlink matches, False if it doesn't match.
2867 """
2868 try:
2869 current_target = os.readlink(full_path)
2870 blob_obj = repo_object_store[entry_sha]
2871 expected_target = blob_obj.as_raw_string()
2872 if isinstance(current_target, str):
2873 current_target = current_target.encode()
2874 return current_target == expected_target
2875 except FileNotFoundError:
2876 # Symlink doesn't exist
2877 return False
2878 except OSError as e:
2879 if e.errno == errno.EINVAL:
2880 # Not a symlink
2881 return False
2882 raise
2885def _check_file_matches(
2886 repo_object_store: "BaseObjectStore",
2887 full_path: bytes,
2888 entry_sha: ObjectID,
2889 entry_mode: int,
2890 current_stat: os.stat_result,
2891 honor_filemode: bool,
2892 blob_normalizer: "FilterBlobNormalizer | None" = None,
2893 tree_path: bytes | None = None,
2894) -> bool:
2895 """Check if a file on disk matches the expected git object.
2897 Returns True if file matches, False if it doesn't match.
2898 """
2899 # Check mode first (if honor_filemode is True)
2900 if honor_filemode:
2901 current_mode = stat.S_IMODE(current_stat.st_mode)
2902 expected_mode = stat.S_IMODE(entry_mode)
2904 # For regular files, only check the user executable bit, not group/other permissions
2905 # This matches Git's behavior where umask differences don't count as modifications
2906 if stat.S_ISREG(current_stat.st_mode):
2907 # Normalize regular file modes to ignore group/other write permissions
2908 current_mode_normalized = (
2909 current_mode & 0o755
2910 ) # Keep only user rwx and all read+execute
2911 expected_mode_normalized = expected_mode & 0o755
2913 # For Git compatibility, regular files should be either 644 or 755
2914 if expected_mode_normalized not in (0o644, 0o755):
2915 expected_mode_normalized = 0o644 # Default for regular files
2916 if current_mode_normalized not in (0o644, 0o755):
2917 # Determine if it should be executable based on user execute bit
2918 if current_mode & 0o100: # User execute bit is set
2919 current_mode_normalized = 0o755
2920 else:
2921 current_mode_normalized = 0o644
2923 if current_mode_normalized != expected_mode_normalized:
2924 return False
2925 else:
2926 # For non-regular files (symlinks, etc.), check mode exactly
2927 if current_mode != expected_mode:
2928 return False
2930 blob_obj = repo_object_store[entry_sha]
2931 normalize = blob_normalizer is not None and tree_path is not None
2932 # Filters can change the size, so only take the shortcut without them.
2933 if not normalize and current_stat.st_size != blob_obj.raw_length():
2934 return False
2936 try:
2937 with open(full_path, "rb") as f:
2938 current_content = f.read()
2939 except (FileNotFoundError, PermissionError, IsADirectoryError):
2940 return False
2942 # Like git, clean the working tree file and compare it with the blob.
2943 if normalize:
2944 assert blob_normalizer is not None and tree_path is not None
2945 current_content = blob_normalizer.checkin_normalize(
2946 Blob.from_string(current_content), tree_path
2947 ).as_raw_string()
2948 return current_content == blob_obj.as_raw_string()
2951def _transition_to_submodule(
2952 repo: "Repo",
2953 path: bytes,
2954 full_path: bytes,
2955 current_stat: os.stat_result | None,
2956 entry: IndexEntry | TreeEntry,
2957 index: Index,
2958) -> None:
2959 """Transition any type to submodule."""
2960 from .submodule import ensure_submodule_placeholder
2962 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2963 # Already a directory, just ensure .git file exists
2964 ensure_submodule_placeholder(repo, path)
2965 else:
2966 # Remove whatever is there and create submodule
2967 if current_stat is not None:
2968 _remove_file_with_readonly_handling(full_path)
2969 ensure_submodule_placeholder(repo, path)
2971 st = os.lstat(full_path)
2972 assert entry.sha is not None and entry.mode is not None
2973 flags, extended_flags = unchanged_entry_flags(index, path, entry.sha, entry.mode)
2974 index[path] = index_entry_from_stat(
2975 st, entry.sha, flags=flags, extended_flags=extended_flags
2976 )
2979def _transition_to_file(
2980 object_store: "BaseObjectStore",
2981 path: bytes,
2982 full_path: bytes,
2983 current_stat: os.stat_result | None,
2984 entry: IndexEntry | TreeEntry,
2985 index: Index,
2986 honor_filemode: bool,
2987 symlink_fn: Callable[
2988 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2989 ]
2990 | None,
2991 blob_normalizer: "FilterBlobNormalizer | None",
2992 tree_encoding: str = "utf-8",
2993) -> None:
2994 """Transition any type to regular file or symlink."""
2995 assert entry.sha is not None and entry.mode is not None
2996 # Without core.filemode the working tree's executable bit is not
2997 # trusted, so record the tree's mode in the index like git does.
2998 index_mode = None if honor_filemode else entry.mode
2999 flags, extended_flags = unchanged_entry_flags(index, path, entry.sha, entry.mode)
3000 # Check if we need to update
3001 if (
3002 current_stat is not None
3003 and stat.S_ISREG(current_stat.st_mode)
3004 and not stat.S_ISLNK(entry.mode)
3005 ):
3006 # File to file - check if update needed
3007 file_matches = _check_file_matches(
3008 object_store,
3009 full_path,
3010 entry.sha,
3011 entry.mode,
3012 current_stat,
3013 honor_filemode,
3014 blob_normalizer,
3015 path,
3016 )
3017 needs_update = not file_matches
3018 elif (
3019 current_stat is not None
3020 and stat.S_ISLNK(current_stat.st_mode)
3021 and stat.S_ISLNK(entry.mode)
3022 ):
3023 # Symlink to symlink - check if update needed
3024 symlink_matches = _check_symlink_matches(full_path, object_store, entry.sha)
3025 needs_update = not symlink_matches
3026 else:
3027 needs_update = True
3029 if not needs_update:
3030 # Just update index - current_stat should always be valid here since we're not updating
3031 assert current_stat is not None
3032 index[path] = index_entry_from_stat(
3033 current_stat,
3034 entry.sha,
3035 mode=index_mode,
3036 flags=flags,
3037 extended_flags=extended_flags,
3038 )
3039 return
3041 # Remove existing entry if needed
3042 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
3043 # Remove directory
3044 dir_contents = set(os.listdir(full_path))
3045 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
3047 if git_file_name in dir_contents:
3048 if dir_contents != {git_file_name}:
3049 raise IsADirectoryError(
3050 f"Cannot replace submodule with untracked files: {full_path!r}"
3051 )
3052 shutil.rmtree(full_path)
3053 else:
3054 try:
3055 os.rmdir(full_path)
3056 except OSError as e:
3057 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
3058 raise IsADirectoryError(
3059 f"Cannot replace non-empty directory with file: {full_path!r}"
3060 )
3061 raise
3062 elif current_stat is not None:
3063 _remove_file_with_readonly_handling(full_path)
3065 # Ensure parent directory exists
3066 _ensure_parent_dir_exists(full_path)
3068 # Write the file
3069 blob_obj = object_store[entry.sha]
3070 assert isinstance(blob_obj, Blob)
3071 if blob_normalizer:
3072 blob_obj = blob_normalizer.checkout_normalize(blob_obj, path)
3073 st = build_file_from_blob(
3074 blob_obj,
3075 entry.mode,
3076 full_path,
3077 honor_filemode=honor_filemode,
3078 tree_encoding=tree_encoding,
3079 symlink_fn=symlink_fn,
3080 )
3081 index[path] = index_entry_from_stat(
3082 st, entry.sha, mode=index_mode, flags=flags, extended_flags=extended_flags
3083 )
3086def _transition_to_absent(
3087 repo: "Repo",
3088 path: bytes,
3089 full_path: bytes,
3090 current_stat: os.stat_result | None,
3091 index: Index,
3092) -> None:
3093 """Remove any type of entry."""
3094 # The index entry has to go even if there is nothing on disk, e.g. for
3095 # sparse-excluded paths or files deleted without "git rm".
3096 try:
3097 del index[path]
3098 except KeyError:
3099 pass
3101 if current_stat is None:
3102 return
3104 if stat.S_ISDIR(current_stat.st_mode):
3105 # Check if it's a submodule directory
3106 dir_contents = set(os.listdir(full_path))
3107 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
3109 if git_file_name in dir_contents and dir_contents == {git_file_name}:
3110 shutil.rmtree(full_path)
3111 else:
3112 try:
3113 os.rmdir(full_path)
3114 except OSError as e:
3115 if e.errno not in (errno.ENOTEMPTY, errno.EEXIST):
3116 raise
3117 else:
3118 _remove_file_with_readonly_handling(full_path)
3120 # Try to remove empty parent directories
3121 _remove_empty_parents(
3122 full_path, repo.path if isinstance(repo.path, bytes) else repo.path.encode()
3123 )
3126def detect_case_only_renames(
3127 changes: Sequence["TreeChange"],
3128 config: "Config",
3129) -> list["TreeChange"]:
3130 """Detect and transform case-only renames in a list of tree changes.
3132 This function identifies file renames that only differ in case (e.g.,
3133 README.txt -> readme.txt) and transforms matching ADD/DELETE pairs into
3134 CHANGE_RENAME operations. It uses filesystem-appropriate path normalization
3135 based on the repository configuration.
3137 Args:
3138 changes: List of TreeChange objects representing file changes
3139 config: Repository configuration object
3141 Returns:
3142 New list of TreeChange objects with case-only renames converted to CHANGE_RENAME
3143 """
3144 from .diff_tree import (
3145 CHANGE_ADD,
3146 CHANGE_COPY,
3147 CHANGE_DELETE,
3148 CHANGE_MODIFY,
3149 CHANGE_RENAME,
3150 TreeChange,
3151 )
3153 # Build dictionaries of old and new paths with their normalized forms
3154 old_paths_normalized = {}
3155 new_paths_normalized = {}
3156 old_changes = {} # Map from old path to change object
3157 new_changes = {} # Map from new path to change object
3159 # Get the appropriate normalizer based on config
3160 normalize_func = get_path_element_normalizer(config)
3162 def normalize_path(path: bytes) -> bytes:
3163 """Normalize entire path using element normalization."""
3164 return b"/".join(normalize_func(part) for part in path.split(b"/"))
3166 # Pre-normalize all paths once to avoid repeated normalization
3167 for change in changes:
3168 if change.type == CHANGE_DELETE and change.old:
3169 assert change.old.path is not None
3170 try:
3171 normalized = normalize_path(change.old.path)
3172 except UnicodeDecodeError:
3173 logger.warning(
3174 "Skipping case-only rename detection for path with invalid UTF-8: %r",
3175 change.old.path,
3176 )
3177 else:
3178 old_paths_normalized[normalized] = change.old.path
3179 old_changes[change.old.path] = change
3180 elif change.type == CHANGE_RENAME and change.old:
3181 assert change.old.path is not None
3182 # Treat RENAME as DELETE + ADD for case-only detection
3183 try:
3184 normalized = normalize_path(change.old.path)
3185 except UnicodeDecodeError:
3186 logger.warning(
3187 "Skipping case-only rename detection for path with invalid UTF-8: %r",
3188 change.old.path,
3189 )
3190 else:
3191 old_paths_normalized[normalized] = change.old.path
3192 old_changes[change.old.path] = change
3194 if (
3195 change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY)
3196 and change.new
3197 ):
3198 assert change.new.path is not None
3199 try:
3200 normalized = normalize_path(change.new.path)
3201 except UnicodeDecodeError:
3202 logger.warning(
3203 "Skipping case-only rename detection for path with invalid UTF-8: %r",
3204 change.new.path,
3205 )
3206 else:
3207 new_paths_normalized[normalized] = change.new.path
3208 new_changes[change.new.path] = change
3210 # Find case-only renames and transform changes
3211 case_only_renames = set()
3212 new_rename_changes = []
3214 for norm_path, old_path in old_paths_normalized.items():
3215 if norm_path in new_paths_normalized:
3216 new_path = new_paths_normalized[norm_path]
3217 if old_path != new_path:
3218 # Found a case-only rename
3219 old_change = old_changes[old_path]
3220 new_change = new_changes[new_path]
3222 # Create a CHANGE_RENAME to replace the DELETE and ADD/MODIFY pair
3223 if new_change.type == CHANGE_ADD:
3224 # Simple case: DELETE + ADD becomes RENAME
3225 rename_change = TreeChange(
3226 CHANGE_RENAME, old_change.old, new_change.new
3227 )
3228 else:
3229 # Complex case: DELETE + MODIFY becomes RENAME
3230 # Use the old file from DELETE and new file from MODIFY
3231 rename_change = TreeChange(
3232 CHANGE_RENAME, old_change.old, new_change.new
3233 )
3235 new_rename_changes.append(rename_change)
3237 # Mark the old changes for removal
3238 case_only_renames.add(old_change)
3239 case_only_renames.add(new_change)
3241 # Return new list with original ADD/DELETE changes replaced by renames
3242 result = [change for change in changes if change not in case_only_renames]
3243 result.extend(new_rename_changes)
3244 return result
3247def _sparse_checkout_matcher(
3248 repo: "Repo", config: "Config", tree_encoding: str
3249) -> Callable[[bytes], bool] | None:
3250 """Return a predicate for paths inside the sparse checkout, if enabled."""
3251 if not config.get_boolean((b"core",), b"sparseCheckout", False):
3252 return None
3253 worktree = repo.get_worktree()
3254 if not os.path.exists(worktree._sparse_checkout_file_path()):
3255 # Without a patterns file, sparse checkout is not in effect.
3256 return None
3257 from .sparse_patterns import parse_sparse_patterns, sparse_path_matcher
3259 lines = worktree.get_sparse_checkout_patterns()
3260 if not parse_sparse_patterns(lines):
3261 # An existing file without patterns excludes everything.
3262 return lambda path: False
3263 matcher = sparse_path_matcher(
3264 lines,
3265 worktree.infer_cone_mode(),
3266 ignorecase=config.get_boolean((b"core",), b"ignorecase", False),
3267 )
3268 return lambda path: matcher(path.decode(tree_encoding))
3271def _skip_worktree_entry(sha: ObjectID, mode: int) -> IndexEntry:
3272 entry = IndexEntry(
3273 ctime=0,
3274 mtime=0,
3275 dev=0,
3276 ino=0,
3277 mode=mode,
3278 uid=0,
3279 gid=0,
3280 size=0,
3281 sha=sha,
3282 )
3283 entry.set_skip_worktree(True)
3284 return entry
3287def update_working_tree(
3288 repo: "Repo",
3289 old_tree_id: bytes | None,
3290 new_tree_id: bytes,
3291 change_iterator: Iterator["TreeChange"],
3292 honor_filemode: bool = True,
3293 validate_path_element: Callable[[bytes], bool] | None = None,
3294 symlink_fn: Callable[
3295 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
3296 ]
3297 | None = None,
3298 force_remove_untracked: bool = False,
3299 blob_normalizer: "FilterBlobNormalizer | None" = None,
3300 tree_encoding: str = "utf-8",
3301 allow_overwrite_modified: bool = False,
3302 *,
3303 config: "Config | None" = None,
3304) -> None:
3305 """Update the working tree and index to match a new tree.
3307 This function handles:
3308 - Adding new files
3309 - Updating modified files
3310 - Removing deleted files
3311 - Cleaning up empty directories
3313 If ``core.sparseCheckout`` is enabled, paths outside the sparse-checkout
3314 patterns are recorded in the index with the skip-worktree bit set and
3315 kept out of the working tree.
3317 Args:
3318 repo: Repository object
3319 old_tree_id: SHA of the tree before the update
3320 new_tree_id: SHA of the tree to update to
3321 change_iterator: Iterator of TreeChange objects to apply
3322 honor_filemode: An optional flag to honor core.filemode setting
3323 validate_path_element: Function to validate path elements to check out.
3324 If None, derived from ``config`` so that ``core.protectNTFS`` and
3325 ``core.protectHFS`` are honored by default.
3326 symlink_fn: Function to use for creating symlinks
3327 force_remove_untracked: If True, remove files that exist in working
3328 directory but not in target tree, even if old_tree_id is None
3329 blob_normalizer: An optional BlobNormalizer to use for converting line
3330 endings when writing blobs to the working directory.
3331 tree_encoding: Encoding used for tree paths (default: utf-8)
3332 allow_overwrite_modified: If False, raise an error when attempting to
3333 overwrite files that have been modified compared to old_tree_id
3334 config: Repository configuration. If None, falls back to
3335 ``repo.get_config_stack()``.
3336 """
3337 from .diff_tree import (
3338 CHANGE_ADD,
3339 CHANGE_COPY,
3340 CHANGE_DELETE,
3341 CHANGE_MODIFY,
3342 CHANGE_RENAME,
3343 CHANGE_UNCHANGED,
3344 )
3346 if force_remove_untracked:
3347 import warnings
3349 warnings.warn(
3350 "force_remove_untracked is a no-op and will be removed in a future release",
3351 DeprecationWarning,
3352 )
3353 repo_path = repo.path if isinstance(repo.path, bytes) else repo.path.encode()
3354 if config is None:
3355 config = repo.get_config_stack()
3357 if validate_path_element is None:
3358 # Derive the validator from config so callers cannot accidentally
3359 # skip ``core.protectNTFS``/``core.protectHFS`` enforcement by
3360 # omitting this argument.
3361 validate_path_element = get_path_element_validator(config)
3363 index = repo.open_index(config=config)
3364 in_sparse_checkout = _sparse_checkout_matcher(repo, config, tree_encoding)
3366 # Convert iterator to list since we need multiple passes
3367 changes = list(change_iterator)
3369 # Transform case-only renames on case-insensitive filesystems
3370 import platform
3372 default_ignore_case = platform.system() in ("Windows", "Darwin")
3373 config = repo.get_config()
3374 ignore_case = config.get_boolean((b"core",), b"ignorecase", default_ignore_case)
3376 if ignore_case:
3377 config = repo.get_config()
3378 changes = detect_case_only_renames(changes, config)
3380 if not allow_overwrite_modified:
3381 # Check for path conflicts where files need to become directories
3382 paths_becoming_dirs = set()
3383 for change in changes:
3384 if change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY):
3385 assert change.new is not None
3386 path = change.new.path
3387 assert path is not None
3388 if b"/" in path: # This is a file inside a directory
3389 # Check if any parent path exists as a file in the old tree or changes
3390 parts = path.split(b"/")
3391 for i in range(1, len(parts)):
3392 parent = b"/".join(parts[:i])
3393 # See if this parent path is being deleted (was a file, becoming a dir)
3394 for other_change in changes:
3395 if (
3396 other_change.type == CHANGE_DELETE
3397 and other_change.old
3398 and other_change.old.path == parent
3399 ):
3400 paths_becoming_dirs.add(parent)
3402 # Check if any path that needs to become a directory has been modified
3403 for path in paths_becoming_dirs:
3404 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3405 try:
3406 current_stat = os.lstat(full_path)
3407 except FileNotFoundError:
3408 continue # File doesn't exist, nothing to check
3409 except OSError as e:
3410 raise OSError(
3411 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3412 ) from e
3414 if stat.S_ISREG(current_stat.st_mode):
3415 # Find the old entry for this path
3416 old_change = None
3417 for change in changes:
3418 if (
3419 change.type == CHANGE_DELETE
3420 and change.old
3421 and change.old.path == path
3422 ):
3423 old_change = change
3424 break
3426 if old_change:
3427 # Check if file has been modified
3428 assert old_change.old is not None
3429 assert (
3430 old_change.old.sha is not None
3431 and old_change.old.mode is not None
3432 )
3433 file_matches = _check_file_matches(
3434 repo.object_store,
3435 full_path,
3436 old_change.old.sha,
3437 old_change.old.mode,
3438 current_stat,
3439 honor_filemode,
3440 blob_normalizer,
3441 path,
3442 )
3443 if not file_matches:
3444 raise OSError(
3445 f"Cannot replace modified file with directory: {path!r}"
3446 )
3448 # Check for uncommitted modifications before making any changes
3449 if not allow_overwrite_modified and old_tree_id:
3450 for change in changes:
3451 # Only check files that are being modified or deleted
3452 if change.type in (CHANGE_MODIFY, CHANGE_DELETE) and change.old:
3453 path = change.old.path
3454 assert path is not None
3455 if not validate_path(path, validate_path_element):
3456 continue
3458 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3459 try:
3460 current_stat = os.lstat(full_path)
3461 except FileNotFoundError:
3462 continue # File doesn't exist, nothing to check
3463 except OSError as e:
3464 raise OSError(
3465 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3466 ) from e
3468 if stat.S_ISREG(current_stat.st_mode):
3469 # Check if working tree file differs from old tree
3470 assert change.old.sha is not None and change.old.mode is not None
3471 file_matches = _check_file_matches(
3472 repo.object_store,
3473 full_path,
3474 change.old.sha,
3475 change.old.mode,
3476 current_stat,
3477 honor_filemode,
3478 blob_normalizer,
3479 path,
3480 )
3481 if not file_matches:
3482 from .errors import WorkingTreeModifiedError
3484 raise WorkingTreeModifiedError(
3485 f"Your local changes to '{path.decode('utf-8', errors='replace')}' "
3486 f"would be overwritten by checkout. "
3487 f"Please commit your changes or stash them before you switch branches."
3488 )
3490 # Apply deletes before adds so that a directory being replaced by a file
3491 # has its contents removed first (and vice versa, a file being replaced by
3492 # a directory is removed before its new children are written).
3493 for change in changes:
3494 if change.type in (CHANGE_DELETE, CHANGE_RENAME):
3495 # Remove file/directory
3496 assert change.old is not None and change.old.path is not None
3497 path = change.old.path
3498 if not validate_path(path, validate_path_element):
3499 continue
3501 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3502 try:
3503 verify_leading_dirs(path, [], repo_path)
3504 except InvalidPathError:
3505 # Removing through the symlink would unlink a path outside
3506 # the work tree; only drop the index entry, as git does.
3507 logger.warning(
3508 "Not removing %r: leading path is not a real directory", path
3509 )
3510 _transition_to_absent(repo, path, full_path, None, index)
3511 continue
3512 try:
3513 delete_stat: os.stat_result | None = os.lstat(full_path)
3514 except FileNotFoundError:
3515 delete_stat = None
3516 except OSError as e:
3517 raise OSError(
3518 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3519 ) from e
3521 _transition_to_absent(repo, path, full_path, delete_stat, index)
3523 for change in changes:
3524 if change.type in (
3525 CHANGE_ADD,
3526 CHANGE_MODIFY,
3527 CHANGE_UNCHANGED,
3528 CHANGE_COPY,
3529 CHANGE_RENAME,
3530 ):
3531 # Add or modify file
3532 assert (
3533 change.new is not None
3534 and change.new.path is not None
3535 and change.new.mode is not None
3536 and change.new.sha is not None
3537 )
3538 path = change.new.path
3539 # Validate as we go and abort on the first invalid path,
3540 # leaving any changes already applied in place.
3541 verify_tree_path(path, validate_path_element, [], repo_path)
3542 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3543 try:
3544 modify_stat: os.stat_result | None = os.lstat(full_path)
3545 except FileNotFoundError:
3546 modify_stat = None
3547 except OSError as e:
3548 raise OSError(
3549 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3550 ) from e
3552 if in_sparse_checkout is not None and not in_sparse_checkout(path):
3553 # Only record paths outside the sparse checkout in the index.
3554 # Remove them from disk only if they were checked out before,
3555 # leaving untracked files alone.
3556 try:
3557 old_entry = index[path]
3558 except KeyError:
3559 pass
3560 else:
3561 if not (
3562 isinstance(old_entry, IndexEntry) and old_entry.skip_worktree
3563 ):
3564 _transition_to_absent(repo, path, full_path, modify_stat, index)
3565 index[path] = _skip_worktree_entry(change.new.sha, change.new.mode)
3566 elif S_ISGITLINK(change.new.mode):
3567 _transition_to_submodule(
3568 repo, path, full_path, modify_stat, change.new, index
3569 )
3570 else:
3571 _transition_to_file(
3572 repo.object_store,
3573 path,
3574 full_path,
3575 modify_stat,
3576 change.new,
3577 index,
3578 honor_filemode,
3579 symlink_fn,
3580 blob_normalizer,
3581 tree_encoding,
3582 )
3583 if in_sparse_checkout is not None and in_sparse_checkout(path):
3584 # The path is now checked out, so a skip-worktree bit kept
3585 # from the old entry no longer applies.
3586 checked_out = index[path]
3587 assert isinstance(checked_out, IndexEntry)
3588 checked_out.set_skip_worktree(False)
3590 index.write()
3593def _stat_matches_entry(
3594 st: os.stat_result, entry: IndexEntry, trust_ctime: bool = True
3595) -> bool:
3596 """Check if filesystem stat matches index entry stat.
3598 This is used to determine if a file might have changed without reading its content.
3599 Git uses this optimization to avoid expensive filter operations on unchanged files.
3601 Args:
3602 st: Filesystem stat result
3603 entry: Index entry to compare against
3604 trust_ctime: If True, also check ctime (default: True, matching Git behavior)
3605 Returns: True if stat matches and file is likely unchanged
3606 """
3607 # Compare change time (ctime) if trust_ctime is enabled
3608 if trust_ctime:
3609 # Get entry ctime with nanosecond precision if available
3610 if isinstance(entry.ctime, tuple):
3611 entry_ctime_sec = entry.ctime[0]
3612 entry_ctime_nsec = entry.ctime[1]
3613 else:
3614 entry_ctime_sec = int(entry.ctime)
3615 entry_ctime_nsec = 0
3617 if hasattr(st, "st_ctime_ns"):
3618 # Use nanosecond precision when available
3619 st_ctime_nsec = st.st_ctime_ns
3620 entry_ctime_nsec_total = entry_ctime_sec * 1_000_000_000 + entry_ctime_nsec
3621 if st_ctime_nsec != entry_ctime_nsec_total:
3622 return False
3623 else:
3624 # Fall back to second precision
3625 if int(st.st_ctime) != entry_ctime_sec:
3626 return False
3628 # Get entry mtime with nanosecond precision if available
3629 if isinstance(entry.mtime, tuple):
3630 entry_mtime_sec = entry.mtime[0]
3631 entry_mtime_nsec = entry.mtime[1]
3632 else:
3633 entry_mtime_sec = int(entry.mtime)
3634 entry_mtime_nsec = 0
3636 # Compare modification time with nanosecond precision if available
3637 # This is important for fast workflows (e.g., stash) where files can be
3638 # modified multiple times within the same second
3639 if hasattr(st, "st_mtime_ns"):
3640 # Use nanosecond precision when available
3641 st_mtime_nsec = st.st_mtime_ns
3642 entry_mtime_nsec_total = entry_mtime_sec * 1_000_000_000 + entry_mtime_nsec
3643 if st_mtime_nsec != entry_mtime_nsec_total:
3644 return False
3645 else:
3646 # Fall back to second precision
3647 if int(st.st_mtime) != entry_mtime_sec:
3648 return False
3650 # Compare file size
3651 if st.st_size != entry.size:
3652 return False
3654 # If all checks pass, file is likely unchanged
3655 return True
3658def _check_entry_for_changes(
3659 tree_path: bytes,
3660 entry: IndexEntry | ConflictedIndexEntry,
3661 root_path: bytes,
3662 filter_blob_callback: Callable[[Blob, bytes], Blob] | None = None,
3663 trust_ctime: bool = True,
3664 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
3665 refresh_stat: list[tuple[bytes, os.stat_result]] | None = None,
3666) -> bytes | None:
3667 """Check a single index entry for changes.
3669 Args:
3670 tree_path: Path in the tree
3671 entry: Index entry to check
3672 root_path: Root filesystem path
3673 filter_blob_callback: Optional callback to filter blobs
3674 trust_ctime: If True, use ctime for change detection (default: True)
3675 object_format: Object format to hash the working tree blob with
3676 refresh_stat: If provided, entries with drifted stat but matching
3677 content have their (tree_path, stat_result) appended so the
3678 caller can update the index stat cache in a single pass.
3679 Returns: tree_path if changed, None otherwise
3680 """
3681 if isinstance(entry, ConflictedIndexEntry):
3682 # Conflicted files are always unstaged
3683 return tree_path
3685 if entry.skip_worktree:
3686 # Paths outside the sparse checkout are not expected on disk.
3687 return None
3689 full_path = _tree_to_fs_path(root_path, tree_path)
3690 try:
3691 st = os.lstat(full_path)
3692 if stat.S_ISDIR(st.st_mode):
3693 if _has_directory_changed(tree_path, entry):
3694 return tree_path
3695 return None
3697 if not stat.S_ISREG(st.st_mode) and not stat.S_ISLNK(st.st_mode):
3698 return None
3700 # Optimization: If stat matches index entry (mtime and size unchanged),
3701 # we can skip reading and filtering the file entirely. This is a significant
3702 # performance improvement for repositories with many unchanged files.
3703 # Even with filters (e.g., LFS), if the file hasn't been modified (stat unchanged),
3704 # the filter output would be the same, so we can safely skip the expensive
3705 # filter operation. This addresses performance issues with LFS repositories
3706 # where filter operations can be very slow.
3707 if _stat_matches_entry(st, entry, trust_ctime):
3708 return None
3710 blob = blob_from_path_and_stat(full_path, st)
3712 if filter_blob_callback is not None:
3713 blob = filter_blob_callback(blob, tree_path)
3714 except FileNotFoundError:
3715 # The file was removed, so we assume that counts as
3716 # different from whatever file used to exist.
3717 return tree_path
3718 else:
3719 if blob.get_id(object_format) != entry.sha:
3720 return tree_path
3721 if refresh_stat is not None:
3722 refresh_stat.append((tree_path, st))
3723 return None
3726def get_unstaged_changes(
3727 index: Index,
3728 root_path: str | bytes,
3729 filter_blob_callback: Callable[..., Any] | None = None,
3730 preload_index: bool = False,
3731 trust_ctime: bool = True,
3732 max_stat: int | None = None,
3733 refresh_stat: list[tuple[bytes, os.stat_result]] | None = None,
3734) -> Generator[bytes, None, None]:
3735 """Walk through an index and check for differences against working tree.
3737 Args:
3738 index: index to check
3739 root_path: path in which to find files
3740 filter_blob_callback: Optional callback to filter blobs
3741 preload_index: If True, use parallel threads to check files (requires threading support)
3742 trust_ctime: If True, use ctime for change detection (default: True)
3743 max_stat: If set, limit the number of stat operations performed.
3744 When the limit is reached, remaining files are assumed unchanged.
3745 refresh_stat: If provided, entries whose stat has drifted but whose
3746 content is unchanged are appended as ``(tree_path, stat_result)``
3747 pairs so the caller can update the index stat cache without a
3748 second pass.
3749 Returns: iterator over paths with unstaged changes
3750 """
3751 # For each entry in the index check the sha1 & ensure not staged
3752 if not isinstance(root_path, bytes):
3753 root_path = os.fsencode(root_path)
3755 stat_count = 0
3757 if preload_index:
3758 # Use parallel processing for better performance on slow filesystems
3759 try:
3760 import multiprocessing
3761 from concurrent.futures import ThreadPoolExecutor
3762 except ImportError:
3763 # If threading is not available, fall back to serial processing
3764 preload_index = False
3765 else:
3766 # Collect all entries first
3767 entries = list(index.iteritems())
3769 if max_stat is not None:
3770 # When max_stat is set, limit the entries we process
3771 entries = entries[:max_stat]
3773 # Use number of CPUs but cap at 8 threads to avoid overhead
3774 num_workers = min(multiprocessing.cpu_count(), 8)
3776 # Per-thread refresh accumulators so appends stay lock-free,
3777 # then merged into the caller's list once workers finish.
3778 worker_refresh: list[list[tuple[bytes, os.stat_result]] | None] = [
3779 [] if refresh_stat is not None else None for _ in entries
3780 ]
3782 # Process entries in parallel
3783 with ThreadPoolExecutor(max_workers=num_workers) as executor:
3784 # Submit all tasks
3785 futures = [
3786 executor.submit(
3787 _check_entry_for_changes,
3788 tree_path,
3789 entry,
3790 root_path,
3791 filter_blob_callback,
3792 trust_ctime,
3793 index.object_format,
3794 worker_refresh[i],
3795 )
3796 for i, (tree_path, entry) in enumerate(entries)
3797 ]
3799 # Yield results as they complete
3800 for future in futures:
3801 result = future.result()
3802 if result is not None:
3803 yield result
3805 if refresh_stat is not None:
3806 for bucket in worker_refresh:
3807 if bucket:
3808 refresh_stat.extend(bucket)
3810 if not preload_index:
3811 # Serial processing
3812 for tree_path, entry in index.iteritems():
3813 if max_stat is not None and stat_count >= max_stat:
3814 return
3815 result = _check_entry_for_changes(
3816 tree_path,
3817 entry,
3818 root_path,
3819 filter_blob_callback,
3820 trust_ctime,
3821 index.object_format,
3822 refresh_stat,
3823 )
3824 stat_count += 1
3825 if result is not None:
3826 yield result
3829def _decode_utf8_with_fallback(data: bytes) -> str:
3830 """Decode bytes as UTF-8, with lossy fallbacks for invalid sequences.
3832 Mirrors the behaviour of git-for-windows's ``xutftowcsn`` (in
3833 ``compat/mingw.c``) so that tree paths containing legacy-encoded or
3834 otherwise invalid UTF-8 produce the same on-disk filename as C git.
3836 Rules:
3837 * Valid UTF-8 (1-4 byte sequences, excluding overlongs and codepoints
3838 > U+10FFFF) is decoded normally.
3839 * Invalid bytes in 0xa0-0xff map 1:1 to U+00A0-U+00FF.
3840 * Invalid bytes in 0x80-0x9f are expanded to two lowercase ASCII hex
3841 digits (e.g. byte 0x80 -> "80").
3842 * Truncated multi-byte sequences and overlong/out-of-range encodings
3843 cause the lead byte to fall through to the above invalid-byte rules
3844 (the trail bytes are re-evaluated on the next iteration).
3845 """
3846 out: list[str] = []
3847 i = 0
3848 n = len(data)
3849 while i < n:
3850 c = data[i]
3851 if c < 0x80:
3852 out.append(chr(c))
3853 i += 1
3854 elif 0xC2 <= c < 0xE0 and i + 1 < n and (data[i + 1] & 0xC0) == 0x80:
3855 cp = ((c & 0x1F) << 6) | (data[i + 1] & 0x3F)
3856 out.append(chr(cp))
3857 i += 2
3858 elif (
3859 0xE0 <= c < 0xF0
3860 and i + 2 < n
3861 and not (c == 0xE0 and data[i + 1] < 0xA0)
3862 and (data[i + 1] & 0xC0) == 0x80
3863 and (data[i + 2] & 0xC0) == 0x80
3864 ):
3865 cp = ((c & 0x0F) << 12) | ((data[i + 1] & 0x3F) << 6) | (data[i + 2] & 0x3F)
3866 out.append(chr(cp))
3867 i += 3
3868 elif (
3869 0xF0 <= c < 0xF5
3870 and i + 3 < n
3871 and not (c == 0xF0 and data[i + 1] < 0x90)
3872 and not (c == 0xF4 and data[i + 1] >= 0x90)
3873 and (data[i + 1] & 0xC0) == 0x80
3874 and (data[i + 2] & 0xC0) == 0x80
3875 and (data[i + 3] & 0xC0) == 0x80
3876 ):
3877 cp = (
3878 ((c & 0x07) << 18)
3879 | ((data[i + 1] & 0x3F) << 12)
3880 | ((data[i + 2] & 0x3F) << 6)
3881 | (data[i + 3] & 0x3F)
3882 )
3883 out.append(chr(cp))
3884 i += 4
3885 elif c >= 0xA0:
3886 out.append(chr(c))
3887 i += 1
3888 else:
3889 out.append(f"{c:02x}")
3890 i += 1
3891 return "".join(out)
3894def _tree_to_fs_path(
3895 root_path: bytes, tree_path: bytes, tree_encoding: str = "utf-8"
3896) -> bytes:
3897 """Convert a git tree path to a file system path.
3899 Args:
3900 root_path: Root filesystem path
3901 tree_path: Git tree path as bytes (encoded with tree_encoding)
3902 tree_encoding: Encoding used for tree paths (default: utf-8)
3904 Returns: File system path.
3905 """
3906 assert isinstance(tree_path, bytes)
3907 if os_sep_bytes != b"/":
3908 sep_corrected_path = tree_path.replace(b"/", os_sep_bytes)
3909 else:
3910 sep_corrected_path = tree_path
3912 # On Windows, decode tree-encoded bytes to a str so they can flow into
3913 # the wide-char Win32 APIs via Python's filesystem layer. For UTF-8
3914 # (the default tree encoding) we use a lossy decoder that matches C
3915 # git's xutftowcsn fallbacks; for other encodings we let UnicodeDecodeError
3916 # propagate rather than silently producing a corrupt path.
3917 if sys.platform == "win32":
3918 if tree_encoding == "utf-8":
3919 tree_path_str = _decode_utf8_with_fallback(sep_corrected_path)
3920 else:
3921 tree_path_str = sep_corrected_path.decode(tree_encoding)
3922 sep_corrected_path = os.fsencode(tree_path_str)
3924 return os.path.join(root_path, sep_corrected_path)
3927def _fs_to_tree_path(fs_path: str | bytes, tree_encoding: str = "utf-8") -> bytes:
3928 """Convert a file system path to a git tree path.
3930 Args:
3931 fs_path: File system path.
3932 tree_encoding: Encoding to use for tree paths (default: utf-8)
3934 Returns: Git tree path as bytes (encoded with tree_encoding)
3935 """
3936 if not isinstance(fs_path, bytes):
3937 fs_path_bytes = os.fsencode(fs_path)
3938 else:
3939 fs_path_bytes = fs_path
3941 # On Windows the on-disk filename is a UTF-16 wide string; Python gives
3942 # us either str (already decoded) or bytes encoded via the filesystem
3943 # codec. Normalise to str, then encode under the tree encoding so the
3944 # resulting tree path is plain UTF-8. This matches C git's xwcstoutf,
3945 # which is just WideCharToMultiByte(CP_UTF8); it makes no attempt to
3946 # reverse the xutftowcsn fallbacks, so a file that was checked out from
3947 # a tree path with invalid UTF-8 will read back as the lossy form (the
3948 # same divergence C git exhibits, documented as a one-way mapping).
3949 if sys.platform == "win32":
3950 fs_path_str = os.fsdecode(fs_path_bytes)
3951 fs_path_bytes = fs_path_str.encode(tree_encoding)
3953 if os_sep_bytes != b"/":
3954 tree_path = fs_path_bytes.replace(os_sep_bytes, b"/")
3955 else:
3956 tree_path = fs_path_bytes
3957 return tree_path
3960def index_entry_from_directory(st: os.stat_result, path: bytes) -> IndexEntry | None:
3961 """Create an index entry for a directory.
3963 This is only used for submodules (directories containing .git).
3965 Args:
3966 st: Stat result for the directory
3967 path: Path to the directory
3969 Returns:
3970 IndexEntry for a submodule, or None if not a submodule
3971 """
3972 if os.path.exists(os.path.join(path, b".git")):
3973 head = read_submodule_head(path)
3974 if head is None:
3975 return None
3976 return index_entry_from_stat(st, head, mode=S_IFGITLINK)
3977 return None
3980def index_entry_from_path(
3981 path: bytes,
3982 object_store: ObjectContainer | None = None,
3983 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
3984) -> IndexEntry | None:
3985 """Create an index from a filesystem path.
3987 This returns an index value for files, symlinks
3988 and tree references. for directories and
3989 non-existent files it returns None
3991 Args:
3992 path: Path to create an index entry for
3993 object_store: Optional object store to
3994 save new blobs in
3995 object_format: Object format to name new blobs with
3996 Returns: An index entry; None for directories
3997 """
3998 assert isinstance(path, bytes)
3999 st = os.lstat(path)
4000 if stat.S_ISDIR(st.st_mode):
4001 return index_entry_from_directory(st, path)
4003 if stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode):
4004 blob = blob_from_path_and_stat(path, st)
4005 if object_store is not None:
4006 object_store.add_object(blob)
4007 return index_entry_from_stat(st, ObjectID(blob.get_id(object_format)))
4009 return None
4012def iter_fresh_entries(
4013 paths: Iterable[bytes],
4014 root_path: bytes,
4015 object_store: ObjectContainer | None = None,
4016 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
4017) -> Iterator[tuple[bytes, IndexEntry | None]]:
4018 """Iterate over current versions of index entries on disk.
4020 Args:
4021 paths: Paths to iterate over
4022 root_path: Root path to access from
4023 object_store: Optional store to save new blobs in
4024 object_format: Object format to name new blobs with
4025 Returns: Iterator over path, index_entry
4026 """
4027 for path in paths:
4028 p = _tree_to_fs_path(root_path, path)
4029 try:
4030 entry = index_entry_from_path(
4031 p, object_store=object_store, object_format=object_format
4032 )
4033 except (FileNotFoundError, IsADirectoryError):
4034 entry = None
4035 yield path, entry
4038def iter_fresh_objects(
4039 paths: Iterable[bytes],
4040 root_path: bytes,
4041 include_deleted: bool = False,
4042 object_store: ObjectContainer | None = None,
4043 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
4044) -> Iterator[tuple[bytes, ObjectID | None, int | None]]:
4045 """Iterate over versions of objects on disk referenced by index.
4047 Args:
4048 paths: Paths to check
4049 root_path: Root path to access from
4050 include_deleted: Include deleted entries with sha and
4051 mode set to None
4052 object_store: Optional object store to report new items to
4053 object_format: Object format to name new blobs with
4054 Returns: Iterator over path, sha, mode
4055 """
4056 for path, entry in iter_fresh_entries(
4057 paths, root_path, object_store=object_store, object_format=object_format
4058 ):
4059 if entry is None:
4060 if include_deleted:
4061 yield path, None, None
4062 else:
4063 yield path, entry.sha, cleanup_mode(entry.mode)
4066def refresh_index(index: Index, root_path: bytes) -> None:
4067 """Refresh the contents of an index.
4069 This is the equivalent to running 'git commit -a'.
4071 Args:
4072 index: Index to update
4073 root_path: Root filesystem path
4074 """
4075 for path, entry in iter_fresh_entries(
4076 index, root_path, object_format=index.object_format
4077 ):
4078 if entry:
4079 index[path] = entry
4082def apply_stat_refresh(
4083 index: Index, refreshed: Iterable[tuple[bytes, os.stat_result]]
4084) -> bool:
4085 """Apply refreshed stat info collected during an index walk.
4087 ``refreshed`` should be the list populated by passing ``refresh_stat``
4088 into `get_unstaged_changes`: it contains (tree_path, stat_result)
4089 pairs for entries whose on-disk stat drifted but whose content still
4090 matches the index entry sha.
4092 Returns True if any entries were updated (caller may want to write
4093 the index).
4094 """
4095 updated = False
4096 for tree_path, st in refreshed:
4097 entry = index[tree_path]
4098 if isinstance(entry, ConflictedIndexEntry):
4099 continue
4100 index[tree_path] = index_entry_from_stat(
4101 st,
4102 entry.sha,
4103 mode=entry.mode,
4104 flags=entry.flags,
4105 extended_flags=entry.extended_flags,
4106 )
4107 updated = True
4108 return updated
4111class locked_index:
4112 """Lock the index while making modifications.
4114 Works as a context manager.
4115 """
4117 _file: "_GitFile"
4119 def __init__(
4120 self,
4121 path: bytes | str,
4122 object_format: ObjectFormat = DEFAULT_OBJECT_FORMAT,
4123 ) -> None:
4124 """Initialize locked_index.
4126 Args:
4127 path: Path to the index file
4128 object_format: Object format of the repository owning the index
4129 """
4130 self._path = path
4131 self._object_format = object_format
4133 def __enter__(self) -> Index:
4134 """Enter context manager and lock index."""
4135 f = GitFile(self._path, "wb")
4136 self._file = f
4137 self._index = Index(self._path, object_format=self._object_format)
4138 return self._index
4140 def __exit__(
4141 self,
4142 exc_type: type | None,
4143 exc_value: BaseException | None,
4144 traceback: types.TracebackType | None,
4145 ) -> None:
4146 """Exit context manager and unlock index."""
4147 if exc_type is not None:
4148 self._file.abort()
4149 return
4150 try:
4151 f = IndexChecksumWriter(self._file, self._object_format)
4152 write_index_dict(f, self._index._byname, object_format=self._object_format)
4153 except BaseException:
4154 self._file.abort()
4155 else:
4156 f.close()