dulwich/dulwich/objects.py

# objects.py -- Access to base git objects
# Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
# Copyright (C) 2008-2013 Jelmer Vernooij <jelmer@jelmer.uk>
#
# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
# General Public License as public by the Free Software Foundation; version 2.0
# or (at your option) any later version. You can redistribute it and/or
# modify it under the terms of either of these two licenses.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# You should have received a copy of the licenses; if not, see
# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
# License, Version 2.0.
#

"""Access to base git objects."""

import binascii
from io import BytesIO
from collections import namedtuple
import os
import posixpath
import stat
from typing import (
    Optional,
    Dict,
    Union,
    Type,
)
import warnings
import zlib
from hashlib import sha1

from dulwich.errors import (
    ChecksumMismatch,
    NotBlobError,
    NotCommitError,
    NotTagError,
    NotTreeError,
    ObjectFormatException,
    FileFormatException,
)
from dulwich.file import GitFile


ZERO_SHA = b"0" * 40

# Header fields for commits
_TREE_HEADER = b"tree"
_PARENT_HEADER = b"parent"
_AUTHOR_HEADER = b"author"
_COMMITTER_HEADER = b"committer"
_ENCODING_HEADER = b"encoding"
_MERGETAG_HEADER = b"mergetag"
_GPGSIG_HEADER = b"gpgsig"

# Header fields for objects
_OBJECT_HEADER = b"object"
_TYPE_HEADER = b"type"
_TAG_HEADER = b"tag"
_TAGGER_HEADER = b"tagger"


S_IFGITLINK = 0o160000


MAX_TIME = 9223372036854775807  # (2**63) - 1 - signed long int max

BEGIN_PGP_SIGNATURE = b"-----BEGIN PGP SIGNATURE-----"


class EmptyFileException(FileFormatException):
    """An unexpectedly empty file was encountered."""


def S_ISGITLINK(m):
    """Check if a mode indicates a submodule.

    Args:
      m: Mode to check
    Returns: a ``boolean``
    """
    return stat.S_IFMT(m) == S_IFGITLINK


def _decompress(string):
    dcomp = zlib.decompressobj()
    dcomped = dcomp.decompress(string)
    dcomped += dcomp.flush()
    return dcomped


def sha_to_hex(sha):
    """Takes a string and returns the hex of the sha within"""
    hexsha = binascii.hexlify(sha)
    assert len(hexsha) == 40, "Incorrect length of sha1 string: %d" % hexsha
    return hexsha


def hex_to_sha(hex):
    """Takes a hex sha and returns a binary sha"""
    assert len(hex) == 40, "Incorrect length of hexsha: %s" % hex
    try:
        return binascii.unhexlify(hex)
    except TypeError as exc:
        if not isinstance(hex, bytes):
            raise
        raise ValueError(exc.args[0])


def valid_hexsha(hex):
    if len(hex) != 40:
        return False
    try:
        binascii.unhexlify(hex)
    except (TypeError, binascii.Error):
        return False
    else:
        return True


def hex_to_filename(path, hex):
    """Takes a hex sha and returns its filename relative to the given path."""
    # os.path.join accepts bytes or unicode, but all args must be of the same
    # type. Make sure that hex which is expected to be bytes, is the same type
    # as path.
    if getattr(path, "encode", None) is not None:
        hex = hex.decode("ascii")
    dir = hex[:2]
    file = hex[2:]
    # Check from object dir
    return os.path.join(path, dir, file)


def filename_to_hex(filename):
    """Takes an object filename and returns its corresponding hex sha."""
    # grab the last (up to) two path components
    names = filename.rsplit(os.path.sep, 2)[-2:]
    errmsg = "Invalid object filename: %s" % filename
    assert len(names) == 2, errmsg
    base, rest = names
    assert len(base) == 2 and len(rest) == 38, errmsg
    hex = (base + rest).encode("ascii")
    hex_to_sha(hex)
    return hex


def object_header(num_type: int, length: int) -> bytes:
    """Return an object header for the given numeric type and text length."""
    return object_class(num_type).type_name + b" " + str(length).encode("ascii") + b"\0"


def serializable_property(name: str, docstring: Optional[str] = None):
    """A property that helps tracking whether serialization is necessary."""

    def set(obj, value):
        setattr(obj, "_" + name, value)
        obj._needs_serialization = True

    def get(obj):
        return getattr(obj, "_" + name)

    return property(get, set, doc=docstring)


def object_class(type):
    """Get the object class corresponding to the given type.

    Args:
      type: Either a type name string or a numeric type.
    Returns: The ShaFile subclass corresponding to the given type, or None if
        type is not a valid type name/number.
    """
    return _TYPE_MAP.get(type, None)


def check_hexsha(hex, error_msg):
    """Check if a string is a valid hex sha string.

    Args:
      hex: Hex string to check
      error_msg: Error message to use in exception
    Raises:
      ObjectFormatException: Raised when the string is not valid
    """
    if not valid_hexsha(hex):
        raise ObjectFormatException("%s %s" % (error_msg, hex))


def check_identity(identity, error_msg):
    """Check if the specified identity is valid.

    This will raise an exception if the identity is not valid.

    Args:
      identity: Identity string
      error_msg: Error message to use in exception
    """
    email_start = identity.find(b"<")
    email_end = identity.find(b">")
    if (
        email_start < 0
        or email_end < 0
        or email_end <= email_start
        or identity.find(b"<", email_start + 1) >= 0
        or identity.find(b">", email_end + 1) >= 0
        or not identity.endswith(b">")
    ):
        raise ObjectFormatException(error_msg)


