Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/index.py: 29%
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 "REUC_EXTENSION",
38 "SDIR_EXTENSION",
39 "TREE_EXTENSION",
40 "UNTR_EXTENSION",
41 "Index",
42 "IndexEntry",
43 "IndexExtension",
44 "InvalidPathError",
45 "ResolveUndoExtension",
46 "SerializedIndexEntry",
47 "SparseDirExtension",
48 "Stage",
49 "TreeDict",
50 "TreeExtension",
51 "UnmergedEntries",
52 "UnsupportedIndexFormat",
53 "UntrackedExtension",
54 "blob_from_path_and_mode",
55 "blob_from_path_and_stat",
56 "build_file_from_blob",
57 "build_index_from_tree",
58 "changes_from_tree",
59 "cleanup_mode",
60 "commit_index",
61 "commit_tree",
62 "detect_case_only_renames",
63 "get_path_element_normalizer",
64 "get_path_element_validator",
65 "get_unstaged_changes",
66 "index_entry_from_stat",
67 "index_entry_from_tree_entry",
68 "make_path_normalizer",
69 "pathjoin",
70 "pathsplit",
71 "read_cache_entry",
72 "read_cache_time",
73 "read_index",
74 "read_index_dict",
75 "read_index_dict_with_version",
76 "read_index_header",
77 "read_submodule_head",
78 "update_working_tree",
79 "validate_path",
80 "validate_path_element_default",
81 "validate_path_element_hfs",
82 "validate_path_element_ntfs",
83 "write_cache_entry",
84 "write_cache_time",
85 "write_index",
86 "write_index_dict",
87 "write_index_extension",
88]
90import errno
91import logging
92import os
93import shutil
94import stat
95import struct
96import sys
97import types
98from collections.abc import (
99 Callable,
100 Generator,
101 Iterable,
102 Iterator,
103 Mapping,
104 Sequence,
105 Set,
106)
107from dataclasses import dataclass
108from enum import Enum
109from typing import (
110 IO,
111 TYPE_CHECKING,
112 Any,
113 BinaryIO,
114)
116if TYPE_CHECKING:
117 from .config import Config
118 from .diff_tree import TreeChange
119 from .file import _GitFile
120 from .filters import FilterBlobNormalizer
121 from .object_store import BaseObjectStore
122 from .repo import Repo
124from .file import GitFile
125from .object_store import iter_tree_contents
126from .objects import (
127 S_IFGITLINK,
128 S_ISGITLINK,
129 Blob,
130 ObjectID,
131 Tree,
132 TreeEntry,
133 hex_to_sha,
134 sha_to_hex,
135)
136from .pack import ObjectContainer, SHA1Reader, SHA1Writer
138logger = logging.getLogger(__name__)
140# Type alias for recursive tree structure used in commit_tree
141TreeDict = dict[bytes, "TreeDict | tuple[int, ObjectID]"]
143# 2-bit stage (during merge)
144FLAG_STAGEMASK = 0x3000
145FLAG_STAGESHIFT = 12
146FLAG_NAMEMASK = 0x0FFF
148# assume-valid
149FLAG_VALID = 0x8000
151# extended flag (must be zero in version 2)
152FLAG_EXTENDED = 0x4000
154# used by sparse checkout
155EXTENDED_FLAG_SKIP_WORKTREE = 0x4000
157# used by "git add -N"
158EXTENDED_FLAG_INTEND_TO_ADD = 0x2000
160DEFAULT_VERSION = 2
162# Index extension signatures
163TREE_EXTENSION = b"TREE"
164REUC_EXTENSION = b"REUC"
165UNTR_EXTENSION = b"UNTR"
166EOIE_EXTENSION = b"EOIE"
167IEOT_EXTENSION = b"IEOT"
168SDIR_EXTENSION = b"sdir" # Sparse directory extension
171def _encode_varint(value: int) -> bytes:
172 """Encode an integer using variable-width encoding.
174 Same format as used for OFS_DELTA pack entries and index v4 path compression.
175 Uses 7 bits per byte, with the high bit indicating continuation.
177 Args:
178 value: Integer to encode
179 Returns:
180 Encoded bytes
181 """
182 if value == 0:
183 return b"\x00"
185 result = []
186 while value > 0:
187 byte = value & 0x7F # Take lower 7 bits
188 value >>= 7
189 if value > 0:
190 byte |= 0x80 # Set continuation bit
191 result.append(byte)
193 return bytes(result)
196def _decode_varint(data: bytes, offset: int = 0) -> tuple[int, int]:
197 """Decode a variable-width encoded integer.
199 Args:
200 data: Bytes to decode from
201 offset: Starting offset in data
202 Returns:
203 tuple of (decoded_value, new_offset)
204 """
205 value = 0
206 shift = 0
207 pos = offset
209 while pos < len(data):
210 byte = data[pos]
211 pos += 1
212 value |= (byte & 0x7F) << shift
213 shift += 7
214 if not (byte & 0x80): # No continuation bit
215 break
217 return value, pos
220def _compress_path(path: bytes, previous_path: bytes) -> bytes:
221 """Compress a path relative to the previous path for index version 4.
223 Args:
224 path: Path to compress
225 previous_path: Previous path for comparison
226 Returns:
227 Compressed path data (varint prefix_len + suffix)
228 """
229 # Find the common prefix length
230 common_len = 0
231 min_len = min(len(path), len(previous_path))
233 for i in range(min_len):
234 if path[i] == previous_path[i]:
235 common_len += 1
236 else:
237 break
239 # The number of bytes to remove from the end of previous_path
240 # to get the common prefix
241 remove_len = len(previous_path) - common_len
243 # The suffix to append
244 suffix = path[common_len:]
246 # Encode: varint(remove_len) + suffix + NUL
247 return _encode_varint(remove_len) + suffix + b"\x00"
250def _decompress_path(
251 data: bytes, offset: int, previous_path: bytes
252) -> tuple[bytes, int]:
253 """Decompress a path from index version 4 compressed format.
255 Args:
256 data: Raw data containing compressed path
257 offset: Starting offset in data
258 previous_path: Previous path for decompression
259 Returns:
260 tuple of (decompressed_path, new_offset)
261 """
262 # Decode the number of bytes to remove from previous path
263 remove_len, new_offset = _decode_varint(data, offset)
265 # Find the NUL terminator for the suffix
266 suffix_start = new_offset
267 suffix_end = suffix_start
268 while suffix_end < len(data) and data[suffix_end] != 0:
269 suffix_end += 1
271 if suffix_end >= len(data):
272 raise ValueError("Unterminated path suffix in compressed entry")
274 suffix = data[suffix_start:suffix_end]
275 new_offset = suffix_end + 1 # Skip the NUL terminator
277 # Reconstruct the path
278 if remove_len > len(previous_path):
279 raise ValueError(
280 f"Invalid path compression: trying to remove {remove_len} bytes from {len(previous_path)}-byte path"
281 )
283 prefix = previous_path[:-remove_len] if remove_len > 0 else previous_path
284 path = prefix + suffix
286 return path, new_offset
289def _decompress_path_from_stream(
290 f: BinaryIO, previous_path: bytes
291) -> tuple[bytes, int]:
292 """Decompress a path from index version 4 compressed format, reading from stream.
294 Args:
295 f: File-like object to read from
296 previous_path: Previous path for decompression
297 Returns:
298 tuple of (decompressed_path, bytes_consumed)
299 """
300 # Decode the varint for remove_len by reading byte by byte
301 remove_len = 0
302 shift = 0
303 bytes_consumed = 0
305 while True:
306 byte_data = f.read(1)
307 if not byte_data:
308 raise ValueError("Unexpected end of file while reading varint")
309 byte = byte_data[0]
310 bytes_consumed += 1
311 remove_len |= (byte & 0x7F) << shift
312 shift += 7
313 if not (byte & 0x80): # No continuation bit
314 break
316 # Read the suffix until NUL terminator
317 suffix = b""
318 while True:
319 byte_data = f.read(1)
320 if not byte_data:
321 raise ValueError("Unexpected end of file while reading path suffix")
322 byte = byte_data[0]
323 bytes_consumed += 1
324 if byte == 0: # NUL terminator
325 break
326 suffix += bytes([byte])
328 # Reconstruct the path
329 if remove_len > len(previous_path):
330 raise ValueError(
331 f"Invalid path compression: trying to remove {remove_len} bytes from {len(previous_path)}-byte path"
332 )
334 prefix = previous_path[:-remove_len] if remove_len > 0 else previous_path
335 path = prefix + suffix
337 return path, bytes_consumed
340class Stage(Enum):
341 """Represents the stage of an index entry during merge conflicts."""
343 NORMAL = 0
344 MERGE_CONFLICT_ANCESTOR = 1
345 MERGE_CONFLICT_THIS = 2
346 MERGE_CONFLICT_OTHER = 3
349@dataclass
350class SerializedIndexEntry:
351 """Represents a serialized index entry as stored in the index file.
353 This dataclass holds the raw data for an index entry before it's
354 parsed into the more user-friendly IndexEntry format.
355 """
357 name: bytes
358 ctime: int | float | tuple[int, int]
359 mtime: int | float | tuple[int, int]
360 dev: int
361 ino: int
362 mode: int
363 uid: int
364 gid: int
365 size: int
366 sha: ObjectID
367 flags: int
368 extended_flags: int
370 def stage(self) -> Stage:
371 """Extract the stage from the flags field.
373 Returns:
374 Stage enum value indicating merge conflict state
375 """
376 return Stage((self.flags & FLAG_STAGEMASK) >> FLAG_STAGESHIFT)
378 def is_sparse_dir(self) -> bool:
379 """Check if this entry represents a sparse directory.
381 A sparse directory entry is a collapsed representation of an entire
382 directory tree in a sparse index. It has:
383 - Directory mode (0o040000)
384 - SKIP_WORKTREE flag set
385 - Path ending with '/'
386 - SHA pointing to a tree object
388 Returns:
389 True if entry is a sparse directory entry
390 """
391 return (
392 stat.S_ISDIR(self.mode)
393 and bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
394 and self.name.endswith(b"/")
395 )
398@dataclass
399class IndexExtension:
400 """Base class for index extensions."""
402 signature: bytes
403 data: bytes
405 @classmethod
406 def from_raw(cls, signature: bytes, data: bytes) -> "IndexExtension":
407 """Create an extension from raw data.
409 Args:
410 signature: 4-byte extension signature
411 data: Extension data
412 Returns:
413 Parsed extension object
414 """
415 if signature == TREE_EXTENSION:
416 return TreeExtension.from_bytes(data)
417 elif signature == REUC_EXTENSION:
418 return ResolveUndoExtension.from_bytes(data)
419 elif signature == UNTR_EXTENSION:
420 return UntrackedExtension.from_bytes(data)
421 elif signature == SDIR_EXTENSION:
422 return SparseDirExtension.from_bytes(data)
423 else:
424 # Unknown extension - just store raw data
425 return cls(signature, data)
427 def to_bytes(self) -> bytes:
428 """Serialize extension to bytes."""
429 return self.data
432class TreeExtension(IndexExtension):
433 """Tree cache extension."""
435 def __init__(self, entries: list[tuple[bytes, bytes, int]]) -> None:
436 """Initialize TreeExtension.
438 Args:
439 entries: List of tree cache entries (path, sha, flags)
440 """
441 self.entries = entries
442 super().__init__(TREE_EXTENSION, b"")
444 @classmethod
445 def from_bytes(cls, data: bytes) -> "TreeExtension":
446 """Parse TreeExtension from bytes.
448 Args:
449 data: Raw bytes to parse
451 Returns:
452 TreeExtension instance
453 """
454 # TODO: Implement tree cache parsing
455 return cls([])
457 def to_bytes(self) -> bytes:
458 """Serialize TreeExtension to bytes.
460 Returns:
461 Serialized extension data
462 """
463 # TODO: Implement tree cache serialization
464 return b""
467class ResolveUndoExtension(IndexExtension):
468 """Resolve undo extension for recording merge conflicts."""
470 def __init__(self, entries: list[tuple[bytes, list[tuple[int, bytes]]]]) -> None:
471 """Initialize ResolveUndoExtension.
473 Args:
474 entries: List of (path, stages) where stages is a list of (stage, sha) tuples
475 """
476 self.entries = entries
477 super().__init__(REUC_EXTENSION, b"")
479 @classmethod
480 def from_bytes(cls, data: bytes) -> "ResolveUndoExtension":
481 """Parse ResolveUndoExtension from bytes.
483 Args:
484 data: Raw bytes to parse
486 Returns:
487 ResolveUndoExtension instance
488 """
489 # TODO: Implement resolve undo parsing
490 return cls([])
492 def to_bytes(self) -> bytes:
493 """Serialize ResolveUndoExtension to bytes.
495 Returns:
496 Serialized extension data
497 """
498 # TODO: Implement resolve undo serialization
499 return b""
502class UntrackedExtension(IndexExtension):
503 """Untracked cache extension."""
505 def __init__(self, data: bytes) -> None:
506 """Initialize UntrackedExtension.
508 Args:
509 data: Raw untracked cache data
510 """
511 super().__init__(UNTR_EXTENSION, data)
513 @classmethod
514 def from_bytes(cls, data: bytes) -> "UntrackedExtension":
515 """Parse UntrackedExtension from bytes.
517 Args:
518 data: Raw bytes to parse
520 Returns:
521 UntrackedExtension instance
522 """
523 return cls(data)
526class SparseDirExtension(IndexExtension):
527 """Sparse directory extension.
529 This extension indicates that the index contains sparse directory entries.
530 Tools that don't understand sparse index should avoid interacting with
531 the index when this extension is present.
533 The extension data is empty - its presence is the signal.
534 """
536 def __init__(self) -> None:
537 """Initialize SparseDirExtension."""
538 super().__init__(SDIR_EXTENSION, b"")
540 @classmethod
541 def from_bytes(cls, data: bytes) -> "SparseDirExtension":
542 """Parse SparseDirExtension from bytes.
544 Args:
545 data: Raw bytes to parse (should be empty)
547 Returns:
548 SparseDirExtension instance
549 """
550 return cls()
552 def to_bytes(self) -> bytes:
553 """Serialize SparseDirExtension to bytes.
555 Returns:
556 Empty bytes (extension presence is the signal)
557 """
558 return b""
561@dataclass
562class IndexEntry:
563 """Represents an entry in the Git index.
565 This is a higher-level representation of an index entry that includes
566 parsed data and convenience methods.
567 """
569 ctime: int | float | tuple[int, int]
570 mtime: int | float | tuple[int, int]
571 dev: int
572 ino: int
573 mode: int
574 uid: int
575 gid: int
576 size: int
577 sha: ObjectID
578 flags: int = 0
579 extended_flags: int = 0
581 @classmethod
582 def from_serialized(cls, serialized: SerializedIndexEntry) -> "IndexEntry":
583 """Create an IndexEntry from a SerializedIndexEntry.
585 Args:
586 serialized: SerializedIndexEntry to convert
588 Returns:
589 New IndexEntry instance
590 """
591 return cls(
592 ctime=serialized.ctime,
593 mtime=serialized.mtime,
594 dev=serialized.dev,
595 ino=serialized.ino,
596 mode=serialized.mode,
597 uid=serialized.uid,
598 gid=serialized.gid,
599 size=serialized.size,
600 sha=serialized.sha,
601 flags=serialized.flags,
602 extended_flags=serialized.extended_flags,
603 )
605 def serialize(self, name: bytes, stage: Stage) -> SerializedIndexEntry:
606 """Serialize this entry with a given name and stage.
608 Args:
609 name: Path name for the entry
610 stage: Merge conflict stage
612 Returns:
613 SerializedIndexEntry ready for writing to disk
614 """
615 # Clear out any existing stage bits, then set them from the Stage.
616 new_flags = self.flags & ~FLAG_STAGEMASK
617 new_flags |= stage.value << FLAG_STAGESHIFT
618 return SerializedIndexEntry(
619 name=name,
620 ctime=self.ctime,
621 mtime=self.mtime,
622 dev=self.dev,
623 ino=self.ino,
624 mode=self.mode,
625 uid=self.uid,
626 gid=self.gid,
627 size=self.size,
628 sha=self.sha,
629 flags=new_flags,
630 extended_flags=self.extended_flags,
631 )
633 def stage(self) -> Stage:
634 """Get the merge conflict stage of this entry.
636 Returns:
637 Stage enum value
638 """
639 return Stage((self.flags & FLAG_STAGEMASK) >> FLAG_STAGESHIFT)
641 @property
642 def skip_worktree(self) -> bool:
643 """Return True if the skip-worktree bit is set in extended_flags."""
644 return bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
646 def set_skip_worktree(self, skip: bool = True) -> None:
647 """Helper method to set or clear the skip-worktree bit in extended_flags.
649 Also sets FLAG_EXTENDED in self.flags if needed.
650 """
651 if skip:
652 # Turn on the skip-worktree bit
653 self.extended_flags |= EXTENDED_FLAG_SKIP_WORKTREE
654 # Also ensure the main 'extended' bit is set in flags
655 self.flags |= FLAG_EXTENDED
656 else:
657 # Turn off the skip-worktree bit
658 self.extended_flags &= ~EXTENDED_FLAG_SKIP_WORKTREE
659 # Optionally unset the main extended bit if no extended flags remain
660 if self.extended_flags == 0:
661 self.flags &= ~FLAG_EXTENDED
663 def is_sparse_dir(self, name: bytes) -> bool:
664 """Check if this entry represents a sparse directory.
666 A sparse directory entry is a collapsed representation of an entire
667 directory tree in a sparse index. It has:
668 - Directory mode (0o040000)
669 - SKIP_WORKTREE flag set
670 - Path ending with '/'
671 - SHA pointing to a tree object
673 Args:
674 name: The path name for this entry (IndexEntry doesn't store name)
676 Returns:
677 True if entry is a sparse directory entry
678 """
679 return (
680 stat.S_ISDIR(self.mode)
681 and bool(self.extended_flags & EXTENDED_FLAG_SKIP_WORKTREE)
682 and name.endswith(b"/")
683 )
686class ConflictedIndexEntry:
687 """Index entry that represents a conflict."""
689 ancestor: IndexEntry | None
690 this: IndexEntry | None
691 other: IndexEntry | None
693 def __init__(
694 self,
695 ancestor: IndexEntry | None = None,
696 this: IndexEntry | None = None,
697 other: IndexEntry | None = None,
698 ) -> None:
699 """Initialize ConflictedIndexEntry.
701 Args:
702 ancestor: The common ancestor entry
703 this: The current branch entry
704 other: The other branch entry
705 """
706 self.ancestor = ancestor
707 self.this = this
708 self.other = other
711class UnmergedEntries(Exception):
712 """Unmerged entries exist in the index."""
715def pathsplit(path: bytes) -> tuple[bytes, bytes]:
716 """Split a /-delimited path into a directory part and a basename.
718 Args:
719 path: The path to split.
721 Returns:
722 Tuple with directory name and basename
723 """
724 try:
725 (dirname, basename) = path.rsplit(b"/", 1)
726 except ValueError:
727 return (b"", path)
728 else:
729 return (dirname, basename)
732def pathjoin(*args: bytes) -> bytes:
733 """Join a /-delimited path."""
734 return b"/".join([p for p in args if p])
737def read_cache_time(f: BinaryIO) -> tuple[int, int]:
738 """Read a cache time.
740 Args:
741 f: File-like object to read from
742 Returns:
743 Tuple with seconds and nanoseconds
744 """
745 return struct.unpack(">LL", f.read(8))
748def write_cache_time(f: IO[bytes], t: int | float | tuple[int, int]) -> None:
749 """Write a cache time.
751 Args:
752 f: File-like object to write to
753 t: Time to write (as int, float or tuple with secs and nsecs)
754 """
755 if isinstance(t, int):
756 t = (t, 0)
757 elif isinstance(t, float):
758 (secs, nsecs) = divmod(t, 1.0)
759 t = (int(secs), int(nsecs * 1000000000))
760 elif not isinstance(t, tuple):
761 raise TypeError(t)
762 f.write(struct.pack(">LL", *t))
765def read_cache_entry(
766 f: BinaryIO, version: int, previous_path: bytes = b""
767) -> SerializedIndexEntry:
768 """Read an entry from a cache file.
770 Args:
771 f: File-like object to read from
772 version: Index version
773 previous_path: Previous entry's path (for version 4 compression)
774 """
775 beginoffset = f.tell()
776 ctime = read_cache_time(f)
777 mtime = read_cache_time(f)
778 (
779 dev,
780 ino,
781 mode,
782 uid,
783 gid,
784 size,
785 sha,
786 flags,
787 ) = struct.unpack(">LLLLLL20sH", f.read(20 + 4 * 6 + 2))
788 if flags & FLAG_EXTENDED:
789 if version < 3:
790 raise AssertionError("extended flag set in index with version < 3")
791 (extended_flags,) = struct.unpack(">H", f.read(2))
792 else:
793 extended_flags = 0
795 if version >= 4:
796 # Version 4: paths are always compressed (name_len should be 0)
797 name, _consumed = _decompress_path_from_stream(f, previous_path)
798 else:
799 # Versions < 4: regular name reading
800 name = f.read(flags & FLAG_NAMEMASK)
802 # Padding:
803 if version < 4:
804 real_size = (f.tell() - beginoffset + 8) & ~7
805 f.read((beginoffset + real_size) - f.tell())
807 return SerializedIndexEntry(
808 name,
809 ctime,
810 mtime,
811 dev,
812 ino,
813 mode,
814 uid,
815 gid,
816 size,
817 sha_to_hex(sha),
818 flags & ~FLAG_NAMEMASK,
819 extended_flags,
820 )
823def write_cache_entry(
824 f: IO[bytes], entry: SerializedIndexEntry, version: int, previous_path: bytes = b""
825) -> None:
826 """Write an index entry to a file.
828 Args:
829 f: File object
830 entry: IndexEntry to write
831 version: Index format version
832 previous_path: Previous entry's path (for version 4 compression)
833 """
834 beginoffset = f.tell()
835 write_cache_time(f, entry.ctime)
836 write_cache_time(f, entry.mtime)
838 compressed_path = b""
839 if version >= 4:
840 # Version 4: use compression but set name_len to actual filename length
841 # This matches how C Git implements index v4 flags
842 compressed_path = _compress_path(entry.name, previous_path)
843 flags = len(entry.name) | (entry.flags & ~FLAG_NAMEMASK)
845 if entry.extended_flags:
846 flags |= FLAG_EXTENDED
847 if flags & FLAG_EXTENDED and version is not None and version < 3:
848 raise AssertionError("unable to use extended flags in version < 3")
850 f.write(
851 struct.pack(
852 b">LLLLLL20sH",
853 entry.dev & 0xFFFFFFFF,
854 entry.ino & 0xFFFFFFFF,
855 entry.mode,
856 entry.uid,
857 entry.gid,
858 entry.size,
859 hex_to_sha(entry.sha),
860 flags,
861 )
862 )
863 if flags & FLAG_EXTENDED:
864 f.write(struct.pack(b">H", entry.extended_flags))
866 if version >= 4:
867 # Version 4: always write compressed path
868 f.write(compressed_path)
869 else:
870 # Versions < 4: write regular path and padding
871 f.write(entry.name)
872 real_size = (f.tell() - beginoffset + 8) & ~7
873 f.write(b"\0" * ((beginoffset + real_size) - f.tell()))
876class UnsupportedIndexFormat(Exception):
877 """An unsupported index format was encountered."""
879 def __init__(self, version: int) -> None:
880 """Initialize UnsupportedIndexFormat exception.
882 Args:
883 version: The unsupported index format version
884 """
885 self.index_format_version = version
888def read_index_header(f: BinaryIO) -> tuple[int, int]:
889 """Read an index header from a file.
891 Returns:
892 tuple of (version, num_entries)
893 """
894 header = f.read(4)
895 if header != b"DIRC":
896 raise AssertionError(f"Invalid index file header: {header!r}")
897 (version, num_entries) = struct.unpack(b">LL", f.read(4 * 2))
898 if version not in (1, 2, 3, 4):
899 raise UnsupportedIndexFormat(version)
900 return version, num_entries
903def write_index_extension(f: IO[bytes], extension: IndexExtension) -> None:
904 """Write an index extension.
906 Args:
907 f: File-like object to write to
908 extension: Extension to write
909 """
910 data = extension.to_bytes()
911 f.write(extension.signature)
912 f.write(struct.pack(">I", len(data)))
913 f.write(data)
916def read_index(f: BinaryIO) -> Iterator[SerializedIndexEntry]:
917 """Read an index file, yielding the individual entries."""
918 version, num_entries = read_index_header(f)
919 previous_path = b""
920 for i in range(num_entries):
921 entry = read_cache_entry(f, version, previous_path)
922 previous_path = entry.name
923 yield entry
926def read_index_dict_with_version(
927 f: BinaryIO,
928) -> tuple[dict[bytes, IndexEntry | ConflictedIndexEntry], int, list[IndexExtension]]:
929 """Read an index file and return it as a dictionary along with the version.
931 Returns:
932 tuple of (entries_dict, version, extensions)
933 """
934 version, num_entries = read_index_header(f)
936 ret: dict[bytes, IndexEntry | ConflictedIndexEntry] = {}
937 previous_path = b""
938 for i in range(num_entries):
939 entry = read_cache_entry(f, version, previous_path)
940 previous_path = entry.name
941 stage = entry.stage()
942 if stage == Stage.NORMAL:
943 ret[entry.name] = IndexEntry.from_serialized(entry)
944 else:
945 existing = ret.setdefault(entry.name, ConflictedIndexEntry())
946 if isinstance(existing, IndexEntry):
947 raise AssertionError(f"Non-conflicted entry for {entry.name!r} exists")
948 if stage == Stage.MERGE_CONFLICT_ANCESTOR:
949 existing.ancestor = IndexEntry.from_serialized(entry)
950 elif stage == Stage.MERGE_CONFLICT_THIS:
951 existing.this = IndexEntry.from_serialized(entry)
952 elif stage == Stage.MERGE_CONFLICT_OTHER:
953 existing.other = IndexEntry.from_serialized(entry)
955 # Read extensions
956 extensions = []
957 while True:
958 # Check if we're at the end (20 bytes before EOF for SHA checksum)
959 current_pos = f.tell()
960 f.seek(0, 2) # EOF
961 eof_pos = f.tell()
962 f.seek(current_pos)
964 if current_pos >= eof_pos - 20:
965 break
967 # Try to read extension signature
968 signature = f.read(4)
969 if len(signature) < 4:
970 break
972 # Check if it's a valid extension signature (4 uppercase letters)
973 if not all(65 <= b <= 90 for b in signature):
974 # Not an extension, seek back
975 f.seek(-4, 1)
976 break
978 # Read extension size
979 size_data = f.read(4)
980 if len(size_data) < 4:
981 break
982 size = struct.unpack(">I", size_data)[0]
984 # Read extension data
985 data = f.read(size)
986 if len(data) < size:
987 break
989 extension = IndexExtension.from_raw(signature, data)
990 extensions.append(extension)
992 return ret, version, extensions
995def read_index_dict(
996 f: BinaryIO,
997) -> dict[bytes, IndexEntry | ConflictedIndexEntry]:
998 """Read an index file and return it as a dictionary.
1000 Dict Key is tuple of path and stage number, as
1001 path alone is not unique
1002 Args:
1003 f: File object to read fromls.
1004 """
1005 ret: dict[bytes, IndexEntry | ConflictedIndexEntry] = {}
1006 for entry in read_index(f):
1007 stage = entry.stage()
1008 if stage == Stage.NORMAL:
1009 ret[entry.name] = IndexEntry.from_serialized(entry)
1010 else:
1011 existing = ret.setdefault(entry.name, ConflictedIndexEntry())
1012 if isinstance(existing, IndexEntry):
1013 raise AssertionError(f"Non-conflicted entry for {entry.name!r} exists")
1014 if stage == Stage.MERGE_CONFLICT_ANCESTOR:
1015 existing.ancestor = IndexEntry.from_serialized(entry)
1016 elif stage == Stage.MERGE_CONFLICT_THIS:
1017 existing.this = IndexEntry.from_serialized(entry)
1018 elif stage == Stage.MERGE_CONFLICT_OTHER:
1019 existing.other = IndexEntry.from_serialized(entry)
1020 return ret
1023def write_index(
1024 f: IO[bytes],
1025 entries: Sequence[SerializedIndexEntry],
1026 version: int | None = None,
1027 extensions: Sequence[IndexExtension] | None = None,
1028) -> None:
1029 """Write an index file.
1031 Args:
1032 f: File-like object to write to
1033 version: Version number to write
1034 entries: Iterable over the entries to write
1035 extensions: Optional list of extensions to write
1036 """
1037 if version is None:
1038 version = DEFAULT_VERSION
1039 # STEP 1: check if any extended_flags are set
1040 uses_extended_flags = any(e.extended_flags != 0 for e in entries)
1041 if uses_extended_flags and version < 3:
1042 # Force or bump the version to 3
1043 version = 3
1044 # The rest is unchanged, but you might insert a final check:
1045 if version < 3:
1046 # Double-check no extended flags appear
1047 for e in entries:
1048 if e.extended_flags != 0:
1049 raise AssertionError("Attempt to use extended flags in index < v3")
1050 # Proceed with the existing code to write the header and entries.
1051 f.write(b"DIRC")
1052 f.write(struct.pack(b">LL", version, len(entries)))
1053 previous_path = b""
1054 for entry in entries:
1055 write_cache_entry(f, entry, version=version, previous_path=previous_path)
1056 previous_path = entry.name
1058 # Write extensions
1059 if extensions:
1060 for extension in extensions:
1061 write_index_extension(f, extension)
1064def write_index_dict(
1065 f: IO[bytes],
1066 entries: Mapping[bytes, IndexEntry | ConflictedIndexEntry],
1067 version: int | None = None,
1068 extensions: Sequence[IndexExtension] | None = None,
1069) -> None:
1070 """Write an index file based on the contents of a dictionary.
1072 being careful to sort by path and then by stage.
1073 """
1074 entries_list = []
1075 for key in sorted(entries):
1076 value = entries[key]
1077 if isinstance(value, ConflictedIndexEntry):
1078 if value.ancestor is not None:
1079 entries_list.append(
1080 value.ancestor.serialize(key, Stage.MERGE_CONFLICT_ANCESTOR)
1081 )
1082 if value.this is not None:
1083 entries_list.append(
1084 value.this.serialize(key, Stage.MERGE_CONFLICT_THIS)
1085 )
1086 if value.other is not None:
1087 entries_list.append(
1088 value.other.serialize(key, Stage.MERGE_CONFLICT_OTHER)
1089 )
1090 else:
1091 entries_list.append(value.serialize(key, Stage.NORMAL))
1093 write_index(f, entries_list, version=version, extensions=extensions)
1096def cleanup_mode(mode: int) -> int:
1097 """Cleanup a mode value.
1099 This will return a mode that can be stored in a tree object.
1101 Args:
1102 mode: Mode to clean up.
1104 Returns:
1105 mode
1106 """
1107 if stat.S_ISLNK(mode):
1108 return stat.S_IFLNK
1109 elif stat.S_ISDIR(mode):
1110 return stat.S_IFDIR
1111 elif S_ISGITLINK(mode):
1112 return S_IFGITLINK
1113 ret = stat.S_IFREG | 0o644
1114 if mode & 0o100:
1115 ret |= 0o111
1116 return ret
1119class Index:
1120 """A Git Index file."""
1122 _byname: dict[bytes, IndexEntry | ConflictedIndexEntry]
1124 def __init__(
1125 self,
1126 filename: bytes | str | os.PathLike[str],
1127 read: bool = True,
1128 skip_hash: bool = False,
1129 version: int | None = None,
1130 *,
1131 file_mode: int | None = None,
1132 path_normalizer: Callable[[bytes], bytes] | None = None,
1133 ) -> None:
1134 """Create an index object associated with the given filename.
1136 Args:
1137 filename: Path to the index file
1138 read: Whether to initialize the index from the given file, should it exist.
1139 skip_hash: Whether to skip SHA1 hash when writing (for manyfiles feature)
1140 version: Index format version to use (None = auto-detect from file or use default)
1141 file_mode: Optional file permission mask for shared repository
1142 path_normalizer: Optional function mapping a filesystem path to a
1143 canonical form (e.g. case-folded, NFC-normalized). When provided,
1144 lookups (``index[path]``, ``path in index``, ``del index[path]``)
1145 transparently match paths that normalize to the same form as an
1146 existing entry.
1147 """
1148 self._filename = os.fspath(filename)
1149 # TODO(jelmer): Store the version returned by read_index
1150 self._version = version
1151 self._skip_hash = skip_hash
1152 self._file_mode = file_mode
1153 self._extensions: list[IndexExtension] = []
1154 self._path_normalizer = path_normalizer
1155 self._normalized: dict[bytes, bytes] | None = (
1156 {} if path_normalizer is not None else None
1157 )
1158 self.clear()
1159 if read:
1160 self.read()
1162 def canonical_path(self, name: bytes) -> bytes:
1163 """Resolve ``name`` to the canonical key stored in the index.
1165 If an entry already exists under ``name`` (or no normalizer is
1166 configured), ``name`` is returned unchanged. Otherwise the
1167 normalizer is applied and the key of any entry with the same
1168 normalized form is returned. Falls back to ``name`` if none.
1170 Normally callers do not need this because ``index[name]``,
1171 ``name in index``, and ``del index[name]`` already apply
1172 normalization transparently. Use this when the path is also
1173 being used outside the index (for example to look up the same
1174 entry in a commit tree), so that both sides agree on the key.
1175 """
1176 if self._normalized is None or name in self._byname:
1177 return name
1178 assert self._path_normalizer is not None
1179 return self._normalized.get(self._path_normalizer(name), name)
1181 @property
1182 def path(self) -> bytes | str:
1183 """Get the path to the index file.
1185 Returns:
1186 Path to the index file
1187 """
1188 return self._filename
1190 def __repr__(self) -> str:
1191 """Return string representation of Index."""
1192 return f"{self.__class__.__name__}({self._filename!r})"
1194 def write(self) -> None:
1195 """Write current contents of index to disk."""
1196 mask = self._file_mode if self._file_mode is not None else 0o644
1197 f = GitFile(self._filename, "wb", mask=mask)
1198 try:
1199 # Filter out extensions with no meaningful data
1200 meaningful_extensions = []
1201 for ext in self._extensions:
1202 # Skip extensions that have empty data
1203 ext_data = ext.to_bytes()
1204 if ext_data:
1205 meaningful_extensions.append(ext)
1207 if self._skip_hash:
1208 # When skipHash is enabled, write the index without computing SHA1
1209 write_index_dict(
1210 f,
1211 self._byname,
1212 version=self._version,
1213 extensions=meaningful_extensions,
1214 )
1215 # Write 20 zero bytes instead of SHA1
1216 f.write(b"\x00" * 20)
1217 f.close()
1218 else:
1219 sha1_writer = SHA1Writer(f)
1220 write_index_dict(
1221 sha1_writer,
1222 self._byname,
1223 version=self._version,
1224 extensions=meaningful_extensions,
1225 )
1226 sha1_writer.close()
1227 except:
1228 f.close()
1229 raise
1231 def read(self) -> None:
1232 """Read current contents of index from disk."""
1233 if not os.path.exists(self._filename):
1234 return
1235 f = GitFile(self._filename, "rb")
1236 try:
1237 sha1_reader = SHA1Reader(f)
1238 entries, version, extensions = read_index_dict_with_version(sha1_reader)
1239 self._version = version
1240 self._extensions = extensions
1241 self.update(entries)
1242 # Extensions have already been read by read_index_dict_with_version
1243 sha1_reader.check_sha(allow_empty=True)
1244 finally:
1245 f.close()
1247 def __len__(self) -> int:
1248 """Number of entries in this index file."""
1249 return len(self._byname)
1251 def __getitem__(self, key: bytes) -> IndexEntry | ConflictedIndexEntry:
1252 """Retrieve entry by relative path and stage.
1254 Returns: Either a IndexEntry or a ConflictedIndexEntry
1255 Raises KeyError: if the entry does not exist
1256 """
1257 return self._byname[self.canonical_path(key)]
1259 def __iter__(self) -> Iterator[bytes]:
1260 """Iterate over the paths and stages in this index."""
1261 return iter(self._byname)
1263 def __contains__(self, key: bytes) -> bool:
1264 """Check if a path exists in the index."""
1265 return self.canonical_path(key) in self._byname
1267 def get_sha1(self, path: bytes) -> ObjectID:
1268 """Return the (git object) SHA1 for the object at a path."""
1269 value = self[path]
1270 if isinstance(value, ConflictedIndexEntry):
1271 raise UnmergedEntries
1272 return value.sha
1274 def get_mode(self, path: bytes) -> int:
1275 """Return the POSIX file mode for the object at a path."""
1276 value = self[path]
1277 if isinstance(value, ConflictedIndexEntry):
1278 raise UnmergedEntries
1279 return value.mode
1281 def iterobjects(self) -> Iterable[tuple[bytes, ObjectID, int]]:
1282 """Iterate over path, sha, mode tuples for use with commit_tree."""
1283 for path in self:
1284 entry = self[path]
1285 if isinstance(entry, ConflictedIndexEntry):
1286 raise UnmergedEntries
1287 yield path, entry.sha, cleanup_mode(entry.mode)
1289 def has_conflicts(self) -> bool:
1290 """Check if the index contains any conflicted entries.
1292 Returns:
1293 True if any entries are conflicted, False otherwise
1294 """
1295 for value in self._byname.values():
1296 if isinstance(value, ConflictedIndexEntry):
1297 return True
1298 return False
1300 def clear(self) -> None:
1301 """Remove all contents from this index."""
1302 self._byname = {}
1303 if self._normalized is not None:
1304 self._normalized = {}
1306 def __setitem__(
1307 self, name: bytes, value: IndexEntry | ConflictedIndexEntry
1308 ) -> None:
1309 """Set an entry in the index."""
1310 assert isinstance(name, bytes)
1311 name = self.canonical_path(name)
1312 is_new = name not in self._byname
1313 self._byname[name] = value
1314 if is_new and self._normalized is not None:
1315 assert self._path_normalizer is not None
1316 self._normalized.setdefault(self._path_normalizer(name), name)
1318 def __delitem__(self, name: bytes) -> None:
1319 """Delete an entry from the index."""
1320 name = self.canonical_path(name)
1321 del self._byname[name]
1322 if self._normalized is not None:
1323 assert self._path_normalizer is not None
1324 normalized_key = self._path_normalizer(name)
1325 if self._normalized.get(normalized_key) == name:
1326 del self._normalized[normalized_key]
1328 def iteritems(
1329 self,
1330 ) -> Iterator[tuple[bytes, IndexEntry | ConflictedIndexEntry]]:
1331 """Iterate over (path, entry) pairs in the index.
1333 Returns:
1334 Iterator of (path, entry) tuples
1335 """
1336 return iter(self._byname.items())
1338 def items(self) -> Iterator[tuple[bytes, IndexEntry | ConflictedIndexEntry]]:
1339 """Get an iterator over (path, entry) pairs.
1341 Returns:
1342 Iterator of (path, entry) tuples
1343 """
1344 return iter(self._byname.items())
1346 def update(self, entries: dict[bytes, IndexEntry | ConflictedIndexEntry]) -> None:
1347 """Update the index with multiple entries.
1349 Args:
1350 entries: Dictionary mapping paths to index entries
1351 """
1352 for key, value in entries.items():
1353 self[key] = value
1355 def paths(self) -> Generator[bytes, None, None]:
1356 """Generate all paths in the index.
1358 Yields:
1359 Path names as bytes
1360 """
1361 yield from self._byname.keys()
1363 def changes_from_tree(
1364 self,
1365 object_store: ObjectContainer,
1366 tree: ObjectID | None,
1367 want_unchanged: bool = False,
1368 ) -> Generator[
1369 tuple[
1370 tuple[bytes | None, bytes | None],
1371 tuple[int | None, int | None],
1372 tuple[bytes | None, bytes | None],
1373 ],
1374 None,
1375 None,
1376 ]:
1377 """Find the differences between the contents of this index and a tree.
1379 Args:
1380 object_store: Object store to use for retrieving tree contents
1381 tree: SHA1 of the root tree
1382 want_unchanged: Whether unchanged files should be reported
1383 Returns: Iterator over tuples with (oldpath, newpath), (oldmode,
1384 newmode), (oldsha, newsha)
1385 """
1387 def lookup_entry(path: bytes) -> tuple[bytes, int]:
1388 entry = self[path]
1389 if isinstance(entry, IndexEntry):
1390 return entry.sha, cleanup_mode(entry.mode)
1391 else:
1392 # Handle ConflictedIndexEntry case
1393 return b"", 0
1395 yield from changes_from_tree(
1396 self.paths(),
1397 lookup_entry,
1398 object_store,
1399 tree,
1400 want_unchanged=want_unchanged,
1401 )
1403 def commit(self, object_store: ObjectContainer) -> ObjectID:
1404 """Create a new tree from an index.
1406 Args:
1407 object_store: Object store to save the tree in
1408 Returns:
1409 Root tree SHA
1410 """
1411 return commit_tree(object_store, self.iterobjects())
1413 def is_sparse(self) -> bool:
1414 """Check if this index contains sparse directory entries.
1416 Returns:
1417 True if any sparse directory extension is present
1418 """
1419 return any(isinstance(ext, SparseDirExtension) for ext in self._extensions)
1421 def ensure_full_index(self, object_store: "BaseObjectStore") -> None:
1422 """Expand all sparse directory entries into full file entries.
1424 This converts a sparse index into a full index by recursively
1425 expanding any sparse directory entries into their constituent files.
1427 Args:
1428 object_store: Object store to read tree objects from
1430 Raises:
1431 KeyError: If a tree object referenced by a sparse dir entry doesn't exist
1432 """
1433 if not self.is_sparse():
1434 return
1436 # Find all sparse directory entries
1437 sparse_dirs = []
1438 for path, entry in list(self._byname.items()):
1439 if isinstance(entry, IndexEntry) and entry.is_sparse_dir(path):
1440 sparse_dirs.append((path, entry))
1442 # Expand each sparse directory
1443 for path, entry in sparse_dirs:
1444 # Remove the sparse directory entry
1445 del self[path]
1447 # Get the tree object
1448 tree = object_store[entry.sha]
1449 if not isinstance(tree, Tree):
1450 raise ValueError(f"Sparse directory {path!r} points to non-tree object")
1452 # Recursively add all entries from the tree
1453 self._expand_tree(path.rstrip(b"/"), tree, object_store, entry)
1455 # Remove the sparse directory extension
1456 self._extensions = [
1457 ext for ext in self._extensions if not isinstance(ext, SparseDirExtension)
1458 ]
1460 def _expand_tree(
1461 self,
1462 prefix: bytes,
1463 tree: Tree,
1464 object_store: "BaseObjectStore",
1465 template_entry: IndexEntry,
1466 ) -> None:
1467 """Recursively expand a tree into index entries.
1469 Args:
1470 prefix: Path prefix for entries (without trailing slash)
1471 tree: Tree object to expand
1472 object_store: Object store to read nested trees from
1473 template_entry: Template entry to copy metadata from
1474 """
1475 for name, mode, sha in tree.items():
1476 if prefix:
1477 full_path = prefix + b"/" + name
1478 else:
1479 full_path = name
1481 if stat.S_ISDIR(mode):
1482 # Recursively expand subdirectories
1483 subtree = object_store[sha]
1484 if not isinstance(subtree, Tree):
1485 raise ValueError(
1486 f"Directory entry {full_path!r} points to non-tree object"
1487 )
1488 self._expand_tree(full_path, subtree, object_store, template_entry)
1489 else:
1490 # Create an index entry for this file
1491 # Use the template entry for metadata but with the file's sha and mode
1492 new_entry = IndexEntry(
1493 ctime=template_entry.ctime,
1494 mtime=template_entry.mtime,
1495 dev=template_entry.dev,
1496 ino=template_entry.ino,
1497 mode=mode,
1498 uid=template_entry.uid,
1499 gid=template_entry.gid,
1500 size=0, # Size is unknown from tree
1501 sha=sha,
1502 flags=0,
1503 extended_flags=0, # Don't copy skip-worktree flag
1504 )
1505 self[full_path] = new_entry
1507 def convert_to_sparse(
1508 self,
1509 object_store: "BaseObjectStore",
1510 tree_sha: ObjectID,
1511 sparse_dirs: Set[bytes],
1512 ) -> None:
1513 """Convert full index entries to sparse directory entries.
1515 This collapses directories that are entirely outside the sparse
1516 checkout cone into single sparse directory entries.
1518 Args:
1519 object_store: Object store to read tree objects
1520 tree_sha: SHA of the tree (usually HEAD) to base sparse dirs on
1521 sparse_dirs: Set of directory paths (with trailing /) to collapse
1523 Raises:
1524 KeyError: If tree_sha or a subdirectory doesn't exist
1525 """
1526 if not sparse_dirs:
1527 return
1529 # Get the base tree
1530 tree = object_store[tree_sha]
1531 if not isinstance(tree, Tree):
1532 raise ValueError(f"tree_sha {tree_sha!r} is not a tree object")
1534 # For each sparse directory, find its tree SHA and create sparse entry
1535 for dir_path in sparse_dirs:
1536 dir_path_stripped = dir_path.rstrip(b"/")
1538 # Find the tree SHA for this directory
1539 subtree_sha = self._find_subtree_sha(tree, dir_path_stripped, object_store)
1540 if subtree_sha is None:
1541 # Directory doesn't exist in tree, skip it
1542 continue
1544 # Remove all entries under this directory
1545 entries_to_remove = [
1546 path
1547 for path in self._byname
1548 if path.startswith(dir_path) or path == dir_path_stripped
1549 ]
1550 for path in entries_to_remove:
1551 del self[path]
1553 # Create a sparse directory entry
1554 # Use minimal metadata since it's not a real file
1555 from dulwich.objects import ObjectID
1557 sparse_entry = IndexEntry(
1558 ctime=0,
1559 mtime=0,
1560 dev=0,
1561 ino=0,
1562 mode=stat.S_IFDIR,
1563 uid=0,
1564 gid=0,
1565 size=0,
1566 sha=ObjectID(subtree_sha),
1567 flags=0,
1568 extended_flags=EXTENDED_FLAG_SKIP_WORKTREE,
1569 )
1570 self[dir_path] = sparse_entry
1572 # Add sparse directory extension if not present
1573 if not self.is_sparse():
1574 self._extensions.append(SparseDirExtension())
1576 def _find_subtree_sha(
1577 self,
1578 tree: Tree,
1579 path: bytes,
1580 object_store: "BaseObjectStore",
1581 ) -> bytes | None:
1582 """Find the SHA of a subtree at a given path.
1584 Args:
1585 tree: Root tree object to search in
1586 path: Path to the subdirectory (no trailing slash)
1587 object_store: Object store to read nested trees from
1589 Returns:
1590 SHA of the subtree, or None if path doesn't exist
1591 """
1592 if not path:
1593 return tree.id
1595 parts = path.split(b"/")
1596 current_tree = tree
1598 for part in parts:
1599 # Look for this part in the current tree
1600 try:
1601 mode, sha = current_tree[part]
1602 except KeyError:
1603 return None
1605 if not stat.S_ISDIR(mode):
1606 # Path component is a file, not a directory
1607 return None
1609 # Load the next tree
1610 obj = object_store[sha]
1611 if not isinstance(obj, Tree):
1612 return None
1613 current_tree = obj
1615 return current_tree.id
1618def commit_tree(
1619 object_store: ObjectContainer, blobs: Iterable[tuple[bytes, ObjectID, int]]
1620) -> ObjectID:
1621 """Commit a new tree.
1623 Args:
1624 object_store: Object store to add trees to
1625 blobs: Iterable over blob path, sha, mode entries
1626 Returns:
1627 SHA1 of the created tree.
1628 """
1629 trees: dict[bytes, TreeDict] = {b"": {}}
1631 def add_tree(path: bytes) -> TreeDict:
1632 if path in trees:
1633 return trees[path]
1634 dirname, basename = pathsplit(path)
1635 t = add_tree(dirname)
1636 assert isinstance(basename, bytes)
1637 newtree: TreeDict = {}
1638 t[basename] = newtree
1639 trees[path] = newtree
1640 return newtree
1642 for path, sha, mode in blobs:
1643 tree_path, basename = pathsplit(path)
1644 tree = add_tree(tree_path)
1645 tree[basename] = (mode, sha)
1647 def build_tree(path: bytes) -> ObjectID:
1648 tree = Tree()
1649 for basename, entry in trees[path].items():
1650 if isinstance(entry, dict):
1651 mode = stat.S_IFDIR
1652 sha = build_tree(pathjoin(path, basename))
1653 else:
1654 (mode, sha) = entry
1655 tree.add(basename, mode, sha)
1656 object_store.add_object(tree)
1657 return tree.id
1659 return build_tree(b"")
1662def commit_index(object_store: ObjectContainer, index: Index) -> ObjectID:
1663 """Create a new tree from an index.
1665 Args:
1666 object_store: Object store to save the tree in
1667 index: Index file
1668 Note: This function is deprecated, use index.commit() instead.
1669 Returns: Root tree sha.
1670 """
1671 return commit_tree(object_store, index.iterobjects())
1674def changes_from_tree(
1675 names: Iterable[bytes],
1676 lookup_entry: Callable[[bytes], tuple[bytes, int]],
1677 object_store: ObjectContainer,
1678 tree: ObjectID | None,
1679 want_unchanged: bool = False,
1680) -> Iterable[
1681 tuple[
1682 tuple[bytes | None, bytes | None],
1683 tuple[int | None, int | None],
1684 tuple[bytes | None, bytes | None],
1685 ]
1686]:
1687 """Find the differences between the contents of a tree and a working copy.
1689 Args:
1690 names: Iterable of names in the working copy
1691 lookup_entry: Function to lookup an entry in the working copy
1692 object_store: Object store to use for retrieving tree contents
1693 tree: SHA1 of the root tree, or None for an empty tree
1694 want_unchanged: Whether unchanged files should be reported
1695 Returns: Iterator over tuples with (oldpath, newpath), (oldmode, newmode),
1696 (oldsha, newsha)
1697 """
1698 # TODO(jelmer): Support a include_trees option
1699 other_names = set(names)
1701 if tree is not None:
1702 for name, mode, sha in iter_tree_contents(object_store, tree):
1703 assert name is not None and mode is not None and sha is not None
1704 try:
1705 (other_sha, other_mode) = lookup_entry(name)
1706 except KeyError:
1707 # Was removed
1708 yield ((name, None), (mode, None), (sha, None))
1709 else:
1710 other_names.remove(name)
1711 if want_unchanged or other_sha != sha or other_mode != mode:
1712 yield ((name, name), (mode, other_mode), (sha, other_sha))
1714 # Mention added files
1715 for name in other_names:
1716 try:
1717 (other_sha, other_mode) = lookup_entry(name)
1718 except KeyError:
1719 pass
1720 else:
1721 yield ((None, name), (None, other_mode), (None, other_sha))
1724def index_entry_from_stat(
1725 stat_val: os.stat_result,
1726 hex_sha: bytes,
1727 mode: int | None = None,
1728) -> IndexEntry:
1729 """Create a new index entry from a stat value.
1731 Args:
1732 stat_val: POSIX stat_result instance
1733 hex_sha: Hex sha of the object
1734 mode: Optional file mode, will be derived from stat if not provided
1735 """
1736 if mode is None:
1737 mode = cleanup_mode(stat_val.st_mode)
1739 from dulwich.objects import ObjectID
1741 # Use nanosecond precision when available to avoid precision loss
1742 # through float representation
1743 ctime: int | float | tuple[int, int]
1744 mtime: int | float | tuple[int, int]
1745 st_ctime_ns = getattr(stat_val, "st_ctime_ns", None)
1746 if st_ctime_ns is not None:
1747 ctime = (
1748 st_ctime_ns // 1_000_000_000,
1749 st_ctime_ns % 1_000_000_000,
1750 )
1751 else:
1752 ctime = stat_val.st_ctime
1754 st_mtime_ns = getattr(stat_val, "st_mtime_ns", None)
1755 if st_mtime_ns is not None:
1756 mtime = (
1757 st_mtime_ns // 1_000_000_000,
1758 st_mtime_ns % 1_000_000_000,
1759 )
1760 else:
1761 mtime = stat_val.st_mtime
1763 return IndexEntry(
1764 ctime=ctime,
1765 mtime=mtime,
1766 dev=stat_val.st_dev,
1767 ino=stat_val.st_ino,
1768 mode=mode,
1769 uid=stat_val.st_uid,
1770 gid=stat_val.st_gid,
1771 size=stat_val.st_size,
1772 sha=ObjectID(hex_sha),
1773 flags=0,
1774 extended_flags=0,
1775 )
1778def index_entry_from_tree_entry(
1779 mode: int,
1780 hex_sha: bytes,
1781 size: int = 0,
1782) -> IndexEntry:
1783 """Create an index entry from a tree entry, with zeroed stat fields.
1785 Use this when populating an index directly from a tree without touching
1786 the filesystem, matching ``git read-tree``. The stat-derived fields
1787 (ctime/mtime/dev/ino/uid/gid) are zeroed; git treats such entries as
1788 stat-unmerged and refreshes them on the next stat.
1790 Args:
1791 mode: File mode from the tree entry
1792 hex_sha: Hex sha of the object
1793 size: Size of the object (0 for gitlinks or when unknown)
1794 """
1795 from dulwich.objects import ObjectID
1797 return IndexEntry(
1798 ctime=0,
1799 mtime=0,
1800 dev=0,
1801 ino=0,
1802 mode=mode,
1803 uid=0,
1804 gid=0,
1805 size=size,
1806 sha=ObjectID(hex_sha),
1807 flags=0,
1808 extended_flags=0,
1809 )
1812if sys.platform == "win32":
1813 # On Windows, creating symlinks either requires administrator privileges
1814 # or developer mode. Raise a more helpful error when we're unable to
1815 # create symlinks
1817 # https://github.com/jelmer/dulwich/issues/1005
1819 class WindowsSymlinkPermissionError(PermissionError):
1820 """Windows-specific error for symlink creation failures.
1822 This error is raised when symlink creation fails on Windows,
1823 typically due to lack of developer mode or administrator privileges.
1824 """
1826 def __init__(self, errno: int, msg: str, filename: str | None) -> None:
1827 """Initialize WindowsSymlinkPermissionError."""
1828 super().__init__(
1829 errno,
1830 f"Unable to create symlink; do you have developer mode enabled? {msg}",
1831 filename,
1832 )
1834 def symlink(
1835 src: str | bytes,
1836 dst: str | bytes,
1837 target_is_directory: bool = False,
1838 *,
1839 dir_fd: int | None = None,
1840 ) -> None:
1841 """Create a symbolic link on Windows with better error handling.
1843 Args:
1844 src: Source path for the symlink
1845 dst: Destination path where symlink will be created
1846 target_is_directory: Whether the target is a directory
1847 dir_fd: Optional directory file descriptor
1849 Raises:
1850 WindowsSymlinkPermissionError: If symlink creation fails due to permissions
1851 """
1852 try:
1853 return os.symlink(
1854 src, dst, target_is_directory=target_is_directory, dir_fd=dir_fd
1855 )
1856 except PermissionError as e:
1857 raise WindowsSymlinkPermissionError(
1858 e.errno or 0, e.strerror or "", e.filename
1859 ) from e
1860else:
1861 symlink = os.symlink
1864def build_file_from_blob(
1865 blob: Blob,
1866 mode: int,
1867 target_path: bytes,
1868 *,
1869 honor_filemode: bool = True,
1870 tree_encoding: str = "utf-8",
1871 symlink_fn: Callable[
1872 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
1873 ]
1874 | None = None,
1875) -> os.stat_result:
1876 """Build a file or symlink on disk based on a Git object.
1878 Args:
1879 blob: The git object
1880 mode: File mode
1881 target_path: Path to write to
1882 honor_filemode: An optional flag to honor core.filemode setting in
1883 config file, default is core.filemode=True, change executable bit
1884 tree_encoding: Encoding to use for tree contents
1885 symlink_fn: Function to use for creating symlinks
1886 Returns: stat object for the file
1887 """
1888 try:
1889 oldstat = os.lstat(target_path)
1890 except FileNotFoundError:
1891 oldstat = None
1892 contents = blob.as_raw_string()
1893 if stat.S_ISLNK(mode):
1894 if oldstat:
1895 _remove_file_with_readonly_handling(target_path)
1896 if sys.platform == "win32":
1897 # os.readlink on Python3 on Windows requires a unicode string.
1898 contents_str = contents.decode(tree_encoding)
1899 target_path_str = target_path.decode(tree_encoding)
1900 (symlink_fn or symlink)(contents_str, target_path_str)
1901 else:
1902 (symlink_fn or symlink)(contents, target_path)
1903 else:
1904 if oldstat is not None and stat.S_ISLNK(oldstat.st_mode):
1905 # A symlink left at the target path must not be written through.
1906 # open(..., "wb") would follow it and clobber whatever it points
1907 # at, possibly outside the work tree. Replace it with a fresh
1908 # regular file, like git does on checkout.
1909 _remove_file_with_readonly_handling(target_path)
1910 oldstat = None
1911 if oldstat is not None and oldstat.st_size == len(contents):
1912 with open(target_path, "rb") as f:
1913 if f.read() == contents:
1914 return oldstat
1916 with open(target_path, "wb") as f:
1917 # Write out file
1918 f.write(contents)
1920 if honor_filemode:
1921 # Canonicalize to the permission bits git honors (0o644/0o755).
1922 # The mode comes from a tree/index entry, which for an untrusted
1923 # repository can carry setuid/setgid/sticky or world-writable bits;
1924 # git reduces every regular-file mode to 0o644 or 0o755 on checkout.
1925 os.chmod(target_path, cleanup_mode(mode))
1927 return os.lstat(target_path)
1930INVALID_DOTNAMES = (b".git", b".", b"..", b"")
1933def _normalize_path_element_default(element: bytes) -> bytes:
1934 """Normalize path element for default case-insensitive comparison."""
1935 return element.lower()
1938def _normalize_path_element_ntfs(element: bytes) -> bytes:
1939 """Normalize path element for NTFS filesystem."""
1940 return element.rstrip(b". ").lower()
1943def _normalize_path_element_hfs(element: bytes) -> bytes:
1944 """Normalize path element for HFS+ filesystem."""
1945 import unicodedata
1947 # Decode to Unicode (let UnicodeDecodeError bubble up)
1948 element_str = element.decode("utf-8", errors="strict")
1950 # Remove HFS+ ignorable characters
1951 filtered = "".join(c for c in element_str if ord(c) not in HFS_IGNORABLE_CHARS)
1952 # Normalize to NFD
1953 normalized = unicodedata.normalize("NFD", filtered)
1954 return normalized.lower().encode("utf-8", errors="strict")
1957def get_path_element_normalizer(config: "Config") -> Callable[[bytes], bytes]:
1958 """Get the appropriate path element normalization function based on config.
1960 Args:
1961 config: Repository configuration object
1963 Returns:
1964 Function that normalizes path elements for the configured filesystem
1965 """
1966 import os
1967 import sys
1969 if config.get_boolean(b"core", b"protectNTFS", os.name == "nt"):
1970 return _normalize_path_element_ntfs
1971 elif config.get_boolean(b"core", b"protectHFS", sys.platform == "darwin"):
1972 return _normalize_path_element_hfs
1973 else:
1974 return _normalize_path_element_default
1977def make_path_normalizer(
1978 config: "Config",
1979) -> Callable[[bytes], bytes] | None:
1980 """Build a path normalizer honoring ``core.ignorecase`` and ``core.precomposeunicode``.
1982 The returned callable maps a filesystem-form path to a canonical form
1983 used to match equivalent paths (e.g. ``Foo.txt`` ↔ ``foo.txt`` when
1984 ``core.ignorecase=true``, NFD ↔ NFC when ``core.precomposeunicode=true``).
1985 Returns ``None`` when neither option is active so callers can skip the
1986 comparison entirely.
1987 """
1988 ignorecase = config.get_boolean(b"core", b"ignorecase", False)
1989 precompose = config.get_boolean(b"core", b"precomposeunicode", False)
1990 if not ignorecase and not precompose:
1991 return None
1993 def normalize(path: bytes) -> bytes:
1994 if precompose:
1995 import unicodedata
1997 try:
1998 path = unicodedata.normalize("NFC", path.decode("utf-8")).encode(
1999 "utf-8"
2000 )
2001 except UnicodeDecodeError:
2002 pass
2003 if ignorecase:
2004 path = path.lower()
2005 return path
2007 return normalize
2010class InvalidPathError(Exception):
2011 """Raised when a tree entry's path is unsafe to write to the work tree."""
2013 def __init__(self, path: bytes) -> None:
2014 """Initialize the exception with the offending path."""
2015 self.path = path
2016 super().__init__(f"invalid path {path.decode('utf-8', 'replace')!r}")
2019def validate_path_element_default(element: bytes) -> bool:
2020 """Validate a path element using default rules.
2022 Args:
2023 element: Path element to validate
2025 Returns:
2026 True if path element is valid, False otherwise
2027 """
2028 return _normalize_path_element_default(element) not in INVALID_DOTNAMES
2031def _is_ntfs_dotgit(name: bytes) -> bool:
2032 """Match NTFS-dangerous spellings of ``.git`` at the start of ``name``.
2034 Matches ``.git`` or the 8.3 short name ``git~1`` followed only by
2035 dots/spaces and then the end of the element, a separator, or a ``:``
2036 (an alternate-data-stream marker, as in ``.git::$INDEX_ALLOCATION``).
2037 """
2038 if name[:1] == b".":
2039 if name[1:4].lower() != b"git":
2040 return False
2041 i = 4
2042 elif name[:1].lower() == b"g": # ``git~1`` 8.3 short name
2043 if name[1:3].lower() != b"it" or name[3:5] != b"~1":
2044 return False
2045 i = 5
2046 else:
2047 return False
2049 while i < len(name):
2050 c = name[i : i + 1]
2051 if c == b":":
2052 return True
2053 if c != b"." and c != b" ":
2054 return False
2055 i += 1
2056 return True
2059# Reserved Windows device names. Opening any of these on Windows
2060# resolves to a device rather than a file, regardless of any
2061# extension or trailing dots/spaces (``NUL``, ``NUL.txt``,
2062# ``aux.foo.bar`` all hit the device).
2063RESERVED_WINDOWS_DEVICE_NAMES = frozenset(
2064 [b"con", b"prn", b"aux", b"nul"]
2065 + [b"com%d" % i for i in range(1, 10)]
2066 + [b"lpt%d" % i for i in range(1, 10)]
2067)
2070def _is_reserved_windows_device_name(normalized: bytes) -> bool:
2071 """Match Windows reserved device names regardless of extension."""
2072 # The "stem" is the portion before the first ``.``; Windows
2073 # also strips trailing spaces from that stem when resolving.
2074 stem = normalized.split(b".", 1)[0].rstrip(b" ")
2075 return stem in RESERVED_WINDOWS_DEVICE_NAMES
2078def _has_dos_drive_prefix(name: bytes) -> bool:
2079 """Return True if name begins with a DOS drive-letter prefix like 'C:'.
2081 Any ASCII byte followed by a colon counts, matching the ASCII branch of
2082 C git's win32_has_dos_drive_prefix: subst allows non-alphabetic drives.
2083 """
2084 if len(name) < 2 or name[1:2] != b":":
2085 return False
2086 return name[0] < 0x80
2089def validate_path_element_ntfs(element: bytes) -> bool:
2090 """Validate a path element using NTFS filesystem rules.
2092 Args:
2093 element: Path element to validate
2095 Returns:
2096 True if path element is valid for NTFS, False otherwise
2097 """
2098 # A backslash is a separator on Windows but an ordinary filename
2099 # character on POSIX, so only reject it on Windows.
2100 if os.name == "nt" and b"\\" in element:
2101 return False
2102 # Backslash also separates ``.git`` spellings, so check each segment.
2103 for segment in element.split(b"\\"):
2104 if _is_ntfs_dotgit(segment):
2105 return False
2106 normalized = _normalize_path_element_ntfs(element)
2107 if normalized in INVALID_DOTNAMES:
2108 return False
2109 if _is_reserved_windows_device_name(normalized):
2110 return False
2111 return True
2114# HFS+ ignorable Unicode codepoints (from Git's utf8.c)
2115HFS_IGNORABLE_CHARS = {
2116 0x200C, # ZERO WIDTH NON-JOINER
2117 0x200D, # ZERO WIDTH JOINER
2118 0x200E, # LEFT-TO-RIGHT MARK
2119 0x200F, # RIGHT-TO-LEFT MARK
2120 0x202A, # LEFT-TO-RIGHT EMBEDDING
2121 0x202B, # RIGHT-TO-LEFT EMBEDDING
2122 0x202C, # POP DIRECTIONAL FORMATTING
2123 0x202D, # LEFT-TO-RIGHT OVERRIDE
2124 0x202E, # RIGHT-TO-LEFT OVERRIDE
2125 0x206A, # INHIBIT SYMMETRIC SWAPPING
2126 0x206B, # ACTIVATE SYMMETRIC SWAPPING
2127 0x206C, # INHIBIT ARABIC FORM SHAPING
2128 0x206D, # ACTIVATE ARABIC FORM SHAPING
2129 0x206E, # NATIONAL DIGIT SHAPES
2130 0x206F, # NOMINAL DIGIT SHAPES
2131 0xFEFF, # ZERO WIDTH NO-BREAK SPACE
2132}
2135def validate_path_element_hfs(element: bytes) -> bool:
2136 """Validate path element for HFS+ filesystem.
2138 Equivalent to Git's is_hfs_dotgit and related checks.
2139 Uses NFD normalization and ignores HFS+ ignorable characters.
2140 """
2141 try:
2142 normalized = _normalize_path_element_hfs(element)
2143 except UnicodeDecodeError:
2144 # Malformed UTF-8 - be conservative and reject
2145 return False
2147 # Check against invalid names
2148 if normalized in INVALID_DOTNAMES:
2149 return False
2151 # Also check for 8.3 short name
2152 if normalized == b"git~1":
2153 return False
2155 return True
2158def get_path_element_validator(config: "Config") -> Callable[[bytes], bool]:
2159 """Get the path-element validator to use when checking out a tree.
2161 ``core.protectNTFS`` defaults to true on every platform (matching Git's
2162 ``PROTECT_NTFS_DEFAULT=1``) because a repository authored on POSIX can
2163 still be cloned on Windows later; ``core.protectHFS`` defaults to true on
2164 macOS. Both protections are independent and apply together, so on macOS
2165 (where both default on) a path element must satisfy the NTFS and HFS+
2166 checks. With both disabled this falls back to the default validator, which
2167 only refuses ``.git``, ``.`` and ``..``.
2169 Args:
2170 config: Repository configuration object
2172 Returns:
2173 Function that validates a single path element for the configured
2174 filesystem protections.
2175 """
2176 validators: list[Callable[[bytes], bool]] = []
2177 if config.get_boolean(b"core", b"protectNTFS", True):
2178 validators.append(validate_path_element_ntfs)
2179 if config.get_boolean(b"core", b"protectHFS", sys.platform == "darwin"):
2180 validators.append(validate_path_element_hfs)
2181 if not validators:
2182 return validate_path_element_default
2183 if len(validators) == 1:
2184 return validators[0]
2186 def validate_all(element: bytes) -> bool:
2187 return all(validator(element) for validator in validators)
2189 return validate_all
2192def validate_path(
2193 path: bytes,
2194 element_validator: Callable[[bytes], bool] = validate_path_element_default,
2195) -> bool:
2196 """Default path validator that just checks for .git/."""
2197 # A leading drive-letter prefix lets os.path.join discard the work-tree
2198 # root on Windows. Matches C git's verify_path, whose has_dos_drive_prefix
2199 # is a no-op stub on non-Windows.
2200 if os.name == "nt" and _has_dos_drive_prefix(path):
2201 return False
2202 parts = path.split(b"/")
2203 for p in parts:
2204 if not element_validator(p):
2205 return False
2206 else:
2207 return True
2210def verify_leading_dirs(
2211 tree_path: bytes,
2212 safe_prefix: list[bytes],
2213 repo_path: bytes,
2214) -> None:
2215 """Reject writes whose leading path resolves through a symlink.
2217 Callers that materialize many paths in sorted order can pass a shared
2218 ``safe_prefix`` list to cache the deepest chain of directory components
2219 already verified to be real directories (or absent); each call only
2220 ``lstat``s the components that differ from that chain. Callers that only
2221 verify a single path can pass an empty list. Mirrors git's
2222 ``lstat_cache_matchlen`` (see CVE-2021-21300).
2224 Args:
2225 tree_path: Tree-form path (``/``-separated) about to be written.
2226 safe_prefix: Mutable cache of directory components already verified
2227 under ``repo_path``. Updated in place.
2228 repo_path: Filesystem path to the work-tree root.
2230 Raises:
2231 InvalidPathError: If any leading component is a symlink.
2232 """
2233 slash = tree_path.rfind(b"/")
2234 if slash <= 0:
2235 return
2236 components = tree_path[:slash].split(b"/")
2238 common = 0
2239 while (
2240 common < len(safe_prefix)
2241 and common < len(components)
2242 and safe_prefix[common] == components[common]
2243 ):
2244 common += 1
2245 del safe_prefix[common:]
2247 current = repo_path
2248 for part in components[:common]:
2249 current = os.path.join(current, part)
2250 for part in components[common:]:
2251 current = os.path.join(current, part)
2252 try:
2253 st = os.lstat(current)
2254 except FileNotFoundError:
2255 # Anything below here doesn't exist yet; makedirs will create
2256 # it under a verified-real-directory prefix.
2257 break
2258 if stat.S_ISLNK(st.st_mode):
2259 raise InvalidPathError(tree_path)
2260 safe_prefix.append(part)
2263def build_index_from_tree(
2264 root_path: str | bytes,
2265 index_path: str | bytes,
2266 object_store: ObjectContainer,
2267 tree_id: ObjectID,
2268 honor_filemode: bool = True,
2269 validate_path_element: Callable[[bytes], bool] = validate_path_element_default,
2270 symlink_fn: Callable[
2271 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2272 ]
2273 | None = None,
2274 blob_normalizer: "FilterBlobNormalizer | None" = None,
2275 tree_encoding: str = "utf-8",
2276) -> None:
2277 """Generate and materialize index from a tree.
2279 Args:
2280 tree_id: Tree to materialize
2281 root_path: Target dir for materialized index files
2282 index_path: Target path for generated index
2283 object_store: Non-empty object store holding tree contents
2284 honor_filemode: An optional flag to honor core.filemode setting in
2285 config file, default is core.filemode=True, change executable bit
2286 validate_path_element: Function to validate path elements to check
2287 out; default just refuses .git and .. directories.
2288 symlink_fn: Function to use for creating symlinks
2289 blob_normalizer: An optional BlobNormalizer to use for converting line
2290 endings when writing blobs to the working directory.
2291 tree_encoding: Encoding used for tree paths (default: utf-8)
2293 Note: existing index is wiped and contents are not merged
2294 in a working dir. Suitable only for fresh clones.
2295 """
2296 index = Index(index_path, read=False)
2297 if not isinstance(root_path, bytes):
2298 root_path = os.fsencode(root_path)
2300 for entry in iter_tree_contents(object_store, tree_id):
2301 assert (
2302 entry.path is not None and entry.mode is not None and entry.sha is not None
2303 )
2304 # Validate as we go and abort on the first invalid path,
2305 # leaving any files already written in place.
2306 if not validate_path(entry.path, validate_path_element):
2307 raise InvalidPathError(entry.path)
2308 full_path = _tree_to_fs_path(root_path, entry.path, tree_encoding)
2310 if not os.path.exists(os.path.dirname(full_path)):
2311 os.makedirs(os.path.dirname(full_path))
2313 # TODO(jelmer): Merge new index into working tree
2314 if S_ISGITLINK(entry.mode):
2315 if not os.path.isdir(full_path):
2316 os.mkdir(full_path)
2317 st = os.lstat(full_path)
2318 # TODO(jelmer): record and return submodule paths
2319 else:
2320 obj = object_store[entry.sha]
2321 assert isinstance(obj, Blob)
2322 # Apply blob normalization for checkout if normalizer is provided
2323 if blob_normalizer is not None:
2324 obj = blob_normalizer.checkout_normalize(obj, entry.path)
2325 st = build_file_from_blob(
2326 obj,
2327 entry.mode,
2328 full_path,
2329 honor_filemode=honor_filemode,
2330 tree_encoding=tree_encoding,
2331 symlink_fn=symlink_fn,
2332 )
2334 # Add file to index
2335 if not honor_filemode or S_ISGITLINK(entry.mode):
2336 # we can not use tuple slicing to build a new tuple,
2337 # because on windows that will convert the times to
2338 # longs, which causes errors further along
2339 st_tuple = (
2340 entry.mode,
2341 st.st_ino,
2342 st.st_dev,
2343 st.st_nlink,
2344 st.st_uid,
2345 st.st_gid,
2346 st.st_size,
2347 st.st_atime,
2348 st.st_mtime,
2349 st.st_ctime,
2350 )
2351 st = st.__class__(st_tuple)
2352 # default to a stage 0 index entry (normal)
2353 # when reading from the filesystem
2354 index[entry.path] = index_entry_from_stat(st, entry.sha)
2356 index.write()
2359def blob_from_path_and_mode(
2360 fs_path: bytes, mode: int, tree_encoding: str = "utf-8"
2361) -> Blob:
2362 """Create a blob from a path and a stat object.
2364 Args:
2365 fs_path: Full file system path to file
2366 mode: File mode
2367 tree_encoding: Encoding to use for tree contents
2368 Returns: A `Blob` object
2369 """
2370 assert isinstance(fs_path, bytes)
2371 blob = Blob()
2372 if stat.S_ISLNK(mode):
2373 if sys.platform == "win32":
2374 # os.readlink on Python3 on Windows requires a unicode string.
2375 blob.data = os.readlink(os.fsdecode(fs_path)).encode(tree_encoding)
2376 else:
2377 blob.data = os.readlink(fs_path)
2378 else:
2379 with open(fs_path, "rb") as f:
2380 blob.data = f.read()
2381 return blob
2384def blob_from_path_and_stat(
2385 fs_path: bytes, st: os.stat_result, tree_encoding: str = "utf-8"
2386) -> Blob:
2387 """Create a blob from a path and a stat object.
2389 Args:
2390 fs_path: Full file system path to file
2391 st: A stat object
2392 tree_encoding: Encoding to use for tree contents
2393 Returns: A `Blob` object
2394 """
2395 return blob_from_path_and_mode(fs_path, st.st_mode, tree_encoding)
2398def read_submodule_head(path: str | bytes) -> bytes | None:
2399 """Read the head commit of a submodule.
2401 Args:
2402 path: path to the submodule
2403 Returns: HEAD sha, None if not a valid head/repository
2404 """
2405 from .errors import NotGitRepository
2406 from .repo import Repo
2408 # Repo currently expects a "str", so decode if necessary.
2409 # TODO(jelmer): Perhaps move this into Repo() ?
2410 if not isinstance(path, str):
2411 path = os.fsdecode(path)
2412 try:
2413 repo = Repo(path)
2414 except NotGitRepository:
2415 return None
2416 try:
2417 return repo.head()
2418 except KeyError:
2419 return None
2422def _has_directory_changed(tree_path: bytes, entry: IndexEntry) -> bool:
2423 """Check if a directory has changed after getting an error.
2425 When handling an error trying to create a blob from a path, call this
2426 function. It will check if the path is a directory. If it's a directory
2427 and a submodule, check the submodule head to see if it's has changed. If
2428 not, consider the file as changed as Git tracked a file and not a
2429 directory.
2431 Return true if the given path should be considered as changed and False
2432 otherwise or if the path is not a directory.
2433 """
2434 # This is actually a directory
2435 if os.path.exists(os.path.join(tree_path, b".git")):
2436 # Submodule
2437 head = read_submodule_head(tree_path)
2438 if entry.sha != head:
2439 return True
2440 else:
2441 # The file was changed to a directory, so consider it removed.
2442 return True
2444 return False
2447os_sep_bytes = os.sep.encode("ascii")
2450def _ensure_parent_dir_exists(full_path: bytes) -> None:
2451 """Ensure parent directory exists, checking no parent is a file."""
2452 parent_dir = os.path.dirname(full_path)
2453 if parent_dir and not os.path.exists(parent_dir):
2454 # Walk up the directory tree to find the first existing parent
2455 current = parent_dir
2456 parents_to_check: list[bytes] = []
2458 while current and not os.path.exists(current):
2459 parents_to_check.insert(0, current)
2460 new_parent = os.path.dirname(current)
2461 if new_parent == current:
2462 # Reached the root or can't go up further
2463 break
2464 current = new_parent
2466 # Check if the existing parent (if any) is a directory
2467 if current and os.path.exists(current) and not os.path.isdir(current):
2468 raise OSError(
2469 f"Cannot create directory, parent path is a file: {current!r}"
2470 )
2472 # Now check each parent we need to create isn't blocked by an existing file
2473 for parent_path in parents_to_check:
2474 if os.path.exists(parent_path) and not os.path.isdir(parent_path):
2475 raise OSError(
2476 f"Cannot create directory, parent path is a file: {parent_path!r}"
2477 )
2479 os.makedirs(parent_dir)
2482def _remove_file_with_readonly_handling(path: bytes) -> None:
2483 """Remove a file, handling read-only files on Windows.
2485 Args:
2486 path: Path to the file to remove
2487 """
2488 try:
2489 os.unlink(path)
2490 except PermissionError:
2491 # On Windows, remove read-only attribute and retry
2492 if sys.platform == "win32":
2493 os.chmod(path, stat.S_IWRITE | stat.S_IREAD)
2494 os.unlink(path)
2495 else:
2496 raise
2499def _remove_empty_parents(path: bytes, stop_at: bytes) -> None:
2500 """Remove empty parent directories up to stop_at."""
2501 parent = os.path.dirname(path)
2502 while parent and parent != stop_at:
2503 try:
2504 os.rmdir(parent)
2505 parent = os.path.dirname(parent)
2506 except FileNotFoundError:
2507 # Directory doesn't exist - stop trying
2508 break
2509 except OSError as e:
2510 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
2511 # Directory not empty - stop trying
2512 break
2513 raise
2516def _check_symlink_matches(
2517 full_path: bytes, repo_object_store: "BaseObjectStore", entry_sha: ObjectID
2518) -> bool:
2519 """Check if symlink target matches expected target.
2521 Returns True if symlink matches, False if it doesn't match.
2522 """
2523 try:
2524 current_target = os.readlink(full_path)
2525 blob_obj = repo_object_store[entry_sha]
2526 expected_target = blob_obj.as_raw_string()
2527 if isinstance(current_target, str):
2528 current_target = current_target.encode()
2529 return current_target == expected_target
2530 except FileNotFoundError:
2531 # Symlink doesn't exist
2532 return False
2533 except OSError as e:
2534 if e.errno == errno.EINVAL:
2535 # Not a symlink
2536 return False
2537 raise
2540def _check_file_matches(
2541 repo_object_store: "BaseObjectStore",
2542 full_path: bytes,
2543 entry_sha: ObjectID,
2544 entry_mode: int,
2545 current_stat: os.stat_result,
2546 honor_filemode: bool,
2547 blob_normalizer: "FilterBlobNormalizer | None" = None,
2548 tree_path: bytes | None = None,
2549) -> bool:
2550 """Check if a file on disk matches the expected git object.
2552 Returns True if file matches, False if it doesn't match.
2553 """
2554 # Check mode first (if honor_filemode is True)
2555 if honor_filemode:
2556 current_mode = stat.S_IMODE(current_stat.st_mode)
2557 expected_mode = stat.S_IMODE(entry_mode)
2559 # For regular files, only check the user executable bit, not group/other permissions
2560 # This matches Git's behavior where umask differences don't count as modifications
2561 if stat.S_ISREG(current_stat.st_mode):
2562 # Normalize regular file modes to ignore group/other write permissions
2563 current_mode_normalized = (
2564 current_mode & 0o755
2565 ) # Keep only user rwx and all read+execute
2566 expected_mode_normalized = expected_mode & 0o755
2568 # For Git compatibility, regular files should be either 644 or 755
2569 if expected_mode_normalized not in (0o644, 0o755):
2570 expected_mode_normalized = 0o644 # Default for regular files
2571 if current_mode_normalized not in (0o644, 0o755):
2572 # Determine if it should be executable based on user execute bit
2573 if current_mode & 0o100: # User execute bit is set
2574 current_mode_normalized = 0o755
2575 else:
2576 current_mode_normalized = 0o644
2578 if current_mode_normalized != expected_mode_normalized:
2579 return False
2580 else:
2581 # For non-regular files (symlinks, etc.), check mode exactly
2582 if current_mode != expected_mode:
2583 return False
2585 # If mode matches (or we don't care), check content via size first
2586 blob_obj = repo_object_store[entry_sha]
2587 if current_stat.st_size != blob_obj.raw_length():
2588 return False
2590 # Size matches, check actual content
2591 try:
2592 with open(full_path, "rb") as f:
2593 current_content = f.read()
2594 expected_content = blob_obj.as_raw_string()
2595 if blob_normalizer and tree_path is not None:
2596 assert isinstance(blob_obj, Blob)
2597 normalized_blob = blob_normalizer.checkout_normalize(
2598 blob_obj, tree_path
2599 )
2600 expected_content = normalized_blob.as_raw_string()
2601 return current_content == expected_content
2602 except (FileNotFoundError, PermissionError, IsADirectoryError):
2603 return False
2606def _transition_to_submodule(
2607 repo: "Repo",
2608 path: bytes,
2609 full_path: bytes,
2610 current_stat: os.stat_result | None,
2611 entry: IndexEntry | TreeEntry,
2612 index: Index,
2613) -> None:
2614 """Transition any type to submodule."""
2615 from .submodule import ensure_submodule_placeholder
2617 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2618 # Already a directory, just ensure .git file exists
2619 ensure_submodule_placeholder(repo, path)
2620 else:
2621 # Remove whatever is there and create submodule
2622 if current_stat is not None:
2623 _remove_file_with_readonly_handling(full_path)
2624 ensure_submodule_placeholder(repo, path)
2626 st = os.lstat(full_path)
2627 assert entry.sha is not None
2628 index[path] = index_entry_from_stat(st, entry.sha)
2631def _transition_to_file(
2632 object_store: "BaseObjectStore",
2633 path: bytes,
2634 full_path: bytes,
2635 current_stat: os.stat_result | None,
2636 entry: IndexEntry | TreeEntry,
2637 index: Index,
2638 honor_filemode: bool,
2639 symlink_fn: Callable[
2640 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2641 ]
2642 | None,
2643 blob_normalizer: "FilterBlobNormalizer | None",
2644 tree_encoding: str = "utf-8",
2645) -> None:
2646 """Transition any type to regular file or symlink."""
2647 assert entry.sha is not None and entry.mode is not None
2648 # Check if we need to update
2649 if (
2650 current_stat is not None
2651 and stat.S_ISREG(current_stat.st_mode)
2652 and not stat.S_ISLNK(entry.mode)
2653 ):
2654 # File to file - check if update needed
2655 file_matches = _check_file_matches(
2656 object_store,
2657 full_path,
2658 entry.sha,
2659 entry.mode,
2660 current_stat,
2661 honor_filemode,
2662 blob_normalizer,
2663 path,
2664 )
2665 needs_update = not file_matches
2666 elif (
2667 current_stat is not None
2668 and stat.S_ISLNK(current_stat.st_mode)
2669 and stat.S_ISLNK(entry.mode)
2670 ):
2671 # Symlink to symlink - check if update needed
2672 symlink_matches = _check_symlink_matches(full_path, object_store, entry.sha)
2673 needs_update = not symlink_matches
2674 else:
2675 needs_update = True
2677 if not needs_update:
2678 # Just update index - current_stat should always be valid here since we're not updating
2679 assert current_stat is not None
2680 index[path] = index_entry_from_stat(current_stat, entry.sha)
2681 return
2683 # Remove existing entry if needed
2684 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2685 # Remove directory
2686 dir_contents = set(os.listdir(full_path))
2687 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
2689 if git_file_name in dir_contents:
2690 if dir_contents != {git_file_name}:
2691 raise IsADirectoryError(
2692 f"Cannot replace submodule with untracked files: {full_path!r}"
2693 )
2694 shutil.rmtree(full_path)
2695 else:
2696 try:
2697 os.rmdir(full_path)
2698 except OSError as e:
2699 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
2700 raise IsADirectoryError(
2701 f"Cannot replace non-empty directory with file: {full_path!r}"
2702 )
2703 raise
2704 elif current_stat is not None:
2705 _remove_file_with_readonly_handling(full_path)
2707 # Ensure parent directory exists
2708 _ensure_parent_dir_exists(full_path)
2710 # Write the file
2711 blob_obj = object_store[entry.sha]
2712 assert isinstance(blob_obj, Blob)
2713 if blob_normalizer:
2714 blob_obj = blob_normalizer.checkout_normalize(blob_obj, path)
2715 st = build_file_from_blob(
2716 blob_obj,
2717 entry.mode,
2718 full_path,
2719 honor_filemode=honor_filemode,
2720 tree_encoding=tree_encoding,
2721 symlink_fn=symlink_fn,
2722 )
2723 index[path] = index_entry_from_stat(st, entry.sha)
2726def _transition_to_absent(
2727 repo: "Repo",
2728 path: bytes,
2729 full_path: bytes,
2730 current_stat: os.stat_result | None,
2731 index: Index,
2732) -> None:
2733 """Remove any type of entry."""
2734 if current_stat is None:
2735 return
2737 if stat.S_ISDIR(current_stat.st_mode):
2738 # Check if it's a submodule directory
2739 dir_contents = set(os.listdir(full_path))
2740 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
2742 if git_file_name in dir_contents and dir_contents == {git_file_name}:
2743 shutil.rmtree(full_path)
2744 else:
2745 try:
2746 os.rmdir(full_path)
2747 except OSError as e:
2748 if e.errno not in (errno.ENOTEMPTY, errno.EEXIST):
2749 raise
2750 else:
2751 _remove_file_with_readonly_handling(full_path)
2753 try:
2754 del index[path]
2755 except KeyError:
2756 pass
2758 # Try to remove empty parent directories
2759 _remove_empty_parents(
2760 full_path, repo.path if isinstance(repo.path, bytes) else repo.path.encode()
2761 )
2764def detect_case_only_renames(
2765 changes: Sequence["TreeChange"],
2766 config: "Config",
2767) -> list["TreeChange"]:
2768 """Detect and transform case-only renames in a list of tree changes.
2770 This function identifies file renames that only differ in case (e.g.,
2771 README.txt -> readme.txt) and transforms matching ADD/DELETE pairs into
2772 CHANGE_RENAME operations. It uses filesystem-appropriate path normalization
2773 based on the repository configuration.
2775 Args:
2776 changes: List of TreeChange objects representing file changes
2777 config: Repository configuration object
2779 Returns:
2780 New list of TreeChange objects with case-only renames converted to CHANGE_RENAME
2781 """
2782 from .diff_tree import (
2783 CHANGE_ADD,
2784 CHANGE_COPY,
2785 CHANGE_DELETE,
2786 CHANGE_MODIFY,
2787 CHANGE_RENAME,
2788 TreeChange,
2789 )
2791 # Build dictionaries of old and new paths with their normalized forms
2792 old_paths_normalized = {}
2793 new_paths_normalized = {}
2794 old_changes = {} # Map from old path to change object
2795 new_changes = {} # Map from new path to change object
2797 # Get the appropriate normalizer based on config
2798 normalize_func = get_path_element_normalizer(config)
2800 def normalize_path(path: bytes) -> bytes:
2801 """Normalize entire path using element normalization."""
2802 return b"/".join(normalize_func(part) for part in path.split(b"/"))
2804 # Pre-normalize all paths once to avoid repeated normalization
2805 for change in changes:
2806 if change.type == CHANGE_DELETE and change.old:
2807 assert change.old.path is not None
2808 try:
2809 normalized = normalize_path(change.old.path)
2810 except UnicodeDecodeError:
2811 logger.warning(
2812 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2813 change.old.path,
2814 )
2815 else:
2816 old_paths_normalized[normalized] = change.old.path
2817 old_changes[change.old.path] = change
2818 elif change.type == CHANGE_RENAME and change.old:
2819 assert change.old.path is not None
2820 # Treat RENAME as DELETE + ADD for case-only detection
2821 try:
2822 normalized = normalize_path(change.old.path)
2823 except UnicodeDecodeError:
2824 logger.warning(
2825 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2826 change.old.path,
2827 )
2828 else:
2829 old_paths_normalized[normalized] = change.old.path
2830 old_changes[change.old.path] = change
2832 if (
2833 change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY)
2834 and change.new
2835 ):
2836 assert change.new.path is not None
2837 try:
2838 normalized = normalize_path(change.new.path)
2839 except UnicodeDecodeError:
2840 logger.warning(
2841 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2842 change.new.path,
2843 )
2844 else:
2845 new_paths_normalized[normalized] = change.new.path
2846 new_changes[change.new.path] = change
2848 # Find case-only renames and transform changes
2849 case_only_renames = set()
2850 new_rename_changes = []
2852 for norm_path, old_path in old_paths_normalized.items():
2853 if norm_path in new_paths_normalized:
2854 new_path = new_paths_normalized[norm_path]
2855 if old_path != new_path:
2856 # Found a case-only rename
2857 old_change = old_changes[old_path]
2858 new_change = new_changes[new_path]
2860 # Create a CHANGE_RENAME to replace the DELETE and ADD/MODIFY pair
2861 if new_change.type == CHANGE_ADD:
2862 # Simple case: DELETE + ADD becomes RENAME
2863 rename_change = TreeChange(
2864 CHANGE_RENAME, old_change.old, new_change.new
2865 )
2866 else:
2867 # Complex case: DELETE + MODIFY becomes RENAME
2868 # Use the old file from DELETE and new file from MODIFY
2869 rename_change = TreeChange(
2870 CHANGE_RENAME, old_change.old, new_change.new
2871 )
2873 new_rename_changes.append(rename_change)
2875 # Mark the old changes for removal
2876 case_only_renames.add(old_change)
2877 case_only_renames.add(new_change)
2879 # Return new list with original ADD/DELETE changes replaced by renames
2880 result = [change for change in changes if change not in case_only_renames]
2881 result.extend(new_rename_changes)
2882 return result
2885def update_working_tree(
2886 repo: "Repo",
2887 old_tree_id: bytes | None,
2888 new_tree_id: bytes,
2889 change_iterator: Iterator["TreeChange"],
2890 honor_filemode: bool = True,
2891 validate_path_element: Callable[[bytes], bool] | None = None,
2892 symlink_fn: Callable[
2893 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2894 ]
2895 | None = None,
2896 force_remove_untracked: bool = False,
2897 blob_normalizer: "FilterBlobNormalizer | None" = None,
2898 tree_encoding: str = "utf-8",
2899 allow_overwrite_modified: bool = False,
2900 *,
2901 config: "Config | None" = None,
2902) -> None:
2903 """Update the working tree and index to match a new tree.
2905 This function handles:
2906 - Adding new files
2907 - Updating modified files
2908 - Removing deleted files
2909 - Cleaning up empty directories
2911 Args:
2912 repo: Repository object
2913 old_tree_id: SHA of the tree before the update
2914 new_tree_id: SHA of the tree to update to
2915 change_iterator: Iterator of TreeChange objects to apply
2916 honor_filemode: An optional flag to honor core.filemode setting
2917 validate_path_element: Function to validate path elements to check out.
2918 If None, derived from ``config`` so that ``core.protectNTFS`` and
2919 ``core.protectHFS`` are honored by default.
2920 symlink_fn: Function to use for creating symlinks
2921 force_remove_untracked: If True, remove files that exist in working
2922 directory but not in target tree, even if old_tree_id is None
2923 blob_normalizer: An optional BlobNormalizer to use for converting line
2924 endings when writing blobs to the working directory.
2925 tree_encoding: Encoding used for tree paths (default: utf-8)
2926 allow_overwrite_modified: If False, raise an error when attempting to
2927 overwrite files that have been modified compared to old_tree_id
2928 config: Repository configuration. If None, falls back to
2929 ``repo.get_config_stack()``.
2930 """
2931 from .diff_tree import (
2932 CHANGE_ADD,
2933 CHANGE_COPY,
2934 CHANGE_DELETE,
2935 CHANGE_MODIFY,
2936 CHANGE_RENAME,
2937 CHANGE_UNCHANGED,
2938 )
2940 if force_remove_untracked:
2941 import warnings
2943 warnings.warn(
2944 "force_remove_untracked is a no-op and will be removed in a future release",
2945 DeprecationWarning,
2946 )
2947 repo_path = repo.path if isinstance(repo.path, bytes) else repo.path.encode()
2948 if config is None:
2949 config = repo.get_config_stack()
2951 if validate_path_element is None:
2952 # Derive the validator from config so callers cannot accidentally
2953 # skip ``core.protectNTFS``/``core.protectHFS`` enforcement by
2954 # omitting this argument.
2955 validate_path_element = get_path_element_validator(config)
2957 index = repo.open_index(config=config)
2959 # Convert iterator to list since we need multiple passes
2960 changes = list(change_iterator)
2962 # Transform case-only renames on case-insensitive filesystems
2963 import platform
2965 default_ignore_case = platform.system() in ("Windows", "Darwin")
2966 config = repo.get_config()
2967 ignore_case = config.get_boolean((b"core",), b"ignorecase", default_ignore_case)
2969 if ignore_case:
2970 config = repo.get_config()
2971 changes = detect_case_only_renames(changes, config)
2973 # Check for path conflicts where files need to become directories
2974 paths_becoming_dirs = set()
2975 for change in changes:
2976 if change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY):
2977 assert change.new is not None
2978 path = change.new.path
2979 assert path is not None
2980 if b"/" in path: # This is a file inside a directory
2981 # Check if any parent path exists as a file in the old tree or changes
2982 parts = path.split(b"/")
2983 for i in range(1, len(parts)):
2984 parent = b"/".join(parts[:i])
2985 # See if this parent path is being deleted (was a file, becoming a dir)
2986 for other_change in changes:
2987 if (
2988 other_change.type == CHANGE_DELETE
2989 and other_change.old
2990 and other_change.old.path == parent
2991 ):
2992 paths_becoming_dirs.add(parent)
2994 # Check if any path that needs to become a directory has been modified
2995 for path in paths_becoming_dirs:
2996 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
2997 try:
2998 current_stat = os.lstat(full_path)
2999 except FileNotFoundError:
3000 continue # File doesn't exist, nothing to check
3001 except OSError as e:
3002 raise OSError(
3003 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3004 ) from e
3006 if stat.S_ISREG(current_stat.st_mode):
3007 # Find the old entry for this path
3008 old_change = None
3009 for change in changes:
3010 if (
3011 change.type == CHANGE_DELETE
3012 and change.old
3013 and change.old.path == path
3014 ):
3015 old_change = change
3016 break
3018 if old_change:
3019 # Check if file has been modified
3020 assert old_change.old is not None
3021 assert (
3022 old_change.old.sha is not None and old_change.old.mode is not None
3023 )
3024 file_matches = _check_file_matches(
3025 repo.object_store,
3026 full_path,
3027 old_change.old.sha,
3028 old_change.old.mode,
3029 current_stat,
3030 honor_filemode,
3031 blob_normalizer,
3032 path,
3033 )
3034 if not file_matches:
3035 raise OSError(
3036 f"Cannot replace modified file with directory: {path!r}"
3037 )
3039 # Check for uncommitted modifications before making any changes
3040 if not allow_overwrite_modified and old_tree_id:
3041 for change in changes:
3042 # Only check files that are being modified or deleted
3043 if change.type in (CHANGE_MODIFY, CHANGE_DELETE) and change.old:
3044 path = change.old.path
3045 assert path is not None
3046 if not validate_path(path, validate_path_element):
3047 continue
3049 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3050 try:
3051 current_stat = os.lstat(full_path)
3052 except FileNotFoundError:
3053 continue # File doesn't exist, nothing to check
3054 except OSError as e:
3055 raise OSError(
3056 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3057 ) from e
3059 if stat.S_ISREG(current_stat.st_mode):
3060 # Check if working tree file differs from old tree
3061 assert change.old.sha is not None and change.old.mode is not None
3062 file_matches = _check_file_matches(
3063 repo.object_store,
3064 full_path,
3065 change.old.sha,
3066 change.old.mode,
3067 current_stat,
3068 honor_filemode,
3069 blob_normalizer,
3070 path,
3071 )
3072 if not file_matches:
3073 from .errors import WorkingTreeModifiedError
3075 raise WorkingTreeModifiedError(
3076 f"Your local changes to '{path.decode('utf-8', errors='replace')}' "
3077 f"would be overwritten by checkout. "
3078 f"Please commit your changes or stash them before you switch branches."
3079 )
3081 # Apply the changes
3082 for change in changes:
3083 if change.type in (CHANGE_DELETE, CHANGE_RENAME):
3084 # Remove file/directory
3085 assert change.old is not None and change.old.path is not None
3086 path = change.old.path
3087 if not validate_path(path, validate_path_element):
3088 continue
3090 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3091 try:
3092 delete_stat: os.stat_result | None = os.lstat(full_path)
3093 except FileNotFoundError:
3094 delete_stat = None
3095 except OSError as e:
3096 raise OSError(
3097 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3098 ) from e
3100 _transition_to_absent(repo, path, full_path, delete_stat, index)
3102 if change.type in (
3103 CHANGE_ADD,
3104 CHANGE_MODIFY,
3105 CHANGE_UNCHANGED,
3106 CHANGE_COPY,
3107 CHANGE_RENAME,
3108 ):
3109 # Add or modify file
3110 assert (
3111 change.new is not None
3112 and change.new.path is not None
3113 and change.new.mode is not None
3114 )
3115 path = change.new.path
3116 # Validate as we go and abort on the first invalid path,
3117 # leaving any changes already applied in place.
3118 if not validate_path(path, validate_path_element):
3119 raise InvalidPathError(path)
3120 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3121 try:
3122 modify_stat: os.stat_result | None = os.lstat(full_path)
3123 except FileNotFoundError:
3124 modify_stat = None
3125 except OSError as e:
3126 raise OSError(
3127 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3128 ) from e
3130 if S_ISGITLINK(change.new.mode):
3131 _transition_to_submodule(
3132 repo, path, full_path, modify_stat, change.new, index
3133 )
3134 else:
3135 _transition_to_file(
3136 repo.object_store,
3137 path,
3138 full_path,
3139 modify_stat,
3140 change.new,
3141 index,
3142 honor_filemode,
3143 symlink_fn,
3144 blob_normalizer,
3145 tree_encoding,
3146 )
3148 index.write()
3151def _stat_matches_entry(
3152 st: os.stat_result, entry: IndexEntry, trust_ctime: bool = True
3153) -> bool:
3154 """Check if filesystem stat matches index entry stat.
3156 This is used to determine if a file might have changed without reading its content.
3157 Git uses this optimization to avoid expensive filter operations on unchanged files.
3159 Args:
3160 st: Filesystem stat result
3161 entry: Index entry to compare against
3162 trust_ctime: If True, also check ctime (default: True, matching Git behavior)
3163 Returns: True if stat matches and file is likely unchanged
3164 """
3165 # Compare change time (ctime) if trust_ctime is enabled
3166 if trust_ctime:
3167 # Get entry ctime with nanosecond precision if available
3168 if isinstance(entry.ctime, tuple):
3169 entry_ctime_sec = entry.ctime[0]
3170 entry_ctime_nsec = entry.ctime[1]
3171 else:
3172 entry_ctime_sec = int(entry.ctime)
3173 entry_ctime_nsec = 0
3175 if hasattr(st, "st_ctime_ns"):
3176 # Use nanosecond precision when available
3177 st_ctime_nsec = st.st_ctime_ns
3178 entry_ctime_nsec_total = entry_ctime_sec * 1_000_000_000 + entry_ctime_nsec
3179 if st_ctime_nsec != entry_ctime_nsec_total:
3180 return False
3181 else:
3182 # Fall back to second precision
3183 if int(st.st_ctime) != entry_ctime_sec:
3184 return False
3186 # Get entry mtime with nanosecond precision if available
3187 if isinstance(entry.mtime, tuple):
3188 entry_mtime_sec = entry.mtime[0]
3189 entry_mtime_nsec = entry.mtime[1]
3190 else:
3191 entry_mtime_sec = int(entry.mtime)
3192 entry_mtime_nsec = 0
3194 # Compare modification time with nanosecond precision if available
3195 # This is important for fast workflows (e.g., stash) where files can be
3196 # modified multiple times within the same second
3197 if hasattr(st, "st_mtime_ns"):
3198 # Use nanosecond precision when available
3199 st_mtime_nsec = st.st_mtime_ns
3200 entry_mtime_nsec_total = entry_mtime_sec * 1_000_000_000 + entry_mtime_nsec
3201 if st_mtime_nsec != entry_mtime_nsec_total:
3202 return False
3203 else:
3204 # Fall back to second precision
3205 if int(st.st_mtime) != entry_mtime_sec:
3206 return False
3208 # Compare file size
3209 if st.st_size != entry.size:
3210 return False
3212 # If all checks pass, file is likely unchanged
3213 return True
3216def _check_entry_for_changes(
3217 tree_path: bytes,
3218 entry: IndexEntry | ConflictedIndexEntry,
3219 root_path: bytes,
3220 filter_blob_callback: Callable[[Blob, bytes], Blob] | None = None,
3221 trust_ctime: bool = True,
3222) -> bytes | None:
3223 """Check a single index entry for changes.
3225 Args:
3226 tree_path: Path in the tree
3227 entry: Index entry to check
3228 root_path: Root filesystem path
3229 filter_blob_callback: Optional callback to filter blobs
3230 trust_ctime: If True, use ctime for change detection (default: True)
3231 Returns: tree_path if changed, None otherwise
3232 """
3233 if isinstance(entry, ConflictedIndexEntry):
3234 # Conflicted files are always unstaged
3235 return tree_path
3237 full_path = _tree_to_fs_path(root_path, tree_path)
3238 try:
3239 st = os.lstat(full_path)
3240 if stat.S_ISDIR(st.st_mode):
3241 if _has_directory_changed(tree_path, entry):
3242 return tree_path
3243 return None
3245 if not stat.S_ISREG(st.st_mode) and not stat.S_ISLNK(st.st_mode):
3246 return None
3248 # Optimization: If stat matches index entry (mtime and size unchanged),
3249 # we can skip reading and filtering the file entirely. This is a significant
3250 # performance improvement for repositories with many unchanged files.
3251 # Even with filters (e.g., LFS), if the file hasn't been modified (stat unchanged),
3252 # the filter output would be the same, so we can safely skip the expensive
3253 # filter operation. This addresses performance issues with LFS repositories
3254 # where filter operations can be very slow.
3255 if _stat_matches_entry(st, entry, trust_ctime):
3256 return None
3258 blob = blob_from_path_and_stat(full_path, st)
3260 if filter_blob_callback is not None:
3261 blob = filter_blob_callback(blob, tree_path)
3262 except FileNotFoundError:
3263 # The file was removed, so we assume that counts as
3264 # different from whatever file used to exist.
3265 return tree_path
3266 else:
3267 if blob.id != entry.sha:
3268 return tree_path
3269 return None
3272def get_unstaged_changes(
3273 index: Index,
3274 root_path: str | bytes,
3275 filter_blob_callback: Callable[..., Any] | None = None,
3276 preload_index: bool = False,
3277 trust_ctime: bool = True,
3278 max_stat: int | None = None,
3279) -> Generator[bytes, None, None]:
3280 """Walk through an index and check for differences against working tree.
3282 Args:
3283 index: index to check
3284 root_path: path in which to find files
3285 filter_blob_callback: Optional callback to filter blobs
3286 preload_index: If True, use parallel threads to check files (requires threading support)
3287 trust_ctime: If True, use ctime for change detection (default: True)
3288 max_stat: If set, limit the number of stat operations performed.
3289 When the limit is reached, remaining files are assumed unchanged.
3290 Returns: iterator over paths with unstaged changes
3291 """
3292 # For each entry in the index check the sha1 & ensure not staged
3293 if not isinstance(root_path, bytes):
3294 root_path = os.fsencode(root_path)
3296 stat_count = 0
3298 if preload_index:
3299 # Use parallel processing for better performance on slow filesystems
3300 try:
3301 import multiprocessing
3302 from concurrent.futures import ThreadPoolExecutor
3303 except ImportError:
3304 # If threading is not available, fall back to serial processing
3305 preload_index = False
3306 else:
3307 # Collect all entries first
3308 entries = list(index.iteritems())
3310 if max_stat is not None:
3311 # When max_stat is set, limit the entries we process
3312 entries = entries[:max_stat]
3314 # Use number of CPUs but cap at 8 threads to avoid overhead
3315 num_workers = min(multiprocessing.cpu_count(), 8)
3317 # Process entries in parallel
3318 with ThreadPoolExecutor(max_workers=num_workers) as executor:
3319 # Submit all tasks
3320 futures = [
3321 executor.submit(
3322 _check_entry_for_changes,
3323 tree_path,
3324 entry,
3325 root_path,
3326 filter_blob_callback,
3327 trust_ctime,
3328 )
3329 for tree_path, entry in entries
3330 ]
3332 # Yield results as they complete
3333 for future in futures:
3334 result = future.result()
3335 if result is not None:
3336 yield result
3338 if not preload_index:
3339 # Serial processing
3340 for tree_path, entry in index.iteritems():
3341 if max_stat is not None and stat_count >= max_stat:
3342 return
3343 result = _check_entry_for_changes(
3344 tree_path, entry, root_path, filter_blob_callback, trust_ctime
3345 )
3346 stat_count += 1
3347 if result is not None:
3348 yield result
3351def _decode_utf8_with_fallback(data: bytes) -> str:
3352 """Decode bytes as UTF-8, with lossy fallbacks for invalid sequences.
3354 Mirrors the behaviour of git-for-windows's ``xutftowcsn`` (in
3355 ``compat/mingw.c``) so that tree paths containing legacy-encoded or
3356 otherwise invalid UTF-8 produce the same on-disk filename as C git.
3358 Rules:
3359 * Valid UTF-8 (1-4 byte sequences, excluding overlongs and codepoints
3360 > U+10FFFF) is decoded normally.
3361 * Invalid bytes in 0xa0-0xff map 1:1 to U+00A0-U+00FF.
3362 * Invalid bytes in 0x80-0x9f are expanded to two lowercase ASCII hex
3363 digits (e.g. byte 0x80 -> "80").
3364 * Truncated multi-byte sequences and overlong/out-of-range encodings
3365 cause the lead byte to fall through to the above invalid-byte rules
3366 (the trail bytes are re-evaluated on the next iteration).
3367 """
3368 out: list[str] = []
3369 i = 0
3370 n = len(data)
3371 while i < n:
3372 c = data[i]
3373 if c < 0x80:
3374 out.append(chr(c))
3375 i += 1
3376 elif 0xC2 <= c < 0xE0 and i + 1 < n and (data[i + 1] & 0xC0) == 0x80:
3377 cp = ((c & 0x1F) << 6) | (data[i + 1] & 0x3F)
3378 out.append(chr(cp))
3379 i += 2
3380 elif (
3381 0xE0 <= c < 0xF0
3382 and i + 2 < n
3383 and not (c == 0xE0 and data[i + 1] < 0xA0)
3384 and (data[i + 1] & 0xC0) == 0x80
3385 and (data[i + 2] & 0xC0) == 0x80
3386 ):
3387 cp = ((c & 0x0F) << 12) | ((data[i + 1] & 0x3F) << 6) | (data[i + 2] & 0x3F)
3388 out.append(chr(cp))
3389 i += 3
3390 elif (
3391 0xF0 <= c < 0xF5
3392 and i + 3 < n
3393 and not (c == 0xF0 and data[i + 1] < 0x90)
3394 and not (c == 0xF4 and data[i + 1] >= 0x90)
3395 and (data[i + 1] & 0xC0) == 0x80
3396 and (data[i + 2] & 0xC0) == 0x80
3397 and (data[i + 3] & 0xC0) == 0x80
3398 ):
3399 cp = (
3400 ((c & 0x07) << 18)
3401 | ((data[i + 1] & 0x3F) << 12)
3402 | ((data[i + 2] & 0x3F) << 6)
3403 | (data[i + 3] & 0x3F)
3404 )
3405 out.append(chr(cp))
3406 i += 4
3407 elif c >= 0xA0:
3408 out.append(chr(c))
3409 i += 1
3410 else:
3411 out.append(f"{c:02x}")
3412 i += 1
3413 return "".join(out)
3416def _tree_to_fs_path(
3417 root_path: bytes, tree_path: bytes, tree_encoding: str = "utf-8"
3418) -> bytes:
3419 """Convert a git tree path to a file system path.
3421 Args:
3422 root_path: Root filesystem path
3423 tree_path: Git tree path as bytes (encoded with tree_encoding)
3424 tree_encoding: Encoding used for tree paths (default: utf-8)
3426 Returns: File system path.
3427 """
3428 assert isinstance(tree_path, bytes)
3429 if os_sep_bytes != b"/":
3430 sep_corrected_path = tree_path.replace(b"/", os_sep_bytes)
3431 else:
3432 sep_corrected_path = tree_path
3434 # On Windows, decode tree-encoded bytes to a str so they can flow into
3435 # the wide-char Win32 APIs via Python's filesystem layer. For UTF-8
3436 # (the default tree encoding) we use a lossy decoder that matches C
3437 # git's xutftowcsn fallbacks; for other encodings we let UnicodeDecodeError
3438 # propagate rather than silently producing a corrupt path.
3439 if sys.platform == "win32":
3440 if tree_encoding == "utf-8":
3441 tree_path_str = _decode_utf8_with_fallback(sep_corrected_path)
3442 else:
3443 tree_path_str = sep_corrected_path.decode(tree_encoding)
3444 sep_corrected_path = os.fsencode(tree_path_str)
3446 return os.path.join(root_path, sep_corrected_path)
3449def _fs_to_tree_path(fs_path: str | bytes, tree_encoding: str = "utf-8") -> bytes:
3450 """Convert a file system path to a git tree path.
3452 Args:
3453 fs_path: File system path.
3454 tree_encoding: Encoding to use for tree paths (default: utf-8)
3456 Returns: Git tree path as bytes (encoded with tree_encoding)
3457 """
3458 if not isinstance(fs_path, bytes):
3459 fs_path_bytes = os.fsencode(fs_path)
3460 else:
3461 fs_path_bytes = fs_path
3463 # On Windows the on-disk filename is a UTF-16 wide string; Python gives
3464 # us either str (already decoded) or bytes encoded via the filesystem
3465 # codec. Normalise to str, then encode under the tree encoding so the
3466 # resulting tree path is plain UTF-8. This matches C git's xwcstoutf,
3467 # which is just WideCharToMultiByte(CP_UTF8); it makes no attempt to
3468 # reverse the xutftowcsn fallbacks, so a file that was checked out from
3469 # a tree path with invalid UTF-8 will read back as the lossy form (the
3470 # same divergence C git exhibits, documented as a one-way mapping).
3471 if sys.platform == "win32":
3472 fs_path_str = os.fsdecode(fs_path_bytes)
3473 fs_path_bytes = fs_path_str.encode(tree_encoding)
3475 if os_sep_bytes != b"/":
3476 tree_path = fs_path_bytes.replace(os_sep_bytes, b"/")
3477 else:
3478 tree_path = fs_path_bytes
3479 return tree_path
3482def index_entry_from_directory(st: os.stat_result, path: bytes) -> IndexEntry | None:
3483 """Create an index entry for a directory.
3485 This is only used for submodules (directories containing .git).
3487 Args:
3488 st: Stat result for the directory
3489 path: Path to the directory
3491 Returns:
3492 IndexEntry for a submodule, or None if not a submodule
3493 """
3494 if os.path.exists(os.path.join(path, b".git")):
3495 head = read_submodule_head(path)
3496 if head is None:
3497 return None
3498 return index_entry_from_stat(st, head, mode=S_IFGITLINK)
3499 return None
3502def index_entry_from_path(
3503 path: bytes, object_store: ObjectContainer | None = None
3504) -> IndexEntry | None:
3505 """Create an index from a filesystem path.
3507 This returns an index value for files, symlinks
3508 and tree references. for directories and
3509 non-existent files it returns None
3511 Args:
3512 path: Path to create an index entry for
3513 object_store: Optional object store to
3514 save new blobs in
3515 Returns: An index entry; None for directories
3516 """
3517 assert isinstance(path, bytes)
3518 st = os.lstat(path)
3519 if stat.S_ISDIR(st.st_mode):
3520 return index_entry_from_directory(st, path)
3522 if stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode):
3523 blob = blob_from_path_and_stat(path, st)
3524 if object_store is not None:
3525 object_store.add_object(blob)
3526 return index_entry_from_stat(st, blob.id)
3528 return None
3531def iter_fresh_entries(
3532 paths: Iterable[bytes],
3533 root_path: bytes,
3534 object_store: ObjectContainer | None = None,
3535) -> Iterator[tuple[bytes, IndexEntry | None]]:
3536 """Iterate over current versions of index entries on disk.
3538 Args:
3539 paths: Paths to iterate over
3540 root_path: Root path to access from
3541 object_store: Optional store to save new blobs in
3542 Returns: Iterator over path, index_entry
3543 """
3544 for path in paths:
3545 p = _tree_to_fs_path(root_path, path)
3546 try:
3547 entry = index_entry_from_path(p, object_store=object_store)
3548 except (FileNotFoundError, IsADirectoryError):
3549 entry = None
3550 yield path, entry
3553def iter_fresh_objects(
3554 paths: Iterable[bytes],
3555 root_path: bytes,
3556 include_deleted: bool = False,
3557 object_store: ObjectContainer | None = None,
3558) -> Iterator[tuple[bytes, ObjectID | None, int | None]]:
3559 """Iterate over versions of objects on disk referenced by index.
3561 Args:
3562 paths: Paths to check
3563 root_path: Root path to access from
3564 include_deleted: Include deleted entries with sha and
3565 mode set to None
3566 object_store: Optional object store to report new items to
3567 Returns: Iterator over path, sha, mode
3568 """
3569 for path, entry in iter_fresh_entries(paths, root_path, object_store=object_store):
3570 if entry is None:
3571 if include_deleted:
3572 yield path, None, None
3573 else:
3574 yield path, entry.sha, cleanup_mode(entry.mode)
3577def refresh_index(index: Index, root_path: bytes) -> None:
3578 """Refresh the contents of an index.
3580 This is the equivalent to running 'git commit -a'.
3582 Args:
3583 index: Index to update
3584 root_path: Root filesystem path
3585 """
3586 for path, entry in iter_fresh_entries(index, root_path):
3587 if entry:
3588 index[path] = entry
3591class locked_index:
3592 """Lock the index while making modifications.
3594 Works as a context manager.
3595 """
3597 _file: "_GitFile"
3599 def __init__(self, path: bytes | str) -> None:
3600 """Initialize locked_index."""
3601 self._path = path
3603 def __enter__(self) -> Index:
3604 """Enter context manager and lock index."""
3605 f = GitFile(self._path, "wb")
3606 self._file = f
3607 self._index = Index(self._path)
3608 return self._index
3610 def __exit__(
3611 self,
3612 exc_type: type | None,
3613 exc_value: BaseException | None,
3614 traceback: types.TracebackType | None,
3615 ) -> None:
3616 """Exit context manager and unlock index."""
3617 if exc_type is not None:
3618 self._file.abort()
3619 return
3620 try:
3621 f = SHA1Writer(self._file)
3622 write_index_dict(f, self._index._byname)
3623 except BaseException:
3624 self._file.abort()
3625 else:
3626 f.close()