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, SharedPerm
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 shared_perm: "SharedPerm | 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 shared_perm: Optional shared repository permission setting
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._shared_perm = shared_perm
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 f = GitFile(self._filename, "wb", shared_perm=self._shared_perm)
1197 try:
1198 # Filter out extensions with no meaningful data
1199 meaningful_extensions = []
1200 for ext in self._extensions:
1201 # Skip extensions that have empty data
1202 ext_data = ext.to_bytes()
1203 if ext_data:
1204 meaningful_extensions.append(ext)
1206 if self._skip_hash:
1207 # When skipHash is enabled, write the index without computing SHA1
1208 write_index_dict(
1209 f,
1210 self._byname,
1211 version=self._version,
1212 extensions=meaningful_extensions,
1213 )
1214 # Write 20 zero bytes instead of SHA1
1215 f.write(b"\x00" * 20)
1216 f.close()
1217 else:
1218 sha1_writer = SHA1Writer(f)
1219 write_index_dict(
1220 sha1_writer,
1221 self._byname,
1222 version=self._version,
1223 extensions=meaningful_extensions,
1224 )
1225 sha1_writer.close()
1226 except:
1227 f.close()
1228 raise
1230 def read(self) -> None:
1231 """Read current contents of index from disk."""
1232 if not os.path.exists(self._filename):
1233 return
1234 f = GitFile(self._filename, "rb")
1235 try:
1236 sha1_reader = SHA1Reader(f)
1237 entries, version, extensions = read_index_dict_with_version(sha1_reader)
1238 self._version = version
1239 self._extensions = extensions
1240 self.update(entries)
1241 # Extensions have already been read by read_index_dict_with_version
1242 sha1_reader.check_sha(allow_empty=True)
1243 finally:
1244 f.close()
1246 def __len__(self) -> int:
1247 """Number of entries in this index file."""
1248 return len(self._byname)
1250 def __getitem__(self, key: bytes) -> IndexEntry | ConflictedIndexEntry:
1251 """Retrieve entry by relative path and stage.
1253 Returns: Either a IndexEntry or a ConflictedIndexEntry
1254 Raises KeyError: if the entry does not exist
1255 """
1256 return self._byname[self.canonical_path(key)]
1258 def __iter__(self) -> Iterator[bytes]:
1259 """Iterate over the paths and stages in this index."""
1260 return iter(self._byname)
1262 def __contains__(self, key: bytes) -> bool:
1263 """Check if a path exists in the index."""
1264 return self.canonical_path(key) in self._byname
1266 def get_sha1(self, path: bytes) -> ObjectID:
1267 """Return the (git object) SHA1 for the object at a path."""
1268 value = self[path]
1269 if isinstance(value, ConflictedIndexEntry):
1270 raise UnmergedEntries
1271 return value.sha
1273 def get_mode(self, path: bytes) -> int:
1274 """Return the POSIX file mode for the object at a path."""
1275 value = self[path]
1276 if isinstance(value, ConflictedIndexEntry):
1277 raise UnmergedEntries
1278 return value.mode
1280 def iterobjects(self) -> Iterable[tuple[bytes, ObjectID, int]]:
1281 """Iterate over path, sha, mode tuples for use with commit_tree."""
1282 for path in self:
1283 entry = self[path]
1284 if isinstance(entry, ConflictedIndexEntry):
1285 raise UnmergedEntries
1286 yield path, entry.sha, cleanup_mode(entry.mode)
1288 def has_conflicts(self) -> bool:
1289 """Check if the index contains any conflicted entries.
1291 Returns:
1292 True if any entries are conflicted, False otherwise
1293 """
1294 for value in self._byname.values():
1295 if isinstance(value, ConflictedIndexEntry):
1296 return True
1297 return False
1299 def clear(self) -> None:
1300 """Remove all contents from this index."""
1301 self._byname = {}
1302 if self._normalized is not None:
1303 self._normalized = {}
1305 def __setitem__(
1306 self, name: bytes, value: IndexEntry | ConflictedIndexEntry
1307 ) -> None:
1308 """Set an entry in the index."""
1309 assert isinstance(name, bytes)
1310 name = self.canonical_path(name)
1311 is_new = name not in self._byname
1312 self._byname[name] = value
1313 if is_new and self._normalized is not None:
1314 assert self._path_normalizer is not None
1315 self._normalized.setdefault(self._path_normalizer(name), name)
1317 def __delitem__(self, name: bytes) -> None:
1318 """Delete an entry from the index."""
1319 name = self.canonical_path(name)
1320 del self._byname[name]
1321 if self._normalized is not None:
1322 assert self._path_normalizer is not None
1323 normalized_key = self._path_normalizer(name)
1324 if self._normalized.get(normalized_key) == name:
1325 del self._normalized[normalized_key]
1327 def iteritems(
1328 self,
1329 ) -> Iterator[tuple[bytes, IndexEntry | ConflictedIndexEntry]]:
1330 """Iterate over (path, entry) pairs in the index.
1332 Returns:
1333 Iterator of (path, entry) tuples
1334 """
1335 return iter(self._byname.items())
1337 def items(self) -> Iterator[tuple[bytes, IndexEntry | ConflictedIndexEntry]]:
1338 """Get an iterator over (path, entry) pairs.
1340 Returns:
1341 Iterator of (path, entry) tuples
1342 """
1343 return iter(self._byname.items())
1345 def update(self, entries: dict[bytes, IndexEntry | ConflictedIndexEntry]) -> None:
1346 """Update the index with multiple entries.
1348 Args:
1349 entries: Dictionary mapping paths to index entries
1350 """
1351 for key, value in entries.items():
1352 self[key] = value
1354 def paths(self) -> Generator[bytes, None, None]:
1355 """Generate all paths in the index.
1357 Yields:
1358 Path names as bytes
1359 """
1360 yield from self._byname.keys()
1362 def changes_from_tree(
1363 self,
1364 object_store: ObjectContainer,
1365 tree: ObjectID | None,
1366 want_unchanged: bool = False,
1367 ) -> Generator[
1368 tuple[
1369 tuple[bytes | None, bytes | None],
1370 tuple[int | None, int | None],
1371 tuple[bytes | None, bytes | None],
1372 ],
1373 None,
1374 None,
1375 ]:
1376 """Find the differences between the contents of this index and a tree.
1378 Args:
1379 object_store: Object store to use for retrieving tree contents
1380 tree: SHA1 of the root tree
1381 want_unchanged: Whether unchanged files should be reported
1382 Returns: Iterator over tuples with (oldpath, newpath), (oldmode,
1383 newmode), (oldsha, newsha)
1384 """
1386 def lookup_entry(path: bytes) -> tuple[bytes, int]:
1387 entry = self[path]
1388 if isinstance(entry, IndexEntry):
1389 return entry.sha, cleanup_mode(entry.mode)
1390 else:
1391 # Handle ConflictedIndexEntry case
1392 return b"", 0
1394 yield from changes_from_tree(
1395 self.paths(),
1396 lookup_entry,
1397 object_store,
1398 tree,
1399 want_unchanged=want_unchanged,
1400 )
1402 def commit(self, object_store: ObjectContainer) -> ObjectID:
1403 """Create a new tree from an index.
1405 Args:
1406 object_store: Object store to save the tree in
1407 Returns:
1408 Root tree SHA
1409 """
1410 return commit_tree(object_store, self.iterobjects())
1412 def is_sparse(self) -> bool:
1413 """Check if this index contains sparse directory entries.
1415 Returns:
1416 True if any sparse directory extension is present
1417 """
1418 return any(isinstance(ext, SparseDirExtension) for ext in self._extensions)
1420 def ensure_full_index(self, object_store: "BaseObjectStore") -> None:
1421 """Expand all sparse directory entries into full file entries.
1423 This converts a sparse index into a full index by recursively
1424 expanding any sparse directory entries into their constituent files.
1426 Args:
1427 object_store: Object store to read tree objects from
1429 Raises:
1430 KeyError: If a tree object referenced by a sparse dir entry doesn't exist
1431 """
1432 if not self.is_sparse():
1433 return
1435 # Find all sparse directory entries
1436 sparse_dirs = []
1437 for path, entry in list(self._byname.items()):
1438 if isinstance(entry, IndexEntry) and entry.is_sparse_dir(path):
1439 sparse_dirs.append((path, entry))
1441 # Expand each sparse directory
1442 for path, entry in sparse_dirs:
1443 # Remove the sparse directory entry
1444 del self[path]
1446 # Get the tree object
1447 tree = object_store[entry.sha]
1448 if not isinstance(tree, Tree):
1449 raise ValueError(f"Sparse directory {path!r} points to non-tree object")
1451 # Recursively add all entries from the tree
1452 self._expand_tree(path.rstrip(b"/"), tree, object_store, entry)
1454 # Remove the sparse directory extension
1455 self._extensions = [
1456 ext for ext in self._extensions if not isinstance(ext, SparseDirExtension)
1457 ]
1459 def _expand_tree(
1460 self,
1461 prefix: bytes,
1462 tree: Tree,
1463 object_store: "BaseObjectStore",
1464 template_entry: IndexEntry,
1465 ) -> None:
1466 """Recursively expand a tree into index entries.
1468 Args:
1469 prefix: Path prefix for entries (without trailing slash)
1470 tree: Tree object to expand
1471 object_store: Object store to read nested trees from
1472 template_entry: Template entry to copy metadata from
1473 """
1474 for name, mode, sha in tree.items():
1475 if prefix:
1476 full_path = prefix + b"/" + name
1477 else:
1478 full_path = name
1480 if stat.S_ISDIR(mode):
1481 # Recursively expand subdirectories
1482 subtree = object_store[sha]
1483 if not isinstance(subtree, Tree):
1484 raise ValueError(
1485 f"Directory entry {full_path!r} points to non-tree object"
1486 )
1487 self._expand_tree(full_path, subtree, object_store, template_entry)
1488 else:
1489 # Create an index entry for this file
1490 # Use the template entry for metadata but with the file's sha and mode
1491 new_entry = IndexEntry(
1492 ctime=template_entry.ctime,
1493 mtime=template_entry.mtime,
1494 dev=template_entry.dev,
1495 ino=template_entry.ino,
1496 mode=mode,
1497 uid=template_entry.uid,
1498 gid=template_entry.gid,
1499 size=0, # Size is unknown from tree
1500 sha=sha,
1501 flags=0,
1502 extended_flags=0, # Don't copy skip-worktree flag
1503 )
1504 self[full_path] = new_entry
1506 def convert_to_sparse(
1507 self,
1508 object_store: "BaseObjectStore",
1509 tree_sha: ObjectID,
1510 sparse_dirs: Set[bytes],
1511 ) -> None:
1512 """Convert full index entries to sparse directory entries.
1514 This collapses directories that are entirely outside the sparse
1515 checkout cone into single sparse directory entries.
1517 Args:
1518 object_store: Object store to read tree objects
1519 tree_sha: SHA of the tree (usually HEAD) to base sparse dirs on
1520 sparse_dirs: Set of directory paths (with trailing /) to collapse
1522 Raises:
1523 KeyError: If tree_sha or a subdirectory doesn't exist
1524 """
1525 if not sparse_dirs:
1526 return
1528 # Get the base tree
1529 tree = object_store[tree_sha]
1530 if not isinstance(tree, Tree):
1531 raise ValueError(f"tree_sha {tree_sha!r} is not a tree object")
1533 # For each sparse directory, find its tree SHA and create sparse entry
1534 for dir_path in sparse_dirs:
1535 dir_path_stripped = dir_path.rstrip(b"/")
1537 # Find the tree SHA for this directory
1538 subtree_sha = self._find_subtree_sha(tree, dir_path_stripped, object_store)
1539 if subtree_sha is None:
1540 # Directory doesn't exist in tree, skip it
1541 continue
1543 # Remove all entries under this directory
1544 entries_to_remove = [
1545 path
1546 for path in self._byname
1547 if path.startswith(dir_path) or path == dir_path_stripped
1548 ]
1549 for path in entries_to_remove:
1550 del self[path]
1552 # Create a sparse directory entry
1553 # Use minimal metadata since it's not a real file
1554 from dulwich.objects import ObjectID
1556 sparse_entry = IndexEntry(
1557 ctime=0,
1558 mtime=0,
1559 dev=0,
1560 ino=0,
1561 mode=stat.S_IFDIR,
1562 uid=0,
1563 gid=0,
1564 size=0,
1565 sha=ObjectID(subtree_sha),
1566 flags=0,
1567 extended_flags=EXTENDED_FLAG_SKIP_WORKTREE,
1568 )
1569 self[dir_path] = sparse_entry
1571 # Add sparse directory extension if not present
1572 if not self.is_sparse():
1573 self._extensions.append(SparseDirExtension())
1575 def _find_subtree_sha(
1576 self,
1577 tree: Tree,
1578 path: bytes,
1579 object_store: "BaseObjectStore",
1580 ) -> bytes | None:
1581 """Find the SHA of a subtree at a given path.
1583 Args:
1584 tree: Root tree object to search in
1585 path: Path to the subdirectory (no trailing slash)
1586 object_store: Object store to read nested trees from
1588 Returns:
1589 SHA of the subtree, or None if path doesn't exist
1590 """
1591 if not path:
1592 return tree.id
1594 parts = path.split(b"/")
1595 current_tree = tree
1597 for part in parts:
1598 # Look for this part in the current tree
1599 try:
1600 mode, sha = current_tree[part]
1601 except KeyError:
1602 return None
1604 if not stat.S_ISDIR(mode):
1605 # Path component is a file, not a directory
1606 return None
1608 # Load the next tree
1609 obj = object_store[sha]
1610 if not isinstance(obj, Tree):
1611 return None
1612 current_tree = obj
1614 return current_tree.id
1617def commit_tree(
1618 object_store: ObjectContainer, blobs: Iterable[tuple[bytes, ObjectID, int]]
1619) -> ObjectID:
1620 """Commit a new tree.
1622 Args:
1623 object_store: Object store to add trees to
1624 blobs: Iterable over blob path, sha, mode entries
1625 Returns:
1626 SHA1 of the created tree.
1627 """
1628 trees: dict[bytes, TreeDict] = {b"": {}}
1630 def add_tree(path: bytes) -> TreeDict:
1631 if path in trees:
1632 return trees[path]
1633 dirname, basename = pathsplit(path)
1634 t = add_tree(dirname)
1635 assert isinstance(basename, bytes)
1636 newtree: TreeDict = {}
1637 t[basename] = newtree
1638 trees[path] = newtree
1639 return newtree
1641 for path, sha, mode in blobs:
1642 tree_path, basename = pathsplit(path)
1643 tree = add_tree(tree_path)
1644 tree[basename] = (mode, sha)
1646 def build_tree(path: bytes) -> ObjectID:
1647 tree = Tree()
1648 for basename, entry in trees[path].items():
1649 if isinstance(entry, dict):
1650 mode = stat.S_IFDIR
1651 sha = build_tree(pathjoin(path, basename))
1652 else:
1653 (mode, sha) = entry
1654 tree.add(basename, mode, sha)
1655 object_store.add_object(tree)
1656 return tree.id
1658 return build_tree(b"")
1661def commit_index(object_store: ObjectContainer, index: Index) -> ObjectID:
1662 """Create a new tree from an index.
1664 Args:
1665 object_store: Object store to save the tree in
1666 index: Index file
1667 Note: This function is deprecated, use index.commit() instead.
1668 Returns: Root tree sha.
1669 """
1670 return commit_tree(object_store, index.iterobjects())
1673def changes_from_tree(
1674 names: Iterable[bytes],
1675 lookup_entry: Callable[[bytes], tuple[bytes, int]],
1676 object_store: ObjectContainer,
1677 tree: ObjectID | None,
1678 want_unchanged: bool = False,
1679) -> Iterable[
1680 tuple[
1681 tuple[bytes | None, bytes | None],
1682 tuple[int | None, int | None],
1683 tuple[bytes | None, bytes | None],
1684 ]
1685]:
1686 """Find the differences between the contents of a tree and a working copy.
1688 Args:
1689 names: Iterable of names in the working copy
1690 lookup_entry: Function to lookup an entry in the working copy
1691 object_store: Object store to use for retrieving tree contents
1692 tree: SHA1 of the root tree, or None for an empty tree
1693 want_unchanged: Whether unchanged files should be reported
1694 Returns: Iterator over tuples with (oldpath, newpath), (oldmode, newmode),
1695 (oldsha, newsha)
1696 """
1697 # TODO(jelmer): Support a include_trees option
1698 other_names = set(names)
1700 if tree is not None:
1701 for name, mode, sha in iter_tree_contents(object_store, tree):
1702 assert name is not None and mode is not None and sha is not None
1703 try:
1704 (other_sha, other_mode) = lookup_entry(name)
1705 except KeyError:
1706 # Was removed
1707 yield ((name, None), (mode, None), (sha, None))
1708 else:
1709 other_names.remove(name)
1710 if want_unchanged or other_sha != sha or other_mode != mode:
1711 yield ((name, name), (mode, other_mode), (sha, other_sha))
1713 # Mention added files
1714 for name in other_names:
1715 try:
1716 (other_sha, other_mode) = lookup_entry(name)
1717 except KeyError:
1718 pass
1719 else:
1720 yield ((None, name), (None, other_mode), (None, other_sha))
1723def index_entry_from_stat(
1724 stat_val: os.stat_result,
1725 hex_sha: bytes,
1726 mode: int | None = None,
1727) -> IndexEntry:
1728 """Create a new index entry from a stat value.
1730 Args:
1731 stat_val: POSIX stat_result instance
1732 hex_sha: Hex sha of the object
1733 mode: Optional file mode, will be derived from stat if not provided
1734 """
1735 if mode is None:
1736 mode = cleanup_mode(stat_val.st_mode)
1738 from dulwich.objects import ObjectID
1740 # Use nanosecond precision when available to avoid precision loss
1741 # through float representation
1742 ctime: int | float | tuple[int, int]
1743 mtime: int | float | tuple[int, int]
1744 st_ctime_ns = getattr(stat_val, "st_ctime_ns", None)
1745 if st_ctime_ns is not None:
1746 ctime = (
1747 st_ctime_ns // 1_000_000_000,
1748 st_ctime_ns % 1_000_000_000,
1749 )
1750 else:
1751 ctime = stat_val.st_ctime
1753 st_mtime_ns = getattr(stat_val, "st_mtime_ns", None)
1754 if st_mtime_ns is not None:
1755 mtime = (
1756 st_mtime_ns // 1_000_000_000,
1757 st_mtime_ns % 1_000_000_000,
1758 )
1759 else:
1760 mtime = stat_val.st_mtime
1762 return IndexEntry(
1763 ctime=ctime,
1764 mtime=mtime,
1765 dev=stat_val.st_dev,
1766 ino=stat_val.st_ino,
1767 mode=mode,
1768 uid=stat_val.st_uid,
1769 gid=stat_val.st_gid,
1770 size=stat_val.st_size,
1771 sha=ObjectID(hex_sha),
1772 flags=0,
1773 extended_flags=0,
1774 )
1777def index_entry_from_tree_entry(
1778 mode: int,
1779 hex_sha: bytes,
1780 size: int = 0,
1781) -> IndexEntry:
1782 """Create an index entry from a tree entry, with zeroed stat fields.
1784 Use this when populating an index directly from a tree without touching
1785 the filesystem, matching ``git read-tree``. The stat-derived fields
1786 (ctime/mtime/dev/ino/uid/gid) are zeroed; git treats such entries as
1787 stat-unmerged and refreshes them on the next stat.
1789 Args:
1790 mode: File mode from the tree entry
1791 hex_sha: Hex sha of the object
1792 size: Size of the object (0 for gitlinks or when unknown)
1793 """
1794 from dulwich.objects import ObjectID
1796 return IndexEntry(
1797 ctime=0,
1798 mtime=0,
1799 dev=0,
1800 ino=0,
1801 mode=mode,
1802 uid=0,
1803 gid=0,
1804 size=size,
1805 sha=ObjectID(hex_sha),
1806 flags=0,
1807 extended_flags=0,
1808 )
1811if sys.platform == "win32":
1812 # On Windows, creating symlinks either requires administrator privileges
1813 # or developer mode. Raise a more helpful error when we're unable to
1814 # create symlinks
1816 # https://github.com/jelmer/dulwich/issues/1005
1818 class WindowsSymlinkPermissionError(PermissionError):
1819 """Windows-specific error for symlink creation failures.
1821 This error is raised when symlink creation fails on Windows,
1822 typically due to lack of developer mode or administrator privileges.
1823 """
1825 def __init__(self, errno: int, msg: str, filename: str | None) -> None:
1826 """Initialize WindowsSymlinkPermissionError."""
1827 super().__init__(
1828 errno,
1829 f"Unable to create symlink; do you have developer mode enabled? {msg}",
1830 filename,
1831 )
1833 def symlink(
1834 src: str | bytes,
1835 dst: str | bytes,
1836 target_is_directory: bool = False,
1837 *,
1838 dir_fd: int | None = None,
1839 ) -> None:
1840 """Create a symbolic link on Windows with better error handling.
1842 Args:
1843 src: Source path for the symlink
1844 dst: Destination path where symlink will be created
1845 target_is_directory: Whether the target is a directory
1846 dir_fd: Optional directory file descriptor
1848 Raises:
1849 WindowsSymlinkPermissionError: If symlink creation fails due to permissions
1850 """
1851 try:
1852 return os.symlink(
1853 src, dst, target_is_directory=target_is_directory, dir_fd=dir_fd
1854 )
1855 except PermissionError as e:
1856 raise WindowsSymlinkPermissionError(
1857 e.errno or 0, e.strerror or "", e.filename
1858 ) from e
1859else:
1860 symlink = os.symlink
1863def build_file_from_blob(
1864 blob: Blob,
1865 mode: int,
1866 target_path: bytes,
1867 *,
1868 honor_filemode: bool = True,
1869 tree_encoding: str = "utf-8",
1870 symlink_fn: Callable[
1871 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
1872 ]
1873 | None = None,
1874) -> os.stat_result:
1875 """Build a file or symlink on disk based on a Git object.
1877 Args:
1878 blob: The git object
1879 mode: File mode
1880 target_path: Path to write to
1881 honor_filemode: An optional flag to honor core.filemode setting in
1882 config file, default is core.filemode=True, change executable bit
1883 tree_encoding: Encoding to use for tree contents
1884 symlink_fn: Function to use for creating symlinks
1885 Returns: stat object for the file
1886 """
1887 try:
1888 oldstat = os.lstat(target_path)
1889 except FileNotFoundError:
1890 oldstat = None
1891 contents = blob.as_raw_string()
1892 if stat.S_ISLNK(mode):
1893 if oldstat:
1894 _remove_file_with_readonly_handling(target_path)
1895 if sys.platform == "win32":
1896 # os.readlink on Python3 on Windows requires a unicode string.
1897 contents_str = contents.decode(tree_encoding)
1898 target_path_str = target_path.decode(tree_encoding)
1899 (symlink_fn or symlink)(contents_str, target_path_str)
1900 else:
1901 (symlink_fn or symlink)(contents, target_path)
1902 else:
1903 if oldstat is not None and stat.S_ISLNK(oldstat.st_mode):
1904 # A symlink left at the target path must not be written through.
1905 # open(..., "wb") would follow it and clobber whatever it points
1906 # at, possibly outside the work tree. Replace it with a fresh
1907 # regular file, like git does on checkout.
1908 _remove_file_with_readonly_handling(target_path)
1909 oldstat = None
1910 if oldstat is not None and oldstat.st_size == len(contents):
1911 with open(target_path, "rb") as f:
1912 if f.read() == contents:
1913 return oldstat
1915 with open(target_path, "wb") as f:
1916 # Write out file
1917 f.write(contents)
1919 if honor_filemode:
1920 # Canonicalize to the permission bits git honors (0o644/0o755).
1921 # The mode comes from a tree/index entry, which for an untrusted
1922 # repository can carry setuid/setgid/sticky or world-writable bits;
1923 # git reduces every regular-file mode to 0o644 or 0o755 on checkout.
1924 os.chmod(target_path, cleanup_mode(mode))
1926 return os.lstat(target_path)
1929INVALID_DOTNAMES = (b".git", b".", b"..", b"")
1932def _normalize_path_element_default(element: bytes) -> bytes:
1933 """Normalize path element for default case-insensitive comparison."""
1934 return element.lower()
1937def _normalize_path_element_ntfs(element: bytes) -> bytes:
1938 """Normalize path element for NTFS filesystem."""
1939 return element.rstrip(b". ").lower()
1942def _normalize_path_element_hfs(element: bytes) -> bytes:
1943 """Normalize path element for HFS+ filesystem."""
1944 import unicodedata
1946 # Decode to Unicode (let UnicodeDecodeError bubble up)
1947 element_str = element.decode("utf-8", errors="strict")
1949 # Remove HFS+ ignorable characters
1950 filtered = "".join(c for c in element_str if ord(c) not in HFS_IGNORABLE_CHARS)
1951 # Normalize to NFD
1952 normalized = unicodedata.normalize("NFD", filtered)
1953 return normalized.lower().encode("utf-8", errors="strict")
1956def get_path_element_normalizer(config: "Config") -> Callable[[bytes], bytes]:
1957 """Get the appropriate path element normalization function based on config.
1959 Args:
1960 config: Repository configuration object
1962 Returns:
1963 Function that normalizes path elements for the configured filesystem
1964 """
1965 import os
1966 import sys
1968 if config.get_boolean(b"core", b"protectNTFS", os.name == "nt"):
1969 return _normalize_path_element_ntfs
1970 elif config.get_boolean(b"core", b"protectHFS", sys.platform == "darwin"):
1971 return _normalize_path_element_hfs
1972 else:
1973 return _normalize_path_element_default
1976def make_path_normalizer(
1977 config: "Config",
1978) -> Callable[[bytes], bytes] | None:
1979 """Build a path normalizer honoring ``core.ignorecase`` and ``core.precomposeunicode``.
1981 The returned callable maps a filesystem-form path to a canonical form
1982 used to match equivalent paths (e.g. ``Foo.txt`` ↔ ``foo.txt`` when
1983 ``core.ignorecase=true``, NFD ↔ NFC when ``core.precomposeunicode=true``).
1984 Returns ``None`` when neither option is active so callers can skip the
1985 comparison entirely.
1986 """
1987 ignorecase = config.get_boolean(b"core", b"ignorecase", False)
1988 precompose = config.get_boolean(b"core", b"precomposeunicode", False)
1989 if not ignorecase and not precompose:
1990 return None
1992 def normalize(path: bytes) -> bytes:
1993 if precompose:
1994 import unicodedata
1996 try:
1997 path = unicodedata.normalize("NFC", path.decode("utf-8")).encode(
1998 "utf-8"
1999 )
2000 except UnicodeDecodeError:
2001 pass
2002 if ignorecase:
2003 path = path.lower()
2004 return path
2006 return normalize
2009class InvalidPathError(Exception):
2010 """Raised when a tree entry's path is unsafe to write to the work tree."""
2012 def __init__(self, path: bytes) -> None:
2013 """Initialize the exception with the offending path."""
2014 self.path = path
2015 super().__init__(f"invalid path {path.decode('utf-8', 'replace')!r}")
2018def validate_path_element_default(element: bytes) -> bool:
2019 """Validate a path element using default rules.
2021 Args:
2022 element: Path element to validate
2024 Returns:
2025 True if path element is valid, False otherwise
2026 """
2027 return _normalize_path_element_default(element) not in INVALID_DOTNAMES
2030def _is_ntfs_dotgit(name: bytes) -> bool:
2031 """Match NTFS-dangerous spellings of ``.git`` at the start of ``name``.
2033 Matches ``.git`` or the 8.3 short name ``git~1`` followed only by
2034 dots/spaces and then the end of the element, a separator, or a ``:``
2035 (an alternate-data-stream marker, as in ``.git::$INDEX_ALLOCATION``).
2036 """
2037 if name[:1] == b".":
2038 if name[1:4].lower() != b"git":
2039 return False
2040 i = 4
2041 elif name[:1].lower() == b"g": # ``git~1`` 8.3 short name
2042 if name[1:3].lower() != b"it" or name[3:5] != b"~1":
2043 return False
2044 i = 5
2045 else:
2046 return False
2048 while i < len(name):
2049 c = name[i : i + 1]
2050 if c == b":":
2051 return True
2052 if c != b"." and c != b" ":
2053 return False
2054 i += 1
2055 return True
2058# Reserved Windows device names. Opening any of these on Windows
2059# resolves to a device rather than a file, regardless of any
2060# extension or trailing dots/spaces (``NUL``, ``NUL.txt``,
2061# ``aux.foo.bar`` all hit the device). They are ordinary filenames
2062# everywhere else, so this is only enforced when running on Windows.
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 # Like the backslash check above, a reserved device name is only a
2110 # hazard on Windows; C git confines this to is_valid_win32_path in
2111 # compat/mingw.c, and accepts e.g. ``aux`` on POSIX (issue #2351).
2112 if os.name == "nt" and _is_reserved_windows_device_name(normalized):
2113 return False
2114 return True
2117# HFS+ ignorable Unicode codepoints (from Git's utf8.c)
2118HFS_IGNORABLE_CHARS = {
2119 0x200C, # ZERO WIDTH NON-JOINER
2120 0x200D, # ZERO WIDTH JOINER
2121 0x200E, # LEFT-TO-RIGHT MARK
2122 0x200F, # RIGHT-TO-LEFT MARK
2123 0x202A, # LEFT-TO-RIGHT EMBEDDING
2124 0x202B, # RIGHT-TO-LEFT EMBEDDING
2125 0x202C, # POP DIRECTIONAL FORMATTING
2126 0x202D, # LEFT-TO-RIGHT OVERRIDE
2127 0x202E, # RIGHT-TO-LEFT OVERRIDE
2128 0x206A, # INHIBIT SYMMETRIC SWAPPING
2129 0x206B, # ACTIVATE SYMMETRIC SWAPPING
2130 0x206C, # INHIBIT ARABIC FORM SHAPING
2131 0x206D, # ACTIVATE ARABIC FORM SHAPING
2132 0x206E, # NATIONAL DIGIT SHAPES
2133 0x206F, # NOMINAL DIGIT SHAPES
2134 0xFEFF, # ZERO WIDTH NO-BREAK SPACE
2135}
2138def validate_path_element_hfs(element: bytes) -> bool:
2139 """Validate path element for HFS+ filesystem.
2141 Equivalent to Git's is_hfs_dotgit and related checks.
2142 Uses NFD normalization and ignores HFS+ ignorable characters.
2143 """
2144 try:
2145 normalized = _normalize_path_element_hfs(element)
2146 except UnicodeDecodeError:
2147 # Malformed UTF-8 - be conservative and reject
2148 return False
2150 # Check against invalid names
2151 if normalized in INVALID_DOTNAMES:
2152 return False
2154 # Also check for 8.3 short name
2155 if normalized == b"git~1":
2156 return False
2158 return True
2161def get_path_element_validator(config: "Config") -> Callable[[bytes], bool]:
2162 """Get the path-element validator to use when checking out a tree.
2164 ``core.protectNTFS`` defaults to true on every platform (matching Git's
2165 ``PROTECT_NTFS_DEFAULT=1``) because a repository authored on POSIX can
2166 still be cloned on Windows later; ``core.protectHFS`` defaults to true on
2167 macOS. Both protections are independent and apply together, so on macOS
2168 (where both default on) a path element must satisfy the NTFS and HFS+
2169 checks. With both disabled this falls back to the default validator, which
2170 only refuses ``.git``, ``.`` and ``..``.
2172 Args:
2173 config: Repository configuration object
2175 Returns:
2176 Function that validates a single path element for the configured
2177 filesystem protections.
2178 """
2179 validators: list[Callable[[bytes], bool]] = []
2180 if config.get_boolean(b"core", b"protectNTFS", True):
2181 validators.append(validate_path_element_ntfs)
2182 if config.get_boolean(b"core", b"protectHFS", sys.platform == "darwin"):
2183 validators.append(validate_path_element_hfs)
2184 if not validators:
2185 return validate_path_element_default
2186 if len(validators) == 1:
2187 return validators[0]
2189 def validate_all(element: bytes) -> bool:
2190 return all(validator(element) for validator in validators)
2192 return validate_all
2195def validate_path(
2196 path: bytes,
2197 element_validator: Callable[[bytes], bool] = validate_path_element_default,
2198) -> bool:
2199 """Default path validator that just checks for .git/."""
2200 # A leading drive-letter prefix lets os.path.join discard the work-tree
2201 # root on Windows. Matches C git's verify_path, whose has_dos_drive_prefix
2202 # is a no-op stub on non-Windows.
2203 if os.name == "nt" and _has_dos_drive_prefix(path):
2204 return False
2205 parts = path.split(b"/")
2206 for p in parts:
2207 if not element_validator(p):
2208 return False
2209 else:
2210 return True
2213def verify_leading_dirs(
2214 tree_path: bytes,
2215 safe_prefix: list[bytes],
2216 repo_path: bytes,
2217) -> None:
2218 """Reject writes whose leading path resolves through a symlink.
2220 Callers that materialize many paths in sorted order can pass a shared
2221 ``safe_prefix`` list to cache the deepest chain of directory components
2222 already verified to be real directories (or absent); each call only
2223 ``lstat``s the components that differ from that chain. Callers that only
2224 verify a single path can pass an empty list. Mirrors git's
2225 ``lstat_cache_matchlen`` (see CVE-2021-21300).
2227 Args:
2228 tree_path: Tree-form path (``/``-separated) about to be written.
2229 safe_prefix: Mutable cache of directory components already verified
2230 under ``repo_path``. Updated in place.
2231 repo_path: Filesystem path to the work-tree root.
2233 Raises:
2234 InvalidPathError: If any leading component is a symlink.
2235 """
2236 slash = tree_path.rfind(b"/")
2237 if slash <= 0:
2238 return
2239 components = tree_path[:slash].split(b"/")
2241 common = 0
2242 while (
2243 common < len(safe_prefix)
2244 and common < len(components)
2245 and safe_prefix[common] == components[common]
2246 ):
2247 common += 1
2248 del safe_prefix[common:]
2250 current = repo_path
2251 for part in components[:common]:
2252 current = os.path.join(current, part)
2253 for part in components[common:]:
2254 current = os.path.join(current, part)
2255 try:
2256 st = os.lstat(current)
2257 except FileNotFoundError:
2258 # Anything below here doesn't exist yet; makedirs will create
2259 # it under a verified-real-directory prefix.
2260 break
2261 if stat.S_ISLNK(st.st_mode):
2262 raise InvalidPathError(tree_path)
2263 safe_prefix.append(part)
2266def build_index_from_tree(
2267 root_path: str | bytes,
2268 index_path: str | bytes,
2269 object_store: ObjectContainer,
2270 tree_id: ObjectID,
2271 honor_filemode: bool = True,
2272 validate_path_element: Callable[[bytes], bool] = validate_path_element_default,
2273 symlink_fn: Callable[
2274 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2275 ]
2276 | None = None,
2277 blob_normalizer: "FilterBlobNormalizer | None" = None,
2278 tree_encoding: str = "utf-8",
2279) -> None:
2280 """Generate and materialize index from a tree.
2282 Args:
2283 tree_id: Tree to materialize
2284 root_path: Target dir for materialized index files
2285 index_path: Target path for generated index
2286 object_store: Non-empty object store holding tree contents
2287 honor_filemode: An optional flag to honor core.filemode setting in
2288 config file, default is core.filemode=True, change executable bit
2289 validate_path_element: Function to validate path elements to check
2290 out; default just refuses .git and .. directories.
2291 symlink_fn: Function to use for creating symlinks
2292 blob_normalizer: An optional BlobNormalizer to use for converting line
2293 endings when writing blobs to the working directory.
2294 tree_encoding: Encoding used for tree paths (default: utf-8)
2296 Note: existing index is wiped and contents are not merged
2297 in a working dir. Suitable only for fresh clones.
2298 """
2299 index = Index(index_path, read=False)
2300 if not isinstance(root_path, bytes):
2301 root_path = os.fsencode(root_path)
2303 # Cache of leading directory components already verified to be real
2304 # directories, shared across the sorted iteration. See verify_leading_dirs.
2305 safe_prefix: list[bytes] = []
2307 for entry in iter_tree_contents(object_store, tree_id):
2308 assert (
2309 entry.path is not None and entry.mode is not None and entry.sha is not None
2310 )
2311 # Validate as we go and abort on the first invalid path,
2312 # leaving any files already written in place.
2313 if not validate_path(entry.path, validate_path_element):
2314 raise InvalidPathError(entry.path)
2315 # Refuse to write an entry whose leading path resolves through a
2316 # symlink materialized by an earlier entry; open(..., "wb") would
2317 # otherwise follow it and write outside the work tree.
2318 verify_leading_dirs(entry.path, safe_prefix, root_path)
2319 full_path = _tree_to_fs_path(root_path, entry.path, tree_encoding)
2321 if not os.path.exists(os.path.dirname(full_path)):
2322 os.makedirs(os.path.dirname(full_path))
2324 # TODO(jelmer): Merge new index into working tree
2325 if S_ISGITLINK(entry.mode):
2326 if not os.path.isdir(full_path):
2327 os.mkdir(full_path)
2328 st = os.lstat(full_path)
2329 # TODO(jelmer): record and return submodule paths
2330 else:
2331 obj = object_store[entry.sha]
2332 assert isinstance(obj, Blob)
2333 # Apply blob normalization for checkout if normalizer is provided
2334 if blob_normalizer is not None:
2335 obj = blob_normalizer.checkout_normalize(obj, entry.path)
2336 st = build_file_from_blob(
2337 obj,
2338 entry.mode,
2339 full_path,
2340 honor_filemode=honor_filemode,
2341 tree_encoding=tree_encoding,
2342 symlink_fn=symlink_fn,
2343 )
2345 # Add file to index
2346 if not honor_filemode or S_ISGITLINK(entry.mode):
2347 # we can not use tuple slicing to build a new tuple,
2348 # because on windows that will convert the times to
2349 # longs, which causes errors further along
2350 st_tuple = (
2351 entry.mode,
2352 st.st_ino,
2353 st.st_dev,
2354 st.st_nlink,
2355 st.st_uid,
2356 st.st_gid,
2357 st.st_size,
2358 st.st_atime,
2359 st.st_mtime,
2360 st.st_ctime,
2361 )
2362 st = st.__class__(st_tuple)
2363 # default to a stage 0 index entry (normal)
2364 # when reading from the filesystem
2365 index[entry.path] = index_entry_from_stat(st, entry.sha)
2367 index.write()
2370def blob_from_path_and_mode(
2371 fs_path: bytes, mode: int, tree_encoding: str = "utf-8"
2372) -> Blob:
2373 """Create a blob from a path and a stat object.
2375 Args:
2376 fs_path: Full file system path to file
2377 mode: File mode
2378 tree_encoding: Encoding to use for tree contents
2379 Returns: A `Blob` object
2380 """
2381 assert isinstance(fs_path, bytes)
2382 blob = Blob()
2383 if stat.S_ISLNK(mode):
2384 if sys.platform == "win32":
2385 # os.readlink on Python3 on Windows requires a unicode string.
2386 blob.data = os.readlink(os.fsdecode(fs_path)).encode(tree_encoding)
2387 else:
2388 blob.data = os.readlink(fs_path)
2389 else:
2390 with open(fs_path, "rb") as f:
2391 blob.data = f.read()
2392 return blob
2395def blob_from_path_and_stat(
2396 fs_path: bytes, st: os.stat_result, tree_encoding: str = "utf-8"
2397) -> Blob:
2398 """Create a blob from a path and a stat object.
2400 Args:
2401 fs_path: Full file system path to file
2402 st: A stat object
2403 tree_encoding: Encoding to use for tree contents
2404 Returns: A `Blob` object
2405 """
2406 return blob_from_path_and_mode(fs_path, st.st_mode, tree_encoding)
2409def read_submodule_head(path: str | bytes) -> bytes | None:
2410 """Read the head commit of a submodule.
2412 Args:
2413 path: path to the submodule
2414 Returns: HEAD sha, None if not a valid head/repository
2415 """
2416 from .errors import NotGitRepository
2417 from .repo import Repo
2419 # Repo currently expects a "str", so decode if necessary.
2420 # TODO(jelmer): Perhaps move this into Repo() ?
2421 if not isinstance(path, str):
2422 path = os.fsdecode(path)
2423 try:
2424 repo = Repo(path)
2425 except NotGitRepository:
2426 return None
2427 try:
2428 return repo.head()
2429 except KeyError:
2430 return None
2433def _has_directory_changed(tree_path: bytes, entry: IndexEntry) -> bool:
2434 """Check if a directory has changed after getting an error.
2436 When handling an error trying to create a blob from a path, call this
2437 function. It will check if the path is a directory. If it's a directory
2438 and a submodule, check the submodule head to see if it's has changed. If
2439 not, consider the file as changed as Git tracked a file and not a
2440 directory.
2442 Return true if the given path should be considered as changed and False
2443 otherwise or if the path is not a directory.
2444 """
2445 # This is actually a directory
2446 if os.path.exists(os.path.join(tree_path, b".git")):
2447 # Submodule
2448 head = read_submodule_head(tree_path)
2449 if entry.sha != head:
2450 return True
2451 else:
2452 # The file was changed to a directory, so consider it removed.
2453 return True
2455 return False
2458os_sep_bytes = os.sep.encode("ascii")
2461def _ensure_parent_dir_exists(full_path: bytes) -> None:
2462 """Ensure parent directory exists, checking no parent is a file."""
2463 parent_dir = os.path.dirname(full_path)
2464 if parent_dir and not os.path.exists(parent_dir):
2465 # Walk up the directory tree to find the first existing parent
2466 current = parent_dir
2467 parents_to_check: list[bytes] = []
2469 while current and not os.path.exists(current):
2470 parents_to_check.insert(0, current)
2471 new_parent = os.path.dirname(current)
2472 if new_parent == current:
2473 # Reached the root or can't go up further
2474 break
2475 current = new_parent
2477 # Check if the existing parent (if any) is a directory
2478 if current and os.path.exists(current) and not os.path.isdir(current):
2479 raise OSError(
2480 f"Cannot create directory, parent path is a file: {current!r}"
2481 )
2483 # Now check each parent we need to create isn't blocked by an existing file
2484 for parent_path in parents_to_check:
2485 if os.path.exists(parent_path) and not os.path.isdir(parent_path):
2486 raise OSError(
2487 f"Cannot create directory, parent path is a file: {parent_path!r}"
2488 )
2490 os.makedirs(parent_dir)
2493def _remove_file_with_readonly_handling(path: bytes) -> None:
2494 """Remove a file, handling read-only files on Windows.
2496 Args:
2497 path: Path to the file to remove
2498 """
2499 try:
2500 os.unlink(path)
2501 except PermissionError:
2502 # On Windows, remove read-only attribute and retry
2503 if sys.platform == "win32":
2504 os.chmod(path, stat.S_IWRITE | stat.S_IREAD)
2505 os.unlink(path)
2506 else:
2507 raise
2510def _remove_empty_parents(path: bytes, stop_at: bytes) -> None:
2511 """Remove empty parent directories up to stop_at."""
2512 parent = os.path.dirname(path)
2513 while parent and parent != stop_at:
2514 try:
2515 os.rmdir(parent)
2516 parent = os.path.dirname(parent)
2517 except FileNotFoundError:
2518 # Directory doesn't exist - stop trying
2519 break
2520 except OSError as e:
2521 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
2522 # Directory not empty - stop trying
2523 break
2524 raise
2527def _check_symlink_matches(
2528 full_path: bytes, repo_object_store: "BaseObjectStore", entry_sha: ObjectID
2529) -> bool:
2530 """Check if symlink target matches expected target.
2532 Returns True if symlink matches, False if it doesn't match.
2533 """
2534 try:
2535 current_target = os.readlink(full_path)
2536 blob_obj = repo_object_store[entry_sha]
2537 expected_target = blob_obj.as_raw_string()
2538 if isinstance(current_target, str):
2539 current_target = current_target.encode()
2540 return current_target == expected_target
2541 except FileNotFoundError:
2542 # Symlink doesn't exist
2543 return False
2544 except OSError as e:
2545 if e.errno == errno.EINVAL:
2546 # Not a symlink
2547 return False
2548 raise
2551def _check_file_matches(
2552 repo_object_store: "BaseObjectStore",
2553 full_path: bytes,
2554 entry_sha: ObjectID,
2555 entry_mode: int,
2556 current_stat: os.stat_result,
2557 honor_filemode: bool,
2558 blob_normalizer: "FilterBlobNormalizer | None" = None,
2559 tree_path: bytes | None = None,
2560) -> bool:
2561 """Check if a file on disk matches the expected git object.
2563 Returns True if file matches, False if it doesn't match.
2564 """
2565 # Check mode first (if honor_filemode is True)
2566 if honor_filemode:
2567 current_mode = stat.S_IMODE(current_stat.st_mode)
2568 expected_mode = stat.S_IMODE(entry_mode)
2570 # For regular files, only check the user executable bit, not group/other permissions
2571 # This matches Git's behavior where umask differences don't count as modifications
2572 if stat.S_ISREG(current_stat.st_mode):
2573 # Normalize regular file modes to ignore group/other write permissions
2574 current_mode_normalized = (
2575 current_mode & 0o755
2576 ) # Keep only user rwx and all read+execute
2577 expected_mode_normalized = expected_mode & 0o755
2579 # For Git compatibility, regular files should be either 644 or 755
2580 if expected_mode_normalized not in (0o644, 0o755):
2581 expected_mode_normalized = 0o644 # Default for regular files
2582 if current_mode_normalized not in (0o644, 0o755):
2583 # Determine if it should be executable based on user execute bit
2584 if current_mode & 0o100: # User execute bit is set
2585 current_mode_normalized = 0o755
2586 else:
2587 current_mode_normalized = 0o644
2589 if current_mode_normalized != expected_mode_normalized:
2590 return False
2591 else:
2592 # For non-regular files (symlinks, etc.), check mode exactly
2593 if current_mode != expected_mode:
2594 return False
2596 # If mode matches (or we don't care), check content via size first
2597 blob_obj = repo_object_store[entry_sha]
2598 if current_stat.st_size != blob_obj.raw_length():
2599 return False
2601 # Size matches, check actual content
2602 try:
2603 with open(full_path, "rb") as f:
2604 current_content = f.read()
2605 expected_content = blob_obj.as_raw_string()
2606 if blob_normalizer and tree_path is not None:
2607 assert isinstance(blob_obj, Blob)
2608 normalized_blob = blob_normalizer.checkout_normalize(
2609 blob_obj, tree_path
2610 )
2611 expected_content = normalized_blob.as_raw_string()
2612 return current_content == expected_content
2613 except (FileNotFoundError, PermissionError, IsADirectoryError):
2614 return False
2617def _transition_to_submodule(
2618 repo: "Repo",
2619 path: bytes,
2620 full_path: bytes,
2621 current_stat: os.stat_result | None,
2622 entry: IndexEntry | TreeEntry,
2623 index: Index,
2624) -> None:
2625 """Transition any type to submodule."""
2626 from .submodule import ensure_submodule_placeholder
2628 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2629 # Already a directory, just ensure .git file exists
2630 ensure_submodule_placeholder(repo, path)
2631 else:
2632 # Remove whatever is there and create submodule
2633 if current_stat is not None:
2634 _remove_file_with_readonly_handling(full_path)
2635 ensure_submodule_placeholder(repo, path)
2637 st = os.lstat(full_path)
2638 assert entry.sha is not None
2639 index[path] = index_entry_from_stat(st, entry.sha)
2642def _transition_to_file(
2643 object_store: "BaseObjectStore",
2644 path: bytes,
2645 full_path: bytes,
2646 current_stat: os.stat_result | None,
2647 entry: IndexEntry | TreeEntry,
2648 index: Index,
2649 honor_filemode: bool,
2650 symlink_fn: Callable[
2651 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2652 ]
2653 | None,
2654 blob_normalizer: "FilterBlobNormalizer | None",
2655 tree_encoding: str = "utf-8",
2656) -> None:
2657 """Transition any type to regular file or symlink."""
2658 assert entry.sha is not None and entry.mode is not None
2659 # Check if we need to update
2660 if (
2661 current_stat is not None
2662 and stat.S_ISREG(current_stat.st_mode)
2663 and not stat.S_ISLNK(entry.mode)
2664 ):
2665 # File to file - check if update needed
2666 file_matches = _check_file_matches(
2667 object_store,
2668 full_path,
2669 entry.sha,
2670 entry.mode,
2671 current_stat,
2672 honor_filemode,
2673 blob_normalizer,
2674 path,
2675 )
2676 needs_update = not file_matches
2677 elif (
2678 current_stat is not None
2679 and stat.S_ISLNK(current_stat.st_mode)
2680 and stat.S_ISLNK(entry.mode)
2681 ):
2682 # Symlink to symlink - check if update needed
2683 symlink_matches = _check_symlink_matches(full_path, object_store, entry.sha)
2684 needs_update = not symlink_matches
2685 else:
2686 needs_update = True
2688 if not needs_update:
2689 # Just update index - current_stat should always be valid here since we're not updating
2690 assert current_stat is not None
2691 index[path] = index_entry_from_stat(current_stat, entry.sha)
2692 return
2694 # Remove existing entry if needed
2695 if current_stat is not None and stat.S_ISDIR(current_stat.st_mode):
2696 # Remove directory
2697 dir_contents = set(os.listdir(full_path))
2698 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
2700 if git_file_name in dir_contents:
2701 if dir_contents != {git_file_name}:
2702 raise IsADirectoryError(
2703 f"Cannot replace submodule with untracked files: {full_path!r}"
2704 )
2705 shutil.rmtree(full_path)
2706 else:
2707 try:
2708 os.rmdir(full_path)
2709 except OSError as e:
2710 if e.errno in (errno.ENOTEMPTY, errno.EEXIST):
2711 raise IsADirectoryError(
2712 f"Cannot replace non-empty directory with file: {full_path!r}"
2713 )
2714 raise
2715 elif current_stat is not None:
2716 _remove_file_with_readonly_handling(full_path)
2718 # Ensure parent directory exists
2719 _ensure_parent_dir_exists(full_path)
2721 # Write the file
2722 blob_obj = object_store[entry.sha]
2723 assert isinstance(blob_obj, Blob)
2724 if blob_normalizer:
2725 blob_obj = blob_normalizer.checkout_normalize(blob_obj, path)
2726 st = build_file_from_blob(
2727 blob_obj,
2728 entry.mode,
2729 full_path,
2730 honor_filemode=honor_filemode,
2731 tree_encoding=tree_encoding,
2732 symlink_fn=symlink_fn,
2733 )
2734 index[path] = index_entry_from_stat(st, entry.sha)
2737def _transition_to_absent(
2738 repo: "Repo",
2739 path: bytes,
2740 full_path: bytes,
2741 current_stat: os.stat_result | None,
2742 index: Index,
2743) -> None:
2744 """Remove any type of entry."""
2745 if current_stat is None:
2746 return
2748 if stat.S_ISDIR(current_stat.st_mode):
2749 # Check if it's a submodule directory
2750 dir_contents = set(os.listdir(full_path))
2751 git_file_name = b".git" if isinstance(full_path, bytes) else ".git"
2753 if git_file_name in dir_contents and dir_contents == {git_file_name}:
2754 shutil.rmtree(full_path)
2755 else:
2756 try:
2757 os.rmdir(full_path)
2758 except OSError as e:
2759 if e.errno not in (errno.ENOTEMPTY, errno.EEXIST):
2760 raise
2761 else:
2762 _remove_file_with_readonly_handling(full_path)
2764 try:
2765 del index[path]
2766 except KeyError:
2767 pass
2769 # Try to remove empty parent directories
2770 _remove_empty_parents(
2771 full_path, repo.path if isinstance(repo.path, bytes) else repo.path.encode()
2772 )
2775def detect_case_only_renames(
2776 changes: Sequence["TreeChange"],
2777 config: "Config",
2778) -> list["TreeChange"]:
2779 """Detect and transform case-only renames in a list of tree changes.
2781 This function identifies file renames that only differ in case (e.g.,
2782 README.txt -> readme.txt) and transforms matching ADD/DELETE pairs into
2783 CHANGE_RENAME operations. It uses filesystem-appropriate path normalization
2784 based on the repository configuration.
2786 Args:
2787 changes: List of TreeChange objects representing file changes
2788 config: Repository configuration object
2790 Returns:
2791 New list of TreeChange objects with case-only renames converted to CHANGE_RENAME
2792 """
2793 from .diff_tree import (
2794 CHANGE_ADD,
2795 CHANGE_COPY,
2796 CHANGE_DELETE,
2797 CHANGE_MODIFY,
2798 CHANGE_RENAME,
2799 TreeChange,
2800 )
2802 # Build dictionaries of old and new paths with their normalized forms
2803 old_paths_normalized = {}
2804 new_paths_normalized = {}
2805 old_changes = {} # Map from old path to change object
2806 new_changes = {} # Map from new path to change object
2808 # Get the appropriate normalizer based on config
2809 normalize_func = get_path_element_normalizer(config)
2811 def normalize_path(path: bytes) -> bytes:
2812 """Normalize entire path using element normalization."""
2813 return b"/".join(normalize_func(part) for part in path.split(b"/"))
2815 # Pre-normalize all paths once to avoid repeated normalization
2816 for change in changes:
2817 if change.type == CHANGE_DELETE and change.old:
2818 assert change.old.path is not None
2819 try:
2820 normalized = normalize_path(change.old.path)
2821 except UnicodeDecodeError:
2822 logger.warning(
2823 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2824 change.old.path,
2825 )
2826 else:
2827 old_paths_normalized[normalized] = change.old.path
2828 old_changes[change.old.path] = change
2829 elif change.type == CHANGE_RENAME and change.old:
2830 assert change.old.path is not None
2831 # Treat RENAME as DELETE + ADD for case-only detection
2832 try:
2833 normalized = normalize_path(change.old.path)
2834 except UnicodeDecodeError:
2835 logger.warning(
2836 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2837 change.old.path,
2838 )
2839 else:
2840 old_paths_normalized[normalized] = change.old.path
2841 old_changes[change.old.path] = change
2843 if (
2844 change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY)
2845 and change.new
2846 ):
2847 assert change.new.path is not None
2848 try:
2849 normalized = normalize_path(change.new.path)
2850 except UnicodeDecodeError:
2851 logger.warning(
2852 "Skipping case-only rename detection for path with invalid UTF-8: %r",
2853 change.new.path,
2854 )
2855 else:
2856 new_paths_normalized[normalized] = change.new.path
2857 new_changes[change.new.path] = change
2859 # Find case-only renames and transform changes
2860 case_only_renames = set()
2861 new_rename_changes = []
2863 for norm_path, old_path in old_paths_normalized.items():
2864 if norm_path in new_paths_normalized:
2865 new_path = new_paths_normalized[norm_path]
2866 if old_path != new_path:
2867 # Found a case-only rename
2868 old_change = old_changes[old_path]
2869 new_change = new_changes[new_path]
2871 # Create a CHANGE_RENAME to replace the DELETE and ADD/MODIFY pair
2872 if new_change.type == CHANGE_ADD:
2873 # Simple case: DELETE + ADD becomes RENAME
2874 rename_change = TreeChange(
2875 CHANGE_RENAME, old_change.old, new_change.new
2876 )
2877 else:
2878 # Complex case: DELETE + MODIFY becomes RENAME
2879 # Use the old file from DELETE and new file from MODIFY
2880 rename_change = TreeChange(
2881 CHANGE_RENAME, old_change.old, new_change.new
2882 )
2884 new_rename_changes.append(rename_change)
2886 # Mark the old changes for removal
2887 case_only_renames.add(old_change)
2888 case_only_renames.add(new_change)
2890 # Return new list with original ADD/DELETE changes replaced by renames
2891 result = [change for change in changes if change not in case_only_renames]
2892 result.extend(new_rename_changes)
2893 return result
2896def update_working_tree(
2897 repo: "Repo",
2898 old_tree_id: bytes | None,
2899 new_tree_id: bytes,
2900 change_iterator: Iterator["TreeChange"],
2901 honor_filemode: bool = True,
2902 validate_path_element: Callable[[bytes], bool] | None = None,
2903 symlink_fn: Callable[
2904 [str | bytes | os.PathLike[str], str | bytes | os.PathLike[str]], None
2905 ]
2906 | None = None,
2907 force_remove_untracked: bool = False,
2908 blob_normalizer: "FilterBlobNormalizer | None" = None,
2909 tree_encoding: str = "utf-8",
2910 allow_overwrite_modified: bool = False,
2911 *,
2912 config: "Config | None" = None,
2913) -> None:
2914 """Update the working tree and index to match a new tree.
2916 This function handles:
2917 - Adding new files
2918 - Updating modified files
2919 - Removing deleted files
2920 - Cleaning up empty directories
2922 Args:
2923 repo: Repository object
2924 old_tree_id: SHA of the tree before the update
2925 new_tree_id: SHA of the tree to update to
2926 change_iterator: Iterator of TreeChange objects to apply
2927 honor_filemode: An optional flag to honor core.filemode setting
2928 validate_path_element: Function to validate path elements to check out.
2929 If None, derived from ``config`` so that ``core.protectNTFS`` and
2930 ``core.protectHFS`` are honored by default.
2931 symlink_fn: Function to use for creating symlinks
2932 force_remove_untracked: If True, remove files that exist in working
2933 directory but not in target tree, even if old_tree_id is None
2934 blob_normalizer: An optional BlobNormalizer to use for converting line
2935 endings when writing blobs to the working directory.
2936 tree_encoding: Encoding used for tree paths (default: utf-8)
2937 allow_overwrite_modified: If False, raise an error when attempting to
2938 overwrite files that have been modified compared to old_tree_id
2939 config: Repository configuration. If None, falls back to
2940 ``repo.get_config_stack()``.
2941 """
2942 from .diff_tree import (
2943 CHANGE_ADD,
2944 CHANGE_COPY,
2945 CHANGE_DELETE,
2946 CHANGE_MODIFY,
2947 CHANGE_RENAME,
2948 CHANGE_UNCHANGED,
2949 )
2951 if force_remove_untracked:
2952 import warnings
2954 warnings.warn(
2955 "force_remove_untracked is a no-op and will be removed in a future release",
2956 DeprecationWarning,
2957 )
2958 repo_path = repo.path if isinstance(repo.path, bytes) else repo.path.encode()
2959 if config is None:
2960 config = repo.get_config_stack()
2962 if validate_path_element is None:
2963 # Derive the validator from config so callers cannot accidentally
2964 # skip ``core.protectNTFS``/``core.protectHFS`` enforcement by
2965 # omitting this argument.
2966 validate_path_element = get_path_element_validator(config)
2968 index = repo.open_index(config=config)
2970 # Convert iterator to list since we need multiple passes
2971 changes = list(change_iterator)
2973 # Transform case-only renames on case-insensitive filesystems
2974 import platform
2976 default_ignore_case = platform.system() in ("Windows", "Darwin")
2977 config = repo.get_config()
2978 ignore_case = config.get_boolean((b"core",), b"ignorecase", default_ignore_case)
2980 if ignore_case:
2981 config = repo.get_config()
2982 changes = detect_case_only_renames(changes, config)
2984 # Check for path conflicts where files need to become directories
2985 paths_becoming_dirs = set()
2986 for change in changes:
2987 if change.type in (CHANGE_ADD, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY):
2988 assert change.new is not None
2989 path = change.new.path
2990 assert path is not None
2991 if b"/" in path: # This is a file inside a directory
2992 # Check if any parent path exists as a file in the old tree or changes
2993 parts = path.split(b"/")
2994 for i in range(1, len(parts)):
2995 parent = b"/".join(parts[:i])
2996 # See if this parent path is being deleted (was a file, becoming a dir)
2997 for other_change in changes:
2998 if (
2999 other_change.type == CHANGE_DELETE
3000 and other_change.old
3001 and other_change.old.path == parent
3002 ):
3003 paths_becoming_dirs.add(parent)
3005 # Check if any path that needs to become a directory has been modified
3006 for path in paths_becoming_dirs:
3007 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3008 try:
3009 current_stat = os.lstat(full_path)
3010 except FileNotFoundError:
3011 continue # File doesn't exist, nothing to check
3012 except OSError as e:
3013 raise OSError(
3014 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3015 ) from e
3017 if stat.S_ISREG(current_stat.st_mode):
3018 # Find the old entry for this path
3019 old_change = None
3020 for change in changes:
3021 if (
3022 change.type == CHANGE_DELETE
3023 and change.old
3024 and change.old.path == path
3025 ):
3026 old_change = change
3027 break
3029 if old_change:
3030 # Check if file has been modified
3031 assert old_change.old is not None
3032 assert (
3033 old_change.old.sha is not None and old_change.old.mode is not None
3034 )
3035 file_matches = _check_file_matches(
3036 repo.object_store,
3037 full_path,
3038 old_change.old.sha,
3039 old_change.old.mode,
3040 current_stat,
3041 honor_filemode,
3042 blob_normalizer,
3043 path,
3044 )
3045 if not file_matches:
3046 raise OSError(
3047 f"Cannot replace modified file with directory: {path!r}"
3048 )
3050 # Check for uncommitted modifications before making any changes
3051 if not allow_overwrite_modified and old_tree_id:
3052 for change in changes:
3053 # Only check files that are being modified or deleted
3054 if change.type in (CHANGE_MODIFY, CHANGE_DELETE) and change.old:
3055 path = change.old.path
3056 assert path is not None
3057 if not validate_path(path, validate_path_element):
3058 continue
3060 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3061 try:
3062 current_stat = os.lstat(full_path)
3063 except FileNotFoundError:
3064 continue # File doesn't exist, nothing to check
3065 except OSError as e:
3066 raise OSError(
3067 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3068 ) from e
3070 if stat.S_ISREG(current_stat.st_mode):
3071 # Check if working tree file differs from old tree
3072 assert change.old.sha is not None and change.old.mode is not None
3073 file_matches = _check_file_matches(
3074 repo.object_store,
3075 full_path,
3076 change.old.sha,
3077 change.old.mode,
3078 current_stat,
3079 honor_filemode,
3080 blob_normalizer,
3081 path,
3082 )
3083 if not file_matches:
3084 from .errors import WorkingTreeModifiedError
3086 raise WorkingTreeModifiedError(
3087 f"Your local changes to '{path.decode('utf-8', errors='replace')}' "
3088 f"would be overwritten by checkout. "
3089 f"Please commit your changes or stash them before you switch branches."
3090 )
3092 # Apply the changes
3093 for change in changes:
3094 if change.type in (CHANGE_DELETE, CHANGE_RENAME):
3095 # Remove file/directory
3096 assert change.old is not None and change.old.path is not None
3097 path = change.old.path
3098 if not validate_path(path, validate_path_element):
3099 continue
3101 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3102 try:
3103 delete_stat: os.stat_result | None = os.lstat(full_path)
3104 except FileNotFoundError:
3105 delete_stat = None
3106 except OSError as e:
3107 raise OSError(
3108 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3109 ) from e
3111 _transition_to_absent(repo, path, full_path, delete_stat, index)
3113 if change.type in (
3114 CHANGE_ADD,
3115 CHANGE_MODIFY,
3116 CHANGE_UNCHANGED,
3117 CHANGE_COPY,
3118 CHANGE_RENAME,
3119 ):
3120 # Add or modify file
3121 assert (
3122 change.new is not None
3123 and change.new.path is not None
3124 and change.new.mode is not None
3125 )
3126 path = change.new.path
3127 # Validate as we go and abort on the first invalid path,
3128 # leaving any changes already applied in place.
3129 if not validate_path(path, validate_path_element):
3130 raise InvalidPathError(path)
3131 # Refuse to write through a symlinked leading directory that
3132 # would let open(..., "wb") escape the work tree.
3133 verify_leading_dirs(path, [], repo_path)
3134 full_path = _tree_to_fs_path(repo_path, path, tree_encoding)
3135 try:
3136 modify_stat: os.stat_result | None = os.lstat(full_path)
3137 except FileNotFoundError:
3138 modify_stat = None
3139 except OSError as e:
3140 raise OSError(
3141 f"Cannot access {path.decode('utf-8', errors='replace')}: {e}"
3142 ) from e
3144 if S_ISGITLINK(change.new.mode):
3145 _transition_to_submodule(
3146 repo, path, full_path, modify_stat, change.new, index
3147 )
3148 else:
3149 _transition_to_file(
3150 repo.object_store,
3151 path,
3152 full_path,
3153 modify_stat,
3154 change.new,
3155 index,
3156 honor_filemode,
3157 symlink_fn,
3158 blob_normalizer,
3159 tree_encoding,
3160 )
3162 index.write()
3165def _stat_matches_entry(
3166 st: os.stat_result, entry: IndexEntry, trust_ctime: bool = True
3167) -> bool:
3168 """Check if filesystem stat matches index entry stat.
3170 This is used to determine if a file might have changed without reading its content.
3171 Git uses this optimization to avoid expensive filter operations on unchanged files.
3173 Args:
3174 st: Filesystem stat result
3175 entry: Index entry to compare against
3176 trust_ctime: If True, also check ctime (default: True, matching Git behavior)
3177 Returns: True if stat matches and file is likely unchanged
3178 """
3179 # Compare change time (ctime) if trust_ctime is enabled
3180 if trust_ctime:
3181 # Get entry ctime with nanosecond precision if available
3182 if isinstance(entry.ctime, tuple):
3183 entry_ctime_sec = entry.ctime[0]
3184 entry_ctime_nsec = entry.ctime[1]
3185 else:
3186 entry_ctime_sec = int(entry.ctime)
3187 entry_ctime_nsec = 0
3189 if hasattr(st, "st_ctime_ns"):
3190 # Use nanosecond precision when available
3191 st_ctime_nsec = st.st_ctime_ns
3192 entry_ctime_nsec_total = entry_ctime_sec * 1_000_000_000 + entry_ctime_nsec
3193 if st_ctime_nsec != entry_ctime_nsec_total:
3194 return False
3195 else:
3196 # Fall back to second precision
3197 if int(st.st_ctime) != entry_ctime_sec:
3198 return False
3200 # Get entry mtime with nanosecond precision if available
3201 if isinstance(entry.mtime, tuple):
3202 entry_mtime_sec = entry.mtime[0]
3203 entry_mtime_nsec = entry.mtime[1]
3204 else:
3205 entry_mtime_sec = int(entry.mtime)
3206 entry_mtime_nsec = 0
3208 # Compare modification time with nanosecond precision if available
3209 # This is important for fast workflows (e.g., stash) where files can be
3210 # modified multiple times within the same second
3211 if hasattr(st, "st_mtime_ns"):
3212 # Use nanosecond precision when available
3213 st_mtime_nsec = st.st_mtime_ns
3214 entry_mtime_nsec_total = entry_mtime_sec * 1_000_000_000 + entry_mtime_nsec
3215 if st_mtime_nsec != entry_mtime_nsec_total:
3216 return False
3217 else:
3218 # Fall back to second precision
3219 if int(st.st_mtime) != entry_mtime_sec:
3220 return False
3222 # Compare file size
3223 if st.st_size != entry.size:
3224 return False
3226 # If all checks pass, file is likely unchanged
3227 return True
3230def _check_entry_for_changes(
3231 tree_path: bytes,
3232 entry: IndexEntry | ConflictedIndexEntry,
3233 root_path: bytes,
3234 filter_blob_callback: Callable[[Blob, bytes], Blob] | None = None,
3235 trust_ctime: bool = True,
3236) -> bytes | None:
3237 """Check a single index entry for changes.
3239 Args:
3240 tree_path: Path in the tree
3241 entry: Index entry to check
3242 root_path: Root filesystem path
3243 filter_blob_callback: Optional callback to filter blobs
3244 trust_ctime: If True, use ctime for change detection (default: True)
3245 Returns: tree_path if changed, None otherwise
3246 """
3247 if isinstance(entry, ConflictedIndexEntry):
3248 # Conflicted files are always unstaged
3249 return tree_path
3251 full_path = _tree_to_fs_path(root_path, tree_path)
3252 try:
3253 st = os.lstat(full_path)
3254 if stat.S_ISDIR(st.st_mode):
3255 if _has_directory_changed(tree_path, entry):
3256 return tree_path
3257 return None
3259 if not stat.S_ISREG(st.st_mode) and not stat.S_ISLNK(st.st_mode):
3260 return None
3262 # Optimization: If stat matches index entry (mtime and size unchanged),
3263 # we can skip reading and filtering the file entirely. This is a significant
3264 # performance improvement for repositories with many unchanged files.
3265 # Even with filters (e.g., LFS), if the file hasn't been modified (stat unchanged),
3266 # the filter output would be the same, so we can safely skip the expensive
3267 # filter operation. This addresses performance issues with LFS repositories
3268 # where filter operations can be very slow.
3269 if _stat_matches_entry(st, entry, trust_ctime):
3270 return None
3272 blob = blob_from_path_and_stat(full_path, st)
3274 if filter_blob_callback is not None:
3275 blob = filter_blob_callback(blob, tree_path)
3276 except FileNotFoundError:
3277 # The file was removed, so we assume that counts as
3278 # different from whatever file used to exist.
3279 return tree_path
3280 else:
3281 if blob.id != entry.sha:
3282 return tree_path
3283 return None
3286def get_unstaged_changes(
3287 index: Index,
3288 root_path: str | bytes,
3289 filter_blob_callback: Callable[..., Any] | None = None,
3290 preload_index: bool = False,
3291 trust_ctime: bool = True,
3292 max_stat: int | None = None,
3293) -> Generator[bytes, None, None]:
3294 """Walk through an index and check for differences against working tree.
3296 Args:
3297 index: index to check
3298 root_path: path in which to find files
3299 filter_blob_callback: Optional callback to filter blobs
3300 preload_index: If True, use parallel threads to check files (requires threading support)
3301 trust_ctime: If True, use ctime for change detection (default: True)
3302 max_stat: If set, limit the number of stat operations performed.
3303 When the limit is reached, remaining files are assumed unchanged.
3304 Returns: iterator over paths with unstaged changes
3305 """
3306 # For each entry in the index check the sha1 & ensure not staged
3307 if not isinstance(root_path, bytes):
3308 root_path = os.fsencode(root_path)
3310 stat_count = 0
3312 if preload_index:
3313 # Use parallel processing for better performance on slow filesystems
3314 try:
3315 import multiprocessing
3316 from concurrent.futures import ThreadPoolExecutor
3317 except ImportError:
3318 # If threading is not available, fall back to serial processing
3319 preload_index = False
3320 else:
3321 # Collect all entries first
3322 entries = list(index.iteritems())
3324 if max_stat is not None:
3325 # When max_stat is set, limit the entries we process
3326 entries = entries[:max_stat]
3328 # Use number of CPUs but cap at 8 threads to avoid overhead
3329 num_workers = min(multiprocessing.cpu_count(), 8)
3331 # Process entries in parallel
3332 with ThreadPoolExecutor(max_workers=num_workers) as executor:
3333 # Submit all tasks
3334 futures = [
3335 executor.submit(
3336 _check_entry_for_changes,
3337 tree_path,
3338 entry,
3339 root_path,
3340 filter_blob_callback,
3341 trust_ctime,
3342 )
3343 for tree_path, entry in entries
3344 ]
3346 # Yield results as they complete
3347 for future in futures:
3348 result = future.result()
3349 if result is not None:
3350 yield result
3352 if not preload_index:
3353 # Serial processing
3354 for tree_path, entry in index.iteritems():
3355 if max_stat is not None and stat_count >= max_stat:
3356 return
3357 result = _check_entry_for_changes(
3358 tree_path, entry, root_path, filter_blob_callback, trust_ctime
3359 )
3360 stat_count += 1
3361 if result is not None:
3362 yield result
3365def _decode_utf8_with_fallback(data: bytes) -> str:
3366 """Decode bytes as UTF-8, with lossy fallbacks for invalid sequences.
3368 Mirrors the behaviour of git-for-windows's ``xutftowcsn`` (in
3369 ``compat/mingw.c``) so that tree paths containing legacy-encoded or
3370 otherwise invalid UTF-8 produce the same on-disk filename as C git.
3372 Rules:
3373 * Valid UTF-8 (1-4 byte sequences, excluding overlongs and codepoints
3374 > U+10FFFF) is decoded normally.
3375 * Invalid bytes in 0xa0-0xff map 1:1 to U+00A0-U+00FF.
3376 * Invalid bytes in 0x80-0x9f are expanded to two lowercase ASCII hex
3377 digits (e.g. byte 0x80 -> "80").
3378 * Truncated multi-byte sequences and overlong/out-of-range encodings
3379 cause the lead byte to fall through to the above invalid-byte rules
3380 (the trail bytes are re-evaluated on the next iteration).
3381 """
3382 out: list[str] = []
3383 i = 0
3384 n = len(data)
3385 while i < n:
3386 c = data[i]
3387 if c < 0x80:
3388 out.append(chr(c))
3389 i += 1
3390 elif 0xC2 <= c < 0xE0 and i + 1 < n and (data[i + 1] & 0xC0) == 0x80:
3391 cp = ((c & 0x1F) << 6) | (data[i + 1] & 0x3F)
3392 out.append(chr(cp))
3393 i += 2
3394 elif (
3395 0xE0 <= c < 0xF0
3396 and i + 2 < n
3397 and not (c == 0xE0 and data[i + 1] < 0xA0)
3398 and (data[i + 1] & 0xC0) == 0x80
3399 and (data[i + 2] & 0xC0) == 0x80
3400 ):
3401 cp = ((c & 0x0F) << 12) | ((data[i + 1] & 0x3F) << 6) | (data[i + 2] & 0x3F)
3402 out.append(chr(cp))
3403 i += 3
3404 elif (
3405 0xF0 <= c < 0xF5
3406 and i + 3 < n
3407 and not (c == 0xF0 and data[i + 1] < 0x90)
3408 and not (c == 0xF4 and data[i + 1] >= 0x90)
3409 and (data[i + 1] & 0xC0) == 0x80
3410 and (data[i + 2] & 0xC0) == 0x80
3411 and (data[i + 3] & 0xC0) == 0x80
3412 ):
3413 cp = (
3414 ((c & 0x07) << 18)
3415 | ((data[i + 1] & 0x3F) << 12)
3416 | ((data[i + 2] & 0x3F) << 6)
3417 | (data[i + 3] & 0x3F)
3418 )
3419 out.append(chr(cp))
3420 i += 4
3421 elif c >= 0xA0:
3422 out.append(chr(c))
3423 i += 1
3424 else:
3425 out.append(f"{c:02x}")
3426 i += 1
3427 return "".join(out)
3430def _tree_to_fs_path(
3431 root_path: bytes, tree_path: bytes, tree_encoding: str = "utf-8"
3432) -> bytes:
3433 """Convert a git tree path to a file system path.
3435 Args:
3436 root_path: Root filesystem path
3437 tree_path: Git tree path as bytes (encoded with tree_encoding)
3438 tree_encoding: Encoding used for tree paths (default: utf-8)
3440 Returns: File system path.
3441 """
3442 assert isinstance(tree_path, bytes)
3443 if os_sep_bytes != b"/":
3444 sep_corrected_path = tree_path.replace(b"/", os_sep_bytes)
3445 else:
3446 sep_corrected_path = tree_path
3448 # On Windows, decode tree-encoded bytes to a str so they can flow into
3449 # the wide-char Win32 APIs via Python's filesystem layer. For UTF-8
3450 # (the default tree encoding) we use a lossy decoder that matches C
3451 # git's xutftowcsn fallbacks; for other encodings we let UnicodeDecodeError
3452 # propagate rather than silently producing a corrupt path.
3453 if sys.platform == "win32":
3454 if tree_encoding == "utf-8":
3455 tree_path_str = _decode_utf8_with_fallback(sep_corrected_path)
3456 else:
3457 tree_path_str = sep_corrected_path.decode(tree_encoding)
3458 sep_corrected_path = os.fsencode(tree_path_str)
3460 return os.path.join(root_path, sep_corrected_path)
3463def _fs_to_tree_path(fs_path: str | bytes, tree_encoding: str = "utf-8") -> bytes:
3464 """Convert a file system path to a git tree path.
3466 Args:
3467 fs_path: File system path.
3468 tree_encoding: Encoding to use for tree paths (default: utf-8)
3470 Returns: Git tree path as bytes (encoded with tree_encoding)
3471 """
3472 if not isinstance(fs_path, bytes):
3473 fs_path_bytes = os.fsencode(fs_path)
3474 else:
3475 fs_path_bytes = fs_path
3477 # On Windows the on-disk filename is a UTF-16 wide string; Python gives
3478 # us either str (already decoded) or bytes encoded via the filesystem
3479 # codec. Normalise to str, then encode under the tree encoding so the
3480 # resulting tree path is plain UTF-8. This matches C git's xwcstoutf,
3481 # which is just WideCharToMultiByte(CP_UTF8); it makes no attempt to
3482 # reverse the xutftowcsn fallbacks, so a file that was checked out from
3483 # a tree path with invalid UTF-8 will read back as the lossy form (the
3484 # same divergence C git exhibits, documented as a one-way mapping).
3485 if sys.platform == "win32":
3486 fs_path_str = os.fsdecode(fs_path_bytes)
3487 fs_path_bytes = fs_path_str.encode(tree_encoding)
3489 if os_sep_bytes != b"/":
3490 tree_path = fs_path_bytes.replace(os_sep_bytes, b"/")
3491 else:
3492 tree_path = fs_path_bytes
3493 return tree_path
3496def index_entry_from_directory(st: os.stat_result, path: bytes) -> IndexEntry | None:
3497 """Create an index entry for a directory.
3499 This is only used for submodules (directories containing .git).
3501 Args:
3502 st: Stat result for the directory
3503 path: Path to the directory
3505 Returns:
3506 IndexEntry for a submodule, or None if not a submodule
3507 """
3508 if os.path.exists(os.path.join(path, b".git")):
3509 head = read_submodule_head(path)
3510 if head is None:
3511 return None
3512 return index_entry_from_stat(st, head, mode=S_IFGITLINK)
3513 return None
3516def index_entry_from_path(
3517 path: bytes, object_store: ObjectContainer | None = None
3518) -> IndexEntry | None:
3519 """Create an index from a filesystem path.
3521 This returns an index value for files, symlinks
3522 and tree references. for directories and
3523 non-existent files it returns None
3525 Args:
3526 path: Path to create an index entry for
3527 object_store: Optional object store to
3528 save new blobs in
3529 Returns: An index entry; None for directories
3530 """
3531 assert isinstance(path, bytes)
3532 st = os.lstat(path)
3533 if stat.S_ISDIR(st.st_mode):
3534 return index_entry_from_directory(st, path)
3536 if stat.S_ISREG(st.st_mode) or stat.S_ISLNK(st.st_mode):
3537 blob = blob_from_path_and_stat(path, st)
3538 if object_store is not None:
3539 object_store.add_object(blob)
3540 return index_entry_from_stat(st, blob.id)
3542 return None
3545def iter_fresh_entries(
3546 paths: Iterable[bytes],
3547 root_path: bytes,
3548 object_store: ObjectContainer | None = None,
3549) -> Iterator[tuple[bytes, IndexEntry | None]]:
3550 """Iterate over current versions of index entries on disk.
3552 Args:
3553 paths: Paths to iterate over
3554 root_path: Root path to access from
3555 object_store: Optional store to save new blobs in
3556 Returns: Iterator over path, index_entry
3557 """
3558 for path in paths:
3559 p = _tree_to_fs_path(root_path, path)
3560 try:
3561 entry = index_entry_from_path(p, object_store=object_store)
3562 except (FileNotFoundError, IsADirectoryError):
3563 entry = None
3564 yield path, entry
3567def iter_fresh_objects(
3568 paths: Iterable[bytes],
3569 root_path: bytes,
3570 include_deleted: bool = False,
3571 object_store: ObjectContainer | None = None,
3572) -> Iterator[tuple[bytes, ObjectID | None, int | None]]:
3573 """Iterate over versions of objects on disk referenced by index.
3575 Args:
3576 paths: Paths to check
3577 root_path: Root path to access from
3578 include_deleted: Include deleted entries with sha and
3579 mode set to None
3580 object_store: Optional object store to report new items to
3581 Returns: Iterator over path, sha, mode
3582 """
3583 for path, entry in iter_fresh_entries(paths, root_path, object_store=object_store):
3584 if entry is None:
3585 if include_deleted:
3586 yield path, None, None
3587 else:
3588 yield path, entry.sha, cleanup_mode(entry.mode)
3591def refresh_index(index: Index, root_path: bytes) -> None:
3592 """Refresh the contents of an index.
3594 This is the equivalent to running 'git commit -a'.
3596 Args:
3597 index: Index to update
3598 root_path: Root filesystem path
3599 """
3600 for path, entry in iter_fresh_entries(index, root_path):
3601 if entry:
3602 index[path] = entry
3605class locked_index:
3606 """Lock the index while making modifications.
3608 Works as a context manager.
3609 """
3611 _file: "_GitFile"
3613 def __init__(self, path: bytes | str) -> None:
3614 """Initialize locked_index."""
3615 self._path = path
3617 def __enter__(self) -> Index:
3618 """Enter context manager and lock index."""
3619 f = GitFile(self._path, "wb")
3620 self._file = f
3621 self._index = Index(self._path)
3622 return self._index
3624 def __exit__(
3625 self,
3626 exc_type: type | None,
3627 exc_value: BaseException | None,
3628 traceback: types.TracebackType | None,
3629 ) -> None:
3630 """Exit context manager and unlock index."""
3631 if exc_type is not None:
3632 self._file.abort()
3633 return
3634 try:
3635 f = SHA1Writer(self._file)
3636 write_index_dict(f, self._index._byname)
3637 except BaseException:
3638 self._file.abort()
3639 else:
3640 f.close()