def check_time(time_seconds):
    """Check if the specified time is not prone to overflow error.

    This will raise an exception if the time is not valid.

    Args:
      time_info: author/committer/tagger info

    """
    # Prevent overflow error
    if time_seconds > MAX_TIME:
        raise ObjectFormatException("Date field should not exceed %s" % MAX_TIME)


def git_line(*items):
    """Formats items into a space separated line."""
    return b" ".join(items) + b"\n"


class FixedSha(object):
    """SHA object that behaves like hashlib's but is given a fixed value."""

    __slots__ = ("_hexsha", "_sha")

    def __init__(self, hexsha):
        if getattr(hexsha, "encode", None) is not None:
            hexsha = hexsha.encode("ascii")
        if not isinstance(hexsha, bytes):
            raise TypeError("Expected bytes for hexsha, got %r" % hexsha)
        self._hexsha = hexsha
        self._sha = hex_to_sha(hexsha)

    def digest(self):
        """Return the raw SHA digest."""
        return self._sha

    def hexdigest(self):
        """Return the hex SHA digest."""
        return self._hexsha.decode("ascii")


class ShaFile(object):
    """A git SHA file."""

    __slots__ = ("_chunked_text", "_sha", "_needs_serialization")

    type_name = None  # type: bytes
    type_num = None  # type: int

    @staticmethod
    def _parse_legacy_object_header(magic, f):
        """Parse a legacy object, creating it but not reading the file."""
        bufsize = 1024
        decomp = zlib.decompressobj()
        header = decomp.decompress(magic)
        start = 0
        end = -1
        while end < 0:
            extra = f.read(bufsize)
            header += decomp.decompress(extra)
            magic += extra
            end = header.find(b"\0", start)
            start = len(header)
        header = header[:end]
        type_name, size = header.split(b" ", 1)
        try:
            int(size)  # sanity check
        except ValueError as e:
            raise ObjectFormatException("Object size not an integer: %s" % e)
        obj_class = object_class(type_name)
        if not obj_class:
            raise ObjectFormatException("Not a known type: %s" % type_name)
        return obj_class()

    def _parse_legacy_object(self, map):
        """Parse a legacy object, setting the raw string."""
        text = _decompress(map)
        header_end = text.find(b"\0")
        if header_end < 0:
            raise ObjectFormatException("Invalid object header, no \\0")
        self.set_raw_string(text[header_end + 1 :])

    def as_legacy_object_chunks(self, compression_level=-1):
        """Return chunks representing the object in the experimental format.

        Returns: List of strings
        """
        compobj = zlib.compressobj(compression_level)
        yield compobj.compress(self._header())
        for chunk in self.as_raw_chunks():
            yield compobj.compress(chunk)
        yield compobj.flush()

    def as_legacy_object(self, compression_level=-1):
        """Return string representing the object in the experimental format."""
        return b"".join(
            self.as_legacy_object_chunks(compression_level=compression_level)
        )

    def as_raw_chunks(self):
        """Return chunks with serialization of the object.

        Returns: List of strings, not necessarily one per line
        """
        if self._needs_serialization:
            self._sha = None
            self._chunked_text = self._serialize()
            self._needs_serialization = False
        return self._chunked_text

    def as_raw_string(self):
        """Return raw string with serialization of the object.

        Returns: String object
        """
        return b"".join(self.as_raw_chunks())

    def __bytes__(self):
        """Return raw string serialization of this object."""
        return self.as_raw_string()

    def __hash__(self):
        """Return unique hash for this object."""
        return hash(self.id)

    def as_pretty_string(self):
        """Return a string representing this object, fit for display."""
        return self.as_raw_string()

    def set_raw_string(self, text, sha=None):
        """Set the contents of this object from a serialized string."""
        if not isinstance(text, bytes):
            raise TypeError("Expected bytes for text, got %r" % text)
        self.set_raw_chunks([text], sha)

    def set_raw_chunks(self, chunks, sha=None):
        """Set the contents of this object from a list of chunks."""
        self._chunked_text = chunks
        self._deserialize(chunks)
        if sha is None:
            self._sha = None
        else:
            self._sha = FixedSha(sha)
        self._needs_serialization = False

    @staticmethod
    def _parse_object_header(magic, f):
        """Parse a new style object, creating it but not reading the file."""
        num_type = (ord(magic[0:1]) >> 4) & 7
        obj_class = object_class(num_type)
        if not obj_class:
            raise ObjectFormatException("Not a known type %d" % num_type)
        return obj_class()

    def _parse_object(self, map):
        """Parse a new style object, setting self._text."""
        # skip type and size; type must have already been determined, and
        # we trust zlib to fail if it's otherwise corrupted
        byte = ord(map[0:1])
        used = 1
        while (byte & 0x80) != 0:
            byte = ord(map[used : used + 1])
            used += 1
        raw = map[used:]
        self.set_raw_string(_decompress(raw))

    @classmethod
    def _is_legacy_object(cls, magic):
        b0 = ord(magic[0:1])
        b1 = ord(magic[1:2])
        word = (b0 << 8) + b1
        return (b0 & 0x8F) == 0x08 and (word % 31) == 0

    @classmethod
    def _parse_file(cls, f):
        map = f.read()
        if not map:
            raise EmptyFileException("Corrupted empty file detected")

        if cls._is_legacy_object(map):
            obj = cls._parse_legacy_object_header(map, f)
            obj._parse_legacy_object(map)
        else:
            obj = cls._parse_object_header(map, f)
            obj._parse_object(map)
        return obj

    def __init__(self):
        """Don't call this directly"""
        self._sha = None
        self._chunked_text = []
        self._needs_serialization = True

    def _deserialize(self, chunks):
        raise NotImplementedError(self._deserialize)

    def _serialize(self):
        raise NotImplementedError(self._serialize)

    @classmethod
    def from_path(cls, path):
        """Open a SHA file from disk."""
        with GitFile(path, "rb") as f:
            return cls.from_file(f)

    @classmethod
    def from_file(cls, f):
        """Get the contents of a SHA file on disk."""
        try:
            obj = cls._parse_file(f)
            obj._sha = None
            return obj
        except (IndexError, ValueError):
            raise ObjectFormatException("invalid object header")

    @staticmethod
    def from_raw_string(type_num, string, sha=None):
        """Creates an object of the indicated type from the raw string given.

        Args:
          type_num: The numeric type of the object.
          string: The raw uncompressed contents.
          sha: Optional known sha for the object
        """
        obj = object_class(type_num)()
        obj.set_raw_string(string, sha)
        return obj

    @staticmethod
    def from_raw_chunks(type_num, chunks, sha=None):
        """Creates an object of the indicated type from the raw chunks given.

        Args:
          type_num: The numeric type of the object.
          chunks: An iterable of the raw uncompressed contents.
          sha: Optional known sha for the object
        """
        obj = object_class(type_num)()
        obj.set_raw_chunks(chunks, sha)
        return obj

    @classmethod
    def from_string(cls, string):
        """Create a ShaFile from a string."""
        obj = cls()
        obj.set_raw_string(string)
        return obj

    def _check_has_member(self, member, error_msg):
        """Check that the object has a given member variable.

        Args:
          member: the member variable to check for
          error_msg: the message for an error if the member is missing
        Raises:
          ObjectFormatException: with the given error_msg if member is
            missing or is None
        """
        if getattr(self, member, None) is None:
            raise ObjectFormatException(error_msg)

    def check(self):
        """Check this object for internal consistency.

        Raises:
          ObjectFormatException: if the object is malformed in some way
          ChecksumMismatch: if the object was created with a SHA that does
            not match its contents
        """
        # TODO: if we find that error-checking during object parsing is a
        # performance bottleneck, those checks should be moved to the class's
        # check() method during optimization so we can still check the object
        # when necessary.
        old_sha = self.id
        try:
            self._deserialize(self.as_raw_chunks())
            self._sha = None
            new_sha = self.id
        except Exception as e:
            raise ObjectFormatException(e)
        if old_sha != new_sha:
            raise ChecksumMismatch(new_sha, old_sha)

    def _header(self):
        return object_header(self.type, self.raw_length())

    def raw_length(self):
        """Returns the length of the raw string of this object."""
        ret = 0
        for chunk in self.as_raw_chunks():
            ret += len(chunk)
        return ret

    def sha(self):
        """The SHA1 object that is the name of this object."""
        if self._sha is None or self._needs_serialization:
            # this is a local because as_raw_chunks() overwrites self._sha
            new_sha = sha1()
            new_sha.update(self._header())
            for chunk in self.as_raw_chunks():
                new_sha.update(chunk)
            self._sha = new_sha
        return self._sha

    def copy(self):
        """Create a new copy of this SHA1 object from its raw string"""
        obj_class = object_class(self.get_type())
        return obj_class.from_raw_string(self.get_type(), self.as_raw_string(), self.id)

    @property
    def id(self):
        """The hex SHA of this object."""
        return self.sha().hexdigest().encode("ascii")

    def get_type(self):
        """Return the type number for this object class."""
        return self.type_num

    def set_type(self, type):
        """Set the type number for this object class."""
        self.type_num = type

    # DEPRECATED: use type_num or type_name as needed.
    type = property(get_type, set_type)

    def __repr__(self):
        return "<%s %s>" % (self.__class__.__name__, self.id)

    def __ne__(self, other):
        """Check whether this object does not match the other."""
        return not isinstance(other, ShaFile) or self.id != other.id

    def __eq__(self, other):
        """Return True if the SHAs of the two objects match."""
        return isinstance(other, ShaFile) and self.id == other.id

    def __lt__(self, other):
        """Return whether SHA of this object is less than the other."""
        if not isinstance(other, ShaFile):
            raise TypeError
        return self.id < other.id

    def __le__(self, other):
        """Check whether SHA of this object is less than or equal to the other."""
        if not isinstance(other, ShaFile):
            raise TypeError
        return self.id <= other.id

    def __cmp__(self, other):
        """Compare the SHA of this object with that of the other object."""
        if not isinstance(other, ShaFile):
            raise TypeError
        return cmp(self.id, other.id)  # noqa: F821


