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