Allow reading from fifo in *PcapReader* (#4428)

scapy/utils.py

@@ -1427,12 +1427,14 @@ def open(fname  # type: Union[IO[bytes], str]
         """Open (if necessary) filename, and read the magic."""
         if isinstance(fname, str):
             filename = fname
-            try:
-                fdesc = gzip.open(filename, "rb")  # type: _ByteStream
-                magic = fdesc.read(4)
-            except IOError:
-                fdesc = open(filename, "rb")
-                magic = fdesc.read(4)
+            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
             filename = getattr(fdesc, "name", "No name")
@@ -1558,8 +1560,10 @@ def fileno(self):
         return -1 if WINDOWS else self.f.fileno()
 
     def close(self):
-        # type: () -> Optional[Any]
-        return self.f.close()
+        # type: () -> None
+        if isinstance(self.f, gzip.GzipFile):
+            self.f.fileobj.close()  # type: ignore
+        self.f.close()
 
     def __exit__(self, exc_type, exc_value, tracback):
         # type: (Optional[Any], Optional[Any], Optional[Any]) -> None

test/regression.uts

@@ -2225,6 +2225,18 @@ wrpcapng(tmpfile, p)
 l = rdpcap(tmpfile)
 assert l[0].comment == p.comment
 
+= rdpcap on fifo
+~ linux
+f = get_temp_file()
+os.unlink(f)
+os.mkfifo(f)
+p = Ether(bytes(Ether(dst="ff:ff:ff:ff:ff:ff")/"Hello Scapy!!!"))
+s = AsyncSniffer(offline=f)
+s.start()
+wrpcap(f, p)
+s.join(timeout=1)
+assert s.results[0] == p
+
 = Check multiple packets with different combination of linktype,comment,direction,sniffed_on fields. test both wrpcap() and wrpcapng()
 import random,string
 random.seed(0x2807)