class Blob(ShaFile):
    """A Git Blob object."""

    __slots__ = ()

    type_name = b"blob"
    type_num = 3

    def __init__(self):
        super(Blob, self).__init__()
        self._chunked_text = []
        self._needs_serialization = False

    def _get_data(self):
        return self.as_raw_string()

    def _set_data(self, data):
        self.set_raw_string(data)

    data = property(
        _get_data, _set_data, doc="The text contained within the blob object."
    )

    def _get_chunked(self):
        return self._chunked_text

    def _set_chunked(self, chunks):
        self._chunked_text = chunks

    def _serialize(self):
        return self._chunked_text

    def _deserialize(self, chunks):
        self._chunked_text = chunks

    chunked = property(
        _get_chunked,
        _set_chunked,
        doc="The text in the blob object, as chunks (not necessarily lines)",
    )

    @classmethod
    def from_path(cls, path):
        blob = ShaFile.from_path(path)
        if not isinstance(blob, cls):
            raise NotBlobError(path)
        return blob

    def check(self):
        """Check this object for internal consistency.

        Raises:
          ObjectFormatException: if the object is malformed in some way
        """
        super(Blob, self).check()

    def splitlines(self):
        """Return list of lines in this blob.

        This preserves the original line endings.
        """
        chunks = self.chunked
        if not chunks:
            return []
        if len(chunks) == 1:
            return chunks[0].splitlines(True)
        remaining = None
        ret = []
        for chunk in chunks:
            lines = chunk.splitlines(True)
            if len(lines) > 1:
                ret.append((remaining or b"") + lines[0])
                ret.extend(lines[1:-1])
                remaining = lines[-1]
            elif len(lines) == 1:
                if remaining is None:
                    remaining = lines.pop()
                else:
                    remaining += lines.pop()
        if remaining is not None:
            ret.append(remaining)
        return ret


