Added support of usbmon.

Author: KOLANICH

scapy/data.py

@@ -115,6 +115,7 @@
 DLT_BLUETOOTH_HCI_H4_WITH_PHDR = 201
 DLT_AX25_KISS = 202
 DLT_PPP_WITH_DIR = 204
+DLT_USB_LINUX_MMAPPED = 220
 DLT_FC_2 = 224
 DLT_CAN_SOCKETCAN = 227
 if OPENBSD:

scapy/layers/usb.py

@@ -8,22 +8,26 @@
 Default USB frames & Basic implementation
 """
 
-# TODO: support USB headers for Linux and Darwin (usbmon/netmon)
+# TODO: support USB headers for Darwin (netmon)
 # https://github.com/wireshark/wireshark/blob/master/epan/dissectors/packet-usb.c  # noqa: E501
 
 import re
 import subprocess
+from enum import Enum, IntEnum
 
 from scapy.config import conf
 from scapy.consts import WINDOWS
 from scapy.compat import chb, plain_str
-from scapy.data import MTU, DLT_USBPCAP
+from scapy.data import DLT_USB_LINUX, DLT_USB_LINUX_MMAPPED, DLT_USBPCAP, MTU
 from scapy.error import warning
-from scapy.fields import ByteField, XByteField, ByteEnumField, LEShortField, \
-    LEShortEnumField, LEIntField, LEIntEnumField, XLELongField, \
-    LenField
+from scapy.fields import ByteEnumField, ByteField, CharEnumField, \
+    ConditionalField, EnumField, LEIntEnumField, LEIntField, LELongField, \
+    LenField, LEShortEnumField, LEShortField, MultipleTypeField, \
+    PacketLenField, StrLenField, XByteField, XLELongField
 from scapy.interfaces import NetworkInterface, InterfaceProvider, \
     network_name, IFACES
+
+from scapy.interfaces import IFACES, InterfaceProvider, NetworkInterface, network_name
 from scapy.packet import Packet, bind_top_down
 from scapy.supersocket import SuperSocket
 from scapy.utils import PcapReader
@@ -89,15 +93,15 @@
 class USBpcap(Packet):
     name = "USBpcap URB"
     fields_desc = [ByteField("headerLen", None),
-                   ByteField("res", 0),
-                   XLELongField("irpId", 0),
-                   LEIntEnumField("usbd_status", 0x0, _usbd_status_codes),
-                   LEShortEnumField("function", 0, _urb_functions),
-                   XByteField("info", 0),
-                   LEShortField("bus", 0),
-                   LEShortField("device", 0),
-                   XByteField("endpoint", 0),
-                   ByteEnumField("transfer", 0, _transfer_types),
+        ByteField("res", 0),
+        XLELongField("irpId", 0),
+        LEIntEnumField("usbd_status", 0x0, _usbd_status_codes),
+        LEShortEnumField("function", 0, _urb_functions),
+        XByteField("info", 0),
+        LEShortField("bus", 0),
+        LEShortField("device", 0),
+        XByteField("endpoint", 0),
+        ByteEnumField("transfer", 0, _transfer_types),
                    LenField("dataLength", None, fmt="<I")]
 
     def post_build(self, p, pay):
@@ -284,3 +288,198 @@ def close(self):
             self.usbpcap_proc.kill()
 
     conf.USBsocket = USBpcapSocket
+
+
+class EndpointNumber(ByteField):
+    def any2i(self, pkt, x):
+        if isinstance(x, tuple):
+            return self.h2i(pkt, x)
+
+        if isinstance(x, int):
+            return x
+
+        return super(EndpointNumber, self).any2i(pkt, x)
+
+    def h2i(self, pkt, x):
+        is_input, endpoint = x
+        return endpoint | (int(is_input) << 7)
+
+    def i2h(self, pkt, x):
+        return bool(x >> 7), x & 0x7F
+
+    def i2repr(self, pkt, val):
+        is_input, endpoint = self.i2h(pkt, val)
+        return "%s 0x%x" % (u"\u2190" if is_input else u"\u2192", endpoint)
+
+
+# we have to do it since when using `PacketLenField`:
+# `.parent` is not populated, so we cannot use `MultipleTypeField`
+# alternatives are always evaluated no matter what
+# no way to provide parameters even with severe perversions
+setup_common_fields = [
+    LEIntField("interval", None),
+    LEIntField("start_frame", None),
+    LEIntField("copy_of_urb_transfer_flags", None),
+    LEIntField("iso_descriptors_count", None),
+]
+
+
+class SetupSetup(Packet):
+    name = "Setup"
+
+    class PcapUsbSetup(Packet):
+        """USB setup header as defined in USB specification.
+        Appears at the front of each Control S-type packet in DLT_USB captures.
+        """
+
+        name = "Setup"
+        fields_desc = [
+            XByteField("request_type", None),  # 1
+            XByteField("request", None),  # 1
+            LEShortField("value", None),  # 2
+            LEShortField("index", None),  # 2
+            LEShortField("length", None),  # 2
+        ]
+
+    fields_desc = [
+        PacketLenField("s", None, PcapUsbSetup, length_from=lambda pkt: 8),
+    ] + setup_common_fields
+
+
+class SetupIsocr(Packet):
+    name = "Setup"
+
+    class IsoRec(Packet):
+        """Information from the URB for Isochronous transfers.
+
+        .. seealso::
+        Source - https://github.com/the-tcpdump-group/libpcap/blob/ba0ef0353ed9f9f49a1edcfb49fefaf12dec54de/pcap/usb.h#L70
+        """
+
+        name = "IsoRec"
+        fields_desc = [
+            LEIntField("error_count", None),  # 4
+            LEIntField("descriptors_count", None),  # 4
+        ]
+
+    fields_desc = [
+        PacketLenField("s", None, IsoRec, length_from=lambda pkt: 8),
+    ] + setup_common_fields
+
+
+class USBMon(Packet):
+    """A native pcap header of `usbmon <https://www.kernel.org/doc/Documentation/usb/usbmon.txt>`__ part of libpcap and Linux kernel.
+
+    .. seealso::
+    Source - https://github.com/the-tcpdump-group/libpcap/blob/ba0ef0353ed9f9f49a1edcfb49fefaf12dec54de/pcap/usb.h#L94
+
+
+    .. seealso::
+    Source - https://www.kernel.org/doc/Documentation/usb/usbmon.txt
+
+
+    .. seealso::
+    Source - https://www.kernel.org/doc/html/latest/driver-api/usb/URB.html
+
+
+    .. seealso::
+    Source - https://wiki.wireshark.org/USB
+    """
+
+    HEADER_SIZE = None
+    name = "USBMonHeader"
+
+    class EventType(Enum):
+        completion = b"C"
+        error = b"E"
+        submit = b"S"
+
+    class TransferType(IntEnum):
+        isochronous = 0
+        interrupt = 1
+        control = 2
+        bulk = 3
+
+    class SetupFlag(Enum):
+        relevant = b"\0"
+        irrelevant = b"-"
+
+    class DataFlag(Enum):
+        urb = b"\0"
+        incoming = b"<"
+        outgoing = b">"
+        error = b"E"
+
+    class TimeStamp(Packet):
+        name = "TimeStamp"
+
+        @staticmethod
+        def _getSize(*_args):
+            return 12
+
+        fields_desc = [
+            LELongField("seconds", None),  # 8
+            LEIntField("microseconds", None),  # 4
+        ]
+
+    @property
+    def needs_setup(self):
+        SetupFlag = self.__class__.SetupFlag
+        return SetupFlag(self.setup_flag) == SetupFlag.relevant
+
+    @property
+    def is_isochr(self):
+        TransferType = self.__class__.TransferType
+        return TransferType(self.transfer_type) == TransferType.isochronous
+
+    HEADER_STATIC_PART_SIZE = (
+        8 + 1 + 1 + 1 + 1 + 2 + 1 + 1
+        + TimeStamp._getSize() +  # pylint:disable=protected-access
+        4 + 4 + 4
+    )
+
+    def _getOptionalPartSize(pkt=None):
+        return 24
+
+    def _getPaddingSize(pkt):
+        res = pkt.__class__.HEADER_SIZE - __class__._getOptionalPartSize() * int(pkt.needs_setup) - __class__.HEADER_STATIC_PART_SIZE  # pylint:disable=protected-access
+        return res
+
+    fields_desc = [
+        LELongField("urb_id", None),  # 8
+        CharEnumField("event_type", b"\0", EventType),  # 1
+        EnumField("transfer_type", 0, TransferType, "<B"),  # 1
+        EndpointNumber("endpoint_number", 0),  # 1
+        XByteField("device_address", None),  # 1
+        LEShortField("bus_id", None),  # 2
+        CharEnumField("setup_flag", b"\0", SetupFlag),  # 1
+        CharEnumField("data_flag", b"\0", DataFlag),  # 1
+        # just PacketField doesn't work and breaks everything after it
+        PacketLenField("timestamp", None, TimeStamp, length_from=TimeStamp._getSize),  # 12, pylint:disable=protected-access
+        LEIntEnumField("status", 0x0, _usbd_status_codes),  # 4
+        LEIntField("urb_size", 0),  # 4
+        LenField("data_size", 0, fmt="<i"),  # 4
+        ConditionalField(
+            MultipleTypeField(
+                [
+                    # just PacketField doesn't work and breaks everything after it
+                    (PacketLenField("setup", None, SetupIsocr, length_from=_getOptionalPartSize), lambda pkt: pkt.is_isochr,),
+                ],
+                PacketLenField("setup", None, SetupSetup, length_from=_getOptionalPartSize),
+            ),
+            lambda pkt: pkt.needs_setup,
+        ),  # 24
+        StrLenField("padding", None, length_from=_getPaddingSize),
+    ]
+
+
+class USBMonSimple(USBMon):
+    HEADER_SIZE = 48
+
+
+class USBMonMMapped(USBMon):
+    HEADER_SIZE = 64
+
+
+conf.l2types.register(DLT_USB_LINUX, USBMonSimple)
+conf.l2types.register(DLT_USB_LINUX_MMAPPED, USBMonMMapped)

setup.py

@@ -59,6 +59,9 @@ def process_ignore_tags(buffer):
             'sphinx>=3.0.0',
             'sphinx_rtd_theme>=0.4.3',
             'tox>=3.0.0'
+        ],
+        'usbmon': [
+            'enum34 ; python_version < "3.5"'
         ]
     },
     # We use __file__ in scapy/__init__.py, therefore Scapy isn't zip safe

test/pcaps/usbmon.pcap

@@ -0,0 +1,5 @@
+SPDX-FileCopyrightText: 2019 The usbmon-tools Authors
+
+SPDX-License-Identifier: Unlicense
+
+Source: https://github.com/Flameeyes/usbmon-tools/blob/main/testdata/test1.pcap