Fixing #4608

Author: polybassa

scapy/contrib/automotive/someip.py

@@ -8,6 +8,7 @@
 # scapy.contrib.status = loads
 
 import struct
+from typing import Type
 
 from scapy.layers.inet import TCP, UDP
 from scapy.layers.inet6 import IP6Field
@@ -19,7 +20,8 @@
                           ShortField, X3BytesField, StrLenField, IPField,
                           FieldLenField, PacketListField, XIntField,
                           MultipleTypeField, FlagsField, IntField,
-                          XByteEnumField, BitScalingField)
+                          XByteEnumField, BitScalingField, LenField)
+from scapy.utils import hexdump
 
 
 class SOMEIP(Packet):
@@ -73,7 +75,7 @@ class SOMEIP(Packet):
             ],
             XShortField("sub_id", 0),
         ),
-        IntField("len", None),
+        LenField("len", None, fmt=">I", adjust=lambda x: x+8),
         XShortField("client_id", 0),
         XShortField("session_id", 0),
         XByteField("proto_ver", PROTOCOL_VERSION),
@@ -117,12 +119,10 @@ class SOMEIP(Packet):
         ConditionalField(
             BitField("more_seg", 0, 1),
             lambda pkt: SOMEIP._is_tp(pkt)),  # noqa: E501
-        ConditionalField(PacketListField(
+        PacketListField(
             "data", [Raw()], Raw,
             length_from=lambda pkt: pkt.len - (SOMEIP.LEN_OFFSET_TP if (SOMEIP._is_tp(pkt) and (pkt.len is None or pkt.len >= SOMEIP.LEN_OFFSET_TP)) else SOMEIP.LEN_OFFSET),  # noqa: E501
-            next_cls_cb=lambda pkt, lst, cur, remain:
-                SOMEIP.get_payload_cls_by_srv_id(pkt, lst, cur, remain)),
-            lambda pkt: SOMEIP._is_tp(pkt))  # noqa: E501
+            next_cls_cb=lambda pkt, lst, cur, remain: SOMEIP.get_payload_cls_by_srv_id(pkt, lst, cur, remain))  # noqa: E501
     ]
 
     payload_cls_by_srv_id = dict()  # To be customized
@@ -131,6 +131,10 @@ class SOMEIP(Packet):
     def get_payload_cls_by_srv_id(pkt, lst, cur, remain):
         return SOMEIP.payload_cls_by_srv_id.get(pkt.srv_id, Raw)
 
+    def default_payload_class(self, payload):
+        # type: (bytes) -> Type[Packet]
+        return SOMEIP
+
     def post_build(self, pkt, pay):
         length = self.len
         if length is None:

test/contrib/automotive/someip.uts

@@ -727,3 +727,43 @@ _opts_check(opts + opts[::-1])
 p = SOMEIP(srv_id=1234, sub_id=4321, msg_type=0xff, retcode=0xff, offset=4294967040, data=[Raw(b"deadbeef")])
 
 assert p.data[0].load == b"deadbeef"
+
+= SOMEIP multiple frames in one TCP/UDP
+
+payload_3 = bytes.fromhex("deadbeef")
+someip_3 = SOMEIP(srv_id=0xabcd, sub_id=0x8001, len=8 + len(payload_3))
+someip_3.payload = Raw(load=payload_3)
+
+payload_2 = bytes.fromhex("ff")
+someip_23 = SOMEIP(srv_id=0x5678, sub_id=0x8002, len=8 + len(payload_2))
+someip_23.payload = Raw(load=payload_2 + bytes(someip_3))
+
+payload_1 = bytes.fromhex("0000")
+someip_123 = SOMEIP(srv_id=0x1234, sub_id=0x8001, len=8 + len(payload_1))
+someip_123.payload = Raw(load=payload_1 + bytes(someip_23))
+
+eth_frame = (
+    Ether(src="00:11:22:33:44:55", dst="AA:BB:CC:DD:EE:FF")
+    / IP(src="192.168.0.10", dst="192.168.0.20")
+    / UDP(sport=30501, dport=30491)
+    / someip_123
+)
+
+pkt = Ether(bytes(eth_frame))
+
+
+pkt.show()
+layers = pkt.layers()
+assert len(layers) == 6
+assert layers[-1] == SOMEIP
+assert layers[-2] == SOMEIP
+assert layers[-3] == SOMEIP
+
+
+someip_123_x = pkt[SOMEIP]
+
+assert someip_123_x.data[0].load == payload_1
+someip_23_x = someip_123_x.payload
+assert someip_23_x.data[0].load == payload_2
+someip_3_x = someip_23_x.payload
+assert someip_3_x.data[0].load == payload_3