def _parse_message(chunks):
    """Parse a message with a list of fields and a body.

    Args:
      chunks: the raw chunks of the tag or commit object.
    Returns: iterator of tuples of (field, value), one per header line, in the
        order read from the text, possibly including duplicates. Includes a
        field named None for the freeform tag/commit text.
    """
    f = BytesIO(b"".join(chunks))
    k = None
    v = ""
    eof = False

    def _strip_last_newline(value):
        """Strip the last newline from value"""
        if value and value.endswith(b"\n"):
            return value[:-1]
        return value

    # Parse the headers
    #
    # Headers can contain newlines. The next line is indented with a space.
    # We store the latest key as 'k', and the accumulated value as 'v'.
    for line in f:
        if line.startswith(b" "):
            # Indented continuation of the previous line
            v += line[1:]
        else:
            if k is not None:
                # We parsed a new header, return its value
                yield (k, _strip_last_newline(v))
            if line == b"\n":
                # Empty line indicates end of headers
                break
            (k, v) = line.split(b" ", 1)

    else:
        # We reached end of file before the headers ended. We still need to
        # return the previous header, then we need to return a None field for
        # the text.
        eof = True
        if k is not None:
            yield (k, _strip_last_newline(v))
        yield (None, None)

    if not eof:
        # We didn't reach the end of file while parsing headers. We can return
        # the rest of the file as a message.
        yield (None, f.read())

    f.close()


