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 # Cache of leading directory components already verified to be real
2301 # directories, shared across the sorted iteration. See verify_leading_dirs.
2302 safe_prefix: list[bytes] = []
2304 for entry in iter_tree_contents(object_store, tree_id):
2305 assert (
2306 entry.path is not None and entry.mode is not None and entry.sha is not None
2307 )
2308 # Validate as we go and abort on the first invalid path,
2309 # leaving any files already written in place.
2310 if not validate_path(entry.path, validate_path_element):
2311 raise InvalidPathError(entry.path)
2312 # Refuse to write an entry whose leading path resolves through a
2313 # symlink materialized by an earlier entry; open(..., "wb") would
2314 # otherwise follow it and write outside the work tree.
2315 verify_leading_dirs(entry.path, safe_prefix, root_path)
2316 full_path = _tree_to_fs_path(root_path, entry.path, tree_encoding)
2318 if not os.path.exists(os.path.dirname(full_path)):
2319 os.makedirs(os.path.dirname(full_path))
2321 # TODO(jelmer): Merge new index into working tree
2322 if S_ISGITLINK(entry.mode):
2323 if not os.path.isdir(full_path):
2324 os.mkdir(full_path)
2325 st = os.lstat(full_path)
2326 # TODO(jelmer): record and return submodule paths
2327 else:
2328 obj = object_store[entry.sha]
2329 assert isinstance(obj, Blob)
2330 # Apply blob normalization for checkout if normalizer is provided
2331 if blob_normalizer is not None:
2332 obj = blob_normalizer.checkout_normalize(obj, entry.path)
2333 st = build_file_from_blob(
2334 obj,
2335 entry.mode,
2336 full_path,
2337 honor_filemode=honor_filemode,
2338 tree_encoding=tree_encoding,
2339 symlink_fn=symlink_fn,
2340 )
2342 # Add file to index
2343 if not honor_filemode or S_ISGITLINK(entry.mode):
2344 # we can not use tuple slicing to build a new tuple,
2345 # because on windows that will convert the times to
2346 # longs, which causes errors further along
2347 st_tuple = (
2348 entry.mode,
2349 st.st_ino,
2350 st.st_dev,
2351 st.st_nlink,
2352 st.st_uid,
2353 st.st_gid,
2354 st.st_size,
2355 st.st_atime,
2356 st.st_mtime,
2357 st.st_ctime,
2358 )
2359 st = st.__class__(st_tuple)
2360 # default to a stage 0 index entry (normal)
2361 # when reading from the filesystem
2362 index[entry.path] = index_entry_from_stat(st, entry.sha)
2364 index.write()
2367def blob_from_path_and_mode(
2368 fs_path: bytes, mode: int, tree_encoding: str = "utf-8"
2369) -> Blob:
2370 """Create a blob from a path and a stat object.
2372 Args:
2373 fs_path: Full file system path to file
2374 mode: File mode
2375 tree_encoding: Encoding to use for tree contents
2376 Returns: A `Blob` object
2377 """
2378 assert isinstance(fs_path, bytes)
2379 blob = Blob()
2380 if stat.S_ISLNK(mode):
2381 if sys.platform == "win32":
2382 # os.readlink on Python3 on Windows requires a unicode string.
2383 blob.data = os.readlink(os.fsdecode(fs_path)).encode(tree_encoding)
2384 else:
2385 blob.data = os.readlink(fs_path)
2386 else:
2387 with open(fs_path, "rb") as f:
2388 blob.data = f.read()
2389 return blob
2392def blob_from_path_and_stat(
2393 fs_path: bytes, st: os.stat_result, tree_encoding: str = "utf-8"
2394) -> Blob:
2395 """Create a blob from a path and a stat object.
2397 Args:
2398 fs_path: Full file system path to file
2399 st: A stat object
2400 tree_encoding: Encoding to use for tree contents
2401 Returns: A `Blob` object
2402 """
2403 return blob_from_path_and_mode(fs_path, st.st_mode, tree_encoding)
2406def read_submodule_head(path: str | bytes) -> bytes | None:
2407 """Read the head commit of a submodule.
2409 Args:
2410 path: path to the submodule
2411 Returns: HEAD sha, None if not a valid head/repository
2412 """
2413 from .errors import NotGitRepository
2414 from .repo import Repo
2416 # Repo currently expects a "str", so decode if necessary.
2417 # TODO(jelmer): Perhaps move this into Repo() ?
2418 if not isinstance(path, str):
2419 path = os.fsdecode(path)
2420 try:
2421 repo = Repo(path)
2422 except NotGitRepository:
2423 return None
2424 try:
2425 return repo.head()
2426 except KeyError:
2427 return None
2430def _has_directory_changed(tree_path: bytes, entry: IndexEntry) -> bool:
2431 """Check if a directory has changed after getting an error.
2433 When handling an error trying to create a blob from a path, call this
2434 function. It will check if the path is a directory. If it's a directory
2435 and a submodule, check the submodule head to see if it's has changed. If
2436 not, consider the file as changed as Git tracked a file and not a
2437 directory.
2439 Return true if the given path should be considered as changed and False
2440 otherwise or if the path is not a directory.
2441 """
2442 # This is actually a directory
2443 if os.path.exists(os.path.join(tree_path, b".git")):
2444 # Submodule
2445 head = read_submodule_head(tree_path)
2446 if entry.sha != head:
2447 return True
2448 else:
2449 # The file was changed to a directory, so consider it removed.
2450 return True
2452 return False
2455os_sep_bytes = os.sep.encode("ascii")
2458def _ensure_parent_dir_exists(full_path: bytes) -> None:
2459 """Ensure parent directory exists, checking no parent is a file."""
2460 parent_dir = os.path.dirname(full_path)
2461 if parent_dir and not os.path.exists(parent_dir):
2462 # Walk up the directory tree to find the first existing parent
2463 current = parent_dir
2464 parents_to_check: list[bytes] = []
2466 while current and not os.path.exists(current):
2467 parents_to_check.insert(0, current)
2468 new_parent = os.path.dirname(current)
2469 if new_parent == current:
2470 # Reached the root or can't go up further
2471 break
2472 current = new_parent
2474 # Check if the existing parent (if any) is a directory
2475 if current and os.path.exists(current) and not os.path.isdir(current):
2476 raise OSError(
2477 f"Cannot create directory, parent path is a file: {current!r}"
2478 )
2480 # Now check each parent we need to create isn't blocked by an existing file
2481 for parent_path in parents_to_check:
2482 if os.path.exists(parent_path) and not os.path.isdir(parent_path):
2483 raise OSError(
2484 f"Cannot create directory, parent path is a file: {parent_path!r}"
2485 )
2487 os.makedirs(parent_dir)
2490def _remove_file_with_readonly_handling(path: bytes) -> None:
2491 """Remove a file, handling read-only files on Windows.
2493 Args:
2494 path: Path to the file to remove
2495 """
2496 try:
2497 os.unlink(path)
2498 except PermissionError:
2499 # On Windows, remove read-only attribute and retry
2500 if sys.platform == "win32":
2501 os.chmod(path, stat.S_IWRITE | stat.S_IREAD)
2502 os.unlink(path)
2503 else:
2504 raise
2507def _remove_empty_parents(path: bytes, stop_at: bytes) -> None:
2508 """Remove empty parent directories up to stop_at."""
2509 parent = os.path.dirname(path)
2510 while parent and parent != stop_at:
2511 try:
2512 os.rmdir(parent)
2513 parent = os.path.dirname(parent)
2514 except FileNotFoundError:
2515 # Directory doesn't exist - stop trying
2516 break
2517 except OSError as e:
2518 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
2519 # Directory not empty - stop trying
2520 break
2521 raise
2524def _check_symlink_matches(
2525 full_path: bytes, repo_object_store: "BaseObjectStore", entry_sha: ObjectID
2526) -> bool:
2527 """Check if symlink target matches expected target.
2529 Returns True if symlink matches, False if it doesn't match.
2530 """
2531 try:
2532 current_target = os.readlink(full_path)
2533 blob_obj = repo_object_store[entry_sha]
2534 expected_target = blob_obj.as_raw_string()
2535 if isinstance(current_target, str):
2536 current_target = current_target.encode()
2537 return current_target == expected_target
2538 except FileNotFoundError:
2539 # Symlink doesn't exist
2540 return False
2541 except OSError as e:
2542 if e.errno == errno.EINVAL:
2543 # Not a symlink
2544 return False
2545 raise
2548def _check_file_matches(
2549 repo_object_store: "BaseObjectStore",
2550 full_path: bytes,
2551 entry_sha: ObjectID,
2552 entry_mode: int,
2553 current_stat: os.stat_result,
2554 honor_filemode: bool,
2555 blob_normalizer: "FilterBlobNormalizer | None" = None,
2556 tree_path: bytes | None = None,
2557) -> bool:
2558 """Check if a file on disk matches the expected git object.
2560 Returns True if file matches, False if it doesn't match.
2561 """
2562 # Check mode first (if honor_filemode is True)
2563 if honor_filemode:
2564 current_mode = stat.S_IMODE(current_stat.st_mode)
2565 expected_mode = stat.S_IMODE(entry_mode)
2567 # For regular files, only check the user executable bit, not group/other permissions
2568 # This matches Git's behavior where umask differences don't count as modifications
2569 if stat.S_ISREG(current_stat.st_mode):
2570 # Normalize regular file modes to ignore group/other write permissions
2571 current_mode_normalized = (
2572 current_mode & 0o755
2573 ) # Keep only user rwx and all read+execute
2574 expected_mode_normalized = expected_mode & 0o755
2576 # For Git compatibility, regular files should be either 644 or 755
2577 if expected_mode_normalized not in (0o644, 0o755):
2578 expected_mode_normalized = 0o644 # Default for regular files
2579 if current_mode_normalized not in (0o644, 0o755):
2580 # Determine if it should be executable based on user execute bit
2581 if current_mode & 0o100: # User execute bit is set
2582 current_mode_normalized = 0o755
2583 else:
2584 current_mode_normalized = 0o644
2586 if current_mode_normalized != expected_mode_normalized:
2587 return False
2588 else:
2589 # For non-regular files (symlinks, etc.), check mode exactly
2590 if current_mode != expected_mode:
2591 return False
2593 # If mode matches (or we don't care), check content via size first
2594 blob_obj = repo_object_store[entry_sha]
2595 if current_stat.st_size != blob_obj.raw_length():
2596 return False
2598 # Size matches, check actual content
2599 try:
2600 with open(full_path, "rb") as f:
2601 current_content = f.read()
2602 expected_content = blob_obj.as_raw_string()
2603 if blob_normalizer and tree_path is not None:
2604 assert isinstance(blob_obj, Blob)
2605 normalized_blob = blob_normalizer.checkout_normalize(
2606 blob_obj, tree_path
2607 )
2608 expected_content = normalized_blob.as_raw_string()
2609 return current_content == expected_content
2610 except (FileNotFoundError, PermissionError, IsADirectoryError):
2611 return False
2614def _transition_to_submodule(
2615 repo: "Repo",
2616 path: bytes,
2617 full_path: bytes,
2618 current_stat: os.stat_result | None,
2619 entry: IndexEntry | TreeEntry,
2620 index: Index,
2621) -> None:
2622 """Transition any type to submodule."""
2623 from .submodule import ensure_submodule_placeholder
2625 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2626 # Already a directory, just ensure .git file exists
2627 ensure_submodule_placeholder(repo, path)
2628 else:
2629 # Remove whatever is there and create submodule
2630 if current_stat is not None:
2631 _remove_file_with_readonly_handling(full_path)
2632 ensure_submodule_placeholder(repo, path)
2634 st = os.lstat(full_path)
2635 assert entry.sha is not None
2636 index[path] = index_entry_from_stat(st, entry.sha)
2639def _transition_to_file(
2640 object_store: "BaseObjectStore",
2641 path: bytes,
2642 full_path: bytes,
2643 current_stat: os.stat_result | None,
2644 entry: IndexEntry | TreeEntry,
2645 index: Index,
2646 honor_filemode: bool,
2647 symlink_fn: Callable[
2648 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2649 ]
2650 | None,
2651 blob_normalizer: "FilterBlobNormalizer | None",
2652 tree_encoding: str = "utf-8",
2653) -> None:
2654 """Transition any type to regular file or symlink."""
2655 assert entry.sha is not None and entry.mode is not None
2656 # Check if we need to update
2657 if (
2658 current_stat is not None
2659 and stat.S_ISREG(current_stat.st_mode)
2660 and not stat.S_ISLNK(entry.mode)
2661 ):
2662 # File to file - check if update needed
2663 file_matches = _check_file_matches(
2664 object_store,
2665 full_path,
2666 entry.sha,
2667 entry.mode,
2668 current_stat,
2669 honor_filemode,
2670 blob_normalizer,
2671 path,
2672 )
2673 needs_update = not file_matches
2674 elif (
2675 current_stat is not None
2676 and stat.S_ISLNK(current_stat.st_mode)
2677 and stat.S_ISLNK(entry.mode)
2678 ):
2679 # Symlink to symlink - check if update needed
2680 symlink_matches = _check_symlink_matches(full_path, object_store, entry.sha)
2681 needs_update = not symlink_matches
2682 else:
2683 needs_update = True
2685 if not needs_update:
2686 # Just update index - current_stat should always be valid here since we're not updating
2687 assert current_stat is not None
2688 index[path] = index_entry_from_stat(current_stat, entry.sha)
2689 return
2691 # Remove existing entry if needed
2692 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2693 # Remove directory
2694 dir_contents = set(os.listdir(full_path))
2695 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
2697 if git_file_name in dir_contents:
2698 if dir_contents != {git_file_name}:
2699 raise IsADirectoryError(
2700 f"Cannot replace submodule with untracked files: {full_path!r}"
2701 )
2702 shutil.rmtree(full_path)
2703 else:
2704 try:
2705 os.rmdir(full_path)
2706 except OSError as e:
2707 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
2708 raise IsADirectoryError(
2709 f"Cannot replace non-empty directory with file: {full_path!r}"
2710 )
2711 raise
2712 elif current_stat is not None:
2713 _remove_file_with_readonly_handling(full_path)
2715 # Ensure parent directory exists
2716 _ensure_parent_dir_exists(full_path)
2718 # Write the file
2719 blob_obj = object_store[entry.sha]
2720 assert isinstance(blob_obj, Blob)
2721 if blob_normalizer:
2722 blob_obj = blob_normalizer.checkout_normalize(blob_obj, path)
2723 st = build_file_from_blob(
2724 blob_obj,
2725 entry.mode,
2726 full_path,
2727 honor_filemode=honor_filemode,
2728 tree_encoding=tree_encoding,
2729 symlink_fn=symlink_fn,
2730 )
2731 index[path] = index_entry_from_stat(st, entry.sha)
2734def _transition_to_absent(
2735 repo: "Repo",
2736 path: bytes,
2737 full_path: bytes,
2738 current_stat: os.stat_result | None,
2739 index: Index,
2740) -> None:
2741 """Remove any type of entry."""
2742 if current_stat is None:
2743 return
2745 if stat.S_ISDIR(current_stat.st_mode):
2746 # Check if it's a submodule directory
2747 dir_contents = set(os.listdir(full_path))
2748 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
2750 if git_file_name in dir_contents and dir_contents == {git_file_name}:
2751 shutil.rmtree(full_path)
2752 else:
2753 try:
2754 os.rmdir(full_path)
2755 except OSError as e:
2756 if e.errno not in (errno.ENOTEMPTY, errno.EEXIST):
2757 raise
2758 else:
2759 _remove_file_with_readonly_handling(full_path)
2761 try:
2762 del index[path]
2763 except KeyError:
2764 pass
2766 # Try to remove empty parent directories
2767 _remove_empty_parents(
2768 full_path, repo.path if isinstance(repo.path, bytes) else repo.path.encode()
2769 )
2772def detect_case_only_renames(
2773 changes: Sequence["TreeChange"],
2774 config: "Config",
2775) -> list["TreeChange"]:
2776 """Detect and transform case-only renames in a list of tree changes.
2778 This function identifies file renames that only differ in case (e.g.,
2779 README.txt -> readme.txt) and transforms matching ADD/DELETE pairs into
2780 CHANGE_RENAME operations. It uses filesystem-appropriate path normalization
2781 based on the repository configuration.
2783 Args:
2784 changes: List of TreeChange objects representing file changes
2785 config: Repository configuration object
2787 Returns:
2788 New list of TreeChange objects with case-only renames converted to CHANGE_RENAME
2789 """
2790 from .diff_tree import (
2791 CHANGE_ADD,
2792 CHANGE_COPY,
2793 CHANGE_DELETE,
2794 CHANGE_MODIFY,
2795 CHANGE_RENAME,
2796 TreeChange,
2797 )
2799 # Build dictionaries of old and new paths with their normalized forms
2800 old_paths_normalized = {}
2801 new_paths_normalized = {}
2802 old_changes = {} # Map from old path to change object
2803 new_changes = {} # Map from new path to change object
2805 # Get the appropriate normalizer based on config
2806 normalize_func = get_path_element_normalizer(config)
2808 def normalize_path(path: bytes) -> bytes:
2809 """Normalize entire path using element normalization."""
2810 return b"/".join(normalize_func(part) for part in path.split(b"/"))
2812 # Pre-normalize all paths once to avoid repeated normalization
2813 for change in changes:
2814 if change.type == CHANGE_DELETE and change.old:
2815 assert change.old.path is not None
2816 try:
2817 normalized = normalize_path(change.old.path)
2818 except UnicodeDecodeError:
2819 logger.warning(
2820 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2821 change.old.path,
2822 )
2823 else:
2824 old_paths_normalized[normalized] = change.old.path
2825 old_changes[change.old.path] = change
2826 elif change.type == CHANGE_RENAME and change.old:
2827 assert change.old.path is not None
2828 # Treat RENAME as DELETE + ADD for case-only detection
2829 try:
2830 normalized = normalize_path(change.old.path)
2831 except UnicodeDecodeError:
2832 logger.warning(
2833 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2834 change.old.path,
2835 )
2836 else:
2837 old_paths_normalized[normalized] = change.old.path
2838 old_changes[change.old.path] = change
2840 if (
2841 change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY)
2842 and change.new
2843 ):
2844 assert change.new.path is not None
2845 try:
2846 normalized = normalize_path(change.new.path)
2847 except UnicodeDecodeError:
2848 logger.warning(
2849 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2850 change.new.path,
2851 )
2852 else:
2853 new_paths_normalized[normalized] = change.new.path
2854 new_changes[change.new.path] = change
2856 # Find case-only renames and transform changes
2857 case_only_renames = set()
2858 new_rename_changes = []
2860 for norm_path, old_path in old_paths_normalized.items():
2861 if norm_path in new_paths_normalized:
2862 new_path = new_paths_normalized[norm_path]
2863 if old_path != new_path:
2864 # Found a case-only rename
2865 old_change = old_changes[old_path]
2866 new_change = new_changes[new_path]
2868 # Create a CHANGE_RENAME to replace the DELETE and ADD/MODIFY pair
2869 if new_change.type == CHANGE_ADD:
2870 # Simple case: DELETE + ADD becomes RENAME
2871 rename_change = TreeChange(
2872 CHANGE_RENAME, old_change.old, new_change.new
2873 )
2874 else:
2875 # Complex case: DELETE + MODIFY becomes RENAME
2876 # Use the old file from DELETE and new file from MODIFY
2877 rename_change = TreeChange(
2878 CHANGE_RENAME, old_change.old, new_change.new
2879 )
2881 new_rename_changes.append(rename_change)
2883 # Mark the old changes for removal
2884 case_only_renames.add(old_change)
2885 case_only_renames.add(new_change)
2887 # Return new list with original ADD/DELETE changes replaced by renames
2888 result = [change for change in changes if change not in case_only_renames]
2889 result.extend(new_rename_changes)
2890 return result
2893def update_working_tree(
2894 repo: "Repo",
2895 old_tree_id: bytes | None,
2896 new_tree_id: bytes,
2897 change_iterator: Iterator["TreeChange"],
2898 honor_filemode: bool = True,
2899 validate_path_element: Callable[[bytes], bool] | None = None,
2900 symlink_fn: Callable[
2901 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2902 ]
2903 | None = None,
2904 force_remove_untracked: bool = False,
2905 blob_normalizer: "FilterBlobNormalizer | None" = None,
2906 tree_encoding: str = "utf-8",
2907 allow_overwrite_modified: bool = False,
2908 *,
2909 config: "Config | None" = None,
2910) -> None:
2911 """Update the working tree and index to match a new tree.
2913 This function handles:
2914 - Adding new files
2915 - Updating modified files
2916 - Removing deleted files
2917 - Cleaning up empty directories
2919 Args:
2920 repo: Repository object
2921 old_tree_id: SHA of the tree before the update
2922 new_tree_id: SHA of the tree to update to
2923 change_iterator: Iterator of TreeChange objects to apply
2924 honor_filemode: An optional flag to honor core.filemode setting
2925 validate_path_element: Function to validate path elements to check out.
2926 If None, derived from ``config`` so that ``core.protectNTFS`` and
2927 ``core.protectHFS`` are honored by default.
2928 symlink_fn: Function to use for creating symlinks
2929 force_remove_untracked: If True, remove files that exist in working
2930 directory but not in target tree, even if old_tree_id is None
2931 blob_normalizer: An optional BlobNormalizer to use for converting line
2932 endings when writing blobs to the working directory.
2933 tree_encoding: Encoding used for tree paths (default: utf-8)
2934 allow_overwrite_modified: If False, raise an error when attempting to
2935 overwrite files that have been modified compared to old_tree_id
2936 config: Repository configuration. If None, falls back to
2937 ``repo.get_config_stack()``.
2938 """
2939 from .diff_tree import (
2940 CHANGE_ADD,
2941 CHANGE_COPY,
2942 CHANGE_DELETE,
2943 CHANGE_MODIFY,
2944 CHANGE_RENAME,
2945 CHANGE_UNCHANGED,
2946 )
2948 if force_remove_untracked:
2949 import warnings
2951 warnings.warn(
2952 "force_remove_untracked is a no-op and will be removed in a future release",
2953 DeprecationWarning,
2954 )
2955 repo_path = repo.path if isinstance(repo.path, bytes) else repo.path.encode()
2956 if config is None:
2957 config = repo.get_config_stack()
2959 if validate_path_element is None:
2960 # Derive the validator from config so callers cannot accidentally
2961 # skip ``core.protectNTFS``/``core.protectHFS`` enforcement by
2962 # omitting this argument.
2963 validate_path_element = get_path_element_validator(config)
2965 index = repo.open_index(config=config)
2967 # Convert iterator to list since we need multiple passes
2968 changes = list(change_iterator)
2970 # Transform case-only renames on case-insensitive filesystems
2971 import platform
2973 default_ignore_case = platform.system() in ("Windows", "Darwin")
2974 config = repo.get_config()
2975 ignore_case = config.get_boolean((b"core",), b"ignorecase", default_ignore_case)
2977 if ignore_case:
2978 config = repo.get_config()
2979 changes = detect_case_only_renames(changes, config)
2981 # Check for path conflicts where files need to become directories
2982 paths_becoming_dirs = set()
2983 for change in changes:
2984 if change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY):
2985 assert change.new is not None
2986 path = change.new.path
2987 assert path is not None
2988 if b"/" in path: # This is a file inside a directory
2989 # Check if any parent path exists as a file in the old tree or changes
2990 parts = path.split(b"/")
2991 for i in range(1, len(parts)):
2992 parent = b"/".join(parts[:i])
2993 # See if this parent path is being deleted (was a file, becoming a dir)
2994 for other_change in changes:
2995 if (
2996 other_change.type == CHANGE_DELETE
2997 and other_change.old
2998 and other_change.old.path == parent
2999 ):
3000 paths_becoming_dirs.add(parent)
3002 # Check if any path that needs to become a directory has been modified
3003 for path in paths_becoming_dirs:
3004 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3005 try:
3006 current_stat = os.lstat(full_path)
3007 except FileNotFoundError:
3008 continue # File doesn't exist, nothing to check
3009 except OSError as e:
3010 raise OSError(
3011 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3012 ) from e
3014 if stat.S_ISREG(current_stat.st_mode):
3015 # Find the old entry for this path
3016 old_change = None
3017 for change in changes:
3018 if (
3019 change.type == CHANGE_DELETE
3020 and change.old
3021 and change.old.path == path
3022 ):
3023 old_change = change
3024 break
3026 if old_change:
3027 # Check if file has been modified
3028 assert old_change.old is not None
3029 assert (
3030 old_change.old.sha is not None and old_change.old.mode is not None
3031 )
3032 file_matches = _check_file_matches(
3033 repo.object_store,
3034 full_path,
3035 old_change.old.sha,
3036 old_change.old.mode,
3037 current_stat,
3038 honor_filemode,
3039 blob_normalizer,
3040 path,
3041 )
3042 if not file_matches:
3043 raise OSError(
3044 f"Cannot replace modified file with directory: {path!r}"
3045 )
3047 # Check for uncommitted modifications before making any changes
3048 if not allow_overwrite_modified and old_tree_id:
3049 for change in changes:
3050 # Only check files that are being modified or deleted
3051 if change.type in (CHANGE_MODIFY, CHANGE_DELETE) and change.old:
3052 path = change.old.path
3053 assert path is not None
3054 if not validate_path(path, validate_path_element):
3055 continue
3057 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3058 try:
3059 current_stat = os.lstat(full_path)
3060 except FileNotFoundError:
3061 continue # File doesn't exist, nothing to check
3062 except OSError as e:
3063 raise OSError(
3064 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3065 ) from e
3067 if stat.S_ISREG(current_stat.st_mode):
3068 # Check if working tree file differs from old tree
3069 assert change.old.sha is not None and change.old.mode is not None
3070 file_matches = _check_file_matches(
3071 repo.object_store,
3072 full_path,
3073 change.old.sha,
3074 change.old.mode,
3075 current_stat,
3076 honor_filemode,
3077 blob_normalizer,
3078 path,
3079 )
3080 if not file_matches:
3081 from .errors import WorkingTreeModifiedError
3083 raise WorkingTreeModifiedError(
3084 f"Your local changes to '{path.decode('utf-8', errors='replace')}' "
3085 f"would be overwritten by checkout. "
3086 f"Please commit your changes or stash them before you switch branches."
3087 )
3089 # Apply the changes
3090 for change in changes:
3091 if change.type in (CHANGE_DELETE, CHANGE_RENAME):
3092 # Remove file/directory
3093 assert change.old is not None and change.old.path is not None
3094 path = change.old.path
3095 if not validate_path(path, validate_path_element):
3096 continue
3098 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3099 try:
3100 delete_stat: os.stat_result | None = os.lstat(full_path)
3101 except FileNotFoundError:
3102 delete_stat = None
3103 except OSError as e:
3104 raise OSError(
3105 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3106 ) from e
3108 _transition_to_absent(repo, path, full_path, delete_stat, index)
3110 if change.type in (
3111 CHANGE_ADD,
3112 CHANGE_MODIFY,
3113 CHANGE_UNCHANGED,
3114 CHANGE_COPY,
3115 CHANGE_RENAME,
3116 ):
3117 # Add or modify file
3118 assert (
3119 change.new is not None
3120 and change.new.path is not None
3121 and change.new.mode is not None
3122 )
3123 path = change.new.path
3124 # Validate as we go and abort on the first invalid path,
3125 # leaving any changes already applied in place.
3126 if not validate_path(path, validate_path_element):
3127 raise InvalidPathError(path)
3128 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3129 try:
3130 modify_stat: os.stat_result | None = os.lstat(full_path)
3131 except FileNotFoundError:
3132 modify_stat = None
3133 except OSError as e:
3134 raise OSError(
3135 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3136 ) from e
3138 if S_ISGITLINK(change.new.mode):
3139 _transition_to_submodule(
3140 repo, path, full_path, modify_stat, change.new, index
3141 )
3142 else:
3143 _transition_to_file(
3144 repo.object_store,
3145 path,
3146 full_path,
3147 modify_stat,
3148 change.new,
3149 index,
3150 honor_filemode,
3151 symlink_fn,
3152 blob_normalizer,
3153 tree_encoding,
3154 )
3156 index.write()
3159def _stat_matches_entry(
3160 st: os.stat_result, entry: IndexEntry, trust_ctime: bool = True
3161) -> bool:
3162 """Check if filesystem stat matches index entry stat.
3164 This is used to determine if a file might have changed without reading its content.
3165 Git uses this optimization to avoid expensive filter operations on unchanged files.
3167 Args:
3168 st: Filesystem stat result
3169 entry: Index entry to compare against
3170 trust_ctime: If True, also check ctime (default: True, matching Git behavior)
3171 Returns: True if stat matches and file is likely unchanged
3172 """
3173 # Compare change time (ctime) if trust_ctime is enabled
3174 if trust_ctime:
3175 # Get entry ctime with nanosecond precision if available
3176 if isinstance(entry.ctime, tuple):
3177 entry_ctime_sec = entry.ctime[0]
3178 entry_ctime_nsec = entry.ctime[1]
3179 else:
3180 entry_ctime_sec = int(entry.ctime)
3181 entry_ctime_nsec = 0
3183 if hasattr(st, "st_ctime_ns"):
3184 # Use nanosecond precision when available
3185 st_ctime_nsec = st.st_ctime_ns
3186 entry_ctime_nsec_total = entry_ctime_sec * 1_000_000_000 + entry_ctime_nsec
3187 if st_ctime_nsec != entry_ctime_nsec_total:
3188 return False
3189 else:
3190 # Fall back to second precision
3191 if int(st.st_ctime) != entry_ctime_sec:
3192 return False
3194 # Get entry mtime with nanosecond precision if available
3195 if isinstance(entry.mtime, tuple):
3196 entry_mtime_sec = entry.mtime[0]
3197 entry_mtime_nsec = entry.mtime[1]
3198 else:
3199 entry_mtime_sec = int(entry.mtime)
3200 entry_mtime_nsec = 0
3202 # Compare modification time with nanosecond precision if available
3203 # This is important for fast workflows (e.g., stash) where files can be
3204 # modified multiple times within the same second
3205 if hasattr(st, "st_mtime_ns"):
3206 # Use nanosecond precision when available
3207 st_mtime_nsec = st.st_mtime_ns
3208 entry_mtime_nsec_total = entry_mtime_sec * 1_000_000_000 + entry_mtime_nsec
3209 if st_mtime_nsec != entry_mtime_nsec_total:
3210 return False
3211 else:
3212 # Fall back to second precision
3213 if int(st.st_mtime) != entry_mtime_sec:
3214 return False
3216 # Compare file size
3217 if st.st_size != entry.size:
3218 return False
3220 # If all checks pass, file is likely unchanged
3221 return True
3224def _check_entry_for_changes(
3225 tree_path: bytes,
3226 entry: IndexEntry | ConflictedIndexEntry,
3227 root_path: bytes,
3228 filter_blob_callback: Callable[[Blob, bytes], Blob] | None = None,
3229 trust_ctime: bool = True,
3230) -> bytes | None:
3231 """Check a single index entry for changes.
3233 Args:
3234 tree_path: Path in the tree
3235 entry: Index entry to check
3236 root_path: Root filesystem path
3237 filter_blob_callback: Optional callback to filter blobs
3238 trust_ctime: If True, use ctime for change detection (default: True)
3239 Returns: tree_path if changed, None otherwise
3240 """
3241 if isinstance(entry, ConflictedIndexEntry):
3242 # Conflicted files are always unstaged
3243 return tree_path
3245 full_path = _tree_to_fs_path(root_path, tree_path)
3246 try:
3247 st = os.lstat(full_path)
3248 if stat.S_ISDIR(st.st_mode):
3249 if _has_directory_changed(tree_path, entry):
3250 return tree_path
3251 return None
3253 if not stat.S_ISREG(st.st_mode) and not stat.S_ISLNK(st.st_mode):
3254 return None
3256 # Optimization: If stat matches index entry (mtime and size unchanged),
3257 # we can skip reading and filtering the file entirely. This is a significant
3258 # performance improvement for repositories with many unchanged files.
3259 # Even with filters (e.g., LFS), if the file hasn't been modified (stat unchanged),
3260 # the filter output would be the same, so we can safely skip the expensive
3261 # filter operation. This addresses performance issues with LFS repositories
3262 # where filter operations can be very slow.
3263 if _stat_matches_entry(st, entry, trust_ctime):
3264 return None
3266 blob = blob_from_path_and_stat(full_path, st)
3268 if filter_blob_callback is not None:
3269 blob = filter_blob_callback(blob, tree_path)
3270 except FileNotFoundError:
3271 # The file was removed, so we assume that counts as
3272 # different from whatever file used to exist.
3273 return tree_path
3274 else:
3275 if blob.id != entry.sha:
3276 return tree_path
3277 return None
3280def get_unstaged_changes(
3281 index: Index,
3282 root_path: str | bytes,
3283 filter_blob_callback: Callable[..., Any] | None = None,
3284 preload_index: bool = False,
3285 trust_ctime: bool = True,
3286 max_stat: int | None = None,
3287) -> Generator[bytes, None, None]:
3288 """Walk through an index and check for differences against working tree.
3290 Args:
3291 index: index to check
3292 root_path: path in which to find files
3293 filter_blob_callback: Optional callback to filter blobs
3294 preload_index: If True, use parallel threads to check files (requires threading support)
3295 trust_ctime: If True, use ctime for change detection (default: True)
3296 max_stat: If set, limit the number of stat operations performed.
3297 When the limit is reached, remaining files are assumed unchanged.
3298 Returns: iterator over paths with unstaged changes
3299 """
3300 # For each entry in the index check the sha1 & ensure not staged
3301 if not isinstance(root_path, bytes):
3302 root_path = os.fsencode(root_path)
3304 stat_count = 0
3306 if preload_index:
3307 # Use parallel processing for better performance on slow filesystems
3308 try:
3309 import multiprocessing
3310 from concurrent.futures import ThreadPoolExecutor
3311 except ImportError:
3312 # If threading is not available, fall back to serial processing
3313 preload_index = False
3314 else:
3315 # Collect all entries first
3316 entries = list(index.iteritems())
3318 if max_stat is not None:
3319 # When max_stat is set, limit the entries we process
3320 entries = entries[:max_stat]
3322 # Use number of CPUs but cap at 8 threads to avoid overhead
3323 num_workers = min(multiprocessing.cpu_count(), 8)
3325 # Process entries in parallel
3326 with ThreadPoolExecutor(max_workers=num_workers) as executor:
3327 # Submit all tasks
3328 futures = [
3329 executor.submit(
3330 _check_entry_for_changes,
3331 tree_path,
3332 entry,
3333 root_path,
3334 filter_blob_callback,
3335 trust_ctime,
3336 )
3337 for tree_path, entry in entries
3338 ]
3340 # Yield results as they complete
3341 for future in futures:
3342 result = future.result()
3343 if result is not None:
3344 yield result
3346 if not preload_index:
3347 # Serial processing
3348 for tree_path, entry in index.iteritems():
3349 if max_stat is not None and stat_count >= max_stat:
3350 return
3351 result = _check_entry_for_changes(
3352 tree_path, entry, root_path, filter_blob_callback, trust_ctime
3353 )
3354 stat_count += 1
3355 if result is not None:
3356 yield result
3359def _decode_utf8_with_fallback(data: bytes) -> str:
3360 """Decode bytes as UTF-8, with lossy fallbacks for invalid sequences.
3362 Mirrors the behaviour of git-for-windows's ``xutftowcsn`` (in
3363 ``compat/mingw.c``) so that tree paths containing legacy-encoded or
3364 otherwise invalid UTF-8 produce the same on-disk filename as C git.
3366 Rules:
3367 * Valid UTF-8 (1-4 byte sequences, excluding overlongs and codepoints
3368 > U+10FFFF) is decoded normally.
3369 * Invalid bytes in 0xa0-0xff map 1:1 to U+00A0-U+00FF.
3370 * Invalid bytes in 0x80-0x9f are expanded to two lowercase ASCII hex
3371 digits (e.g. byte 0x80 -> "80").
3372 * Truncated multi-byte sequences and overlong/out-of-range encodings
3373 cause the lead byte to fall through to the above invalid-byte rules
3374 (the trail bytes are re-evaluated on the next iteration).
3375 """
3376 out: list[str] = []
3377 i = 0
3378 n = len(data)
3379 while i < n:
3380 c = data[i]
3381 if c < 0x80:
3382 out.append(chr(c))
3383 i += 1
3384 elif 0xC2 <= c < 0xE0 and i + 1 < n and (data[i + 1] & 0xC0) == 0x80:
3385 cp = ((c & 0x1F) << 6) | (data[i + 1] & 0x3F)
3386 out.append(chr(cp))
3387 i += 2
3388 elif (
3389 0xE0 <= c < 0xF0
3390 and i + 2 < n
3391 and not (c == 0xE0 and data[i + 1] < 0xA0)
3392 and (data[i + 1] & 0xC0) == 0x80
3393 and (data[i + 2] & 0xC0) == 0x80
3394 ):
3395 cp = ((c & 0x0F) << 12) | ((data[i + 1] & 0x3F) << 6) | (data[i + 2] & 0x3F)
3396 out.append(chr(cp))
3397 i += 3
3398 elif (
3399 0xF0 <= c < 0xF5
3400 and i + 3 < n
3401 and not (c == 0xF0 and data[i + 1] < 0x90)
3402 and not (c == 0xF4 and data[i + 1] >= 0x90)
3403 and (data[i + 1] & 0xC0) == 0x80
3404 and (data[i + 2] & 0xC0) == 0x80
3405 and (data[i + 3] & 0xC0) == 0x80
3406 ):
3407 cp = (
3408 ((c & 0x07) << 18)
3409 | ((data[i + 1] & 0x3F) << 12)
3410 | ((data[i + 2] & 0x3F) << 6)
3411 | (data[i + 3] & 0x3F)
3412 )
3413 out.append(chr(cp))
3414 i += 4
3415 elif c >= 0xA0:
3416 out.append(chr(c))
3417 i += 1
3418 else:
3419 out.append(f"{c:02x}")
3420 i += 1
3421 return "".join(out)
3424def _tree_to_fs_path(
3425 root_path: bytes, tree_path: bytes, tree_encoding: str = "utf-8"
3426) -> bytes:
3427 """Convert a git tree path to a file system path.
3429 Args:
3430 root_path: Root filesystem path
3431 tree_path: Git tree path as bytes (encoded with tree_encoding)
3432 tree_encoding: Encoding used for tree paths (default: utf-8)
3434 Returns: File system path.
3435 """
3436 assert isinstance(tree_path, bytes)
3437 if os_sep_bytes != b"/":
3438 sep_corrected_path = tree_path.replace(b"/", os_sep_bytes)
3439 else:
3440 sep_corrected_path = tree_path
3442 # On Windows, decode tree-encoded bytes to a str so they can flow into
3443 # the wide-char Win32 APIs via Python's filesystem layer. For UTF-8
3444 # (the default tree encoding) we use a lossy decoder that matches C
3445 # git's xutftowcsn fallbacks; for other encodings we let UnicodeDecodeError
3446 # propagate rather than silently producing a corrupt path.
3447 if sys.platform == "win32":
3448 if tree_encoding == "utf-8":
3449 tree_path_str = _decode_utf8_with_fallback(sep_corrected_path)
3450 else:
3451 tree_path_str = sep_corrected_path.decode(tree_encoding)
3452 sep_corrected_path = os.fsencode(tree_path_str)
3454 return os.path.join(root_path, sep_corrected_path)
3457def _fs_to_tree_path(fs_path: str | bytes, tree_encoding: str = "utf-8") -> bytes:
3458 """Convert a file system path to a git tree path.
3460 Args:
3461 fs_path: File system path.
3462 tree_encoding: Encoding to use for tree paths (default: utf-8)
3464 Returns: Git tree path as bytes (encoded with tree_encoding)
3465 """
3466 if not isinstance(fs_path, bytes):
3467 fs_path_bytes = os.fsencode(fs_path)
3468 else:
3469 fs_path_bytes = fs_path
3471 # On Windows the on-disk filename is a UTF-16 wide string; Python gives
3472 # us either str (already decoded) or bytes encoded via the filesystem
3473 # codec. Normalise to str, then encode under the tree encoding so the
3474 # resulting tree path is plain UTF-8. This matches C git's xwcstoutf,
3475 # which is just WideCharToMultiByte(CP_UTF8); it makes no attempt to
3476 # reverse the xutftowcsn fallbacks, so a file that was checked out from
3477 # a tree path with invalid UTF-8 will read back as the lossy form (the
3478 # same divergence C git exhibits, documented as a one-way mapping).
3479 if sys.platform == "win32":
3480 fs_path_str = os.fsdecode(fs_path_bytes)
3481 fs_path_bytes = fs_path_str.encode(tree_encoding)
3483 if os_sep_bytes != b"/":
3484 tree_path = fs_path_bytes.replace(os_sep_bytes, b"/")
3485 else:
3486 tree_path = fs_path_bytes
3487 return tree_path
3490def index_entry_from_directory(st: os.stat_result, path: bytes) -> IndexEntry | None:
3491 """Create an index entry for a directory.
3493 This is only used for submodules (directories containing .git).
3495 Args:
3496 st: Stat result for the directory
3497 path: Path to the directory
3499 Returns:
3500 IndexEntry for a submodule, or None if not a submodule
3501 """
3502 if os.path.exists(os.path.join(path, b".git")):
3503 head = read_submodule_head(path)
3504 if head is None:
3505 return None
3506 return index_entry_from_stat(st, head, mode=S_IFGITLINK)
3507 return None
3510def index_entry_from_path(
3511 path: bytes, object_store: ObjectContainer | None = None
3512) -> IndexEntry | None:
3513 """Create an index from a filesystem path.
3515 This returns an index value for files, symlinks
3516 and tree references. for directories and
3517 non-existent files it returns None
3519 Args:
3520 path: Path to create an index entry for
3521 object_store: Optional object store to
3522 save new blobs in
3523 Returns: An index entry; None for directories
3524 """
3525 assert isinstance(path, bytes)
3526 st = os.lstat(path)
3527 if stat.S_ISDIR(st.st_mode):
3528 return index_entry_from_directory(st, path)
3530 if stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode):
3531 blob = blob_from_path_and_stat(path, st)
3532 if object_store is not None:
3533 object_store.add_object(blob)
3534 return index_entry_from_stat(st, blob.id)
3536 return None
3539def iter_fresh_entries(
3540 paths: Iterable[bytes],
3541 root_path: bytes,
3542 object_store: ObjectContainer | None = None,
3543) -> Iterator[tuple[bytes, IndexEntry | None]]:
3544 """Iterate over current versions of index entries on disk.
3546 Args:
3547 paths: Paths to iterate over
3548 root_path: Root path to access from
3549 object_store: Optional store to save new blobs in
3550 Returns: Iterator over path, index_entry
3551 """
3552 for path in paths:
3553 p = _tree_to_fs_path(root_path, path)
3554 try:
3555 entry = index_entry_from_path(p, object_store=object_store)
3556 except (FileNotFoundError, IsADirectoryError):
3557 entry = None
3558 yield path, entry
3561def iter_fresh_objects(
3562 paths: Iterable[bytes],
3563 root_path: bytes,
3564 include_deleted: bool = False,
3565 object_store: ObjectContainer | None = None,
3566) -> Iterator[tuple[bytes, ObjectID | None, int | None]]:
3567 """Iterate over versions of objects on disk referenced by index.
3569 Args:
3570 paths: Paths to check
3571 root_path: Root path to access from
3572 include_deleted: Include deleted entries with sha and
3573 mode set to None
3574 object_store: Optional object store to report new items to
3575 Returns: Iterator over path, sha, mode
3576 """
3577 for path, entry in iter_fresh_entries(paths, root_path, object_store=object_store):
3578 if entry is None:
3579 if include_deleted:
3580 yield path, None, None
3581 else:
3582 yield path, entry.sha, cleanup_mode(entry.mode)
3585def refresh_index(index: Index, root_path: bytes) -> None:
3586 """Refresh the contents of an index.
3588 This is the equivalent to running 'git commit -a'.
3590 Args:
3591 index: Index to update
3592 root_path: Root filesystem path
3593 """
3594 for path, entry in iter_fresh_entries(index, root_path):
3595 if entry:
3596 index[path] = entry
3599class locked_index:
3600 """Lock the index while making modifications.
3602 Works as a context manager.
3603 """
3605 _file: "_GitFile"
3607 def __init__(self, path: bytes | str) -> None:
3608 """Initialize locked_index."""
3609 self._path = path
3611 def __enter__(self) -> Index:
3612 """Enter context manager and lock index."""
3613 f = GitFile(self._path, "wb")
3614 self._file = f
3615 self._index = Index(self._path)
3616 return self._index
3618 def __exit__(
3619 self,
3620 exc_type: type | None,
3621 exc_value: BaseException | None,
3622 traceback: types.TracebackType | None,
3623 ) -> None:
3624 """Exit context manager and unlock index."""
3625 if exc_type is not None:
3626 self._file.abort()
3627 return
3628 try:
3629 f = SHA1Writer(self._file)
3630 write_index_dict(f, self._index._byname)
3631 except BaseException:
3632 self._file.abort()
3633 else:
3634 f.close()