Rewind file pointer and be compatible with 3.13.13 (#84)

Author: twiggler

dissect/hypervisor/util/vmtar.py

@@ -22,9 +22,8 @@ class VisorTarInfo(tarfile.TarInfo):
     fixUpPgs: int | None
 
     @classmethod
-    def frombuf(cls, buf: bytes, encoding: str, errors: str) -> VisorTarInfo:
-        obj = super().frombuf(buf, encoding, errors)
-
+    def _init_visor_attrs(cls, obj: VisorTarInfo, buf: bytes) -> None:
+        """Initialize visor-specific attributes from the raw header buffer."""
         obj.is_visor = buf[257:264] == b"visor  "
         if obj.is_visor:
             obj.offset_data = struct.unpack("<I", buf[496:500])[0]
@@ -35,6 +34,19 @@ def frombuf(cls, buf: bytes, encoding: str, errors: str) -> VisorTarInfo:
             obj.textPgs = None
             obj.fixUpPgs = None
 
+    @classmethod
+    def frombuf(cls, buf: bytes, encoding: str, errors: str) -> VisorTarInfo:
+        obj = super().frombuf(buf, encoding, errors)
+        cls._init_visor_attrs(obj, buf)
+        return obj
+
+    @classmethod
+    def _frombuf(cls, buf: bytes, encoding: str, errors: str, **kwargs) -> VisorTarInfo:
+        # Python 3.13.13+ refactored tarfile to call _frombuf directly instead of
+        # frombuf in fromtarfile and _proc_gnulong/_proc_pax. We must override _frombuf
+        # to ensure visor-specific attributes are initialized for all code paths.
+        obj = super()._frombuf(buf, encoding, errors, **kwargs)
+        cls._init_visor_attrs(obj, buf)
         return obj
 
     def _proc_member(self, tarfile: tarfile.TarFile) -> VisorTarInfo | tarfile.TarInfo:
@@ -76,6 +88,8 @@ def visoropen(cls, name: str, mode: str = "r", fileobj: BinaryIO | None = None,
         try:
             t = cls.taropen(name, mode, fileobj, **kwargs)
         except Exception:
+            if fileobj is not None:
+                fileobj.seek(0)
             try:
                 fileobj = GzipFile(name, mode + "b", fileobj=fileobj)
             except OSError as e:
@@ -103,10 +117,11 @@ def visoropen(cls, name: str, mode: str = "r", fileobj: BinaryIO | None = None,
             compressed = True
 
         # If we get here, we have a valid visor tar file
-        if fileobj is not None and compressed:
-            # Just read the entire file into memory, it's probably small
+        if fileobj is not None:
             fileobj.seek(0)
-            fileobj = BytesIO(fileobj.read())
+            if compressed:
+                # Read the decompressed data into a BytesIO to make random reads faster
+                fileobj = BytesIO(fileobj.read())
 
         t = cls.taropen(name, mode, fileobj, **kwargs)
 

tests/util/test_vmtar.py

@@ -21,6 +21,7 @@ def test_vmtar() -> None:
         # The test file has no textPgs/fixUpPgs
         assert all(member.is_visor for member in members.values())
         assert set(members.keys()) == {
+            "test",
             "test/file1",
             "test/file2",
             "test/file3",