class Tag(ShaFile):
    """A Git Tag object."""

    type_name = b"tag"
    type_num = 4

    __slots__ = (
        "_tag_timezone_neg_utc",
        "_name",
        "_object_sha",
        "_object_class",
        "_tag_time",
        "_tag_timezone",
        "_tagger",
        "_message",
        "_signature",
    )

    def __init__(self):
        super(Tag, self).__init__()
        self._tagger = None
        self._tag_time = None
        self._tag_timezone = None
        self._tag_timezone_neg_utc = False
        self._signature = None

    @classmethod
    def from_path(cls, filename):
        tag = ShaFile.from_path(filename)
        if not isinstance(tag, cls):
            raise NotTagError(filename)
        return tag

    def check(self):
        """Check this object for internal consistency.

        Raises:
          ObjectFormatException: if the object is malformed in some way
        """
        super(Tag, self).check()
        self._check_has_member("_object_sha", "missing object sha")
        self._check_has_member("_object_class", "missing object type")
        self._check_has_member("_name", "missing tag name")

        if not self._name:
            raise ObjectFormatException("empty tag name")

        check_hexsha(self._object_sha, "invalid object sha")

        if getattr(self, "_tagger", None):
            check_identity(self._tagger, "invalid tagger")

        self._check_has_member("_tag_time", "missing tag time")
        check_time(self._tag_time)

        last = None
        for field, _ in _parse_message(self._chunked_text):
            if field == _OBJECT_HEADER and last is not None:
                raise ObjectFormatException("unexpected object")
            elif field == _TYPE_HEADER and last != _OBJECT_HEADER:
                raise ObjectFormatException("unexpected type")
            elif field == _TAG_HEADER and last != _TYPE_HEADER:
                raise ObjectFormatException("unexpected tag name")
            elif field == _TAGGER_HEADER and last != _TAG_HEADER:
                raise ObjectFormatException("unexpected tagger")
            last = field

    def _serialize(self):
        chunks = []
        chunks.append(git_line(_OBJECT_HEADER, self._object_sha))
        chunks.append(git_line(_TYPE_HEADER, self._object_class.type_name))
        chunks.append(git_line(_TAG_HEADER, self._name))
        if self._tagger:
            if self._tag_time is None:
                chunks.append(git_line(_TAGGER_HEADER, self._tagger))
            else:
                chunks.append(
                    git_line(
                        _TAGGER_HEADER,
                        self._tagger,
                        str(self._tag_time).encode("ascii"),
                        format_timezone(self._tag_timezone, self._tag_timezone_neg_utc),
                    )
                )
        if self._message is not None:
            chunks.append(b"\n")  # To close headers
            chunks.append(self._message)
        if self._signature is not None:
            chunks.append(self._signature)
        return chunks

    def _deserialize(self, chunks):
        """Grab the metadata attached to the tag"""
        self._tagger = None
        self._tag_time = None
        self._tag_timezone = None
        self._tag_timezone_neg_utc = False
        for field, value in _parse_message(chunks):
            if field == _OBJECT_HEADER:
                self._object_sha = value
            elif field == _TYPE_HEADER:
                obj_class = object_class(value)
                if not obj_class:
                    raise ObjectFormatException("Not a known type: %s" % value)
                self._object_class = obj_class
            elif field == _TAG_HEADER:
                self._name = value
            elif field == _TAGGER_HEADER:
                (
                    self._tagger,
                    self._tag_time,
                    (self._tag_timezone, self._tag_timezone_neg_utc),
                ) = parse_time_entry(value)
            elif field is None:
                if value is None:
                    self._message = None
                    self._signature = None
                else:
                    try:
                        sig_idx = value.index(BEGIN_PGP_SIGNATURE)
                    except ValueError:
                        self._message = value
                        self._signature = None
                    else:
                        self._message = value[:sig_idx]
                        self._signature = value[sig_idx:]
            else:
                raise ObjectFormatException("Unknown field %s" % field)

    def _get_object(self):
        """Get the object pointed to by this tag.

        Returns: tuple of (object class, sha).
        """
        return (self._object_class, self._object_sha)

    def _set_object(self, value):
        (self._object_class, self._object_sha) = value
        self._needs_serialization = True

    object = property(_get_object, _set_object)

    name = serializable_property("name", "The name of this tag")
    tagger = serializable_property(
        "tagger", "Returns the name of the person who created this tag"
    )
    tag_time = serializable_property(
        "tag_time",
        "The creation timestamp of the tag.  As the number of seconds "
        "since the epoch",
    )
    tag_timezone = serializable_property(
        "tag_timezone", "The timezone that tag_time is in."
    )
    message = serializable_property("message", "the message attached to this tag")

    signature = serializable_property("signature", "Optional detached GPG signature")

    def sign(self, keyid: Optional[str] = None):
        import gpg
        with gpg.Context(armor=True) as c:
            if keyid is not None:
                key = c.get_key(keyid)
                with gpg.Context(armor=True, signers=[key]) as ctx:
                    self.signature, unused_result = ctx.sign(
                        self.as_raw_string(),
                        mode=gpg.constants.sig.mode.DETACH,
                    )
            else:
                self.signature, unused_result = c.sign(
                    self.as_raw_string(), mode=gpg.constants.sig.mode.DETACH
                )


class TreeEntry(namedtuple("TreeEntry", ["path", "mode", "sha"])):
    """Named tuple encapsulating a single tree entry."""

    def in_path(self, path):
        """Return a copy of this entry with the given path prepended."""
        if not isinstance(self.path, bytes):
            raise TypeError("Expected bytes for path, got %r" % path)
        return TreeEntry(posixpath.join(path, self.path), self.mode, self.sha)


def parse_tree(text, strict=False):
    """Parse a tree text.

    Args:
      text: Serialized text to parse
    Returns: iterator of tuples of (name, mode, sha)
    Raises:
      ObjectFormatException: if the object was malformed in some way
    """
    count = 0
    length = len(text)
    while count < length:
        mode_end = text.index(b" ", count)
        mode_text = text[count:mode_end]
        if strict and mode_text.startswith(b"0"):
            raise ObjectFormatException("Invalid mode '%s'" % mode_text)
        try:
            mode = int(mode_text, 8)
        except ValueError:
            raise ObjectFormatException("Invalid mode '%s'" % mode_text)
        name_end = text.index(b"\0", mode_end)
        name = text[mode_end + 1 : name_end]
        count = name_end + 21
        sha = text[name_end + 1 : count]
        if len(sha) != 20:
            raise ObjectFormatException("Sha has invalid length")
        hexsha = sha_to_hex(sha)
        yield (name, mode, hexsha)


def serialize_tree(items):
    """Serialize the items in a tree to a text.

    Args:
      items: Sorted iterable over (name, mode, sha) tuples
    Returns: Serialized tree text as chunks
    """
    for name, mode, hexsha in items:
        yield (
            ("%04o" % mode).encode("ascii") + b" " + name + b"\0" + hex_to_sha(hexsha)
        )


def sorted_tree_items(entries, name_order):
    """Iterate over a tree entries dictionary.

    Args:
      name_order: If True, iterate entries in order of their name. If
        False, iterate entries in tree order, that is, treat subtree entries as
        having '/' appended.
      entries: Dictionary mapping names to (mode, sha) tuples
    Returns: Iterator over (name, mode, hexsha)
    """
    key_func = name_order and key_entry_name_order or key_entry
    for name, entry in sorted(entries.items(), key=key_func):
        mode, hexsha = entry
        # Stricter type checks than normal to mirror checks in the C version.
        mode = int(mode)
        if not isinstance(hexsha, bytes):
            raise TypeError("Expected bytes for SHA, got %r" % hexsha)
        yield TreeEntry(name, mode, hexsha)


