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 "MAX_TIME",
29 "OBJECT_CLASSES",
30 "SIGNATURE_PGP",
31 "SIGNATURE_SSH",
32 "S_IFGITLINK",
33 "S_ISGITLINK",
34 "ZERO_SHA",
35 "Blob",
36 "Commit",
37 "EmptyFileException",
38 "FixedSha",
39 "ObjectID",
40 "RawObjectID",
41 "ShaFile",
42 "SubmoduleEncountered",
43 "Tag",
44 "Tree",
45 "TreeEntry",
46 "check_hexsha",
47 "check_identity",
48 "check_time",
49 "filename_to_hex",
50 "format_time_entry",
51 "format_timezone",
52 "git_line",
53 "hex_to_filename",
54 "hex_to_sha",
55 "is_blob",
56 "is_commit",
57 "is_tag",
58 "is_tree",
59 "key_entry",
60 "key_entry_name_order",
61 "object_class",
62 "object_header",
63 "parse_commit_broken",
64 "parse_tree",
65 "pretty_format_tree_entry",
66 "serializable_property",
67 "serialize_tree",
68 "sha_to_hex",
69 "sorted_tree_items",
70 "valid_hexsha",
71]
73import binascii
74import os
75import posixpath
76import re
77import stat
78import sys
79import zlib
80from collections.abc import Callable, Iterable, Iterator, Sequence
81from hashlib import sha1
82from io import BufferedIOBase, BytesIO
83from typing import (
84 IO,
85 TYPE_CHECKING,
86 NamedTuple,
87 TypeVar,
88)
90if sys.version_info >= (3, 11):
91 from typing import Self
92else:
93 from typing_extensions import Self
95from typing import NewType, TypeGuard
97from .errors import (
98 ChecksumMismatch,
99 FileFormatException,
100 NotBlobError,
101 NotCommitError,
102 NotTagError,
103 NotTreeError,
104 ObjectFormatException,
105)
106from .file import GitFile
107from .object_format import DEFAULT_OBJECT_FORMAT, ObjectFormat
109if TYPE_CHECKING:
110 from _hashlib import HASH
112 from .file import _GitFile
114# Zero SHA constants for backward compatibility - now defined below as ObjectID
117# Header fields for commits
118_TREE_HEADER = b"tree"
119_PARENT_HEADER = b"parent"
120_AUTHOR_HEADER = b"author"
121_COMMITTER_HEADER = b"committer"
122_ENCODING_HEADER = b"encoding"
123_MERGETAG_HEADER = b"mergetag"
124_GPGSIG_HEADER = b"gpgsig"
126# Header fields for objects
127_OBJECT_HEADER = b"object"
128_TYPE_HEADER = b"type"
129_TAG_HEADER = b"tag"
130_TAGGER_HEADER = b"tagger"
133S_IFGITLINK = 0o160000
135# Intentionally flexible regex to support various types of brokenness
136# in commit/tag author/committer/tagger lines
137_TIME_ENTRY_RE = re.compile(
138 b"^(?P<person>.*) (?P<time>-?[0-9]+) (?P<timezone>[+-]{0,2}[0-9]+)$"
139)
142MAX_TIME = 9223372036854775807 # (2**63) - 1 - signed long int max
144BEGIN_PGP_SIGNATURE = b"-----BEGIN PGP SIGNATURE-----"
145BEGIN_SSH_SIGNATURE = b"-----BEGIN SSH SIGNATURE-----"
147# Signature type constants
148SIGNATURE_PGP = b"pgp"
149SIGNATURE_SSH = b"ssh"
152# Hex SHA type
153ObjectID = NewType("ObjectID", bytes)
155# Raw SHA type
156RawObjectID = NewType("RawObjectID", bytes)
158# Zero SHA constant
159ZERO_SHA: ObjectID = ObjectID(b"0" * 40)
162class EmptyFileException(FileFormatException):
163 """An unexpectedly empty file was encountered."""
166def S_ISGITLINK(m: int) -> bool:
167 """Check if a mode indicates a submodule.
169 Args:
170 m: Mode to check
171 Returns: a ``boolean``
172 """
173 return stat.S_IFMT(m) == S_IFGITLINK
176# Git's core.bigFileThreshold default.
177DEFAULT_LOOSE_OBJECT_SIZE_LIMIT = 512 * 1024 * 1024
180def _decompress(
181 string: bytes, max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT
182) -> bytes:
183 """Decompress a zlib-compressed loose object, bounded by max_size.
185 Raises:
186 zlib.error: if the decompressed data would exceed max_size bytes,
187 or on any other decompression error.
188 """
189 dcomp = zlib.decompressobj()
190 # +1 so overrun surfaces as unconsumed_tail rather than being truncated.
191 dcomped = dcomp.decompress(string, max_size + 1)
192 if dcomp.unconsumed_tail:
193 raise zlib.error("decompressed data exceeds maximum size")
194 dcomped += dcomp.flush()
195 if len(dcomped) > max_size:
196 raise zlib.error("decompressed data exceeds maximum size")
197 return dcomped
200def sha_to_hex(sha: RawObjectID) -> ObjectID:
201 """Takes a string and returns the hex of the sha within."""
202 hexsha = binascii.hexlify(sha)
203 # Support both SHA1 (40 chars) and SHA256 (64 chars)
204 if len(hexsha) not in (40, 64):
205 raise ValueError(f"Incorrect length of sha string: {hexsha!r}")
206 return ObjectID(hexsha)
209def hex_to_sha(hex: ObjectID | str) -> RawObjectID:
210 """Takes a hex sha and returns a binary sha."""
211 # Support both SHA1 (40 chars) and SHA256 (64 chars)
212 if len(hex) not in (40, 64):
213 raise ValueError(f"Incorrect length of hexsha: {hex!r}")
214 try:
215 return RawObjectID(binascii.unhexlify(hex))
216 except TypeError as exc:
217 if not isinstance(hex, bytes):
218 raise
219 raise ValueError(exc.args[0]) from exc
222def valid_hexsha(hex: bytes | str) -> bool:
223 """Check if a hex string is a valid SHA1 or SHA256.
225 Args:
226 hex: Hex string to validate
228 Returns:
229 True if valid SHA1 (40 chars) or SHA256 (64 chars), False otherwise
230 """
231 if len(hex) not in (40, 64):
232 return False
233 try:
234 binascii.unhexlify(hex)
235 except (TypeError, binascii.Error):
236 return False
237 else:
238 return True
241PathT = TypeVar("PathT", str, bytes)
244def hex_to_filename(path: PathT, hex: ObjectID | str) -> PathT:
245 """Takes a hex sha and returns its filename relative to the given path."""
246 # os.path.join accepts bytes or unicode, but all args must be of the same
247 # type. Make sure that hex which is expected to be bytes, is the same type
248 # as path.
249 if isinstance(path, str):
250 if isinstance(hex, bytes):
251 hex_str = hex.decode("ascii")
252 else:
253 hex_str = hex
254 dir_name = hex_str[:2]
255 file_name = hex_str[2:]
256 result = os.path.join(path, dir_name, file_name)
257 assert isinstance(result, str)
258 return result
259 else:
260 # path is bytes
261 if isinstance(hex, str):
262 hex_bytes = hex.encode("ascii")
263 else:
264 hex_bytes = hex
265 dir_name_b = hex_bytes[:2]
266 file_name_b = hex_bytes[2:]
267 result_b = os.path.join(path, dir_name_b, file_name_b)
268 assert isinstance(result_b, bytes)
269 return result_b
272def filename_to_hex(filename: str | bytes) -> str:
273 """Takes an object filename and returns its corresponding hex sha."""
274 import warnings
276 warnings.warn(
277 "filename_to_hex is unused and will be removed in a future version of Dulwich",
278 DeprecationWarning,
279 )
280 # grab the last (up to) two path components
281 errmsg = f"Invalid object filename: {filename!r}"
282 if isinstance(filename, str):
283 names = filename.rsplit(os.path.sep, 2)[-2:]
284 assert len(names) == 2, errmsg
285 base, rest = names
286 assert len(base) == 2 and len(rest) == 38, errmsg
287 hex_str = base + rest
288 hex_bytes = hex_str.encode("ascii")
289 else:
290 # filename is bytes
291 sep = (
292 os.path.sep.encode("ascii") if isinstance(os.path.sep, str) else os.path.sep
293 )
294 names_b = filename.rsplit(sep, 2)[-2:]
295 assert len(names_b) == 2, errmsg
296 base_b, rest_b = names_b
297 assert len(base_b) == 2 and len(rest_b) == 38, errmsg
298 hex_bytes = base_b + rest_b
299 hex_to_sha(ObjectID(hex_bytes))
300 return hex_bytes.decode("ascii")
303def object_header(num_type: int, length: int) -> bytes:
304 """Return an object header for the given numeric type and text length."""
305 cls = object_class(num_type)
306 if cls is None:
307 raise AssertionError(f"unsupported class type num: {num_type}")
308 return cls.type_name + b" " + str(length).encode("ascii") + b"\0"
311def serializable_property(name: str, docstring: str | None = None) -> property:
312 """A property that helps tracking whether serialization is necessary."""
314 def set(obj: "ShaFile", value: object) -> None:
315 """Set the property value and mark the object as needing serialization.
317 Args:
318 obj: The ShaFile object
319 value: The value to set
320 """
321 setattr(obj, "_" + name, value)
322 obj._needs_serialization = True
324 def get(obj: "ShaFile") -> object:
325 """Get the property value.
327 Args:
328 obj: The ShaFile object
330 Returns:
331 The property value
332 """
333 return getattr(obj, "_" + name)
335 return property(get, set, doc=docstring)
338def object_class(type: bytes | int) -> type["ShaFile"] | None:
339 """Get the object class corresponding to the given type.
341 Args:
342 type: Either a type name string or a numeric type.
343 Returns: The ShaFile subclass corresponding to the given type, or None if
344 type is not a valid type name/number.
345 """
346 return _TYPE_MAP.get(type, None)
349def check_hexsha(hex: str | bytes, error_msg: str) -> None:
350 """Check if a string is a valid hex sha string.
352 Args:
353 hex: Hex string to check
354 error_msg: Error message to use in exception
355 Raises:
356 ObjectFormatException: Raised when the string is not valid
357 """
358 if not valid_hexsha(hex):
359 raise ObjectFormatException(f"{error_msg} {hex!r}")
362def check_identity(identity: bytes | None, error_msg: str) -> None:
363 """Check if the specified identity is valid.
365 This will raise an exception if the identity is not valid.
367 Args:
368 identity: Identity string
369 error_msg: Error message to use in exception
370 """
371 if identity is None:
372 raise ObjectFormatException(error_msg)
373 email_start = identity.find(b"<")
374 email_end = identity.find(b">")
375 if not all(
376 [
377 email_start >= 1,
378 identity[email_start - 1] == b" "[0],
379 identity.find(b"<", email_start + 1) == -1,
380 email_end == len(identity) - 1,
381 b"\0" not in identity,
382 b"\n" not in identity,
383 ]
384 ):
385 raise ObjectFormatException(error_msg)
388def _path_to_bytes(path: str | bytes) -> bytes:
389 """Convert a path to bytes for use in error messages."""
390 if isinstance(path, str):
391 return path.encode("utf-8", "surrogateescape")
392 return path
395def check_time(time_seconds: int) -> None:
396 """Check if the specified time is not prone to overflow error.
398 This will raise an exception if the time is not valid.
400 Args:
401 time_seconds: time in seconds
403 """
404 # Prevent overflow error
405 if time_seconds > MAX_TIME:
406 raise ObjectFormatException(f"Date field should not exceed {MAX_TIME}")
409def git_line(*items: bytes) -> bytes:
410 """Formats items into a space separated line."""
411 return b" ".join(items) + b"\n"
414class FixedSha:
415 """SHA object that behaves like hashlib's but is given a fixed value."""
417 __slots__ = ("_hexsha", "_sha")
419 def __init__(self, hexsha: str | bytes) -> None:
420 """Initialize FixedSha with a fixed SHA value.
422 Args:
423 hexsha: Hex SHA value as string or bytes
424 """
425 if isinstance(hexsha, str):
426 hexsha = hexsha.encode("ascii")
427 if not isinstance(hexsha, bytes):
428 raise TypeError(f"Expected bytes for hexsha, got {hexsha!r}")
429 self._hexsha = hexsha
430 self._sha = hex_to_sha(ObjectID(hexsha))
432 def digest(self) -> bytes:
433 """Return the raw SHA digest."""
434 return self._sha
436 def hexdigest(self) -> str:
437 """Return the hex SHA digest."""
438 return self._hexsha.decode("ascii")
441# Type guard functions for runtime type narrowing
442if TYPE_CHECKING:
444 def is_commit(obj: "ShaFile") -> TypeGuard["Commit"]:
445 """Check if a ShaFile is a Commit."""
446 return obj.type_name == b"commit"
448 def is_tree(obj: "ShaFile") -> TypeGuard["Tree"]:
449 """Check if a ShaFile is a Tree."""
450 return obj.type_name == b"tree"
452 def is_blob(obj: "ShaFile") -> TypeGuard["Blob"]:
453 """Check if a ShaFile is a Blob."""
454 return obj.type_name == b"blob"
456 def is_tag(obj: "ShaFile") -> TypeGuard["Tag"]:
457 """Check if a ShaFile is a Tag."""
458 return obj.type_name == b"tag"
459else:
460 # Runtime versions without type narrowing
461 def is_commit(obj: "ShaFile") -> bool:
462 """Check if a ShaFile is a Commit."""
463 return obj.type_name == b"commit"
465 def is_tree(obj: "ShaFile") -> bool:
466 """Check if a ShaFile is a Tree."""
467 return obj.type_name == b"tree"
469 def is_blob(obj: "ShaFile") -> bool:
470 """Check if a ShaFile is a Blob."""
471 return obj.type_name == b"blob"
473 def is_tag(obj: "ShaFile") -> bool:
474 """Check if a ShaFile is a Tag."""
475 return obj.type_name == b"tag"
478class ShaFile:
479 """A git SHA file."""
481 __slots__ = ("_chunked_text", "_needs_serialization", "_sha", "object_format")
483 _needs_serialization: bool
484 type_name: bytes
485 type_num: int
486 _chunked_text: list[bytes] | None
487 _sha: "FixedSha | None | HASH"
488 object_format: ObjectFormat
490 def __init__(self) -> None:
491 """Initialize a ShaFile."""
492 self._sha = None
493 self._chunked_text = None
494 self._needs_serialization = True
495 self.object_format = DEFAULT_OBJECT_FORMAT
497 @staticmethod
498 def _parse_legacy_object_header(magic: bytes) -> "ShaFile":
499 """Parse a legacy object header, creating it but not reading the file."""
500 # Legitimate loose object headers are at most a few dozen bytes; a
501 # larger inflated prefix indicates corruption or a decompression bomb.
502 header_max = 8192
503 decomp = zlib.decompressobj()
504 # Inflate only the header prefix. The object content may be far larger
505 # than header_max, so we cap the output and keep pulling from
506 # unconsumed_tail until we find the NUL that terminates the header.
507 header = decomp.decompress(magic, header_max)
508 start = 0
509 end = header.find(b"\0", start)
510 start = len(header)
511 while end < 0:
512 if len(header) >= header_max:
513 raise zlib.error("object header exceeds maximum size")
514 if not decomp.unconsumed_tail:
515 raise ObjectFormatException("Invalid object header, no \\0")
516 header += decomp.decompress(
517 decomp.unconsumed_tail, header_max - len(header)
518 )
519 end = header.find(b"\0", start)
520 start = len(header)
521 header = header[:end]
522 type_name, size = header.split(b" ", 1)
523 try:
524 int(size) # sanity check
525 except ValueError as exc:
526 raise ObjectFormatException(f"Object size not an integer: {exc}") from exc
527 obj_class = object_class(type_name)
528 if not obj_class:
529 raise ObjectFormatException(
530 "Not a known type: {}".format(type_name.decode("ascii"))
531 )
532 return obj_class()
534 def _parse_legacy_object(
535 self,
536 map: bytes,
537 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
538 ) -> None:
539 """Parse a legacy object, setting the raw string."""
540 text = _decompress(map, max_size=max_size)
541 header_end = text.find(b"\0")
542 if header_end < 0:
543 raise ObjectFormatException("Invalid object header, no \\0")
544 self.set_raw_string(text[header_end + 1 :])
546 def as_legacy_object_chunks(self, compression_level: int = -1) -> Iterator[bytes]:
547 """Return chunks representing the object in the experimental format.
549 Returns: List of strings
550 """
551 compobj = zlib.compressobj(compression_level)
552 yield compobj.compress(self._header())
553 for chunk in self.as_raw_chunks():
554 yield compobj.compress(chunk)
555 yield compobj.flush()
557 def as_legacy_object(self, compression_level: int = -1) -> bytes:
558 """Return string representing the object in the experimental format."""
559 return b"".join(
560 self.as_legacy_object_chunks(compression_level=compression_level)
561 )
563 def as_raw_chunks(self) -> list[bytes]:
564 """Return chunks with serialization of the object.
566 Returns: List of strings, not necessarily one per line
567 """
568 if self._needs_serialization:
569 self._sha = None
570 self._chunked_text = self._serialize()
571 self._needs_serialization = False
572 assert self._chunked_text is not None
573 return self._chunked_text
575 def as_raw_string(self) -> bytes:
576 """Return raw string with serialization of the object.
578 Returns: String object
579 """
580 return b"".join(self.as_raw_chunks())
582 def __bytes__(self) -> bytes:
583 """Return raw string serialization of this object."""
584 return self.as_raw_string()
586 def __hash__(self) -> int:
587 """Return unique hash for this object."""
588 return hash(self.id)
590 def as_pretty_string(self) -> str:
591 """Return a string representing this object, fit for display."""
592 return self.as_raw_string().decode("utf-8", "replace")
594 def set_raw_string(
595 self,
596 text: bytes,
597 sha: ObjectID | RawObjectID | None = None,
598 *,
599 verify_sha: ObjectID | RawObjectID | None = None,
600 ) -> None:
601 """Set the contents of this object from a serialized string."""
602 if not isinstance(text, bytes):
603 raise TypeError(f"Expected bytes for text, got {text!r}")
604 self.set_raw_chunks([text], sha, verify_sha=verify_sha)
606 def set_raw_chunks(
607 self,
608 chunks: list[bytes],
609 sha: ObjectID | RawObjectID | None = None,
610 *,
611 object_format: ObjectFormat | None = None,
612 verify_sha: ObjectID | RawObjectID | None = None,
613 ) -> None:
614 """Set the contents of this object from a list of chunks.
616 Args:
617 chunks: The raw uncompressed contents.
618 sha: Optional known, trusted sha for the object. Cached without
619 being checked against the contents.
620 object_format: Optional hash algorithm for the object.
621 verify_sha: Optional sha that the contents are checked against;
622 raises ChecksumMismatch if the object does not hash to it. On
623 success it is cached like ``sha``. Mutually exclusive with ``sha``.
624 """
625 if sha is not None and verify_sha is not None:
626 raise ValueError("sha and verify_sha are mutually exclusive")
627 self._chunked_text = chunks
628 # Set hash algorithm if provided
629 if object_format is not None:
630 self.object_format = object_format
631 # Set SHA before deserialization so Tree can use hash algorithm
632 if sha is None:
633 self._sha = None
634 else:
635 self._sha = FixedSha(sha)
636 self._deserialize(chunks)
637 self._needs_serialization = False
638 if verify_sha is not None:
639 got = self.get_id(self.object_format)
640 if got != verify_sha:
641 raise ChecksumMismatch(verify_sha, got)
642 self._sha = FixedSha(verify_sha)
644 @staticmethod
645 def _parse_object_header(
646 magic: bytes, f: BufferedIOBase | IO[bytes] | "_GitFile"
647 ) -> "ShaFile":
648 """Parse a new style object, creating it but not reading the file."""
649 num_type = (ord(magic[0:1]) >> 4) & 7
650 obj_class = object_class(num_type)
651 if not obj_class:
652 raise ObjectFormatException(f"Not a known type {num_type}")
653 return obj_class()
655 def _parse_object(
656 self,
657 map: bytes,
658 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
659 ) -> None:
660 """Parse a new style object, setting self._text."""
661 # skip type and size; type must have already been determined, and
662 # we trust zlib to fail if it's otherwise corrupted
663 byte = ord(map[0:1])
664 used = 1
665 while (byte & 0x80) != 0:
666 byte = ord(map[used : used + 1])
667 used += 1
668 raw = map[used:]
669 self.set_raw_string(_decompress(raw, max_size=max_size))
671 @classmethod
672 def _is_legacy_object(cls, magic: bytes) -> bool:
673 b0 = ord(magic[0:1])
674 b1 = ord(magic[1:2])
675 word = (b0 << 8) + b1
676 return (b0 & 0x8F) == 0x08 and (word % 31) == 0
678 @classmethod
679 def _parse_file(
680 cls,
681 f: BufferedIOBase | IO[bytes] | "_GitFile",
682 *,
683 object_format: ObjectFormat | None = None,
684 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
685 ) -> "ShaFile":
686 map = f.read()
687 if not map:
688 raise EmptyFileException("Corrupted empty file detected")
690 if cls._is_legacy_object(map):
691 obj = cls._parse_legacy_object_header(map)
692 if object_format is not None:
693 obj.object_format = object_format
694 obj._parse_legacy_object(map, max_size=max_size)
695 else:
696 obj = cls._parse_object_header(map, f)
697 if object_format is not None:
698 obj.object_format = object_format
699 obj._parse_object(map, max_size=max_size)
700 return obj
702 def _deserialize(self, chunks: list[bytes]) -> None:
703 raise NotImplementedError(self._deserialize)
705 def _serialize(self) -> list[bytes]:
706 raise NotImplementedError(self._serialize)
708 @classmethod
709 def from_path(
710 cls,
711 path: str | bytes,
712 sha: ObjectID | None = None,
713 *,
714 object_format: ObjectFormat | None = None,
715 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
716 ) -> "ShaFile":
717 """Open a SHA file from disk."""
718 with GitFile(path, "rb") as f:
719 return cls.from_file(f, sha, object_format=object_format, max_size=max_size)
721 @classmethod
722 def from_file(
723 cls,
724 f: BufferedIOBase | IO[bytes] | "_GitFile",
725 sha: ObjectID | None = None,
726 *,
727 object_format: ObjectFormat | None = None,
728 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
729 ) -> "ShaFile":
730 """Get the contents of a SHA file on disk."""
731 try:
732 # Validate SHA length matches hash algorithm if both provided
733 if sha is not None and object_format is not None:
734 expected_len = object_format.hex_length
735 if len(sha) != expected_len:
736 raise ValueError(
737 f"SHA length {len(sha)} doesn't match hash algorithm "
738 f"{object_format.name} (expected {expected_len})"
739 )
741 obj = cls._parse_file(f, object_format=object_format, max_size=max_size)
742 if sha is not None:
743 obj._sha = FixedSha(sha)
744 else:
745 obj._sha = None
746 return obj
747 except (IndexError, ValueError) as exc:
748 raise ObjectFormatException("invalid object header") from exc
750 @staticmethod
751 def from_raw_string(
752 type_num: int,
753 string: bytes,
754 sha: ObjectID | RawObjectID | None = None,
755 *,
756 object_format: ObjectFormat | None = None,
757 verify_sha: ObjectID | RawObjectID | None = None,
758 ) -> "ShaFile":
759 """Creates an object of the indicated type from the raw string given.
761 Args:
762 type_num: The numeric type of the object.
763 string: The raw uncompressed contents.
764 sha: Optional known, trusted sha for the object.
765 object_format: Optional hash algorithm for the object
766 verify_sha: Optional sha to check the contents against; raises
767 ChecksumMismatch on mismatch. Mutually exclusive with ``sha``.
768 """
769 cls = object_class(type_num)
770 if cls is None:
771 raise AssertionError(f"unsupported class type num: {type_num}")
772 obj = cls()
773 if object_format is not None:
774 obj.object_format = object_format
775 obj.set_raw_string(string, sha, verify_sha=verify_sha)
776 return obj
778 @staticmethod
779 def from_raw_chunks(
780 type_num: int,
781 chunks: list[bytes],
782 sha: ObjectID | RawObjectID | None = None,
783 *,
784 object_format: ObjectFormat | None = None,
785 ) -> "ShaFile":
786 """Creates an object of the indicated type from the raw chunks given.
788 Args:
789 type_num: The numeric type of the object.
790 chunks: An iterable of the raw uncompressed contents.
791 sha: Optional known sha for the object
792 object_format: Optional object format (hash algorithm) for the object.
793 Required for trees in SHA-256 repositories so entry parsing uses
794 the correct OID length.
795 """
796 cls = object_class(type_num)
797 if cls is None:
798 raise AssertionError(f"unsupported class type num: {type_num}")
799 obj = cls()
800 obj.set_raw_chunks(chunks, sha, object_format=object_format)
801 return obj
803 @classmethod
804 def from_string(cls, string: bytes) -> Self:
805 """Create a ShaFile from a string."""
806 obj = cls()
807 obj.set_raw_string(string)
808 return obj
810 def _check_has_member(self, member: str, error_msg: str) -> None:
811 """Check that the object has a given member variable.
813 Args:
814 member: the member variable to check for
815 error_msg: the message for an error if the member is missing
816 Raises:
817 ObjectFormatException: with the given error_msg if member is
818 missing or is None
819 """
820 if getattr(self, member, None) is None:
821 raise ObjectFormatException(error_msg)
823 def check(self) -> None:
824 """Check this object for internal consistency.
826 Raises:
827 ObjectFormatException: if the object is malformed in some way
828 ChecksumMismatch: if the object was created with a SHA that does
829 not match its contents
830 """
831 # TODO: if we find that error-checking during object parsing is a
832 # performance bottleneck, those checks should be moved to the class's
833 # check() method during optimization so we can still check the object
834 # when necessary.
835 old_sha = self.id
836 try:
837 self._deserialize(self.as_raw_chunks())
838 self._sha = None
839 new_sha = self.id
840 except Exception as exc:
841 raise ObjectFormatException(exc) from exc
842 if old_sha != new_sha:
843 raise ChecksumMismatch(new_sha, old_sha)
845 def _header(self) -> bytes:
846 return object_header(self.type_num, self.raw_length())
848 def raw_length(self) -> int:
849 """Returns the length of the raw string of this object."""
850 return sum(map(len, self.as_raw_chunks()))
852 def sha(self, object_format: ObjectFormat | None = None) -> "FixedSha | HASH":
853 """The SHA object that is the name of this object.
855 Args:
856 object_format: Optional HashAlgorithm to use. Defaults to SHA1.
857 """
858 # If using a different hash algorithm, always recalculate
859 if object_format is not None:
860 new_sha = object_format.new_hash()
861 new_sha.update(self._header())
862 for chunk in self.as_raw_chunks():
863 new_sha.update(chunk)
864 return new_sha
866 # Otherwise use cached SHA1 value
867 if self._sha is None or self._needs_serialization:
868 # this is a local because as_raw_chunks() overwrites self._sha
869 new_sha = sha1()
870 new_sha.update(self._header())
871 for chunk in self.as_raw_chunks():
872 new_sha.update(chunk)
873 self._sha = new_sha
874 return self._sha
876 def copy(self) -> "ShaFile":
877 """Create a new copy of this SHA1 object from its raw string."""
878 obj_class = object_class(self.type_num)
879 if obj_class is None:
880 raise AssertionError(f"invalid type num {self.type_num}")
881 return obj_class.from_raw_string(self.type_num, self.as_raw_string(), self.id)
883 @property
884 def id(self) -> ObjectID:
885 """The hex SHA1 of this object.
887 For SHA256 repositories, use get_id(object_format) instead.
888 This property always returns SHA1 for backward compatibility.
889 """
890 return ObjectID(self.sha().hexdigest().encode("ascii"))
892 def get_id(self, object_format: ObjectFormat | None = None) -> bytes:
893 """Get the hex SHA of this object using the specified hash algorithm.
895 Args:
896 object_format: Optional HashAlgorithm to use. Defaults to SHA1.
898 Example:
899 >>> blob = Blob()
900 >>> blob.data = b"Hello, World!"
901 >>> blob.id # Always returns SHA1 for backward compatibility
902 b'4ab299c8ad6ed14f31923dd94f8b5f5cb89dfb54'
903 >>> blob.get_id() # Same as .id
904 b'4ab299c8ad6ed14f31923dd94f8b5f5cb89dfb54'
905 >>> from dulwich.object_format import SHA256
906 >>> blob.get_id(SHA256) # Get SHA256 hash
907 b'03ba204e2f2e707...' # 64-character SHA256
908 """
909 return self.sha(object_format).hexdigest().encode("ascii")
911 def __repr__(self) -> str:
912 """Return string representation of this object."""
913 return f"<{self.__class__.__name__} {self.id!r}>"
915 def __ne__(self, other: object) -> bool:
916 """Check whether this object does not match the other."""
917 return not isinstance(other, ShaFile) or self.id != other.id
919 def __eq__(self, other: object) -> bool:
920 """Return True if the SHAs of the two objects match."""
921 return isinstance(other, ShaFile) and self.id == other.id
923 def __lt__(self, other: object) -> bool:
924 """Return whether SHA of this object is less than the other."""
925 if not isinstance(other, ShaFile):
926 raise TypeError
927 return self.id < other.id
929 def __le__(self, other: object) -> bool:
930 """Check whether SHA of this object is less than or equal to the other."""
931 if not isinstance(other, ShaFile):
932 raise TypeError
933 return self.id <= other.id
936class Blob(ShaFile):
937 """A Git Blob object."""
939 __slots__ = ()
941 type_name = b"blob"
942 type_num = 3
944 def __init__(self) -> None:
945 """Initialize a new Blob object."""
946 super().__init__()
947 self._chunked_text = []
948 self._needs_serialization = False
950 @property
951 def data(self) -> bytes:
952 """The text contained within the blob object."""
953 return self.as_raw_string()
955 @data.setter
956 def data(self, data: bytes) -> None:
957 self.set_raw_string(data)
959 @property
960 def chunked(self) -> list[bytes]:
961 """The text in the blob object, as chunks (not necessarily lines)."""
962 assert self._chunked_text is not None
963 return self._chunked_text
965 @chunked.setter
966 def chunked(self, chunks: list[bytes]) -> None:
967 self._chunked_text = chunks
969 def _serialize(self) -> list[bytes]:
970 assert self._chunked_text is not None
971 return self._chunked_text
973 def _deserialize(self, chunks: list[bytes]) -> None:
974 self._chunked_text = chunks
976 @classmethod
977 def from_path(
978 cls,
979 path: str | bytes,
980 sha: ObjectID | None = None,
981 *,
982 object_format: ObjectFormat | None = None,
983 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
984 ) -> "Blob":
985 """Read a blob from a file on disk.
987 Args:
988 path: Path to the blob file
989 sha: Optional known SHA for the object
990 object_format: Optional object format to use
991 max_size: Maximum inflated size in bytes
993 Returns:
994 A Blob object
996 Raises:
997 NotBlobError: If the file is not a blob
998 """
999 blob = ShaFile.from_path(
1000 path, sha, object_format=object_format, max_size=max_size
1001 )
1002 if not isinstance(blob, cls):
1003 raise NotBlobError(_path_to_bytes(path))
1004 return blob
1006 def check(self) -> None:
1007 """Check this object for internal consistency.
1009 Raises:
1010 ObjectFormatException: if the object is malformed in some way
1011 """
1012 super().check()
1014 def splitlines(self) -> list[bytes]:
1015 """Return list of lines in this blob.
1017 This preserves the original line endings.
1018 """
1019 chunks = self.chunked
1020 if not chunks:
1021 return []
1022 if len(chunks) == 1:
1023 result: list[bytes] = chunks[0].splitlines(True)
1024 return result
1025 remaining = None
1026 ret = []
1027 for chunk in chunks:
1028 lines = chunk.splitlines(True)
1029 if len(lines) > 1:
1030 ret.append((remaining or b"") + lines[0])
1031 ret.extend(lines[1:-1])
1032 remaining = lines[-1]
1033 elif len(lines) == 1:
1034 if remaining is None:
1035 remaining = lines.pop()
1036 else:
1037 remaining += lines.pop()
1038 if remaining is not None:
1039 ret.append(remaining)
1040 return ret
1043def _parse_message(
1044 chunks: Iterable[bytes],
1045) -> Iterator[tuple[None, None] | tuple[bytes | None, bytes]]:
1046 """Parse a message with a list of fields and a body.
1048 Args:
1049 chunks: the raw chunks of the tag or commit object.
1050 Returns: iterator of tuples of (field, value), one per header line, in the
1051 order read from the text, possibly including duplicates. Includes a
1052 field named None for the freeform tag/commit text.
1053 """
1054 f = BytesIO(b"".join(chunks))
1055 k = None
1056 v: list[bytes] = []
1057 eof = False
1059 def _strip_last_newline(value: bytes) -> bytes:
1060 """Strip the last newline from value."""
1061 if value and value.endswith(b"\n"):
1062 return value[:-1]
1063 return value
1065 # Parse the headers
1066 #
1067 # Headers can contain newlines. The next line is indented with a space.
1068 # We store the latest key as 'k', and the accumulated value chunks as 'v'.
1069 # The chunks are joined only once per header (rather than with ``+=`` per
1070 # continuation line) so a value split across many indented lines stays
1071 # linear instead of quadratic.
1072 for line in f:
1073 if line.startswith(b" "):
1074 # Indented continuation of the previous line
1075 v.append(line[1:])
1076 else:
1077 if k is not None:
1078 # We parsed a new header, return its value
1079 yield (k, _strip_last_newline(b"".join(v)))
1080 if line == b"\n":
1081 # Empty line indicates end of headers
1082 break
1083 (k, rest) = line.split(b" ", 1)
1084 v = [rest]
1086 else:
1087 # We reached end of file before the headers ended. We still need to
1088 # return the previous header, then we need to return a None field for
1089 # the text.
1090 eof = True
1091 if k is not None:
1092 yield (k, _strip_last_newline(b"".join(v)))
1093 yield (None, None)
1095 if not eof:
1096 # We didn't reach the end of file while parsing headers. We can return
1097 # the rest of the file as a message.
1098 yield (None, f.read())
1100 f.close()
1103def _format_message(
1104 headers: Sequence[tuple[bytes, bytes]], body: bytes | None
1105) -> Iterator[bytes]:
1106 for field, value in headers:
1107 lines = value.split(b"\n")
1108 yield git_line(field, lines[0])
1109 for line in lines[1:]:
1110 yield b" " + line + b"\n"
1111 yield b"\n" # There must be a new line after the headers
1112 if body:
1113 yield body
1116class Tag(ShaFile):
1117 """A Git Tag object."""
1119 type_name = b"tag"
1120 type_num = 4
1122 __slots__ = (
1123 "_message",
1124 "_name",
1125 "_object_class",
1126 "_object_sha",
1127 "_signature",
1128 "_tag_time",
1129 "_tag_timezone",
1130 "_tag_timezone_neg_utc",
1131 "_tagger",
1132 )
1134 _message: bytes | None
1135 _name: bytes | None
1136 _object_class: "type[ShaFile] | None"
1137 _object_sha: bytes | None
1138 _signature: bytes | None
1139 _tag_time: int | None
1140 _tag_timezone: int | None
1141 _tag_timezone_neg_utc: bool | None
1142 _tagger: bytes | None
1144 def __init__(self) -> None:
1145 """Initialize a new Tag object."""
1146 super().__init__()
1147 self._tagger = None
1148 self._tag_time = None
1149 self._tag_timezone = None
1150 self._tag_timezone_neg_utc = False
1151 self._signature: bytes | None = None
1153 @classmethod
1154 def from_path(
1155 cls,
1156 path: str | bytes,
1157 sha: ObjectID | None = None,
1158 *,
1159 object_format: ObjectFormat | None = None,
1160 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
1161 ) -> "Tag":
1162 """Read a tag from a file on disk.
1164 Args:
1165 path: Path to the tag file
1166 sha: Optional known SHA for the object
1167 object_format: Optional object format to use
1168 max_size: Maximum inflated size in bytes
1170 Returns:
1171 A Tag object
1173 Raises:
1174 NotTagError: If the file is not a tag
1175 """
1176 tag = ShaFile.from_path(
1177 path, sha, object_format=object_format, max_size=max_size
1178 )
1179 if not isinstance(tag, cls):
1180 raise NotTagError(_path_to_bytes(path))
1181 return tag
1183 def check(self) -> None:
1184 """Check this object for internal consistency.
1186 Raises:
1187 ObjectFormatException: if the object is malformed in some way
1188 """
1189 super().check()
1190 assert self._chunked_text is not None
1191 self._check_has_member("_object_sha", "missing object sha")
1192 self._check_has_member("_object_class", "missing object type")
1193 self._check_has_member("_name", "missing tag name")
1195 if not self._name:
1196 raise ObjectFormatException("empty tag name")
1198 if self._object_sha is None:
1199 raise ObjectFormatException("missing object sha")
1200 check_hexsha(self._object_sha, "invalid object sha")
1202 if self._tagger is not None:
1203 check_identity(self._tagger, "invalid tagger")
1205 self._check_has_member("_tag_time", "missing tag time")
1206 if self._tag_time is None:
1207 raise ObjectFormatException("missing tag time")
1208 check_time(self._tag_time)
1210 last = None
1211 for field, _ in _parse_message(self._chunked_text):
1212 if field == _OBJECT_HEADER and last is not None:
1213 raise ObjectFormatException("unexpected object")
1214 elif field == _TYPE_HEADER and last != _OBJECT_HEADER:
1215 raise ObjectFormatException("unexpected type")
1216 elif field == _TAG_HEADER and last != _TYPE_HEADER:
1217 raise ObjectFormatException("unexpected tag name")
1218 elif field == _TAGGER_HEADER and last != _TAG_HEADER:
1219 raise ObjectFormatException("unexpected tagger")
1220 last = field
1222 def _serialize(self) -> list[bytes]:
1223 headers = []
1224 if self._object_sha is None:
1225 raise ObjectFormatException("missing object sha")
1226 headers.append((_OBJECT_HEADER, self._object_sha))
1227 if self._object_class is None:
1228 raise ObjectFormatException("missing object class")
1229 headers.append((_TYPE_HEADER, self._object_class.type_name))
1230 if self._name is None:
1231 raise ObjectFormatException("missing tag name")
1232 headers.append((_TAG_HEADER, self._name))
1233 if self._tagger:
1234 if self._tag_time is None:
1235 headers.append((_TAGGER_HEADER, self._tagger))
1236 else:
1237 if self._tag_timezone is None or self._tag_timezone_neg_utc is None:
1238 raise ObjectFormatException("missing timezone info")
1239 headers.append(
1240 (
1241 _TAGGER_HEADER,
1242 format_time_entry(
1243 self._tagger,
1244 self._tag_time,
1245 (self._tag_timezone, self._tag_timezone_neg_utc),
1246 ),
1247 )
1248 )
1250 if self.message is None and self._signature is None:
1251 body = None
1252 else:
1253 body = (self.message or b"") + (self._signature or b"")
1254 return list(_format_message(headers, body))
1256 def _deserialize(self, chunks: list[bytes]) -> None:
1257 """Grab the metadata attached to the tag."""
1258 self._tagger = None
1259 self._tag_time = None
1260 self._tag_timezone = None
1261 self._tag_timezone_neg_utc = False
1262 for field, value in _parse_message(chunks):
1263 if field == _OBJECT_HEADER:
1264 self._object_sha = value
1265 elif field == _TYPE_HEADER:
1266 assert isinstance(value, bytes)
1267 obj_class = object_class(value)
1268 if not obj_class:
1269 raise ObjectFormatException(f"Not a known type: {value!r}")
1270 self._object_class = obj_class
1271 elif field == _TAG_HEADER:
1272 self._name = value
1273 elif field == _TAGGER_HEADER:
1274 if value is None:
1275 raise ObjectFormatException("missing tagger value")
1276 (
1277 self._tagger,
1278 self._tag_time,
1279 (self._tag_timezone, self._tag_timezone_neg_utc),
1280 ) = parse_time_entry(value)
1281 elif field is None:
1282 if value is None:
1283 self._message = None
1284 self._signature = None
1285 else:
1286 # Try to find either PGP or SSH signature
1287 sig_idx = None
1288 try:
1289 sig_idx = value.index(BEGIN_PGP_SIGNATURE)
1290 except ValueError:
1291 try:
1292 sig_idx = value.index(BEGIN_SSH_SIGNATURE)
1293 except ValueError:
1294 pass
1296 if sig_idx is not None:
1297 self._message = value[:sig_idx]
1298 self._signature = value[sig_idx:]
1299 else:
1300 self._message = value
1301 self._signature = None
1302 else:
1303 raise ObjectFormatException(
1304 f"Unknown field {field.decode('ascii', 'replace')}"
1305 )
1307 @property
1308 def object(self) -> tuple[type[ShaFile], ObjectID]:
1309 """Get the object pointed to by this tag.
1311 Returns: tuple of (object class, sha).
1312 """
1313 if self._object_class is None or self._object_sha is None:
1314 raise ValueError("Tag object is not properly initialized")
1315 return (self._object_class, ObjectID(self._object_sha))
1317 @object.setter
1318 def object(self, value: tuple[type[ShaFile], bytes]) -> None:
1319 self._object_class, self._object_sha = value
1320 self._needs_serialization = True
1322 name = serializable_property("name", "The name of this tag")
1323 tagger = serializable_property(
1324 "tagger", "Returns the name of the person who created this tag"
1325 )
1326 tag_time = serializable_property(
1327 "tag_time",
1328 "The creation timestamp of the tag. As the number of seconds since the epoch",
1329 )
1330 tag_timezone = serializable_property(
1331 "tag_timezone", "The timezone that tag_time is in."
1332 )
1333 message = serializable_property("message", "the message attached to this tag")
1335 signature = serializable_property("signature", "Optional detached GPG signature")
1337 def raw_without_sig(self) -> bytes:
1338 """Return raw string serialization without the GPG/SSH signature.
1340 self.signature is a signature for the returned raw byte string serialization.
1341 """
1342 ret = self.as_raw_string()
1343 if self._signature:
1344 ret = ret[: -len(self._signature)]
1345 return ret
1347 def extract_signature(self) -> tuple[bytes, bytes | None, bytes | None]:
1348 """Extract the payload, signature, and signature type from this tag.
1350 Returns:
1351 tuple of (``payload``, ``signature``, ``signature_type``) where:
1353 - ``payload``: The raw tag data without the signature
1354 - ``signature``: The signature bytes if present, None otherwise
1355 - ``signature_type``: SIGNATURE_PGP for PGP, SIGNATURE_SSH for SSH, None if no signature
1357 Raises:
1358 ObjectFormatException: If signature has unknown format
1359 """
1360 if self._signature is None:
1361 return self.as_raw_string(), None, None
1363 payload = self.raw_without_sig()
1365 # Determine signature type
1366 if self._signature.startswith(BEGIN_PGP_SIGNATURE):
1367 sig_type = SIGNATURE_PGP
1368 elif self._signature.startswith(BEGIN_SSH_SIGNATURE):
1369 sig_type = SIGNATURE_SSH
1370 else:
1371 raise ObjectFormatException("Unknown signature format")
1373 return payload, self._signature, sig_type
1376class TreeEntry(NamedTuple):
1377 """Named tuple encapsulating a single tree entry."""
1379 path: bytes
1380 mode: int
1381 sha: ObjectID
1383 def in_path(self, path: bytes) -> "TreeEntry":
1384 """Return a copy of this entry with the given path prepended."""
1385 if not isinstance(self.path, bytes):
1386 raise TypeError(f"Expected bytes for path, got {path!r}")
1387 return TreeEntry(posixpath.join(path, self.path), self.mode, self.sha)
1390def parse_tree(
1391 text: bytes, sha_len: int | None = None, *, strict: bool = False
1392) -> Iterator[tuple[bytes, int, bytes]]:
1393 """Parse a tree text.
1395 Args:
1396 text: Serialized text to parse
1397 sha_len: Length of the object IDs in bytes
1398 strict: Whether to be strict about format
1399 Returns: iterator of tuples of (name, mode, sha)
1401 Raises:
1402 ObjectFormatException: if the object was malformed in some way
1403 """
1404 count = 0
1405 length = len(text)
1407 while count < length:
1408 mode_end = text.index(b" ", count)
1409 mode_text = text[count:mode_end]
1410 if strict and mode_text.startswith(b"0"):
1411 raise ObjectFormatException(f"Invalid mode {mode_text!r}")
1412 try:
1413 mode = int(mode_text, 8)
1414 except ValueError as exc:
1415 raise ObjectFormatException(f"Invalid mode {mode_text!r}") from exc
1416 name_end = text.index(b"\0", mode_end)
1417 name = text[mode_end + 1 : name_end]
1419 if sha_len is None:
1420 raise ObjectFormatException("sha_len must be specified")
1421 count = name_end + 1 + sha_len
1422 if count > length:
1423 raise ObjectFormatException(
1424 f"Tree entry extends beyond tree length: {count} > {length}"
1425 )
1427 sha = text[name_end + 1 : count]
1428 if len(sha) != sha_len:
1429 raise ObjectFormatException(
1430 f"Sha has invalid length: {len(sha)} != {sha_len}"
1431 )
1432 hexsha = sha_to_hex(RawObjectID(sha))
1433 yield (name, mode, hexsha)
1436def serialize_tree(items: Iterable[tuple[bytes, int, ObjectID]]) -> Iterator[bytes]:
1437 """Serialize the items in a tree to a text.
1439 Args:
1440 items: Sorted iterable over (name, mode, sha) tuples
1441 Returns: Serialized tree text as chunks
1442 """
1443 for name, mode, hexsha in items:
1444 yield (
1445 (f"{mode:04o}").encode("ascii") + b" " + name + b"\0" + hex_to_sha(hexsha)
1446 )
1449def sorted_tree_items(
1450 entries: dict[bytes, tuple[int, ObjectID]], name_order: bool
1451) -> Iterator[TreeEntry]:
1452 """Iterate over a tree entries dictionary.
1454 Args:
1455 name_order: If True, iterate entries in order of their name. If
1456 False, iterate entries in tree order, that is, treat subtree entries as
1457 having '/' appended.
1458 entries: Dictionary mapping names to (mode, sha) tuples
1459 Returns: Iterator over (name, mode, hexsha)
1460 """
1461 if name_order:
1462 key_func = key_entry_name_order
1463 else:
1464 key_func = key_entry
1465 for name, entry in sorted(entries.items(), key=key_func):
1466 mode, hexsha = entry
1467 # Stricter type checks than normal to mirror checks in the Rust version.
1468 mode = int(mode)
1469 if not isinstance(hexsha, bytes):
1470 raise TypeError(f"Expected bytes for SHA, got {hexsha!r}")
1471 yield TreeEntry(name, mode, hexsha)
1474def key_entry(entry: tuple[bytes, tuple[int, ObjectID]]) -> bytes:
1475 """Sort key for tree entry.
1477 Args:
1478 entry: (name, value) tuple
1479 """
1480 (name, (mode, _sha)) = entry
1481 if stat.S_ISDIR(mode):
1482 name += b"/"
1483 return name
1486def key_entry_name_order(entry: tuple[bytes, tuple[int, ObjectID]]) -> bytes:
1487 """Sort key for tree entry in name order."""
1488 return entry[0]
1491def pretty_format_tree_entry(
1492 name: bytes, mode: int, hexsha: ObjectID, encoding: str = "utf-8"
1493) -> str:
1494 """Pretty format tree entry.
1496 Args:
1497 name: Name of the directory entry
1498 mode: Mode of entry
1499 hexsha: Hexsha of the referenced object
1500 encoding: Character encoding for the name
1501 Returns: string describing the tree entry
1502 """
1503 if mode & stat.S_IFDIR:
1504 kind = "tree"
1505 else:
1506 kind = "blob"
1507 return "{:04o} {} {}\t{}\n".format(
1508 mode,
1509 kind,
1510 hexsha.decode("ascii"),
1511 name.decode(encoding, "replace"),
1512 )
1515class SubmoduleEncountered(Exception):
1516 """A submodule was encountered while resolving a path."""
1518 def __init__(self, path: bytes, sha: ObjectID) -> None:
1519 """Initialize SubmoduleEncountered exception.
1521 Args:
1522 path: Path where the submodule was encountered
1523 sha: SHA of the submodule
1524 """
1525 self.path = path
1526 self.sha = sha
1529class Tree(ShaFile):
1530 """A Git tree object."""
1532 type_name = b"tree"
1533 type_num = 2
1535 __slots__ = "_entries"
1537 def __init__(self) -> None:
1538 """Initialize an empty Tree."""
1539 super().__init__()
1540 self._entries: dict[bytes, tuple[int, ObjectID]] = {}
1542 @classmethod
1543 def from_path(
1544 cls,
1545 path: str | bytes,
1546 sha: ObjectID | None = None,
1547 *,
1548 object_format: ObjectFormat | None = None,
1549 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
1550 ) -> "Tree":
1551 """Read a tree from a file on disk.
1553 Args:
1554 path: Path to the tree file
1555 sha: Optional known SHA for the object
1556 object_format: Optional object format to use
1557 max_size: Maximum inflated size in bytes
1559 Returns:
1560 A Tree object
1562 Raises:
1563 NotTreeError: If the file is not a tree
1564 """
1565 tree = ShaFile.from_path(
1566 path, sha, object_format=object_format, max_size=max_size
1567 )
1568 if not isinstance(tree, cls):
1569 raise NotTreeError(_path_to_bytes(path))
1570 return tree
1572 def __contains__(self, name: bytes) -> bool:
1573 """Check if name exists in tree."""
1574 return name in self._entries
1576 def __getitem__(self, name: bytes) -> tuple[int, ObjectID]:
1577 """Get tree entry by name."""
1578 return self._entries[name]
1580 def __setitem__(self, name: bytes, value: tuple[int, ObjectID]) -> None:
1581 """Set a tree entry by name.
1583 Args:
1584 name: The name of the entry, as a string.
1585 value: A tuple of (mode, hexsha), where mode is the mode of the
1586 entry as an integral type and hexsha is the hex SHA of the entry as
1587 a string.
1588 """
1589 mode, hexsha = value
1590 self._entries[name] = (mode, hexsha)
1591 self._needs_serialization = True
1593 def __delitem__(self, name: bytes) -> None:
1594 """Delete tree entry by name."""
1595 del self._entries[name]
1596 self._needs_serialization = True
1598 def __len__(self) -> int:
1599 """Return number of entries in tree."""
1600 return len(self._entries)
1602 def __iter__(self) -> Iterator[bytes]:
1603 """Iterate over tree entry names."""
1604 return iter(self._entries)
1606 def add(self, name: bytes, mode: int, hexsha: ObjectID) -> None:
1607 """Add an entry to the tree.
1609 Args:
1610 mode: The mode of the entry as an integral type. Not all
1611 possible modes are supported by git; see check() for details.
1612 name: The name of the entry, as a string.
1613 hexsha: The hex SHA of the entry as a string.
1614 """
1615 self._entries[name] = mode, hexsha
1616 self._needs_serialization = True
1618 def iteritems(self, name_order: bool = False) -> Iterator[TreeEntry]:
1619 """Iterate over entries.
1621 Args:
1622 name_order: If True, iterate in name order instead of tree
1623 order.
1624 Returns: Iterator over (name, mode, sha) tuples
1625 """
1626 return sorted_tree_items(self._entries, name_order)
1628 def items(self) -> list[TreeEntry]:
1629 """Return the sorted entries in this tree.
1631 Returns: List with (name, mode, sha) tuples
1632 """
1633 return list(self.iteritems())
1635 def _deserialize(self, chunks: list[bytes]) -> None:
1636 """Grab the entries in the tree."""
1637 try:
1638 parsed_entries = parse_tree(
1639 b"".join(chunks),
1640 sha_len=self.object_format.oid_length,
1641 )
1642 except ValueError as exc:
1643 raise ObjectFormatException(exc) from exc
1644 # TODO: list comprehension is for efficiency in the common (small)
1645 # case; if memory efficiency in the large case is a concern, use a
1646 # genexp.
1647 self._entries = {n: (m, ObjectID(s)) for n, m, s in parsed_entries}
1649 def check(self) -> None:
1650 """Check this object for internal consistency.
1652 Raises:
1653 ObjectFormatException: if the object is malformed in some way
1654 """
1655 super().check()
1656 assert self._chunked_text is not None
1657 last = None
1658 allowed_modes = (
1659 stat.S_IFREG | 0o755,
1660 stat.S_IFREG | 0o644,
1661 stat.S_IFLNK,
1662 stat.S_IFDIR,
1663 S_IFGITLINK,
1664 # TODO: optionally exclude as in git fsck --strict
1665 stat.S_IFREG | 0o664,
1666 )
1667 for name, mode, sha in parse_tree(
1668 b"".join(self._chunked_text),
1669 strict=True,
1670 sha_len=self.object_format.oid_length,
1671 ):
1672 check_hexsha(sha, f"invalid sha {sha!r}")
1673 if b"/" in name or name in (b"", b".", b"..", b".git"):
1674 raise ObjectFormatException(
1675 "invalid name {}".format(name.decode("utf-8", "replace"))
1676 )
1678 if mode not in allowed_modes:
1679 raise ObjectFormatException(f"invalid mode {mode:06o}")
1681 entry = (name, (mode, ObjectID(sha)))
1682 if last:
1683 if key_entry(last) > key_entry(entry):
1684 raise ObjectFormatException("entries not sorted")
1685 if name == last[0]:
1686 raise ObjectFormatException(f"duplicate entry {name!r}")
1687 last = entry
1689 def _serialize(self) -> list[bytes]:
1690 return list(serialize_tree(self.iteritems()))
1692 def as_pretty_string(self) -> str:
1693 """Return a human-readable string representation of this tree.
1695 Returns:
1696 Pretty-printed tree entries
1697 """
1698 text: list[str] = []
1699 for entry in self.iteritems():
1700 if (
1701 entry.path is not None
1702 and entry.mode is not None
1703 and entry.sha is not None
1704 ):
1705 text.append(pretty_format_tree_entry(entry.path, entry.mode, entry.sha))
1706 return "".join(text)
1708 def lookup_path(
1709 self, lookup_obj: Callable[[ObjectID], ShaFile], path: bytes
1710 ) -> tuple[int, ObjectID]:
1711 """Look up an object in a Git tree.
1713 Args:
1714 lookup_obj: Callback for retrieving object by SHA1
1715 path: Path to lookup
1716 Returns: A tuple of (mode, SHA) of the resulting path.
1717 """
1718 # Handle empty path - return the tree itself
1719 if not path:
1720 return stat.S_IFDIR, self.id
1722 parts = path.split(b"/")
1723 sha = self.id
1724 mode: int | None = None
1725 for i, p in enumerate(parts):
1726 if not p:
1727 continue
1728 if mode is not None and S_ISGITLINK(mode):
1729 raise SubmoduleEncountered(b"/".join(parts[:i]), sha)
1730 obj = lookup_obj(sha)
1731 if not isinstance(obj, Tree):
1732 raise NotTreeError(sha)
1733 mode, sha = obj[p]
1734 if mode is None:
1735 raise ValueError("No valid path found")
1736 return mode, sha
1739def parse_timezone(text: bytes) -> tuple[int, bool]:
1740 """Parse a timezone text fragment (e.g. '+0100').
1742 Args:
1743 text: Text to parse.
1744 Returns: Tuple with timezone as seconds difference to UTC
1745 and a boolean indicating whether this was a UTC timezone
1746 prefixed with a negative sign (-0000).
1747 """
1748 # cgit parses the first character as the sign, and the rest
1749 # as an integer (using strtol), which could also be negative.
1750 # We do the same for compatibility. See #697828.
1751 if text[0] not in b"+-":
1752 raise ValueError("Timezone must start with + or - ({text})".format(**vars()))
1753 sign = text[:1]
1754 offset = int(text[1:])
1755 if sign == b"-":
1756 offset = -offset
1757 unnecessary_negative_timezone = offset >= 0 and sign == b"-"
1758 signum = ((offset < 0) and -1) or 1
1759 offset = abs(offset)
1760 hours = int(offset / 100)
1761 minutes = offset % 100
1762 return (
1763 signum * (hours * 3600 + minutes * 60),
1764 unnecessary_negative_timezone,
1765 )
1768def parse_timezone_broken(text: bytes) -> tuple[int, bool]:
1769 """Parse a timezone text fragment, accepting broken formats.
1771 This function handles various broken timezone formats found in the wild:
1772 - Missing sign prefix (e.g., '0000' instead of '+0000')
1773 - Double negative (e.g., '--700')
1775 Args:
1776 text: Text to parse.
1777 Returns: Tuple with timezone as seconds difference to UTC
1778 and a boolean indicating whether this was a UTC timezone
1779 prefixed with a negative sign (-0000).
1780 """
1781 if text[0] not in b"+-":
1782 # Some (broken) commits do not have a sign
1783 text = b"+" + text
1785 # cgit parses the first character as the sign, and the rest
1786 # as an integer (using strtol), which could also be negative.
1787 # We do the same for compatibility. See #697828.
1788 sign = text[:1]
1789 offset = int(text[1:])
1790 if sign == b"-":
1791 offset = -offset
1792 unnecessary_negative_timezone = offset >= 0 and sign == b"-"
1793 signum = ((offset < 0) and -1) or 1
1794 offset = abs(offset)
1795 hours = int(offset / 100)
1796 minutes = offset % 100
1797 return (
1798 signum * (hours * 3600 + minutes * 60),
1799 unnecessary_negative_timezone,
1800 )
1803def format_timezone(offset: int, unnecessary_negative_timezone: bool = False) -> bytes:
1804 """Format a timezone for Git serialization.
1806 Args:
1807 offset: Timezone offset as seconds difference to UTC
1808 unnecessary_negative_timezone: Whether to use a minus sign for
1809 UTC or positive timezones (-0000 and --700 rather than +0000 / +0700).
1810 """
1811 if offset % 60 != 0:
1812 raise ValueError("Unable to handle non-minute offset.")
1813 if offset < 0 or unnecessary_negative_timezone:
1814 sign = "-"
1815 offset = -offset
1816 else:
1817 sign = "+"
1818 return ("%c%02d%02d" % (sign, offset / 3600, (offset / 60) % 60)).encode("ascii") # noqa: UP031
1821def parse_time_entry(
1822 value: bytes,
1823) -> tuple[bytes, int | None, tuple[int | None, bool]]:
1824 """Parse event.
1826 Args:
1827 value: Bytes representing a git commit/tag line
1828 Raises:
1829 ObjectFormatException in case of parsing error (malformed
1830 field date)
1831 Returns: Tuple of (author, time, (timezone, timezone_neg_utc))
1832 """
1833 try:
1834 sep = value.rindex(b"> ")
1835 except ValueError:
1836 return (value, None, (None, False))
1837 try:
1838 person = value[0 : sep + 1]
1839 rest = value[sep + 2 :]
1840 timetext, timezonetext = rest.rsplit(b" ", 1)
1841 time = int(timetext)
1842 timezone, timezone_neg_utc = parse_timezone(timezonetext)
1843 except ValueError as exc:
1844 raise ObjectFormatException(exc) from exc
1845 return person, time, (timezone, timezone_neg_utc)
1848def parse_time_entry_broken(
1849 value: bytes,
1850) -> tuple[bytes, int | None, tuple[int | None, bool]]:
1851 """Parse event, accepting broken formats.
1853 This function handles various broken author/committer/tagger line formats:
1854 - Missing angle brackets around email
1855 - Unsigned timezones
1856 - Double-negative timezones
1858 Args:
1859 value: Bytes representing a git commit/tag line
1860 Raises:
1861 ObjectFormatException in case of parsing error
1862 Returns: Tuple of (author, time, (timezone, timezone_neg_utc))
1863 """
1864 m = _TIME_ENTRY_RE.match(value)
1865 if not m:
1866 raise ObjectFormatException(f"Unable to parse time entry: {value!r}")
1868 person = m.group("person")
1869 timetext = m.group("time")
1870 timezonetext = m.group("timezone")
1871 time = int(timetext)
1872 timezone, timezone_neg_utc = parse_timezone_broken(timezonetext)
1874 return person, time, (timezone, timezone_neg_utc)
1877def format_time_entry(
1878 person: bytes, time: int, timezone_info: tuple[int, bool]
1879) -> bytes:
1880 """Format an event."""
1881 (timezone, timezone_neg_utc) = timezone_info
1882 return b" ".join(
1883 [person, str(time).encode("ascii"), format_timezone(timezone, timezone_neg_utc)]
1884 )
1887def _parse_commit(
1888 chunks: Iterable[bytes],
1889) -> tuple[
1890 bytes | None,
1891 list[bytes],
1892 tuple[bytes | None, int | None, tuple[int | None, bool | None]],
1893 tuple[bytes | None, int | None, tuple[int | None, bool | None]],
1894 bytes | None,
1895 list[Tag],
1896 bytes | None,
1897 bytes | None,
1898 list[tuple[bytes, bytes]],
1899]:
1900 """Parse a commit object from chunks.
1902 Args:
1903 chunks: Chunks to parse
1904 Returns: Tuple of (tree, parents, author_info, commit_info,
1905 encoding, mergetag, gpgsig, message, extra)
1906 """
1907 parents = []
1908 extra: list[tuple[bytes, bytes]] = []
1909 tree = None
1910 author_info: tuple[bytes | None, int | None, tuple[int | None, bool | None]] = (
1911 None,
1912 None,
1913 (None, None),
1914 )
1915 commit_info: tuple[bytes | None, int | None, tuple[int | None, bool | None]] = (
1916 None,
1917 None,
1918 (None, None),
1919 )
1920 encoding = None
1921 mergetag = []
1922 message = None
1923 gpgsig = None
1925 for field, value in _parse_message(chunks):
1926 # TODO(jelmer): Enforce ordering
1927 if field == _TREE_HEADER:
1928 tree = value
1929 elif field == _PARENT_HEADER:
1930 if value is None:
1931 raise ObjectFormatException("missing parent value")
1932 parents.append(value)
1933 elif field == _AUTHOR_HEADER:
1934 if value is None:
1935 raise ObjectFormatException("missing author value")
1936 author_info = parse_time_entry(value)
1937 elif field == _COMMITTER_HEADER:
1938 if value is None:
1939 raise ObjectFormatException("missing committer value")
1940 commit_info = parse_time_entry(value)
1941 elif field == _ENCODING_HEADER:
1942 encoding = value
1943 elif field == _MERGETAG_HEADER:
1944 if value is None:
1945 raise ObjectFormatException("missing mergetag value")
1946 tag = Tag.from_string(value + b"\n")
1947 assert isinstance(tag, Tag)
1948 mergetag.append(tag)
1949 elif field == _GPGSIG_HEADER:
1950 gpgsig = value
1951 elif field is None:
1952 message = value
1953 else:
1954 if value is None:
1955 raise ObjectFormatException(f"missing value for field {field!r}")
1956 extra.append((field, value))
1957 return (
1958 tree,
1959 parents,
1960 author_info,
1961 commit_info,
1962 encoding,
1963 mergetag,
1964 gpgsig,
1965 message,
1966 extra,
1967 )
1970def _parse_commit_broken(
1971 chunks: Iterable[bytes],
1972) -> tuple[
1973 bytes | None,
1974 list[bytes],
1975 tuple[bytes | None, int | None, tuple[int | None, bool | None]],
1976 tuple[bytes | None, int | None, tuple[int | None, bool | None]],
1977 bytes | None,
1978 list[Tag],
1979 bytes | None,
1980 bytes | None,
1981 list[tuple[bytes, bytes]],
1982]:
1983 """Parse a commit object from chunks, accepting broken formats.
1985 This function handles various broken author/committer line formats:
1986 - Missing angle brackets around email
1987 - Unsigned timezones
1988 - Double-negative timezones
1990 Args:
1991 chunks: Chunks to parse
1992 Returns: Tuple of (tree, parents, author_info, commit_info,
1993 encoding, mergetag, gpgsig, message, extra)
1994 """
1995 parents = []
1996 extra: list[tuple[bytes, bytes]] = []
1997 tree = None
1998 author_info: tuple[bytes | None, int | None, tuple[int | None, bool | None]] = (
1999 None,
2000 None,
2001 (None, None),
2002 )
2003 commit_info: tuple[bytes | None, int | None, tuple[int | None, bool | None]] = (
2004 None,
2005 None,
2006 (None, None),
2007 )
2008 encoding = None
2009 mergetag = []
2010 message = None
2011 gpgsig = None
2013 for field, value in _parse_message(chunks):
2014 # TODO(jelmer): Enforce ordering
2015 if field == _TREE_HEADER:
2016 tree = value
2017 elif field == _PARENT_HEADER:
2018 if value is None:
2019 raise ObjectFormatException("missing parent value")
2020 parents.append(value)
2021 elif field == _AUTHOR_HEADER:
2022 if value is None:
2023 raise ObjectFormatException("missing author value")
2024 author_info = parse_time_entry_broken(value)
2025 elif field == _COMMITTER_HEADER:
2026 if value is None:
2027 raise ObjectFormatException("missing committer value")
2028 commit_info = parse_time_entry_broken(value)
2029 elif field == _ENCODING_HEADER:
2030 encoding = value
2031 elif field == _MERGETAG_HEADER:
2032 if value is None:
2033 raise ObjectFormatException("missing mergetag value")
2034 tag = Tag.from_string(value + b"\n")
2035 assert isinstance(tag, Tag)
2036 mergetag.append(tag)
2037 elif field == _GPGSIG_HEADER:
2038 gpgsig = value
2039 elif field is None:
2040 message = value
2041 else:
2042 if value is None:
2043 raise ObjectFormatException(f"missing value for field {field!r}")
2044 extra.append((field, value))
2045 return (
2046 tree,
2047 parents,
2048 author_info,
2049 commit_info,
2050 encoding,
2051 mergetag,
2052 gpgsig,
2053 message,
2054 extra,
2055 )
2058class Commit(ShaFile):
2059 """A git commit object."""
2061 type_name = b"commit"
2062 type_num = 1
2064 __slots__ = (
2065 "_author",
2066 "_author_time",
2067 "_author_timezone",
2068 "_author_timezone_neg_utc",
2069 "_commit_time",
2070 "_commit_timezone",
2071 "_commit_timezone_neg_utc",
2072 "_committer",
2073 "_encoding",
2074 "_extra",
2075 "_gpgsig",
2076 "_mergetag",
2077 "_message",
2078 "_parents",
2079 "_tree",
2080 )
2082 def __init__(self) -> None:
2083 """Initialize an empty Commit."""
2084 super().__init__()
2085 self._parents: list[ObjectID] = []
2086 self._encoding: bytes | None = None
2087 self._mergetag: list[Tag] = []
2088 self._gpgsig: bytes | None = None
2089 self._extra: list[tuple[bytes, bytes]] = []
2090 self._author_timezone_neg_utc: bool | None = False
2091 self._commit_timezone_neg_utc: bool | None = False
2093 @classmethod
2094 def from_path(
2095 cls,
2096 path: str | bytes,
2097 sha: ObjectID | None = None,
2098 *,
2099 object_format: ObjectFormat | None = None,
2100 max_size: int = DEFAULT_LOOSE_OBJECT_SIZE_LIMIT,
2101 ) -> "Commit":
2102 """Read a commit from a file on disk.
2104 Args:
2105 path: Path to the commit file
2106 sha: Optional known SHA for the object
2107 object_format: Optional object format to use
2108 max_size: Maximum inflated size in bytes
2110 Returns:
2111 A Commit object
2113 Raises:
2114 NotCommitError: If the file is not a commit
2115 """
2116 commit = ShaFile.from_path(
2117 path, sha, object_format=object_format, max_size=max_size
2118 )
2119 if not isinstance(commit, cls):
2120 raise NotCommitError(_path_to_bytes(path))
2121 return commit
2123 def _deserialize(self, chunks: list[bytes]) -> None:
2124 (
2125 tree,
2126 parents,
2127 author_info,
2128 commit_info,
2129 encoding,
2130 mergetag,
2131 gpgsig,
2132 message,
2133 extra,
2134 ) = _parse_commit(chunks)
2136 self._tree = tree
2137 self._parents = [ObjectID(p) for p in parents]
2138 self._encoding = encoding
2139 self._mergetag = mergetag
2140 self._gpgsig = gpgsig
2141 self._message = message
2142 self._extra = extra
2144 (
2145 self._author,
2146 self._author_time,
2147 (self._author_timezone, self._author_timezone_neg_utc),
2148 ) = author_info
2149 (
2150 self._committer,
2151 self._commit_time,
2152 (self._commit_timezone, self._commit_timezone_neg_utc),
2153 ) = commit_info
2155 def check(self) -> None:
2156 """Check this object for internal consistency.
2158 Raises:
2159 ObjectFormatException: if the object is malformed in some way
2160 """
2161 super().check()
2162 assert self._chunked_text is not None
2163 self._check_has_member("_tree", "missing tree")
2164 self._check_has_member("_author", "missing author")
2165 self._check_has_member("_committer", "missing committer")
2166 self._check_has_member("_author_time", "missing author time")
2167 self._check_has_member("_commit_time", "missing commit time")
2169 for parent in self._parents:
2170 check_hexsha(parent, "invalid parent sha")
2171 assert self._tree is not None # checked by _check_has_member above
2172 check_hexsha(self._tree, "invalid tree sha")
2174 assert self._author is not None # checked by _check_has_member above
2175 assert self._committer is not None # checked by _check_has_member above
2176 check_identity(self._author, "invalid author")
2177 check_identity(self._committer, "invalid committer")
2179 assert self._author_time is not None # checked by _check_has_member above
2180 assert self._commit_time is not None # checked by _check_has_member above
2181 check_time(self._author_time)
2182 check_time(self._commit_time)
2184 last = None
2185 for field, _ in _parse_message(self._chunked_text):
2186 if field == _TREE_HEADER and last is not None:
2187 raise ObjectFormatException("unexpected tree")
2188 elif field == _PARENT_HEADER and last not in (
2189 _PARENT_HEADER,
2190 _TREE_HEADER,
2191 ):
2192 raise ObjectFormatException("unexpected parent")
2193 elif field == _AUTHOR_HEADER and last not in (
2194 _TREE_HEADER,
2195 _PARENT_HEADER,
2196 ):
2197 raise ObjectFormatException("unexpected author")
2198 elif field == _COMMITTER_HEADER and last != _AUTHOR_HEADER:
2199 raise ObjectFormatException("unexpected committer")
2200 elif field == _ENCODING_HEADER and last != _COMMITTER_HEADER:
2201 raise ObjectFormatException("unexpected encoding")
2202 last = field
2204 # TODO: optionally check for duplicate parents
2206 def raw_without_sig(self) -> bytes:
2207 """Return raw string serialization without the GPG/SSH signature.
2209 self.gpgsig is a signature for the returned raw byte string serialization.
2210 """
2211 tmp = self.copy()
2212 assert isinstance(tmp, Commit)
2213 tmp._gpgsig = None
2214 tmp.gpgsig = None
2215 return tmp.as_raw_string()
2217 def extract_signature(self) -> tuple[bytes, bytes | None, bytes | None]:
2218 """Extract the payload, signature, and signature type from this commit.
2220 Returns:
2221 tuple of (``payload``, ``signature``, ``signature_type``) where:
2223 - ``payload``: The raw commit data without the signature
2224 - ``signature``: The signature bytes if present, None otherwise
2225 - ``signature_type``: SIGNATURE_PGP for PGP, SIGNATURE_SSH for SSH, None if no signature
2227 Raises:
2228 ObjectFormatException: If signature has unknown format
2229 """
2230 if self._gpgsig is None:
2231 return self.as_raw_string(), None, None
2233 payload = self.raw_without_sig()
2235 # Determine signature type
2236 if self._gpgsig.startswith(BEGIN_PGP_SIGNATURE):
2237 sig_type = SIGNATURE_PGP
2238 elif self._gpgsig.startswith(BEGIN_SSH_SIGNATURE):
2239 sig_type = SIGNATURE_SSH
2240 else:
2241 raise ObjectFormatException("Unknown signature format")
2243 return payload, self._gpgsig, sig_type
2245 def _serialize(self) -> list[bytes]:
2246 headers = []
2247 assert self._tree is not None
2248 tree_bytes = self._tree.id if isinstance(self._tree, Tree) else self._tree
2249 headers.append((_TREE_HEADER, tree_bytes))
2250 for p in self._parents:
2251 headers.append((_PARENT_HEADER, p))
2252 assert self._author is not None
2253 assert self._author_time is not None
2254 assert self._author_timezone is not None
2255 assert self._author_timezone_neg_utc is not None
2256 headers.append(
2257 (
2258 _AUTHOR_HEADER,
2259 format_time_entry(
2260 self._author,
2261 self._author_time,
2262 (self._author_timezone, self._author_timezone_neg_utc),
2263 ),
2264 )
2265 )
2266 assert self._committer is not None
2267 assert self._commit_time is not None
2268 assert self._commit_timezone is not None
2269 assert self._commit_timezone_neg_utc is not None
2270 headers.append(
2271 (
2272 _COMMITTER_HEADER,
2273 format_time_entry(
2274 self._committer,
2275 self._commit_time,
2276 (self._commit_timezone, self._commit_timezone_neg_utc),
2277 ),
2278 )
2279 )
2280 if self.encoding:
2281 headers.append((_ENCODING_HEADER, self.encoding))
2282 for mergetag in self.mergetag:
2283 headers.append((_MERGETAG_HEADER, mergetag.as_raw_string()[:-1]))
2284 headers.extend(
2285 (field, value) for field, value in self._extra if value is not None
2286 )
2287 if self.gpgsig:
2288 headers.append((_GPGSIG_HEADER, self.gpgsig))
2289 return list(_format_message(headers, self._message))
2291 tree = serializable_property("tree", "Tree that is the state of this commit")
2293 @property
2294 def parents(self) -> list[ObjectID]:
2295 """Parents of this commit, by their SHA1."""
2296 return self._parents
2298 @parents.setter
2299 def parents(self, value: list[ObjectID]) -> None:
2300 """Set a list of parents of this commit."""
2301 self._needs_serialization = True
2302 self._parents = value
2304 author = serializable_property("author", "The name of the author of the commit")
2306 committer = serializable_property(
2307 "committer", "The name of the committer of the commit"
2308 )
2310 message = serializable_property("message", "The commit message")
2312 commit_time = serializable_property(
2313 "commit_time",
2314 "The timestamp of the commit. As the number of seconds since the epoch.",
2315 )
2317 commit_timezone = serializable_property(
2318 "commit_timezone", "The zone the commit time is in"
2319 )
2321 author_time = serializable_property(
2322 "author_time",
2323 "The timestamp the commit was written. As the number of "
2324 "seconds since the epoch.",
2325 )
2327 author_timezone = serializable_property(
2328 "author_timezone", "Returns the zone the author time is in."
2329 )
2331 encoding = serializable_property("encoding", "Encoding of the commit message.")
2333 mergetag = serializable_property("mergetag", "Associated signed tag.")
2335 gpgsig = serializable_property("gpgsig", "GPG Signature.")
2338OBJECT_CLASSES = (
2339 Commit,
2340 Tree,
2341 Blob,
2342 Tag,
2343)
2345_TYPE_MAP: dict[bytes | int, type[ShaFile]] = {}
2347for cls in OBJECT_CLASSES:
2348 _TYPE_MAP[cls.type_name] = cls
2349 _TYPE_MAP[cls.type_num] = cls
2352# Public API functions
2355def parse_commit_broken(data: bytes) -> Commit:
2356 """Parse a commit with broken author/committer lines.
2358 This function handles various broken formats found in the wild:
2359 - Missing angle brackets around email addresses
2360 - Unsigned timezones (e.g., "0000" instead of "+0000")
2361 - Double-negative timezones (e.g., "--700")
2362 - Negative timestamps
2363 - Long/short/nonsensical timezone values
2365 Warning: Commits parsed with this function may not round-trip correctly
2366 through serialization, as the broken formatting is normalized during parsing.
2367 The .check() method will likely fail for commits with malformed identity fields.
2369 Args:
2370 data: Raw commit data as bytes
2372 Returns:
2373 A Commit object with normalized fields
2375 Example:
2376 >>> data = b'''tree d80c186a03f423a81b39df39dc87fd269736ca86
2377 ... author user@example.com 1234567890 -0500
2378 ... committer user@example.com 1234567890 -0500
2379 ...
2380 ... Commit message
2381 ... '''
2382 >>> commit = parse_commit_broken(data)
2383 >>> commit.author
2384 b'user@example.com'
2385 """
2386 commit = Commit()
2387 (
2388 tree,
2389 parents,
2390 author_info,
2391 commit_info,
2392 encoding,
2393 mergetag,
2394 gpgsig,
2395 message,
2396 extra,
2397 ) = _parse_commit_broken([data])
2399 commit._tree = tree
2400 commit._parents = [ObjectID(p) for p in parents]
2401 commit._encoding = encoding
2402 commit._mergetag = mergetag
2403 commit._gpgsig = gpgsig
2404 commit._message = message
2405 commit._extra = extra
2407 (
2408 commit._author,
2409 commit._author_time,
2410 (commit._author_timezone, commit._author_timezone_neg_utc),
2411 ) = author_info
2412 (
2413 commit._committer,
2414 commit._commit_time,
2415 (commit._commit_timezone, commit._commit_timezone_neg_utc),
2416 ) = commit_info
2418 return commit
2421# Hold on to the pure-python implementations for testing
2422_parse_tree_py = parse_tree
2423_sorted_tree_items_py = sorted_tree_items
2424try:
2425 # Try to import Rust versions
2426 from dulwich._objects import (
2427 parse_tree as _parse_tree_rs,
2428 )
2429 from dulwich._objects import (
2430 sorted_tree_items as _sorted_tree_items_rs,
2431 )
2432except ImportError:
2433 pass
2434else:
2435 parse_tree = _parse_tree_rs
2436 sorted_tree_items = _sorted_tree_items_rs