Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/objects.py: 46%
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# objects.py -- Access to base git objects
2# Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
3# Copyright (C) 2008-2013 Jelmer Vernooij <jelmer@jelmer.uk>
4#
5# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
6# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
7# General Public License as published by the Free Software Foundation; version 2.0
8# or (at your option) any later version. You can redistribute it and/or
9# modify it under the terms of either of these two licenses.
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16#
17# You should have received a copy of the licenses; if not, see
18# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
19# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
20# License, Version 2.0.
21#
23"""Access to base git objects."""
25__all__ = [
26 "BEGIN_PGP_SIGNATURE",
27 "BEGIN_SSH_SIGNATURE",
28 "DEFAULT_LOOSE_OBJECT_SIZE_LIMIT",
29 "MAX_TIME",
30 "OBJECT_CLASSES",
31 "SIGNATURE_PGP",
32 "SIGNATURE_SSH",
33 "S_IFGITLINK",
34 "S_ISGITLINK",
35 "ZERO_SHA",
36 "Blob",
37 "Commit",
38 "EmptyFileException",
39 "FixedSha",
40 "ObjectID",
41 "RawObjectID",
42 "ShaFile",
43 "SubmoduleEncountered",
44 "Tag",
45 "Tree",
46 "TreeEntry",
47 "check_hexsha",
48 "check_identity",
49 "check_time",
50 "filename_to_hex",
51 "format_time_entry",
52 "format_timezone",
53 "git_line",
54 "hex_to_filename",
55 "hex_to_sha",
56 "is_blob",
57 "is_commit",
58 "is_tag",
59 "is_tree",
60 "key_entry",
61 "key_entry_name_order",
62 "object_class",
63 "object_header",
64 "parse_commit_broken",
65 "parse_time_entry",
66 "parse_time_entry_broken",
67 "parse_timezone",
68 "parse_timezone_broken",
69 "parse_tree",
70 "pretty_format_tree_entry",
71 "serializable_property",
72 "serialize_tree",
73 "sha_to_hex",
74 "sorted_tree_items",
75 "valid_hexsha",
76]
78import binascii
79import os
80import posixpath
81import re
82import stat
83import sys
84import zlib
85from collections.abc import Callable, Iterable, Iterator, Sequence
86from hashlib import sha1
87from io import BufferedIOBase, BytesIO
88from typing import (
89 IO,
90 TYPE_CHECKING,
91 NamedTuple,
92 TypeVar,
93)
95if sys.version_info >= (3, 11):
96 from typing import Self
97else:
98 from typing_extensions import Self
100from typing import NewType, TypeGuard
102from .errors import (
103 ChecksumMismatch,
104 FileFormatException,
105 NotBlobError,
106 NotCommitError,
107 NotTagError,
108 NotTreeError,
109 ObjectFormatException,
110)
111from .file import GitFile
112from .object_format import DEFAULT_OBJECT_FORMAT, ObjectFormat
114if TYPE_CHECKING:
115 from _hashlib import HASH
117 from .file import _GitFile
119# Zero SHA constants for backward compatibility - now defined below as ObjectID
122# Header fields for commits
123_TREE_HEADER = b"tree"
124_PARENT_HEADER = b"parent"
125_AUTHOR_HEADER = b"author"
126_COMMITTER_HEADER = b"committer"
127_ENCODING_HEADER = b"encoding"
128_MERGETAG_HEADER = b"mergetag"
129_GPGSIG_HEADER = b"gpgsig"
131# Header fields for objects
132_OBJECT_HEADER = b"object"
133_TYPE_HEADER = b"type"
134_TAG_HEADER = b"tag"
135_TAGGER_HEADER = b"tagger"
138S_IFGITLINK = 0o160000
140# Intentionally flexible regex to support various types of brokenness
141# in commit/tag author/committer/tagger lines
142_TIME_ENTRY_RE = re.compile(
143 b"^(?P<person>.*) (?P<time>-?[0-9]+) (?P<timezone>[+-]{0,2}[0-9]+)$"
144)
147MAX_TIME = 9223372036854775807 # (2**63) - 1 - signed long int max
149BEGIN_PGP_SIGNATURE = b"-----BEGIN PGP SIGNATURE-----"
150BEGIN_SSH_SIGNATURE = b"-----BEGIN SSH SIGNATURE-----"
152# Signature type constants
153SIGNATURE_PGP = b"pgp"
154SIGNATURE_SSH = b"ssh"
157# Hex SHA type
158ObjectID = NewType("ObjectID", bytes)
160# Raw SHA type
161RawObjectID = NewType("RawObjectID", bytes)
163# Zero SHA constant
164ZERO_SHA: ObjectID = ObjectID(b"0" * 40)
167class EmptyFileException(FileFormatException):
168 """An unexpectedly empty file was encountered."""
171def S_ISGITLINK(m: int) -> bool:
172 """Check if a mode indicates a submodule.
174 Args:
175 m: Mode to check
176 Returns: a ``boolean``
177 """
178 return stat.S_IFMT(m) == S_IFGITLINK
181# Git's core.bigFileThreshold default.
182DEFAULT_LOOSE_OBJECT_SIZE_LIMIT = 512 * 1024 * 1024
185def _decompress(
186 string: bytes, max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT
187) -> bytes:
188 """Decompress a zlib-compressed loose object, bounded by max_size.
190 Raises:
191 zlib.error: if the decompressed data would exceed max_size bytes,
192 or on any other decompression error.
193 """
194 dcomp = zlib.decompressobj()
195 # +1 so overrun surfaces as unconsumed_tail rather than being truncated.
196 dcomped = dcomp.decompress(string, max_size + 1)
197 if dcomp.unconsumed_tail:
198 raise zlib.error("decompressed data exceeds maximum size")
199 dcomped += dcomp.flush()
200 if len(dcomped) > max_size:
201 raise zlib.error("decompressed data exceeds maximum size")
202 return dcomped
205def sha_to_hex(sha: RawObjectID) -> ObjectID:
206 """Takes a string and returns the hex of the sha within."""
207 hexsha = binascii.hexlify(sha)
208 # Support both SHA1 (40 chars) and SHA256 (64 chars)
209 if len(hexsha) not in (40, 64):
210 raise ValueError(f"Incorrect length of sha string: {hexsha!r}")
211 return ObjectID(hexsha)
214def hex_to_sha(hex: ObjectID | str) -> RawObjectID:
215 """Takes a hex sha and returns a binary sha."""
216 # Support both SHA1 (40 chars) and SHA256 (64 chars)
217 if len(hex) not in (40, 64):
218 raise ValueError(f"Incorrect length of hexsha: {hex!r}")
219 try:
220 return RawObjectID(binascii.unhexlify(hex))
221 except TypeError as exc:
222 if not isinstance(hex, bytes):
223 raise
224 raise ValueError(exc.args[0]) from exc
227def valid_hexsha(hex: bytes | str) -> bool:
228 """Check if a hex string is a valid SHA1 or SHA256.
230 Args:
231 hex: Hex string to validate
233 Returns:
234 True if valid SHA1 (40 chars) or SHA256 (64 chars), False otherwise
235 """
236 if len(hex) not in (40, 64):
237 return False
238 try:
239 binascii.unhexlify(hex)
240 except (TypeError, binascii.Error):
241 return False
242 else:
243 return True
246PathT = TypeVar("PathT", str, bytes)
249def hex_to_filename(path: PathT, hex: ObjectID | str) -> PathT:
250 """Takes a hex sha and returns its filename relative to the given path."""
251 # os.path.join accepts bytes or unicode, but all args must be of the same
252 # type. Make sure that hex which is expected to be bytes, is the same type
253 # as path.
254 if isinstance(path, str):
255 if isinstance(hex, bytes):
256 hex_str = hex.decode("ascii")
257 else:
258 hex_str = hex
259 dir_name = hex_str[:2]
260 file_name = hex_str[2:]
261 result = os.path.join(path, dir_name, file_name)
262 assert isinstance(result, str)
263 return result
264 else:
265 # path is bytes
266 if isinstance(hex, str):
267 hex_bytes = hex.encode("ascii")
268 else:
269 hex_bytes = hex
270 dir_name_b = hex_bytes[:2]
271 file_name_b = hex_bytes[2:]
272 result_b = os.path.join(path, dir_name_b, file_name_b)
273 assert isinstance(result_b, bytes)
274 return result_b
277def filename_to_hex(filename: str | bytes) -> str:
278 """Takes an object filename and returns its corresponding hex sha."""
279 import warnings
281 warnings.warn(
282 "filename_to_hex is unused and will be removed in a future version of Dulwich",
283 DeprecationWarning,
284 )
285 # grab the last (up to) two path components
286 errmsg = f"Invalid object filename: {filename!r}"
287 if isinstance(filename, str):
288 names = filename.rsplit(os.path.sep, 2)[-2:]
289 assert len(names) == 2, errmsg
290 base, rest = names
291 assert len(base) == 2 and len(rest) == 38, errmsg
292 hex_str = base + rest
293 hex_bytes = hex_str.encode("ascii")
294 else:
295 # filename is bytes
296 sep = (
297 os.path.sep.encode("ascii") if isinstance(os.path.sep, str) else os.path.sep
298 )
299 names_b = filename.rsplit(sep, 2)[-2:]
300 assert len(names_b) == 2, errmsg
301 base_b, rest_b = names_b
302 assert len(base_b) == 2 and len(rest_b) == 38, errmsg
303 hex_bytes = base_b + rest_b
304 hex_to_sha(ObjectID(hex_bytes))
305 return hex_bytes.decode("ascii")
308def object_header(num_type: int, length: int) -> bytes:
309 """Return an object header for the given numeric type and text length."""
310 cls = object_class(num_type)
311 if cls is None:
312 raise AssertionError(f"unsupported class type num: {num_type}")
313 return cls.type_name + b" " + str(length).encode("ascii") + b"\0"
316def serializable_property(name: str, docstring: str | None = None) -> property:
317 """A property that helps tracking whether serialization is necessary."""
319 def set(obj: "ShaFile", value: object) -> None:
320 """Set the property value and mark the object as needing serialization.
322 Args:
323 obj: The ShaFile object
324 value: The value to set
325 """
326 setattr(obj, "_" + name, value)
327 obj._needs_serialization = True
329 def get(obj: "ShaFile") -> object:
330 """Get the property value.
332 Args:
333 obj: The ShaFile object
335 Returns:
336 The property value
337 """
338 return getattr(obj, "_" + name)
340 return property(get, set, doc=docstring)
343def object_class(type: bytes | int) -> type["ShaFile"] | None:
344 """Get the object class corresponding to the given type.
346 Args:
347 type: Either a type name string or a numeric type.
348 Returns: The ShaFile subclass corresponding to the given type, or None if
349 type is not a valid type name/number.
350 """
351 return _TYPE_MAP.get(type, None)
354def check_hexsha(hex: str | bytes, error_msg: str) -> None:
355 """Check if a string is a valid hex sha string.
357 Args:
358 hex: Hex string to check
359 error_msg: Error message to use in exception
360 Raises:
361 ObjectFormatException: Raised when the string is not valid
362 """
363 if not valid_hexsha(hex):
364 raise ObjectFormatException(f"{error_msg} {hex!r}")
367def check_identity(identity: bytes | None, error_msg: str) -> None:
368 """Check if the specified identity is valid.
370 This will raise an exception if the identity is not valid.
372 Args:
373 identity: Identity string
374 error_msg: Error message to use in exception
375 """
376 if identity is None:
377 raise ObjectFormatException(error_msg)
378 email_start = identity.find(b"<")
379 email_end = identity.find(b">")
380 if not all(
381 [
382 email_start >= 1,
383 identity[email_start - 1] == b" "[0],
384 identity.find(b"<", email_start + 1) == -1,
385 email_end == len(identity) - 1,
386 b"\0" not in identity,
387 b"\n" not in identity,
388 ]
389 ):
390 raise ObjectFormatException(error_msg)
393def _path_to_bytes(path: str | bytes) -> bytes:
394 """Convert a path to bytes for use in error messages."""
395 if isinstance(path, str):
396 return path.encode("utf-8", "surrogateescape")
397 return path
400def check_time(time_seconds: int) -> None:
401 """Check if the specified time is not prone to overflow error.
403 This will raise an exception if the time is not valid.
405 Args:
406 time_seconds: time in seconds
408 """
409 # Prevent overflow error
410 if time_seconds > MAX_TIME:
411 raise ObjectFormatException(f"Date field should not exceed {MAX_TIME}")
414def git_line(*items: bytes) -> bytes:
415 """Formats items into a space separated line."""
416 return b" ".join(items) + b"\n"
419class FixedSha:
420 """SHA object that behaves like hashlib's but is given a fixed value."""
422 __slots__ = ("_hexsha", "_sha")
424 def __init__(self, hexsha: str | bytes) -> None:
425 """Initialize FixedSha with a fixed SHA value.
427 Args:
428 hexsha: Hex SHA value as string or bytes
429 """
430 if isinstance(hexsha, str):
431 hexsha = hexsha.encode("ascii")
432 if not isinstance(hexsha, bytes):
433 raise TypeError(f"Expected bytes for hexsha, got {hexsha!r}")
434 self._hexsha = hexsha
435 self._sha = hex_to_sha(ObjectID(hexsha))
437 def digest(self) -> bytes:
438 """Return the raw SHA digest."""
439 return self._sha
441 def hexdigest(self) -> str:
442 """Return the hex SHA digest."""
443 return self._hexsha.decode("ascii")
446# Type guard functions for runtime type narrowing
447if TYPE_CHECKING:
449 def is_commit(obj: "ShaFile") -> TypeGuard["Commit"]:
450 """Check if a ShaFile is a Commit."""
451 return obj.type_name == b"commit"
453 def is_tree(obj: "ShaFile") -> TypeGuard["Tree"]:
454 """Check if a ShaFile is a Tree."""
455 return obj.type_name == b"tree"
457 def is_blob(obj: "ShaFile") -> TypeGuard["Blob"]:
458 """Check if a ShaFile is a Blob."""
459 return obj.type_name == b"blob"
461 def is_tag(obj: "ShaFile") -> TypeGuard["Tag"]:
462 """Check if a ShaFile is a Tag."""
463 return obj.type_name == b"tag"
464else:
465 # Runtime versions without type narrowing
466 def is_commit(obj: "ShaFile") -> bool:
467 """Check if a ShaFile is a Commit."""
468 return obj.type_name == b"commit"
470 def is_tree(obj: "ShaFile") -> bool:
471 """Check if a ShaFile is a Tree."""
472 return obj.type_name == b"tree"
474 def is_blob(obj: "ShaFile") -> bool:
475 """Check if a ShaFile is a Blob."""
476 return obj.type_name == b"blob"
478 def is_tag(obj: "ShaFile") -> bool:
479 """Check if a ShaFile is a Tag."""
480 return obj.type_name == b"tag"
483class ShaFile:
484 """A git SHA file."""
486 __slots__ = ("_chunked_text", "_needs_serialization", "_sha", "object_format")
488 _needs_serialization: bool
489 type_name: bytes
490 type_num: int
491 _chunked_text: list[bytes] | None
492 _sha: "FixedSha | HASH | None"
493 object_format: ObjectFormat
495 def __init__(self) -> None:
496 """Initialize a ShaFile."""
497 self._sha = None
498 self._chunked_text = None
499 self._needs_serialization = True
500 self.object_format = DEFAULT_OBJECT_FORMAT
502 @staticmethod
503 def _parse_legacy_object_header(magic: bytes) -> "ShaFile":
504 """Parse a legacy object header, creating it but not reading the file."""
505 # Legitimate loose object headers are at most a few dozen bytes; a
506 # larger inflated prefix indicates corruption or a decompression bomb.
507 header_max = 8192
508 decomp = zlib.decompressobj()
509 # Inflate only the header prefix. The object content may be far larger
510 # than header_max, so we cap the output and keep pulling from
511 # unconsumed_tail until we find the NUL that terminates the header.
512 header = decomp.decompress(magic, header_max)
513 start = 0
514 end = header.find(b"\0", start)
515 start = len(header)
516 while end < 0:
517 if len(header) >= header_max:
518 raise zlib.error("object header exceeds maximum size")
519 if not decomp.unconsumed_tail:
520 raise ObjectFormatException("Invalid object header, no \\0")
521 header += decomp.decompress(
522 decomp.unconsumed_tail, header_max - len(header)
523 )
524 end = header.find(b"\0", start)
525 start = len(header)
526 header = header[:end]
527 type_name, size = header.split(b" ", 1)
528 try:
529 int(size) # sanity check
530 except ValueError as exc:
531 raise ObjectFormatException(f"Object size not an integer: {exc}") from exc
532 obj_class = object_class(type_name)
533 if not obj_class:
534 raise ObjectFormatException(
535 "Not a known type: {}".format(type_name.decode("ascii"))
536 )
537 return obj_class()
539 def _parse_legacy_object(
540 self,
541 map: bytes,
542 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
543 ) -> None:
544 """Parse a legacy object, setting the raw string."""
545 text = _decompress(map, max_size=max_size)
546 header_end = text.find(b"\0")
547 if header_end < 0:
548 raise ObjectFormatException("Invalid object header, no \\0")
549 self.set_raw_string(text[header_end + 1 :])
551 def as_legacy_object_chunks(self, compression_level: int = -1) -> Iterator[bytes]:
552 """Return chunks representing the object in the experimental format.
554 Returns: List of strings
555 """
556 compobj = zlib.compressobj(compression_level)
557 yield compobj.compress(self._header())
558 for chunk in self.as_raw_chunks():
559 yield compobj.compress(chunk)
560 yield compobj.flush()
562 def as_legacy_object(self, compression_level: int = -1) -> bytes:
563 """Return string representing the object in the experimental format."""
564 return b"".join(
565 self.as_legacy_object_chunks(compression_level=compression_level)
566 )
568 def as_raw_chunks(self) -> list[bytes]:
569 """Return chunks with serialization of the object.
571 Returns: List of strings, not necessarily one per line
572 """
573 if self._needs_serialization:
574 self._sha = None
575 self._chunked_text = self._serialize()
576 self._needs_serialization = False
577 assert self._chunked_text is not None
578 return self._chunked_text
580 def as_raw_string(self) -> bytes:
581 """Return raw string with serialization of the object.
583 Returns: String object
584 """
585 return b"".join(self.as_raw_chunks())
587 def __bytes__(self) -> bytes:
588 """Return raw string serialization of this object."""
589 return self.as_raw_string()
591 def __hash__(self) -> int:
592 """Return unique hash for this object."""
593 return hash(self.id)
595 def as_pretty_string(self) -> str:
596 """Return a string representing this object, fit for display."""
597 return self.as_raw_string().decode("utf-8", "replace")
599 def set_raw_string(
600 self,
601 text: bytes,
602 sha: ObjectID | RawObjectID | None = None,
603 *,
604 verify_sha: ObjectID | RawObjectID | None = None,
605 ) -> None:
606 """Set the contents of this object from a serialized string."""
607 if not isinstance(text, bytes):
608 raise TypeError(f"Expected bytes for text, got {text!r}")
609 self.set_raw_chunks([text], sha, verify_sha=verify_sha)
611 def set_raw_chunks(
612 self,
613 chunks: list[bytes],
614 sha: ObjectID | RawObjectID | None = None,
615 *,
616 object_format: ObjectFormat | None = None,
617 verify_sha: ObjectID | RawObjectID | None = None,
618 ) -> None:
619 """Set the contents of this object from a list of chunks.
621 Args:
622 chunks: The raw uncompressed contents.
623 sha: Optional known, trusted sha for the object. Cached without
624 being checked against the contents.
625 object_format: Optional hash algorithm for the object.
626 verify_sha: Optional sha that the contents are checked against;
627 raises ChecksumMismatch if the object does not hash to it. On
628 success it is cached like ``sha``. Mutually exclusive with ``sha``.
629 """
630 if sha is not None and verify_sha is not None:
631 raise ValueError("sha and verify_sha are mutually exclusive")
632 self._chunked_text = chunks
633 # Set hash algorithm if provided
634 if object_format is not None:
635 self.object_format = object_format
636 # Set SHA before deserialization so Tree can use hash algorithm
637 if sha is None:
638 self._sha = None
639 else:
640 self._sha = FixedSha(sha)
641 self._deserialize(chunks)
642 self._needs_serialization = False
643 if verify_sha is not None:
644 got = self.get_id(self.object_format)
645 if got != verify_sha:
646 raise ChecksumMismatch(verify_sha, got)
647 self._sha = FixedSha(verify_sha)
649 @staticmethod
650 def _parse_object_header(
651 magic: bytes, f: BufferedIOBase | IO[bytes] | "_GitFile"
652 ) -> "ShaFile":
653 """Parse a new style object, creating it but not reading the file."""
654 num_type = (ord(magic[0:1]) >> 4) & 7
655 obj_class = object_class(num_type)
656 if not obj_class:
657 raise ObjectFormatException(f"Not a known type {num_type}")
658 return obj_class()
660 def _parse_object(
661 self,
662 map: bytes,
663 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
664 ) -> None:
665 """Parse a new style object, setting self._text."""
666 # skip type and size; type must have already been determined, and
667 # we trust zlib to fail if it's otherwise corrupted
668 byte = ord(map[0:1])
669 used = 1
670 while (byte & 0x80) != 0:
671 byte = ord(map[used : used + 1])
672 used += 1
673 raw = map[used:]
674 self.set_raw_string(_decompress(raw, max_size=max_size))
676 @classmethod
677 def _is_legacy_object(cls, magic: bytes) -> bool:
678 b0 = ord(magic[0:1])
679 b1 = ord(magic[1:2])
680 word = (b0 << 8) + b1
681 return (b0 & 0x8F) == 0x08 and (word % 31) == 0
683 @classmethod
684 def _parse_file(
685 cls,
686 f: BufferedIOBase | IO[bytes] | "_GitFile",
687 *,
688 object_format: ObjectFormat | None = None,
689 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
690 ) -> "ShaFile":
691 map = f.read()
692 if not map:
693 raise EmptyFileException("Corrupted empty file detected")
695 if cls._is_legacy_object(map):
696 obj = cls._parse_legacy_object_header(map)
697 if object_format is not None:
698 obj.object_format = object_format
699 obj._parse_legacy_object(map, max_size=max_size)
700 else:
701 obj = cls._parse_object_header(map, f)
702 if object_format is not None:
703 obj.object_format = object_format
704 obj._parse_object(map, max_size=max_size)
705 return obj
707 def _deserialize(self, chunks: list[bytes]) -> None:
708 raise NotImplementedError(self._deserialize)
710 def _serialize(self) -> list[bytes]:
711 raise NotImplementedError(self._serialize)
713 @classmethod
714 def from_path(
715 cls,
716 path: str | bytes,
717 sha: ObjectID | None = None,
718 *,
719 object_format: ObjectFormat | None = None,
720 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
721 ) -> "ShaFile":
722 """Open a SHA file from disk."""
723 with GitFile(path, "rb") as f:
724 return cls.from_file(f, sha, object_format=object_format, max_size=max_size)
726 @classmethod
727 def from_file(
728 cls,
729 f: BufferedIOBase | IO[bytes] | "_GitFile",
730 sha: ObjectID | None = None,
731 *,
732 object_format: ObjectFormat | None = None,
733 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
734 ) -> "ShaFile":
735 """Get the contents of a SHA file on disk."""
736 try:
737 # Validate SHA length matches hash algorithm if both provided
738 if sha is not None and object_format is not None:
739 expected_len = object_format.hex_length
740 if len(sha) != expected_len:
741 raise ValueError(
742 f"SHA length {len(sha)} doesn't match hash algorithm "
743 f"{object_format.name} (expected {expected_len})"
744 )
746 obj = cls._parse_file(f, object_format=object_format, max_size=max_size)
747 if sha is not None:
748 obj._sha = FixedSha(sha)
749 else:
750 obj._sha = None
751 return obj
752 except (IndexError, ValueError) as exc:
753 raise ObjectFormatException("invalid object header") from exc
755 @staticmethod
756 def from_raw_string(
757 type_num: int,
758 string: bytes,
759 sha: ObjectID | RawObjectID | None = None,
760 *,
761 object_format: ObjectFormat | None = None,
762 verify_sha: ObjectID | RawObjectID | None = None,
763 ) -> "ShaFile":
764 """Creates an object of the indicated type from the raw string given.
766 Args:
767 type_num: The numeric type of the object.
768 string: The raw uncompressed contents.
769 sha: Optional known, trusted sha for the object.
770 object_format: Optional hash algorithm for the object
771 verify_sha: Optional sha to check the contents against; raises
772 ChecksumMismatch on mismatch. Mutually exclusive with ``sha``.
773 """
774 cls = object_class(type_num)
775 if cls is None:
776 raise AssertionError(f"unsupported class type num: {type_num}")
777 obj = cls()
778 if object_format is not None:
779 obj.object_format = object_format
780 obj.set_raw_string(string, sha, verify_sha=verify_sha)
781 return obj
783 @staticmethod
784 def from_raw_chunks(
785 type_num: int,
786 chunks: list[bytes],
787 sha: ObjectID | RawObjectID | None = None,
788 *,
789 object_format: ObjectFormat | None = None,
790 ) -> "ShaFile":
791 """Creates an object of the indicated type from the raw chunks given.
793 Args:
794 type_num: The numeric type of the object.
795 chunks: An iterable of the raw uncompressed contents.
796 sha: Optional known sha for the object
797 object_format: Optional object format (hash algorithm) for the object.
798 Required for trees in SHA-256 repositories so entry parsing uses
799 the correct OID length.
800 """
801 cls = object_class(type_num)
802 if cls is None:
803 raise AssertionError(f"unsupported class type num: {type_num}")
804 obj = cls()
805 obj.set_raw_chunks(chunks, sha, object_format=object_format)
806 return obj
808 @classmethod
809 def from_string(cls, string: bytes) -> Self:
810 """Create a ShaFile from a string."""
811 obj = cls()
812 obj.set_raw_string(string)
813 return obj
815 def _check_has_member(self, member: str, error_msg: str) -> None:
816 """Check that the object has a given member variable.
818 Args:
819 member: the member variable to check for
820 error_msg: the message for an error if the member is missing
821 Raises:
822 ObjectFormatException: with the given error_msg if member is
823 missing or is None
824 """
825 if getattr(self, member, None) is None:
826 raise ObjectFormatException(error_msg)
828 def check(self) -> None:
829 """Check this object for internal consistency.
831 Raises:
832 ObjectFormatException: if the object is malformed in some way
833 ChecksumMismatch: if the object was created with a SHA that does
834 not match its contents
835 """
836 # TODO: if we find that error-checking during object parsing is a
837 # performance bottleneck, those checks should be moved to the class's
838 # check() method during optimization so we can still check the object
839 # when necessary.
840 old_sha = self.id
841 try:
842 self._deserialize(self.as_raw_chunks())
843 self._sha = None
844 new_sha = self.id
845 except Exception as exc:
846 raise ObjectFormatException(exc) from exc
847 if old_sha != new_sha:
848 raise ChecksumMismatch(new_sha, old_sha)
850 def _header(self) -> bytes:
851 return object_header(self.type_num, self.raw_length())
853 def raw_length(self) -> int:
854 """Returns the length of the raw string of this object."""
855 return sum(map(len, self.as_raw_chunks()))
857 def sha(self, object_format: ObjectFormat | None = None) -> "FixedSha | HASH":
858 """The SHA object that is the name of this object.
860 Args:
861 object_format: Optional HashAlgorithm to use. Defaults to SHA1.
862 """
863 # If using a different hash algorithm, always recalculate
864 if object_format is not None:
865 new_sha = object_format.new_hash()
866 new_sha.update(self._header())
867 for chunk in self.as_raw_chunks():
868 new_sha.update(chunk)
869 return new_sha
871 # Otherwise use cached SHA1 value
872 if self._sha is None or self._needs_serialization:
873 # this is a local because as_raw_chunks() overwrites self._sha
874 new_sha = sha1()
875 new_sha.update(self._header())
876 for chunk in self.as_raw_chunks():
877 new_sha.update(chunk)
878 self._sha = new_sha
879 return self._sha
881 def copy(self) -> "ShaFile":
882 """Create a new copy of this SHA1 object from its raw string."""
883 obj_class = object_class(self.type_num)
884 if obj_class is None:
885 raise AssertionError(f"invalid type num {self.type_num}")
886 return obj_class.from_raw_string(self.type_num, self.as_raw_string(), self.id)
888 @property
889 def id(self) -> ObjectID:
890 """The hex SHA1 of this object.
892 For SHA256 repositories, use get_id(object_format) instead.
893 This property always returns SHA1 for backward compatibility.
894 """
895 return ObjectID(self.sha().hexdigest().encode("ascii"))
897 def get_id(self, object_format: ObjectFormat | None = None) -> bytes:
898 """Get the hex SHA of this object using the specified hash algorithm.
900 Args:
901 object_format: Optional HashAlgorithm to use. Defaults to SHA1.
903 Example:
904 >>> blob = Blob()
905 >>> blob.data = b"Hello, World!"
906 >>> blob.id # Always returns SHA1 for backward compatibility
907 b'4ab299c8ad6ed14f31923dd94f8b5f5cb89dfb54'
908 >>> blob.get_id() # Same as .id
909 b'4ab299c8ad6ed14f31923dd94f8b5f5cb89dfb54'
910 >>> from dulwich.object_format import SHA256
911 >>> blob.get_id(SHA256) # Get SHA256 hash
912 b'03ba204e2f2e707...' # 64-character SHA256
913 """
914 return self.sha(object_format).hexdigest().encode("ascii")
916 def __repr__(self) -> str:
917 """Return string representation of this object."""
918 return f"<{self.__class__.__name__} {self.id!r}>"
920 def __ne__(self, other: object) -> bool:
921 """Check whether this object does not match the other."""
922 return not isinstance(other, ShaFile) or self.id != other.id
924 def __eq__(self, other: object) -> bool:
925 """Return True if the SHAs of the two objects match."""
926 return isinstance(other, ShaFile) and self.id == other.id
928 def __lt__(self, other: object) -> bool:
929 """Return whether SHA of this object is less than the other."""
930 if not isinstance(other, ShaFile):
931 raise TypeError
932 return self.id < other.id
934 def __le__(self, other: object) -> bool:
935 """Check whether SHA of this object is less than or equal to the other."""
936 if not isinstance(other, ShaFile):
937 raise TypeError
938 return self.id <= other.id
941class Blob(ShaFile):
942 """A Git Blob object."""
944 __slots__ = ()
946 type_name = b"blob"
947 type_num = 3
949 def __init__(self) -> None:
950 """Initialize a new Blob object."""
951 super().__init__()
952 self._chunked_text = []
953 self._needs_serialization = False
955 @property
956 def data(self) -> bytes:
957 """The text contained within the blob object."""
958 return self.as_raw_string()
960 @data.setter
961 def data(self, data: bytes) -> None:
962 self.set_raw_string(data)
964 @property
965 def chunked(self) -> list[bytes]:
966 """The text in the blob object, as chunks (not necessarily lines)."""
967 assert self._chunked_text is not None
968 return self._chunked_text
970 @chunked.setter
971 def chunked(self, chunks: list[bytes]) -> None:
972 self._chunked_text = chunks
974 def _serialize(self) -> list[bytes]:
975 assert self._chunked_text is not None
976 return self._chunked_text
978 def _deserialize(self, chunks: list[bytes]) -> None:
979 self._chunked_text = chunks
981 @classmethod
982 def from_path(
983 cls,
984 path: str | bytes,
985 sha: ObjectID | None = None,
986 *,
987 object_format: ObjectFormat | None = None,
988 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
989 ) -> "Blob":
990 """Read a blob from a file on disk.
992 Args:
993 path: Path to the blob file
994 sha: Optional known SHA for the object
995 object_format: Optional object format to use
996 max_size: Maximum inflated size in bytes
998 Returns:
999 A Blob object
1001 Raises:
1002 NotBlobError: If the file is not a blob
1003 """
1004 blob = ShaFile.from_path(
1005 path, sha, object_format=object_format, max_size=max_size
1006 )
1007 if not isinstance(blob, cls):
1008 raise NotBlobError(_path_to_bytes(path))
1009 return blob
1011 def check(self) -> None:
1012 """Check this object for internal consistency.
1014 Raises:
1015 ObjectFormatException: if the object is malformed in some way
1016 """
1017 super().check()
1019 def splitlines(self) -> list[bytes]:
1020 """Return list of lines in this blob.
1022 This preserves the original line endings.
1023 """
1024 chunks = self.chunked
1025 if not chunks:
1026 return []
1027 if len(chunks) == 1:
1028 result: list[bytes] = chunks[0].splitlines(True)
1029 return result
1030 remaining = None
1031 ret = []
1032 for chunk in chunks:
1033 lines = chunk.splitlines(True)
1034 if len(lines) > 1:
1035 ret.append((remaining or b"") + lines[0])
1036 ret.extend(lines[1:-1])
1037 remaining = lines[-1]
1038 elif len(lines) == 1:
1039 if remaining is None:
1040 remaining = lines.pop()
1041 else:
1042 remaining += lines.pop()
1043 if remaining is not None:
1044 ret.append(remaining)
1045 return ret
1048def _parse_message(
1049 chunks: Iterable[bytes],
1050) -> Iterator[tuple[None, None] | tuple[bytes | None, bytes]]:
1051 """Parse a message with a list of fields and a body.
1053 Args:
1054 chunks: the raw chunks of the tag or commit object.
1055 Returns: iterator of tuples of (field, value), one per header line, in the
1056 order read from the text, possibly including duplicates. Includes a
1057 field named None for the freeform tag/commit text.
1058 """
1059 f = BytesIO(b"".join(chunks))
1060 k = None
1061 v: list[bytes] = []
1062 eof = False
1064 def _strip_last_newline(value: bytes) -> bytes:
1065 """Strip the last newline from value."""
1066 if value and value.endswith(b"\n"):
1067 return value[:-1]
1068 return value
1070 # Parse the headers
1071 #
1072 # Headers can contain newlines. The next line is indented with a space.
1073 # We store the latest key as 'k', and the accumulated value chunks as 'v'.
1074 # The chunks are joined only once per header (rather than with ``+=`` per
1075 # continuation line) so a value split across many indented lines stays
1076 # linear instead of quadratic.
1077 for line in f:
1078 if line.startswith(b" "):
1079 # Indented continuation of the previous line
1080 v.append(line[1:])
1081 else:
1082 if k is not None:
1083 # We parsed a new header, return its value
1084 yield (k, _strip_last_newline(b"".join(v)))
1085 if line == b"\n":
1086 # Empty line indicates end of headers
1087 break
1088 (k, rest) = line.split(b" ", 1)
1089 v = [rest]
1091 else:
1092 # We reached end of file before the headers ended. We still need to
1093 # return the previous header, then we need to return a None field for
1094 # the text.
1095 eof = True
1096 if k is not None:
1097 yield (k, _strip_last_newline(b"".join(v)))
1098 yield (None, None)
1100 if not eof:
1101 # We didn't reach the end of file while parsing headers. We can return
1102 # the rest of the file as a message.
1103 yield (None, f.read())
1105 f.close()
1108def _format_message(
1109 headers: Sequence[tuple[bytes, bytes]], body: bytes | None
1110) -> Iterator[bytes]:
1111 for field, value in headers:
1112 lines = value.split(b"\n")
1113 yield git_line(field, lines[0])
1114 for line in lines[1:]:
1115 yield b" " + line + b"\n"
1116 yield b"\n" # There must be a new line after the headers
1117 if body:
1118 yield body
1121class Tag(ShaFile):
1122 """A Git Tag object."""
1124 type_name = b"tag"
1125 type_num = 4
1127 __slots__ = (
1128 "_message",
1129 "_name",
1130 "_object_class",
1131 "_object_sha",
1132 "_signature",
1133 "_tag_time",
1134 "_tag_timezone",
1135 "_tag_timezone_neg_utc",
1136 "_tagger",
1137 )
1139 _message: bytes | None
1140 _name: bytes | None
1141 _object_class: "type[ShaFile] | None"
1142 _object_sha: bytes | None
1143 _signature: bytes | None
1144 _tag_time: int | None
1145 _tag_timezone: int | None
1146 _tag_timezone_neg_utc: bool | None
1147 _tagger: bytes | None
1149 def __init__(self) -> None:
1150 """Initialize a new Tag object."""
1151 super().__init__()
1152 self._tagger = None
1153 self._tag_time = None
1154 self._tag_timezone = None
1155 self._tag_timezone_neg_utc = False
1156 self._signature: bytes | None = None
1158 @classmethod
1159 def from_path(
1160 cls,
1161 path: str | bytes,
1162 sha: ObjectID | None = None,
1163 *,
1164 object_format: ObjectFormat | None = None,
1165 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
1166 ) -> "Tag":
1167 """Read a tag from a file on disk.
1169 Args:
1170 path: Path to the tag file
1171 sha: Optional known SHA for the object
1172 object_format: Optional object format to use
1173 max_size: Maximum inflated size in bytes
1175 Returns:
1176 A Tag object
1178 Raises:
1179 NotTagError: If the file is not a tag
1180 """
1181 tag = ShaFile.from_path(
1182 path, sha, object_format=object_format, max_size=max_size
1183 )
1184 if not isinstance(tag, cls):
1185 raise NotTagError(_path_to_bytes(path))
1186 return tag
1188 def check(self) -> None:
1189 """Check this object for internal consistency.
1191 Raises:
1192 ObjectFormatException: if the object is malformed in some way
1193 """
1194 super().check()
1195 assert self._chunked_text is not None
1196 self._check_has_member("_object_sha", "missing object sha")
1197 self._check_has_member("_object_class", "missing object type")
1198 self._check_has_member("_name", "missing tag name")
1200 if not self._name:
1201 raise ObjectFormatException("empty tag name")
1203 if self._object_sha is None:
1204 raise ObjectFormatException("missing object sha")
1205 check_hexsha(self._object_sha, "invalid object sha")
1207 if self._tagger is not None:
1208 check_identity(self._tagger, "invalid tagger")
1210 self._check_has_member("_tag_time", "missing tag time")
1211 if self._tag_time is None:
1212 raise ObjectFormatException("missing tag time")
1213 check_time(self._tag_time)
1215 last = None
1216 for field, _ in _parse_message(self._chunked_text):
1217 if field == _OBJECT_HEADER and last is not None:
1218 raise ObjectFormatException("unexpected object")
1219 elif field == _TYPE_HEADER and last != _OBJECT_HEADER:
1220 raise ObjectFormatException("unexpected type")
1221 elif field == _TAG_HEADER and last != _TYPE_HEADER:
1222 raise ObjectFormatException("unexpected tag name")
1223 elif field == _TAGGER_HEADER and last != _TAG_HEADER:
1224 raise ObjectFormatException("unexpected tagger")
1225 last = field
1227 def _serialize(self) -> list[bytes]:
1228 headers = []
1229 if self._object_sha is None:
1230 raise ObjectFormatException("missing object sha")
1231 headers.append((_OBJECT_HEADER, self._object_sha))
1232 if self._object_class is None:
1233 raise ObjectFormatException("missing object class")
1234 headers.append((_TYPE_HEADER, self._object_class.type_name))
1235 if self._name is None:
1236 raise ObjectFormatException("missing tag name")
1237 headers.append((_TAG_HEADER, self._name))
1238 if self._tagger:
1239 if self._tag_time is None:
1240 headers.append((_TAGGER_HEADER, self._tagger))
1241 else:
1242 if self._tag_timezone is None or self._tag_timezone_neg_utc is None:
1243 raise ObjectFormatException("missing timezone info")
1244 headers.append(
1245 (
1246 _TAGGER_HEADER,
1247 format_time_entry(
1248 self._tagger,
1249 self._tag_time,
1250 (self._tag_timezone, self._tag_timezone_neg_utc),
1251 ),
1252 )
1253 )
1255 if self.message is None and self._signature is None:
1256 body = None
1257 else:
1258 body = (self.message or b"") + (self._signature or b"")
1259 return list(_format_message(headers, body))
1261 def _deserialize(self, chunks: list[bytes]) -> None:
1262 """Grab the metadata attached to the tag."""
1263 self._tagger = None
1264 self._tag_time = None
1265 self._tag_timezone = None
1266 self._tag_timezone_neg_utc = False
1267 for field, value in _parse_message(chunks):
1268 if field == _OBJECT_HEADER:
1269 self._object_sha = value
1270 elif field == _TYPE_HEADER:
1271 assert isinstance(value, bytes)
1272 obj_class = object_class(value)
1273 if not obj_class:
1274 raise ObjectFormatException(f"Not a known type: {value!r}")
1275 self._object_class = obj_class
1276 elif field == _TAG_HEADER:
1277 self._name = value
1278 elif field == _TAGGER_HEADER:
1279 if value is None:
1280 raise ObjectFormatException("missing tagger value")
1281 (
1282 self._tagger,
1283 self._tag_time,
1284 (self._tag_timezone, self._tag_timezone_neg_utc),
1285 ) = parse_time_entry(value)
1286 elif field is None:
1287 if value is None:
1288 self._message = None
1289 self._signature = None
1290 else:
1291 # Try to find either PGP or SSH signature
1292 sig_idx = None
1293 try:
1294 sig_idx = value.index(BEGIN_PGP_SIGNATURE)
1295 except ValueError:
1296 try:
1297 sig_idx = value.index(BEGIN_SSH_SIGNATURE)
1298 except ValueError:
1299 pass
1301 if sig_idx is not None:
1302 self._message = value[:sig_idx]
1303 self._signature = value[sig_idx:]
1304 else:
1305 self._message = value
1306 self._signature = None
1307 else:
1308 raise ObjectFormatException(
1309 f"Unknown field {field.decode('ascii', 'replace')}"
1310 )
1312 @property
1313 def object(self) -> tuple[type[ShaFile], ObjectID]:
1314 """Get the object pointed to by this tag.
1316 Returns: tuple of (object class, sha).
1317 """
1318 if self._object_class is None or self._object_sha is None:
1319 raise ValueError("Tag object is not properly initialized")
1320 return (self._object_class, ObjectID(self._object_sha))
1322 @object.setter
1323 def object(self, value: tuple[type[ShaFile], bytes]) -> None:
1324 self._object_class, self._object_sha = value
1325 self._needs_serialization = True
1327 name = serializable_property("name", "The name of this tag")
1328 tagger = serializable_property(
1329 "tagger", "Returns the name of the person who created this tag"
1330 )
1331 tag_time = serializable_property(
1332 "tag_time",
1333 "The creation timestamp of the tag. As the number of seconds since the epoch",
1334 )
1335 tag_timezone = serializable_property(
1336 "tag_timezone", "The timezone that tag_time is in."
1337 )
1338 message = serializable_property("message", "the message attached to this tag")
1340 signature = serializable_property("signature", "Optional detached GPG signature")
1342 def raw_without_sig(self) -> bytes:
1343 """Return raw string serialization without the GPG/SSH signature.
1345 self.signature is a signature for the returned raw byte string serialization.
1346 """
1347 ret = self.as_raw_string()
1348 if self._signature:
1349 ret = ret[: -len(self._signature)]
1350 return ret
1352 def extract_signature(self) -> tuple[bytes, bytes | None, bytes | None]:
1353 """Extract the payload, signature, and signature type from this tag.
1355 Returns:
1356 tuple of (``payload``, ``signature``, ``signature_type``) where:
1358 - ``payload``: The raw tag data without the signature
1359 - ``signature``: The signature bytes if present, None otherwise
1360 - ``signature_type``: SIGNATURE_PGP for PGP, SIGNATURE_SSH for SSH, None if no signature
1362 Raises:
1363 ObjectFormatException: If signature has unknown format
1364 """
1365 if self._signature is None:
1366 return self.as_raw_string(), None, None
1368 payload = self.raw_without_sig()
1370 # Determine signature type
1371 if self._signature.startswith(BEGIN_PGP_SIGNATURE):
1372 sig_type = SIGNATURE_PGP
1373 elif self._signature.startswith(BEGIN_SSH_SIGNATURE):
1374 sig_type = SIGNATURE_SSH
1375 else:
1376 raise ObjectFormatException("Unknown signature format")
1378 return payload, self._signature, sig_type
1381class TreeEntry(NamedTuple):
1382 """Named tuple encapsulating a single tree entry."""
1384 path: bytes
1385 mode: int
1386 sha: ObjectID
1388 def in_path(self, path: bytes) -> "TreeEntry":
1389 """Return a copy of this entry with the given path prepended."""
1390 if not isinstance(self.path, bytes):
1391 raise TypeError(f"Expected bytes for path, got {path!r}")
1392 return TreeEntry(posixpath.join(path, self.path), self.mode, self.sha)
1395def parse_tree(
1396 text: bytes, sha_len: int | None = None, *, strict: bool = False
1397) -> Iterator[tuple[bytes, int, bytes]]:
1398 """Parse a tree text.
1400 Args:
1401 text: Serialized text to parse
1402 sha_len: Length of the object IDs in bytes
1403 strict: Whether to be strict about format
1404 Returns: iterator of tuples of (name, mode, sha)
1406 Raises:
1407 ObjectFormatException: if the object was malformed in some way
1408 """
1409 count = 0
1410 length = len(text)
1412 while count < length:
1413 mode_end = text.index(b" ", count)
1414 mode_text = text[count:mode_end]
1415 if strict and mode_text.startswith(b"0"):
1416 raise ObjectFormatException(f"Invalid mode {mode_text!r}")
1417 try:
1418 mode = int(mode_text, 8)
1419 except ValueError as exc:
1420 raise ObjectFormatException(f"Invalid mode {mode_text!r}") from exc
1421 name_end = text.index(b"\0", mode_end)
1422 name = text[mode_end + 1 : name_end]
1424 if sha_len is None:
1425 raise ObjectFormatException("sha_len must be specified")
1426 count = name_end + 1 + sha_len
1427 if count > length:
1428 raise ObjectFormatException(
1429 f"Tree entry extends beyond tree length: {count} > {length}"
1430 )
1432 sha = text[name_end + 1 : count]
1433 if len(sha) != sha_len:
1434 raise ObjectFormatException(
1435 f"Sha has invalid length: {len(sha)} != {sha_len}"
1436 )
1437 hexsha = sha_to_hex(RawObjectID(sha))
1438 yield (name, mode, hexsha)
1441def serialize_tree(items: Iterable[tuple[bytes, int, ObjectID]]) -> Iterator[bytes]:
1442 """Serialize the items in a tree to a text.
1444 Args:
1445 items: Sorted iterable over (name, mode, sha) tuples
1446 Returns: Serialized tree text as chunks
1447 """
1448 for name, mode, hexsha in items:
1449 yield (
1450 (f"{mode:04o}").encode("ascii") + b" " + name + b"\0" + hex_to_sha(hexsha)
1451 )
1454def sorted_tree_items(
1455 entries: dict[bytes, tuple[int, ObjectID]], name_order: bool
1456) -> Iterator[TreeEntry]:
1457 """Iterate over a tree entries dictionary.
1459 Args:
1460 name_order: If True, iterate entries in order of their name. If
1461 False, iterate entries in tree order, that is, treat subtree entries as
1462 having '/' appended.
1463 entries: Dictionary mapping names to (mode, sha) tuples
1464 Returns: Iterator over (name, mode, hexsha)
1465 """
1466 if name_order:
1467 key_func = key_entry_name_order
1468 else:
1469 key_func = key_entry
1470 for name, entry in sorted(entries.items(), key=key_func):
1471 mode, hexsha = entry
1472 # Stricter type checks than normal to mirror checks in the Rust version.
1473 mode = int(mode)
1474 if not isinstance(hexsha, bytes):
1475 raise TypeError(f"Expected bytes for SHA, got {hexsha!r}")
1476 yield TreeEntry(name, mode, hexsha)
1479def key_entry(entry: tuple[bytes, tuple[int, ObjectID]]) -> bytes:
1480 """Sort key for tree entry.
1482 Args:
1483 entry: (name, value) tuple
1484 """
1485 (name, (mode, _sha)) = entry
1486 if stat.S_ISDIR(mode):
1487 name += b"/"
1488 return name
1491def key_entry_name_order(entry: tuple[bytes, tuple[int, ObjectID]]) -> bytes:
1492 """Sort key for tree entry in name order."""
1493 return entry[0]
1496def pretty_format_tree_entry(
1497 name: bytes, mode: int, hexsha: ObjectID, encoding: str = "utf-8"
1498) -> str:
1499 """Pretty format tree entry.
1501 Args:
1502 name: Name of the directory entry
1503 mode: Mode of entry
1504 hexsha: Hexsha of the referenced object
1505 encoding: Character encoding for the name
1506 Returns: string describing the tree entry
1507 """
1508 if mode & stat.S_IFDIR:
1509 kind = "tree"
1510 else:
1511 kind = "blob"
1512 return "{:04o} {} {}\t{}\n".format(
1513 mode,
1514 kind,
1515 hexsha.decode("ascii"),
1516 name.decode(encoding, "replace"),
1517 )
1520class SubmoduleEncountered(Exception):
1521 """A submodule was encountered while resolving a path."""
1523 def __init__(self, path: bytes, sha: ObjectID) -> None:
1524 """Initialize SubmoduleEncountered exception.
1526 Args:
1527 path: Path where the submodule was encountered
1528 sha: SHA of the submodule
1529 """
1530 self.path = path
1531 self.sha = sha
1534class Tree(ShaFile):
1535 """A Git tree object."""
1537 type_name = b"tree"
1538 type_num = 2
1540 __slots__ = "_entries"
1542 def __init__(self) -> None:
1543 """Initialize an empty Tree."""
1544 super().__init__()
1545 self._entries: dict[bytes, tuple[int, ObjectID]] = {}
1547 @classmethod
1548 def from_path(
1549 cls,
1550 path: str | bytes,
1551 sha: ObjectID | None = None,
1552 *,
1553 object_format: ObjectFormat | None = None,
1554 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
1555 ) -> "Tree":
1556 """Read a tree from a file on disk.
1558 Args:
1559 path: Path to the tree file
1560 sha: Optional known SHA for the object
1561 object_format: Optional object format to use
1562 max_size: Maximum inflated size in bytes
1564 Returns:
1565 A Tree object
1567 Raises:
1568 NotTreeError: If the file is not a tree
1569 """
1570 tree = ShaFile.from_path(
1571 path, sha, object_format=object_format, max_size=max_size
1572 )
1573 if not isinstance(tree, cls):
1574 raise NotTreeError(_path_to_bytes(path))
1575 return tree
1577 def __contains__(self, name: bytes) -> bool:
1578 """Check if name exists in tree."""
1579 return name in self._entries
1581 def __getitem__(self, name: bytes) -> tuple[int, ObjectID]:
1582 """Get tree entry by name."""
1583 return self._entries[name]
1585 def __setitem__(self, name: bytes, value: tuple[int, ObjectID]) -> None:
1586 """Set a tree entry by name.
1588 Args:
1589 name: The name of the entry, as a string.
1590 value: A tuple of (mode, hexsha), where mode is the mode of the
1591 entry as an integral type and hexsha is the hex SHA of the entry as
1592 a string.
1593 """
1594 mode, hexsha = value
1595 self._entries[name] = (mode, hexsha)
1596 self._needs_serialization = True
1598 def __delitem__(self, name: bytes) -> None:
1599 """Delete tree entry by name."""
1600 del self._entries[name]
1601 self._needs_serialization = True
1603 def __len__(self) -> int:
1604 """Return number of entries in tree."""
1605 return len(self._entries)
1607 def __iter__(self) -> Iterator[bytes]:
1608 """Iterate over tree entry names."""
1609 return iter(self._entries)
1611 def add(self, name: bytes, mode: int, hexsha: ObjectID) -> None:
1612 """Add an entry to the tree.
1614 Args:
1615 mode: The mode of the entry as an integral type. Not all
1616 possible modes are supported by git; see check() for details.
1617 name: The name of the entry, as a string.
1618 hexsha: The hex SHA of the entry as a string.
1619 """
1620 self._entries[name] = mode, hexsha
1621 self._needs_serialization = True
1623 def iteritems(self, name_order: bool = False) -> Iterator[TreeEntry]:
1624 """Iterate over entries.
1626 Args:
1627 name_order: If True, iterate in name order instead of tree
1628 order.
1629 Returns: Iterator over (name, mode, sha) tuples
1630 """
1631 return sorted_tree_items(self._entries, name_order)
1633 def items(self) -> list[TreeEntry]:
1634 """Return the sorted entries in this tree.
1636 Returns: List with (name, mode, sha) tuples
1637 """
1638 return list(self.iteritems())
1640 def _deserialize(self, chunks: list[bytes]) -> None:
1641 """Grab the entries in the tree."""
1642 try:
1643 parsed_entries = parse_tree(
1644 b"".join(chunks),
1645 sha_len=self.object_format.oid_length,
1646 )
1647 except ValueError as exc:
1648 raise ObjectFormatException(exc) from exc
1649 # TODO: list comprehension is for efficiency in the common (small)
1650 # case; if memory efficiency in the large case is a concern, use a
1651 # genexp.
1652 self._entries = {n: (m, ObjectID(s)) for n, m, s in parsed_entries}
1654 def check(self) -> None:
1655 """Check this object for internal consistency.
1657 Raises:
1658 ObjectFormatException: if the object is malformed in some way
1659 """
1660 super().check()
1661 assert self._chunked_text is not None
1662 last = None
1663 allowed_modes = (
1664 stat.S_IFREG | 0o755,
1665 stat.S_IFREG | 0o644,
1666 stat.S_IFLNK,
1667 stat.S_IFDIR,
1668 S_IFGITLINK,
1669 # TODO: optionally exclude as in git fsck --strict
1670 stat.S_IFREG | 0o664,
1671 )
1672 for name, mode, sha in parse_tree(
1673 b"".join(self._chunked_text),
1674 strict=True,
1675 sha_len=self.object_format.oid_length,
1676 ):
1677 check_hexsha(sha, f"invalid sha {sha!r}")
1678 if b"/" in name or name in (b"", b".", b"..", b".git"):
1679 raise ObjectFormatException(
1680 "invalid name {}".format(name.decode("utf-8", "replace"))
1681 )
1683 if mode not in allowed_modes:
1684 raise ObjectFormatException(f"invalid mode {mode:06o}")
1686 entry = (name, (mode, ObjectID(sha)))
1687 if last:
1688 if key_entry(last) > key_entry(entry):
1689 raise ObjectFormatException("entries not sorted")
1690 if name == last[0]:
1691 raise ObjectFormatException(f"duplicate entry {name!r}")
1692 last = entry
1694 def _serialize(self) -> list[bytes]:
1695 return list(serialize_tree(self.iteritems()))
1697 def as_pretty_string(self) -> str:
1698 """Return a human-readable string representation of this tree.
1700 Returns:
1701 Pretty-printed tree entries
1702 """
1703 text: list[str] = []
1704 for entry in self.iteritems():
1705 if (
1706 entry.path is not None
1707 and entry.mode is not None
1708 and entry.sha is not None
1709 ):
1710 text.append(pretty_format_tree_entry(entry.path, entry.mode, entry.sha))
1711 return "".join(text)
1713 def lookup_path(
1714 self, lookup_obj: Callable[[ObjectID], ShaFile], path: bytes
1715 ) -> tuple[int, ObjectID]:
1716 """Look up an object in a Git tree.
1718 Args:
1719 lookup_obj: Callback for retrieving object by SHA1
1720 path: Path to lookup
1721 Returns: A tuple of (mode, SHA) of the resulting path.
1722 """
1723 # Handle empty path - return the tree itself
1724 if not path:
1725 return stat.S_IFDIR, self.id
1727 parts = path.split(b"/")
1728 sha = self.id
1729 mode: int | None = None
1730 for i, p in enumerate(parts):
1731 if not p:
1732 continue
1733 if mode is not None and S_ISGITLINK(mode):
1734 raise SubmoduleEncountered(b"/".join(parts[:i]), sha)
1735 obj = lookup_obj(sha)
1736 if not isinstance(obj, Tree):
1737 raise NotTreeError(sha)
1738 mode, sha = obj[p]
1739 if mode is None:
1740 raise ValueError("No valid path found")
1741 return mode, sha
1744def parse_timezone(text: bytes) -> tuple[int, bool]:
1745 """Parse a timezone text fragment (e.g. '+0100').
1747 Args:
1748 text: Text to parse.
1749 Returns: Tuple with timezone as seconds difference to UTC
1750 and a boolean indicating whether this was a UTC timezone
1751 prefixed with a negative sign (-0000).
1752 """
1753 # cgit parses the first character as the sign, and the rest
1754 # as an integer (using strtol), which could also be negative.
1755 # We do the same for compatibility. See #697828.
1756 if text[0] not in b"+-":
1757 raise ValueError("Timezone must start with + or - ({text})".format(**vars()))
1758 sign = text[:1]
1759 offset = int(text[1:])
1760 if sign == b"-":
1761 offset = -offset
1762 unnecessary_negative_timezone = offset >= 0 and sign == b"-"
1763 signum = ((offset < 0) and -1) or 1
1764 offset = abs(offset)
1765 hours = int(offset / 100)
1766 minutes = offset % 100
1767 return (
1768 signum * (hours * 3600 + minutes * 60),
1769 unnecessary_negative_timezone,
1770 )
1773def parse_timezone_broken(text: bytes) -> tuple[int, bool]:
1774 """Parse a timezone text fragment, accepting broken formats.
1776 This function handles various broken timezone formats found in the wild:
1777 - Missing sign prefix (e.g., '0000' instead of '+0000')
1778 - Double negative (e.g., '--700')
1780 Args:
1781 text: Text to parse.
1782 Returns: Tuple with timezone as seconds difference to UTC
1783 and a boolean indicating whether this was a UTC timezone
1784 prefixed with a negative sign (-0000).
1785 """
1786 if text[0] not in b"+-":
1787 # Some (broken) commits do not have a sign
1788 text = b"+" + text
1790 # cgit parses the first character as the sign, and the rest
1791 # as an integer (using strtol), which could also be negative.
1792 # We do the same for compatibility. See #697828.
1793 sign = text[:1]
1794 offset = int(text[1:])
1795 if sign == b"-":
1796 offset = -offset
1797 unnecessary_negative_timezone = offset >= 0 and sign == b"-"
1798 signum = ((offset < 0) and -1) or 1
1799 offset = abs(offset)
1800 hours = int(offset / 100)
1801 minutes = offset % 100
1802 return (
1803 signum * (hours * 3600 + minutes * 60),
1804 unnecessary_negative_timezone,
1805 )
1808def format_timezone(offset: int, unnecessary_negative_timezone: bool = False) -> bytes:
1809 """Format a timezone for Git serialization.
1811 Args:
1812 offset: Timezone offset as seconds difference to UTC
1813 unnecessary_negative_timezone: Whether to use a minus sign for
1814 UTC or positive timezones (-0000 and --700 rather than +0000 / +0700).
1815 """
1816 if offset % 60 != 0:
1817 raise ValueError("Unable to handle non-minute offset.")
1818 if offset < 0 or unnecessary_negative_timezone:
1819 sign = "-"
1820 offset = -offset
1821 else:
1822 sign = "+"
1823 return ("%c%02d%02d" % (sign, offset / 3600, (offset / 60) % 60)).encode("ascii") # noqa: UP031
1826def parse_time_entry(
1827 value: bytes,
1828) -> tuple[bytes, int | None, tuple[int | None, bool]]:
1829 """Parse event.
1831 Args:
1832 value: Bytes representing a git commit/tag line
1833 Raises:
1834 ObjectFormatException in case of parsing error (malformed
1835 field date)
1836 Returns: Tuple of (author, time, (timezone, timezone_neg_utc))
1837 """
1838 try:
1839 sep = value.rindex(b"> ")
1840 except ValueError:
1841 return (value, None, (None, False))
1842 try:
1843 person = value[0 : sep + 1]
1844 rest = value[sep + 2 :]
1845 timetext, timezonetext = rest.rsplit(b" ", 1)
1846 time = int(timetext)
1847 timezone, timezone_neg_utc = parse_timezone(timezonetext)
1848 except ValueError as exc:
1849 raise ObjectFormatException(exc) from exc
1850 return person, time, (timezone, timezone_neg_utc)
1853def parse_time_entry_broken(
1854 value: bytes,
1855) -> tuple[bytes, int | None, tuple[int | None, bool]]:
1856 """Parse event, accepting broken formats.
1858 This function handles various broken author/committer/tagger line formats:
1859 - Missing angle brackets around email
1860 - Unsigned timezones
1861 - Double-negative timezones
1863 Args:
1864 value: Bytes representing a git commit/tag line
1865 Raises:
1866 ObjectFormatException in case of parsing error
1867 Returns: Tuple of (author, time, (timezone, timezone_neg_utc))
1868 """
1869 m = _TIME_ENTRY_RE.match(value)
1870 if not m:
1871 raise ObjectFormatException(f"Unable to parse time entry: {value!r}")
1873 person = m.group("person")
1874 timetext = m.group("time")
1875 timezonetext = m.group("timezone")
1876 time = int(timetext)
1877 timezone, timezone_neg_utc = parse_timezone_broken(timezonetext)
1879 return person, time, (timezone, timezone_neg_utc)
1882def format_time_entry(
1883 person: bytes, time: int, timezone_info: tuple[int, bool]
1884) -> bytes:
1885 """Format an event."""
1886 (timezone, timezone_neg_utc) = timezone_info
1887 return b" ".join(
1888 [person, str(time).encode("ascii"), format_timezone(timezone, timezone_neg_utc)]
1889 )
1892def _parse_commit(
1893 chunks: Iterable[bytes],
1894) -> tuple[
1895 bytes | None,
1896 list[bytes],
1897 tuple[bytes | None, int | None, tuple[int | None, bool | None]],
1898 tuple[bytes | None, int | None, tuple[int | None, bool | None]],
1899 bytes | None,
1900 list[Tag],
1901 bytes | None,
1902 bytes | None,
1903 list[tuple[bytes, bytes]],
1904]:
1905 """Parse a commit object from chunks.
1907 Args:
1908 chunks: Chunks to parse
1909 Returns: Tuple of (tree, parents, author_info, commit_info,
1910 encoding, mergetag, gpgsig, message, extra)
1911 """
1912 parents = []
1913 extra: list[tuple[bytes, bytes]] = []
1914 tree = None
1915 author_info: tuple[bytes | None, int | None, tuple[int | None, bool | None]] = (
1916 None,
1917 None,
1918 (None, None),
1919 )
1920 commit_info: tuple[bytes | None, int | None, tuple[int | None, bool | None]] = (
1921 None,
1922 None,
1923 (None, None),
1924 )
1925 encoding = None
1926 mergetag = []
1927 message = None
1928 gpgsig = None
1930 for field, value in _parse_message(chunks):
1931 # TODO(jelmer): Enforce ordering
1932 if field == _TREE_HEADER:
1933 tree = value
1934 elif field == _PARENT_HEADER:
1935 if value is None:
1936 raise ObjectFormatException("missing parent value")
1937 parents.append(value)
1938 elif field == _AUTHOR_HEADER:
1939 if value is None:
1940 raise ObjectFormatException("missing author value")
1941 author_info = parse_time_entry(value)
1942 elif field == _COMMITTER_HEADER:
1943 if value is None:
1944 raise ObjectFormatException("missing committer value")
1945 commit_info = parse_time_entry(value)
1946 elif field == _ENCODING_HEADER:
1947 encoding = value
1948 elif field == _MERGETAG_HEADER:
1949 if value is None:
1950 raise ObjectFormatException("missing mergetag value")
1951 tag = Tag.from_string(value + b"\n")
1952 assert isinstance(tag, Tag)
1953 mergetag.append(tag)
1954 elif field == _GPGSIG_HEADER:
1955 gpgsig = value
1956 elif field is None:
1957 message = value
1958 else:
1959 if value is None:
1960 raise ObjectFormatException(f"missing value for field {field!r}")
1961 extra.append((field, value))
1962 return (
1963 tree,
1964 parents,
1965 author_info,
1966 commit_info,
1967 encoding,
1968 mergetag,
1969 gpgsig,
1970 message,
1971 extra,
1972 )
1975def _parse_commit_broken(
1976 chunks: Iterable[bytes],
1977) -> tuple[
1978 bytes | None,
1979 list[bytes],
1980 tuple[bytes | None, int | None, tuple[int | None, bool | None]],
1981 tuple[bytes | None, int | None, tuple[int | None, bool | None]],
1982 bytes | None,
1983 list[Tag],
1984 bytes | None,
1985 bytes | None,
1986 list[tuple[bytes, bytes]],
1987]:
1988 """Parse a commit object from chunks, accepting broken formats.
1990 This function handles various broken author/committer line formats:
1991 - Missing angle brackets around email
1992 - Unsigned timezones
1993 - Double-negative timezones
1995 Args:
1996 chunks: Chunks to parse
1997 Returns: Tuple of (tree, parents, author_info, commit_info,
1998 encoding, mergetag, gpgsig, message, extra)
1999 """
2000 parents = []
2001 extra: list[tuple[bytes, bytes]] = []
2002 tree = None
2003 author_info: tuple[bytes | None, int | None, tuple[int | None, bool | None]] = (
2004 None,
2005 None,
2006 (None, None),
2007 )
2008 commit_info: tuple[bytes | None, int | None, tuple[int | None, bool | None]] = (
2009 None,
2010 None,
2011 (None, None),
2012 )
2013 encoding = None
2014 mergetag = []
2015 message = None
2016 gpgsig = None
2018 for field, value in _parse_message(chunks):
2019 # TODO(jelmer): Enforce ordering
2020 if field == _TREE_HEADER:
2021 tree = value
2022 elif field == _PARENT_HEADER:
2023 if value is None:
2024 raise ObjectFormatException("missing parent value")
2025 parents.append(value)
2026 elif field == _AUTHOR_HEADER:
2027 if value is None:
2028 raise ObjectFormatException("missing author value")
2029 author_info = parse_time_entry_broken(value)
2030 elif field == _COMMITTER_HEADER:
2031 if value is None:
2032 raise ObjectFormatException("missing committer value")
2033 commit_info = parse_time_entry_broken(value)
2034 elif field == _ENCODING_HEADER:
2035 encoding = value
2036 elif field == _MERGETAG_HEADER:
2037 if value is None:
2038 raise ObjectFormatException("missing mergetag value")
2039 tag = Tag.from_string(value + b"\n")
2040 assert isinstance(tag, Tag)
2041 mergetag.append(tag)
2042 elif field == _GPGSIG_HEADER:
2043 gpgsig = value
2044 elif field is None:
2045 message = value
2046 else:
2047 if value is None:
2048 raise ObjectFormatException(f"missing value for field {field!r}")
2049 extra.append((field, value))
2050 return (
2051 tree,
2052 parents,
2053 author_info,
2054 commit_info,
2055 encoding,
2056 mergetag,
2057 gpgsig,
2058 message,
2059 extra,
2060 )
2063class Commit(ShaFile):
2064 """A git commit object."""
2066 type_name = b"commit"
2067 type_num = 1
2069 __slots__ = (
2070 "_author",
2071 "_author_time",
2072 "_author_timezone",
2073 "_author_timezone_neg_utc",
2074 "_commit_time",
2075 "_commit_timezone",
2076 "_commit_timezone_neg_utc",
2077 "_committer",
2078 "_encoding",
2079 "_extra",
2080 "_gpgsig",
2081 "_mergetag",
2082 "_message",
2083 "_parents",
2084 "_tree",
2085 )
2087 def __init__(self) -> None:
2088 """Initialize an empty Commit."""
2089 super().__init__()
2090 self._parents: list[ObjectID] = []
2091 self._encoding: bytes | None = None
2092 self._mergetag: list[Tag] = []
2093 self._gpgsig: bytes | None = None
2094 self._extra: list[tuple[bytes, bytes]] = []
2095 self._author_timezone_neg_utc: bool | None = False
2096 self._commit_timezone_neg_utc: bool | None = False
2098 @classmethod
2099 def from_path(
2100 cls,
2101 path: str | bytes,
2102 sha: ObjectID | None = None,
2103 *,
2104 object_format: ObjectFormat | None = None,
2105 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
2106 ) -> "Commit":
2107 """Read a commit from a file on disk.
2109 Args:
2110 path: Path to the commit file
2111 sha: Optional known SHA for the object
2112 object_format: Optional object format to use
2113 max_size: Maximum inflated size in bytes
2115 Returns:
2116 A Commit object
2118 Raises:
2119 NotCommitError: If the file is not a commit
2120 """
2121 commit = ShaFile.from_path(
2122 path, sha, object_format=object_format, max_size=max_size
2123 )
2124 if not isinstance(commit, cls):
2125 raise NotCommitError(_path_to_bytes(path))
2126 return commit
2128 def _deserialize(self, chunks: list[bytes]) -> None:
2129 (
2130 tree,
2131 parents,
2132 author_info,
2133 commit_info,
2134 encoding,
2135 mergetag,
2136 gpgsig,
2137 message,
2138 extra,
2139 ) = _parse_commit(chunks)
2141 self._tree = tree
2142 self._parents = [ObjectID(p) for p in parents]
2143 self._encoding = encoding
2144 self._mergetag = mergetag
2145 self._gpgsig = gpgsig
2146 self._message = message
2147 self._extra = extra
2149 (
2150 self._author,
2151 self._author_time,
2152 (self._author_timezone, self._author_timezone_neg_utc),
2153 ) = author_info
2154 (
2155 self._committer,
2156 self._commit_time,
2157 (self._commit_timezone, self._commit_timezone_neg_utc),
2158 ) = commit_info
2160 def check(self) -> None:
2161 """Check this object for internal consistency.
2163 Raises:
2164 ObjectFormatException: if the object is malformed in some way
2165 """
2166 super().check()
2167 assert self._chunked_text is not None
2168 self._check_has_member("_tree", "missing tree")
2169 self._check_has_member("_author", "missing author")
2170 self._check_has_member("_committer", "missing committer")
2171 self._check_has_member("_author_time", "missing author time")
2172 self._check_has_member("_commit_time", "missing commit time")
2174 for parent in self._parents:
2175 check_hexsha(parent, "invalid parent sha")
2176 assert self._tree is not None # checked by _check_has_member above
2177 check_hexsha(self._tree, "invalid tree sha")
2179 assert self._author is not None # checked by _check_has_member above
2180 assert self._committer is not None # checked by _check_has_member above
2181 check_identity(self._author, "invalid author")
2182 check_identity(self._committer, "invalid committer")
2184 assert self._author_time is not None # checked by _check_has_member above
2185 assert self._commit_time is not None # checked by _check_has_member above
2186 check_time(self._author_time)
2187 check_time(self._commit_time)
2189 last = None
2190 for field, _ in _parse_message(self._chunked_text):
2191 if field == _TREE_HEADER and last is not None:
2192 raise ObjectFormatException("unexpected tree")
2193 elif field == _PARENT_HEADER and last not in (
2194 _PARENT_HEADER,
2195 _TREE_HEADER,
2196 ):
2197 raise ObjectFormatException("unexpected parent")
2198 elif field == _AUTHOR_HEADER and last not in (
2199 _TREE_HEADER,
2200 _PARENT_HEADER,
2201 ):
2202 raise ObjectFormatException("unexpected author")
2203 elif field == _COMMITTER_HEADER and last != _AUTHOR_HEADER:
2204 raise ObjectFormatException("unexpected committer")
2205 elif field == _ENCODING_HEADER and last != _COMMITTER_HEADER:
2206 raise ObjectFormatException("unexpected encoding")
2207 last = field
2209 # TODO: optionally check for duplicate parents
2211 def raw_without_sig(self) -> bytes:
2212 """Return raw string serialization without the GPG/SSH signature.
2214 self.gpgsig is a signature for the returned raw byte string serialization.
2215 """
2216 tmp = self.copy()
2217 assert isinstance(tmp, Commit)
2218 tmp._gpgsig = None
2219 tmp.gpgsig = None
2220 return tmp.as_raw_string()
2222 def extract_signature(self) -> tuple[bytes, bytes | None, bytes | None]:
2223 """Extract the payload, signature, and signature type from this commit.
2225 Returns:
2226 tuple of (``payload``, ``signature``, ``signature_type``) where:
2228 - ``payload``: The raw commit data without the signature
2229 - ``signature``: The signature bytes if present, None otherwise
2230 - ``signature_type``: SIGNATURE_PGP for PGP, SIGNATURE_SSH for SSH, None if no signature
2232 Raises:
2233 ObjectFormatException: If signature has unknown format
2234 """
2235 if self._gpgsig is None:
2236 return self.as_raw_string(), None, None
2238 payload = self.raw_without_sig()
2240 # Determine signature type
2241 if self._gpgsig.startswith(BEGIN_PGP_SIGNATURE):
2242 sig_type = SIGNATURE_PGP
2243 elif self._gpgsig.startswith(BEGIN_SSH_SIGNATURE):
2244 sig_type = SIGNATURE_SSH
2245 else:
2246 raise ObjectFormatException("Unknown signature format")
2248 return payload, self._gpgsig, sig_type
2250 def _serialize(self) -> list[bytes]:
2251 headers = []
2252 assert self._tree is not None
2253 tree_bytes = self._tree.id if isinstance(self._tree, Tree) else self._tree
2254 headers.append((_TREE_HEADER, tree_bytes))
2255 for p in self._parents:
2256 headers.append((_PARENT_HEADER, p))
2257 assert self._author is not None
2258 assert self._author_time is not None
2259 assert self._author_timezone is not None
2260 assert self._author_timezone_neg_utc is not None
2261 headers.append(
2262 (
2263 _AUTHOR_HEADER,
2264 format_time_entry(
2265 self._author,
2266 self._author_time,
2267 (self._author_timezone, self._author_timezone_neg_utc),
2268 ),
2269 )
2270 )
2271 assert self._committer is not None
2272 assert self._commit_time is not None
2273 assert self._commit_timezone is not None
2274 assert self._commit_timezone_neg_utc is not None
2275 headers.append(
2276 (
2277 _COMMITTER_HEADER,
2278 format_time_entry(
2279 self._committer,
2280 self._commit_time,
2281 (self._commit_timezone, self._commit_timezone_neg_utc),
2282 ),
2283 )
2284 )
2285 if self.encoding:
2286 headers.append((_ENCODING_HEADER, self.encoding))
2287 for mergetag in self.mergetag:
2288 headers.append((_MERGETAG_HEADER, mergetag.as_raw_string()[:-1]))
2289 headers.extend(
2290 (field, value) for field, value in self._extra if value is not None
2291 )
2292 if self.gpgsig:
2293 headers.append((_GPGSIG_HEADER, self.gpgsig))
2294 return list(_format_message(headers, self._message))
2296 tree = serializable_property("tree", "Tree that is the state of this commit")
2298 @property
2299 def parents(self) -> list[ObjectID]:
2300 """Parents of this commit, by their SHA1."""
2301 return self._parents
2303 @parents.setter
2304 def parents(self, value: list[ObjectID]) -> None:
2305 """Set a list of parents of this commit."""
2306 self._needs_serialization = True
2307 self._parents = value
2309 author = serializable_property("author", "The name of the author of the commit")
2311 committer = serializable_property(
2312 "committer", "The name of the committer of the commit"
2313 )
2315 message = serializable_property("message", "The commit message")
2317 commit_time = serializable_property(
2318 "commit_time",
2319 "The timestamp of the commit. As the number of seconds since the epoch.",
2320 )
2322 commit_timezone = serializable_property(
2323 "commit_timezone", "The zone the commit time is in"
2324 )
2326 author_time = serializable_property(
2327 "author_time",
2328 "The timestamp the commit was written. As the number of "
2329 "seconds since the epoch.",
2330 )
2332 author_timezone = serializable_property(
2333 "author_timezone", "Returns the zone the author time is in."
2334 )
2336 encoding = serializable_property("encoding", "Encoding of the commit message.")
2338 mergetag = serializable_property("mergetag", "Associated signed tag.")
2340 gpgsig = serializable_property("gpgsig", "GPG Signature.")
2343OBJECT_CLASSES = (
2344 Commit,
2345 Tree,
2346 Blob,
2347 Tag,
2348)
2350_TYPE_MAP: dict[bytes | int, type[ShaFile]] = {}
2352for cls in OBJECT_CLASSES:
2353 _TYPE_MAP[cls.type_name] = cls
2354 _TYPE_MAP[cls.type_num] = cls
2357# Public API functions
2360def parse_commit_broken(data: bytes) -> Commit:
2361 """Parse a commit with broken author/committer lines.
2363 This function handles various broken formats found in the wild:
2364 - Missing angle brackets around email addresses
2365 - Unsigned timezones (e.g., "0000" instead of "+0000")
2366 - Double-negative timezones (e.g., "--700")
2367 - Negative timestamps
2368 - Long/short/nonsensical timezone values
2370 Warning: Commits parsed with this function may not round-trip correctly
2371 through serialization, as the broken formatting is normalized during parsing.
2372 The .check() method will likely fail for commits with malformed identity fields.
2374 Args:
2375 data: Raw commit data as bytes
2377 Returns:
2378 A Commit object with normalized fields
2380 Example:
2381 >>> data = b'''tree d80c186a03f423a81b39df39dc87fd269736ca86
2382 ... author user@example.com 1234567890 -0500
2383 ... committer user@example.com 1234567890 -0500
2384 ...
2385 ... Commit message
2386 ... '''
2387 >>> commit = parse_commit_broken(data)
2388 >>> commit.author
2389 b'user@example.com'
2390 """
2391 commit = Commit()
2392 (
2393 tree,
2394 parents,
2395 author_info,
2396 commit_info,
2397 encoding,
2398 mergetag,
2399 gpgsig,
2400 message,
2401 extra,
2402 ) = _parse_commit_broken([data])
2404 commit._tree = tree
2405 commit._parents = [ObjectID(p) for p in parents]
2406 commit._encoding = encoding
2407 commit._mergetag = mergetag
2408 commit._gpgsig = gpgsig
2409 commit._message = message
2410 commit._extra = extra
2412 (
2413 commit._author,
2414 commit._author_time,
2415 (commit._author_timezone, commit._author_timezone_neg_utc),
2416 ) = author_info
2417 (
2418 commit._committer,
2419 commit._commit_time,
2420 (commit._commit_timezone, commit._commit_timezone_neg_utc),
2421 ) = commit_info
2423 return commit
2426# Hold on to the pure-python implementations for testing
2427_parse_tree_py = parse_tree
2428_sorted_tree_items_py = sorted_tree_items
2429try:
2430 # Try to import Rust versions
2431 from dulwich._objects import (
2432 parse_tree as _parse_tree_rs,
2433 )
2434 from dulwich._objects import (
2435 sorted_tree_items as _sorted_tree_items_rs,
2436 )
2437except ImportError:
2438 pass
2439else:
2440 parse_tree = _parse_tree_rs
2441 sorted_tree_items = _sorted_tree_items_rs