def key_entry(entry):
    """Sort key for tree entry.

    Args:
      entry: (name, value) tuplee
    """
    (name, value) = entry
    if stat.S_ISDIR(value[0]):
        name += b"/"
    return name


def key_entry_name_order(entry):
    """Sort key for tree entry in name order."""
    return entry[0]


def pretty_format_tree_entry(name, mode, hexsha, encoding="utf-8"):
    """Pretty format tree entry.

    Args:
      name: Name of the directory entry
      mode: Mode of entry
      hexsha: Hexsha of the referenced object
    Returns: string describing the tree entry
    """
    if mode & stat.S_IFDIR:
        kind = "tree"
    else:
        kind = "blob"
    return "%04o %s %s\t%s\n" % (
        mode,
        kind,
        hexsha.decode("ascii"),
        name.decode(encoding, "replace"),
    )


class Tree(ShaFile):
    """A Git tree object"""

    type_name = b"tree"
    type_num = 2

    __slots__ = "_entries"

    def __init__(self):
        super(Tree, self).__init__()
        self._entries = {}

    @classmethod
    def from_path(cls, filename):
        tree = ShaFile.from_path(filename)
        if not isinstance(tree, cls):
            raise NotTreeError(filename)
        return tree

    def __contains__(self, name):
        return name in self._entries

    def __getitem__(self, name):
        return self._entries[name]

    def __setitem__(self, name, value):
        """Set a tree entry by name.

        Args:
          name: The name of the entry, as a string.
          value: A tuple of (mode, hexsha), where mode is the mode of the
            entry as an integral type and hexsha is the hex SHA of the entry as
            a string.
        """
        mode, hexsha = value
        self._entries[name] = (mode, hexsha)
        self._needs_serialization = True

    def __delitem__(self, name):
        del self._entries[name]
        self._needs_serialization = True

    def __len__(self):
        return len(self._entries)

    def __iter__(self):
        return iter(self._entries)

    def add(self, name, mode, hexsha):
        """Add an entry to the tree.

        Args:
          mode: The mode of the entry as an integral type. Not all
            possible modes are supported by git; see check() for details.
          name: The name of the entry, as a string.
          hexsha: The hex SHA of the entry as a string.
        """
        if isinstance(name, int) and isinstance(mode, bytes):
            (name, mode) = (mode, name)
            warnings.warn(
                "Please use Tree.add(name, mode, hexsha)",
                category=DeprecationWarning,
                stacklevel=2,
            )
        self._entries[name] = mode, hexsha
        self._needs_serialization = True

    def iteritems(self, name_order=False):
        """Iterate over entries.

        Args:
          name_order: If True, iterate in name order instead of tree
            order.
        Returns: Iterator over (name, mode, sha) tuples
        """
        return sorted_tree_items(self._entries, name_order)

    def items(self):
        """Return the sorted entries in this tree.

        Returns: List with (name, mode, sha) tuples
        """
        return list(self.iteritems())

    def _deserialize(self, chunks):
        """Grab the entries in the tree"""
        try:
            parsed_entries = parse_tree(b"".join(chunks))
        except ValueError as e:
            raise ObjectFormatException(e)
        # TODO: list comprehension is for efficiency in the common (small)
        # case; if memory efficiency in the large case is a concern, use a
        # genexp.
        self._entries = dict([(n, (m, s)) for n, m, s in parsed_entries])

    def check(self):
        """Check this object for internal consistency.

        Raises:
          ObjectFormatException: if the object is malformed in some way
        """
        super(Tree, self).check()
        last = None
        allowed_modes = (
            stat.S_IFREG | 0o755,
            stat.S_IFREG | 0o644,
            stat.S_IFLNK,
            stat.S_IFDIR,
            S_IFGITLINK,
            # TODO: optionally exclude as in git fsck --strict
            stat.S_IFREG | 0o664,
        )
        for name, mode, sha in parse_tree(b"".join(self._chunked_text), True):
            check_hexsha(sha, "invalid sha %s" % sha)
            if b"/" in name or name in (b"", b".", b"..", b".git"):
                raise ObjectFormatException(
                    "invalid name %s" % name.decode("utf-8", "replace")
                )

            if mode not in allowed_modes:
                raise ObjectFormatException("invalid mode %06o" % mode)

            entry = (name, (mode, sha))
            if last:
                if key_entry(last) > key_entry(entry):
                    raise ObjectFormatException("entries not sorted")
                if name == last[0]:
                    raise ObjectFormatException("duplicate entry %s" % name)
            last = entry

    def _serialize(self):
        return list(serialize_tree(self.iteritems()))

    def as_pretty_string(self):
        text = []
        for name, mode, hexsha in self.iteritems():
            text.append(pretty_format_tree_entry(name, mode, hexsha))
        return "".join(text)

    def lookup_path(self, lookup_obj, path):
        """Look up an object in a Git tree.

        Args:
          lookup_obj: Callback for retrieving object by SHA1
          path: Path to lookup
        Returns: A tuple of (mode, SHA) of the resulting path.
        """
        parts = path.split(b"/")
        sha = self.id
        mode = None
        for p in parts:
            if not p:
                continue
            obj = lookup_obj(sha)
            if not isinstance(obj, Tree):
                raise NotTreeError(sha)
            mode, sha = obj[p]
        return mode, sha


