PcapNg - Apple Process Information Block (#4396)

Author: guedou

scapy/packet.py

@@ -100,7 +100,8 @@ class Packet(
         "direction", "sniffed_on",
         # handle snaplen Vs real length
         "wirelen",
-        "comment"
+        "comment",
+        "process_information"
     ]
     name = None
     fields_desc = []  # type: List[AnyField]
@@ -178,6 +179,7 @@ def __init__(self,
         self.direction = None  # type: Optional[int]
         self.sniffed_on = None  # type: Optional[_GlobInterfaceType]
         self.comment = None  # type: Optional[bytes]
+        self.process_information = None  # type: Optional[Dict[str, Any]]
         self.stop_dissection_after = stop_dissection_after
         if _pkt:
             self.dissect(_pkt)

scapy/utils.py

@@ -11,6 +11,7 @@
 from decimal import Decimal
 from io import StringIO
 from itertools import zip_longest
+from uuid import UUID
 
 import array
 import argparse
@@ -1646,7 +1647,8 @@ class RawPcapNgReader(RawPcapReader):
     PacketMetadata = collections.namedtuple("PacketMetadataNg",  # type: ignore
                                             ["linktype", "tsresol",
                                              "tshigh", "tslow", "wirelen",
-                                             "comment", "ifname", "direction"])
+                                             "comment", "ifname", "direction",
+                                             "process_information"])
 
     def __init__(self, filename, fdesc=None, magic=None):  # type: ignore
         # type: (str, IO[bytes], bytes) -> None
@@ -1668,8 +1670,10 @@ def __init__(self, filename, fdesc=None, magic=None):  # type: ignore
                 3: self._read_block_spb,
                 6: self._read_block_epb,
                 10: self._read_block_dsb,
+                0x80000001: self._read_block_pib,
         }
         self.endian = "!"  # Will be overwritten by first SHB
+        self.process_information = []  # type: List[Dict[str, Any]]
 
         if magic != b"\x0a\x0d\x0d\x0a":  # PcapNg:
             raise Scapy_Exception(
@@ -1868,6 +1872,18 @@ def _read_block_epb(self, block, size):
 
         # Parse options
         options = self._read_options(block[opt_offset:])
+
+        process_information = {}
+        for code, value in options.items():
+            if code in [0x8001, 0x8003]:  # PCAPNG_EPB_PIB_INDEX, PCAPNG_EPB_E_PIB_INDEX
+                proc_index = struct.unpack(self.endian + "I", value)[0]
+                if proc_index < len(self.process_information):
+                    key = "proc" if code == 0x8001 else "eproc"
+                    process_information[key] = self.process_information[proc_index]
+                else:
+                    warning("PcapNg: EPB invalid process information index "
+                            "(%d/%d) !" % (proc_index, len(self.process_information)))
+
         comment = options.get(1, None)
         epb_flags_raw = options.get(2, None)
         if epb_flags_raw:
@@ -1884,6 +1900,7 @@ def _read_block_epb(self, block, size):
 
         self._check_interface_id(intid)
         ifname = self.interfaces[intid][2].get('name', None)
+
         return (block[20:20 + caplen][:size],
                 RawPcapNgReader.PacketMetadata(linktype=self.interfaces[intid][0],  # noqa: E501
                                                tsresol=self.interfaces[intid][2]['tsresol'],  # noqa: E501
@@ -1892,7 +1909,8 @@ def _read_block_epb(self, block, size):
                                                wirelen=wirelen,
                                                comment=comment,
                                                ifname=ifname,
-                                               direction=direction))
+                                               direction=direction,
+                                               process_information=process_information))
 
     def _read_block_spb(self, block, size):
         # type: (bytes, int) -> Tuple[bytes, RawPcapNgReader.PacketMetadata]
@@ -1918,7 +1936,8 @@ def _read_block_spb(self, block, size):
                                                wirelen=wirelen,
                                                comment=None,
                                                ifname=None,
-                                               direction=None))
+                                               direction=None,
+                                               process_information={}))
 
     def _read_block_pkt(self, block, size):
         # type: (bytes, int) -> Tuple[bytes, RawPcapNgReader.PacketMetadata]
@@ -1941,7 +1960,8 @@ def _read_block_pkt(self, block, size):
                                                wirelen=wirelen,
                                                comment=None,
                                                ifname=None,
-                                               direction=None))
+                                               direction=None,
+                                               process_information={}))
 
     def _read_block_dsb(self, block, size):
         # type: (bytes, int) -> None
@@ -1995,6 +2015,35 @@ def _read_block_dsb(self, block, size):
         else:
             warning("PcapNg: Unknown DSB secrets type (0x%x)!", secrets_type)
 
