fix: validate object types when loading binary SHAs

.basedpyright/baseline.json

@@ -77,14 +77,6 @@
           "lineCount": 1
         }
       },
-      {
-        "code": "reportAssignmentType",
-        "range": {
-          "startColumn": 38,
-          "endColumn": 76,
-          "lineCount": 1
-        }
-      },
       {
         "code": "reportArgumentType",
         "range": {

git/diff.py

@@ -11,6 +11,7 @@
 
 from git.cmd import Git, handle_process_output
 from git.compat import defenc
+from git.objects.base import IndexObject
 from git.objects.blob import Blob
 from git.objects.util import mode_str_to_int
 from git.util import finalize_process, hex_to_bin
@@ -35,7 +36,6 @@
 if TYPE_CHECKING:
     from subprocess import Popen
 
-    from git.objects.base import IndexObject
     from git.objects.commit import Commit
     from git.objects.tree import Tree
     from git.repo.base import Repo
@@ -378,6 +378,10 @@ class Diff:
     Diffs keep information about the changed blob objects, the file mode, renames,
     deletions and new files.
 
+    For submodule changes, ``a_blob`` and ``b_blob`` are
+    :class:`~git.objects.base.IndexObject` instances whose SHAs refer to commits in
+    the submodule repository.
+
     There are a few cases where ``None`` has to be expected as member variable value:
 
     New File::
@@ -481,17 +485,22 @@ def __init__(
                         repo = submodule.module()
                     break
 
+        # Gitlinks reference commits; generic index objects preserve their path and mode.
         self.a_blob: Union["IndexObject", None]
         if a_blob_id is None or a_blob_id == self.NULL_HEX_SHA:
             self.a_blob = None
         else:
-            self.a_blob = Blob(repo, hex_to_bin(a_blob_id), mode=self.a_mode, path=self.a_path)
+            self.a_blob = (IndexObject if self.a_mode == 0o160000 else Blob)(
+                repo, hex_to_bin(a_blob_id), mode=self.a_mode, path=self.a_path
+            )
 
         self.b_blob: Union["IndexObject", None]
         if b_blob_id is None or b_blob_id == self.NULL_HEX_SHA:
             self.b_blob = None
         else:
-            self.b_blob = Blob(repo, hex_to_bin(b_blob_id), mode=self.b_mode, path=self.b_path)
+            self.b_blob = (IndexObject if self.b_mode == 0o160000 else Blob)(
+                repo, hex_to_bin(b_blob_id), mode=self.b_mode, path=self.b_path
+            )
 
         self.new_file: bool = new_file
         self.deleted_file: bool = deleted_file

git/index/base.py

@@ -310,7 +310,7 @@ def merge_tree(
         return self
 
     @classmethod
-    def new(cls, repo: "Repo", *tree_sha: Union[str, Tree]) -> "IndexFile":
+    def new(cls, repo: "Repo", *tree_sha: Union[str, bytes, Tree]) -> "IndexFile":
         """Merge the given treeish revisions into a new index which is returned.
 
         This method behaves like ``git-read-tree --aggressive`` when doing the merge.
@@ -326,7 +326,9 @@ def new(cls, repo: "Repo", *tree_sha: Union[str, Tree]) -> "IndexFile":
             If you intend to write such a merged Index, supply an alternate
             ``file_path`` to its :meth:`write` method.
         """
-        tree_sha_bytes: List[bytes] = [to_bin_sha(str(t)) for t in tree_sha]
+        tree_sha_bytes: List[bytes] = [
+            to_bin_sha(t if isinstance(t, bytes) else str(t).encode("ascii")) for t in tree_sha
+        ]
         base_entries = aggressive_tree_merge(repo.odb, tree_sha_bytes)
 
         inst = cls(repo)

git/objects/base.py

@@ -9,6 +9,7 @@
 
 import gitdb.typ as dbtyp
 
+from git.compat import force_text
 from git.exc import WorkTreeRepositoryUnsupported
 from git.util import LazyMixin, bin_to_hex, join_path_native, stream_copy
 
@@ -107,6 +108,10 @@ def __init__(self, repo: "Repo", binsha: bytes) -> None:
 
         :param binsha:
             20 byte SHA1
+
+        :note:
+            Object data is loaded lazily. Loading uncached :attr:`size` metadata
+            raises :exc:`ValueError` if `binsha` refers to a different object type.
         """
         super().__init__()
         self.repo = repo
@@ -155,6 +160,9 @@ def _set_cache_(self, attr: str) -> None:
         """Retrieve object information."""
         if attr == "size":
             oinfo = self.repo.odb.info(self.binsha)
+            typename = force_text(oinfo.type, "ascii")
+            if self.type is not None and typename != self.type:
+                raise ValueError("Object %s is a %s, not a %s" % (self.hexsha, typename, self.type))
             self.size = oinfo.size  # type: int
         else:
             super()._set_cache_(attr)

test/test_base.py

@@ -4,11 +4,15 @@
 # 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
 
 import gc
+from io import BytesIO
 import os
 import os.path as osp
 import sys
 import tempfile
 from unittest import skipIf
+from unittest.mock import patch
+
+from gitdb import OInfo
 
 from git import Repo
 from git.objects import Blob, Commit, TagObject, Tree
@@ -76,6 +80,20 @@ def test_base_object(self):
             # Remove the file this way, instead of with a context manager or "finally",
             # so it is only removed on success, and we can inspect the file on failure.
             os.remove(tmpfile.name)
+
+            for stored_type in (typename, typename.encode("ascii")):
+                with patch.object(self.rorepo.odb, "info", return_value=OInfo(binsha, stored_type, item.size)):
+                    self.assertEqual(obj_type(self.rorepo, binsha).size, item.size)
+                    for wrong_type in types:
+                        if wrong_type is obj_type:
+                            continue
+                        invalid = wrong_type(self.rorepo, binsha)
+                        with self.assertRaisesRegex(ValueError, f"{hexsha}.*{typename}.*{wrong_type.type}"):
+                            invalid.size
+                        self.assertEqual(invalid.data_stream.read(), data)
+                        ostream = BytesIO()
+                        invalid.stream_data(ostream)
+                        self.assertEqual(ostream.getvalue(), data)
         # END for each object type to create
 
         # Each has a unique sha.

test/test_diff.py

@@ -348,7 +348,7 @@ def test_diff_submodule(self):
         with open(self.submodule_dir + "/subfile", "w") as sub_subfile:
             sub_subfile.write("")
         sub.index.add(["subfile"])
-        sub.index.commit("first commit")
+        first_commit = sub.index.commit("first commit")
 
         # Init a temp git repo that will incorporate the submodule.
         repo = Repo.init(self.repo_dir)
@@ -364,7 +364,7 @@ def test_diff_submodule(self):
         with open(self.repo_dir + "/sub/subfile", "w") as foo_sub_subfile:
             foo_sub_subfile.write("blub")
         submodule.module().index.add(["subfile"])
-        submodule.module().index.commit("changed subfile")
+        changed_commit = submodule.module().index.commit("changed subfile")
         submodule.binsha = submodule.module().head.commit.binsha
 
         # Commit submodule updates in parent repo.
@@ -373,11 +373,10 @@ def test_diff_submodule(self):
         repo.create_tag("2")
 
         diff = repo.commit("1").diff(repo.commit("2"))[0]
-        # If diff is unable to find the commit hashes (looks in wrong repo) the
-        # *_blob.size property will be a string containing exception text, an int
-        # indicates success.
-        self.assertIsInstance(diff.a_blob.size, int)
-        self.assertIsInstance(diff.b_blob.size, int)
+        # Gitlinks refer to commits in the submodule's object database.
+        for item, commit in ((diff.a_blob, first_commit), (diff.b_blob, changed_commit)):
+            self.assertEqual(item.size, commit.size)
+            self.assertEqual(item.data_stream.read(), commit.data_stream.read())
 
     def test_diff_rejects_unsafe_output_options(self):
         commit = self.rorepo.head.commit

test/test_index.py

@@ -1030,7 +1030,7 @@ def test_index_new(self):
         H = self.rorepo.tree("25dca42bac17d511b7e2ebdd9d1d679e7626db5f")
         M = self.rorepo.tree("e746f96bcc29238b79118123028ca170adc4ff0f")
 
-        for args in ((B,), (B, H), (B, H, M)):
+        for args in ((B.binsha,), (B.hexsha, H), (B, H.binsha, M.hexsha.encode("ascii"))):
             index = IndexFile.new(self.rorepo, *args)
             assert isinstance(index, IndexFile)
         # END for each arg tuple