def parse_timezone(text):
    """Parse a timezone text fragment (e.g. '+0100').

    Args:
      text: Text to parse.
    Returns: Tuple with timezone as seconds difference to UTC
        and a boolean indicating whether this was a UTC timezone
        prefixed with a negative sign (-0000).
    """
    # cgit parses the first character as the sign, and the rest
    #  as an integer (using strtol), which could also be negative.
    #  We do the same for compatibility. See #697828.
    if not text[0] in b"+-":
        raise ValueError("Timezone must start with + or - (%(text)s)" % vars())
    sign = text[:1]
    offset = int(text[1:])
    if sign == b"-":
        offset = -offset
    unnecessary_negative_timezone = offset >= 0 and sign == b"-"
    signum = (offset < 0) and -1 or 1
    offset = abs(offset)
    hours = int(offset / 100)
    minutes = offset % 100
    return (
        signum * (hours * 3600 + minutes * 60),
        unnecessary_negative_timezone,
    )


def format_timezone(offset, unnecessary_negative_timezone=False):
    """Format a timezone for Git serialization.

    Args:
      offset: Timezone offset as seconds difference to UTC
      unnecessary_negative_timezone: Whether to use a minus sign for
        UTC or positive timezones (-0000 and --700 rather than +0000 / +0700).
    """
    if offset % 60 != 0:
        raise ValueError("Unable to handle non-minute offset.")
    if offset < 0 or unnecessary_negative_timezone:
        sign = "-"
        offset = -offset
    else:
        sign = "+"
    return ("%c%02d%02d" % (sign, offset / 3600, (offset / 60) % 60)).encode("ascii")


def parse_time_entry(value):
    """Parse time entry behavior

    Args:
      value: Bytes representing a git commit/tag line
    Raises:
      ObjectFormatException in case of parsing error (malformed
      field date)
    Returns: Tuple of (author, time, (timezone, timezone_neg_utc))
    """
    try:
        sep = value.rindex(b"> ")
    except ValueError:
        return (value, None, (None, False))
    try:
        person = value[0 : sep + 1]
        rest = value[sep + 2 :]
        timetext, timezonetext = rest.rsplit(b" ", 1)
        time = int(timetext)
        timezone, timezone_neg_utc = parse_timezone(timezonetext)
    except ValueError as e:
        raise ObjectFormatException(e)
    return person, time, (timezone, timezone_neg_utc)


def parse_commit(chunks):
    """Parse a commit object from chunks.

    Args:
      chunks: Chunks to parse
    Returns: Tuple of (tree, parents, author_info, commit_info,
        encoding, mergetag, gpgsig, message, extra)
    """
    parents = []
    extra = []
    tree = None
    author_info = (None, None, (None, None))
    commit_info = (None, None, (None, None))
    encoding = None
    mergetag = []
    message = None
    gpgsig = None

    for field, value in _parse_message(chunks):
        # TODO(jelmer): Enforce ordering
        if field == _TREE_HEADER:
            tree = value
        elif field == _PARENT_HEADER:
            parents.append(value)
        elif field == _AUTHOR_HEADER:
            author_info = parse_time_entry(value)
        elif field == _COMMITTER_HEADER:
            commit_info = parse_time_entry(value)
        elif field == _ENCODING_HEADER:
            encoding = value
        elif field == _MERGETAG_HEADER:
            mergetag.append(Tag.from_string(value + b"\n"))
        elif field == _GPGSIG_HEADER:
            gpgsig = value
        elif field is None:
            message = value
        else:
            extra.append((field, value))
    return (
        tree,
        parents,
        author_info,
        commit_info,
        encoding,
        mergetag,
        gpgsig,
        message,
        extra,
    )


