PcapReader_metaclass: Refactor open

scapy/utils.py

@@ -1410,24 +1410,30 @@ def __call__(cls, filename):
         raise Scapy_Exception("Not a supported capture file")
 
     @staticmethod
-    def open(fname  # type: Union[IO[bytes], str]
-             ):
-        # type: (...) -> Tuple[str, _ByteStream, bytes]
-        """Open (if necessary) filename, and read the magic."""
-        if isinstance(fname, str):
-            filename = fname
+    def _open_stream(inpt  # type: Union[IO[bytes], str]
+                     ):
+        # type: (...) -> Tuple[str, _ByteStream]
+        """Open (if necessary) filename, and obtain a stream."""
+        if isinstance(inpt, str):
+            filename = inpt
             fdesc = open(filename, "rb")  # type: _ByteStream
-            magic = fdesc.read(2)
-            if magic == b"\x1f\x8b":
-                # GZIP header detected.
-                fdesc.seek(0)
-                fdesc = gzip.GzipFile(fileobj=fdesc)
-                magic = fdesc.read(2)
-            magic += fdesc.read(2)
         else:
-            fdesc = fname
+            fdesc = inpt
             filename = getattr(fdesc, "name", "No name")
-            magic = fdesc.read(4)
+        return [filename, fdesc]
+
+    @staticmethod
+    def open(inpt  # type: Union[IO[bytes], str]
+             ):
+        # type: (...) -> Tuple[str, _ByteStream, bytes]
+        """Open (if necessary) filename, obtain a stream and read the magic."""
+        filename, fdesc = PcapReader_metaclass._open_stream(inpt)
+        magic_short = fdesc.read(2)
+        fdesc.seek(0)
+        if magic_short == b"\x1f\x8b":
+            # GZIP header detected.
+            fdesc = gzip.GzipFile(fileobj=fdesc)
+        magic = fdesc.read(4)
         return filename, fdesc, magic