+    def _read_block_pib(self, block, _):
+        # type: (bytes, int) -> None
+        """Apple Process Information Block"""
+
+        # Get the Process ID
+        try:
+            dpeb_pid = struct.unpack(self.endian + "I", block[:4])[0]
+            process_information = {"id": dpeb_pid}
+            block = block[4:]
+        except struct.error:
+            warning("PcapNg: DPEB is too small (%d). Cannot get PID!",
+                    len(block))
+            raise EOFError
+
+        # Get Options
+        options = self._read_options(block)
+        for code, value in options.items():
+            if code == 2:
+                process_information["name"] = value.decode("ascii", "backslashreplace")
+            elif code == 4:
+                if len(value) == 16:
+                    process_information["uuid"] = str(UUID(bytes=value))
+                else:
+                    warning("PcapNg: DPEB UUID length is invalid (%d)!",
+                            len(value))
+
+        # Store process information
+        self.process_information.append(process_information)
+
 
 class PcapNgReader(RawPcapNgReader, PcapReader):
 
@@ -2013,7 +2062,7 @@ def read_packet(self, size=MTU, **kwargs):
         rp = super(PcapNgReader, self)._read_packet(size=size)
         if rp is None:
             raise EOFError
-        s, (linktype, tsresol, tshigh, tslow, wirelen, comment, ifname, direction) = rp
+        s, (linktype, tsresol, tshigh, tslow, wirelen, comment, ifname, direction, process_information) = rp  # noqa: E501
         try:
             cls = conf.l2types.num2layer[linktype]  # type: Type[Packet]
             p = cls(s, **kwargs)  # type: Packet
@@ -2031,6 +2080,7 @@ def read_packet(self, size=MTU, **kwargs):
         p.wirelen = wirelen
         p.comment = comment
         p.direction = direction
+        p.process_information = process_information.copy()
         if ifname is not None:
             p.sniffed_on = ifname.decode('utf-8', 'backslashreplace')
         return p

test/regression.uts

@@ -2274,6 +2274,16 @@ for i in range(len(plist)):
     assert type(plist_check[i]) == type(plist[0])
     assert bytes(plist_check[i]) == bytes(plist[i])
 
+= PcapNg - Process Information Block
+
+pib_pcapng_file = BytesIO(b'\n\r\r\n\xbc\x00\x00\x00M<+\x1a\x01\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\x02\x00\x05\x00arm64\x00\x00\x00\x03\x00f\x00Darwin Kernel Version 23.3.0: Thu Dec 21 02:29:41 PST 2023; root:xnu-10002.81.5~11/RELEASE_ARM64_T8122\x00\x00\x04\x00 \x00tcpdump (libpcap version 1.10.1)\x00\x00\x00\x00\xbc\x00\x00\x00\x01\x00\x00\x00 \x00\x00\x00\x01\x00\x00\x00\x00\x00\x08\x00\x02\x00\x03\x00en0\x00\x00\x00\x00\x00 \x00\x00\x00\x01\x00\x00\x80 \x00\x00\x00$\'\x00\x00\x02\x00\x06\x00trustd\x00\x00\x00\x00\x00\x00 \x00\x00\x00\x01\x00\x00\x80$\x00\x00\x00")\x00\x00\x02\x00\x0c\x00mobileassetd\x00\x00\x00\x00$\x00\x00\x00\x06\x00\x00\x00\x90\x00\x00\x00\x00\x00\x00\x00\xfb\x18\x06\x00EcqdB\x00\x00\x00B\x00\x00\x00\xe8\x9f\x80\xfa\x8c\xc6P\xa6\xd8\xd5\x83v\x08\x00E\x00\x004\x00\x00@\x00@\x06\x90T\nh\x01\xc3\xc0\xe5\xdd_\xf4\xb8\x00P\x95\xc3\xcb\x01\xcb\xeb\x11\xe8\x80\x11\x08\x00\x0c\xe6\x00\x00\x01\x01\x08\n\xbe\xb8\xd4\xb3\xbb\x9b4\xbc\x00\x00\x01\x80\x04\x00\x00\x00\x00\x00\x03\x80\x04\x00\x01\x00\x00\x00\x02\x00\x04\x00\x02\x00\x00\x00\x02\x80\x04\x00\x00\x00\x00\x00\x04\x80\x04\x00\x10\x00\x00\x00\x00\x00\x00\x00\x90\x00\x00\x00')
+
+l = rdpcap(pib_pcapng_file)
+assert(len(l) == 1)
+assert(TCP in l[0])
+assert(len(l[0].process_information) == 2)
+assert(l[0].process_information["proc"]["name"] == "trustd")
+
 = OSS-Fuzz Findings
 
 from io import BytesIO