class Commit(ShaFile):
    """A git commit object"""

    type_name = b"commit"
    type_num = 1

    __slots__ = (
        "_parents",
        "_encoding",
        "_extra",
        "_author_timezone_neg_utc",
        "_commit_timezone_neg_utc",
        "_commit_time",
        "_author_time",
        "_author_timezone",
        "_commit_timezone",
        "_author",
        "_committer",
        "_tree",
        "_message",
        "_mergetag",
        "_gpgsig",
    )

    def __init__(self):
        super(Commit, self).__init__()
        self._parents = []
        self._encoding = None
        self._mergetag = []
        self._gpgsig = None
        self._extra = []
        self._author_timezone_neg_utc = False
        self._commit_timezone_neg_utc = False

    @classmethod
    def from_path(cls, path):
        commit = ShaFile.from_path(path)
        if not isinstance(commit, cls):
            raise NotCommitError(path)
        return commit

    def _deserialize(self, chunks):
        (
            self._tree,
            self._parents,
            author_info,
            commit_info,
            self._encoding,
            self._mergetag,
            self._gpgsig,
            self._message,
            self._extra,
        ) = parse_commit(chunks)
        (
            self._author,
            self._author_time,
            (self._author_timezone, self._author_timezone_neg_utc),
        ) = author_info
        (
            self._committer,
            self._commit_time,
            (self._commit_timezone, self._commit_timezone_neg_utc),
        ) = commit_info

    def check(self):
        """Check this object for internal consistency.

        Raises:
          ObjectFormatException: if the object is malformed in some way
        """
        super(Commit, self).check()
        self._check_has_member("_tree", "missing tree")
        self._check_has_member("_author", "missing author")
        self._check_has_member("_committer", "missing committer")
        self._check_has_member("_author_time", "missing author time")
        self._check_has_member("_commit_time", "missing commit time")

        for parent in self._parents:
            check_hexsha(parent, "invalid parent sha")
        check_hexsha(self._tree, "invalid tree sha")

        check_identity(self._author, "invalid author")
        check_identity(self._committer, "invalid committer")

        check_time(self._author_time)
        check_time(self._commit_time)

        last = None
        for field, _ in _parse_message(self._chunked_text):
            if field == _TREE_HEADER and last is not None:
                raise ObjectFormatException("unexpected tree")
            elif field == _PARENT_HEADER and last not in (
                _PARENT_HEADER,
                _TREE_HEADER,
            ):
                raise ObjectFormatException("unexpected parent")
            elif field == _AUTHOR_HEADER and last not in (
                _TREE_HEADER,
                _PARENT_HEADER,
            ):
                raise ObjectFormatException("unexpected author")
            elif field == _COMMITTER_HEADER and last != _AUTHOR_HEADER:
                raise ObjectFormatException("unexpected committer")
            elif field == _ENCODING_HEADER and last != _COMMITTER_HEADER:
                raise ObjectFormatException("unexpected encoding")
            last = field

        # TODO: optionally check for duplicate parents

    def _serialize(self):
        chunks = []
        tree_bytes = self._tree.id if isinstance(self._tree, Tree) else self._tree
        chunks.append(git_line(_TREE_HEADER, tree_bytes))
        for p in self._parents:
            chunks.append(git_line(_PARENT_HEADER, p))
        chunks.append(
            git_line(
                _AUTHOR_HEADER,
                self._author,
                str(self._author_time).encode("ascii"),
                format_timezone(self._author_timezone, self._author_timezone_neg_utc),
            )
        )
        chunks.append(
            git_line(
                _COMMITTER_HEADER,
                self._committer,
                str(self._commit_time).encode("ascii"),
                format_timezone(self._commit_timezone, self._commit_timezone_neg_utc),
            )
        )
        if self.encoding:
            chunks.append(git_line(_ENCODING_HEADER, self.encoding))
        for mergetag in self.mergetag:
            mergetag_chunks = mergetag.as_raw_string().split(b"\n")

            chunks.append(git_line(_MERGETAG_HEADER, mergetag_chunks[0]))
            # Embedded extra header needs leading space
            for chunk in mergetag_chunks[1:]:
                chunks.append(b" " + chunk + b"\n")

            # No trailing empty line
            if chunks[-1].endswith(b" \n"):
                chunks[-1] = chunks[-1][:-2]
        for k, v in self.extra:
            if b"\n" in k or b"\n" in v:
                raise AssertionError("newline in extra data: %r -> %r" % (k, v))
            chunks.append(git_line(k, v))
        if self.gpgsig:
            sig_chunks = self.gpgsig.split(b"\n")
            chunks.append(git_line(_GPGSIG_HEADER, sig_chunks[0]))
            for chunk in sig_chunks[1:]:
                chunks.append(git_line(b"", chunk))
        chunks.append(b"\n")  # There must be a new line after the headers
        chunks.append(self._message)
        return chunks

    tree = serializable_property("tree", "Tree that is the state of this commit")

    def _get_parents(self):
        """Return a list of parents of this commit."""
        return self._parents

    def _set_parents(self, value):
        """Set a list of parents of this commit."""
        self._needs_serialization = True
        self._parents = value

    parents = property(
        _get_parents,
        _set_parents,
        doc="Parents of this commit, by their SHA1.",
    )

    def _get_extra(self):
        """Return extra settings of this commit."""
        return self._extra

    extra = property(
        _get_extra,
        doc="Extra header fields not understood (presumably added in a "
        "newer version of git). Kept verbatim so the object can "
        "be correctly reserialized. For private commit metadata, use "
        "pseudo-headers in Commit.message, rather than this field.",
    )

    author = serializable_property("author", "The name of the author of the commit")

    committer = serializable_property(
        "committer", "The name of the committer of the commit"
    )

    message = serializable_property("message", "The commit message")

    commit_time = serializable_property(
        "commit_time",
        "The timestamp of the commit. As the number of seconds since the " "epoch.",
    )

    commit_timezone = serializable_property(
        "commit_timezone", "The zone the commit time is in"
    )

    author_time = serializable_property(
        "author_time",
        "The timestamp the commit was written. As the number of "
        "seconds since the epoch.",
    )

    author_timezone = serializable_property(
        "author_timezone", "Returns the zone the author time is in."
    )

    encoding = serializable_property("encoding", "Encoding of the commit message.")

    mergetag = serializable_property("mergetag", "Associated signed tag.")

    gpgsig = serializable_property("gpgsig", "GPG Signature.")


OBJECT_CLASSES = (
    Commit,
    Tree,
    Blob,
    Tag,
)

_TYPE_MAP = {}  # type: Dict[Union[bytes, int], Type[ShaFile]]

for cls in OBJECT_CLASSES:
    _TYPE_MAP[cls.type_name] = cls
    _TYPE_MAP[cls.type_num] = cls


# Hold on to the pure-python implementations for testing
_parse_tree_py = parse_tree
_sorted_tree_items_py = sorted_tree_items
try:
    # Try to import C versions
    from dulwich._objects import parse_tree, sorted_tree_items  # type: ignore
except ImportError:
    pass