pnio_rpc: bound alarm SDU and items by their lengths

scapy/contrib/pnio_rpc.py

@@ -17,9 +17,9 @@
 from scapy.config import conf
 from scapy.fields import BitField, ByteField, BitEnumField, ByteEnumField, \
     ConditionalField, FieldLenField, FieldListField, IntField, IntEnumField, \
-    LenField, MACField, PadField, PacketField, PacketListField, \
-    ShortEnumField, ShortField, StrField, StrFixedLenField, StrLenField, \
-    UUIDField, XByteField, XIntField, XShortEnumField, XShortField
+    LenField, MACField, PadField, PacketField, PacketLenField, \
+    PacketListField, ShortEnumField, ShortField, StrField, StrFixedLenField, \
+    StrLenField, UUIDField, XByteField, XIntField, XShortEnumField, XShortField
 from scapy.layers.dcerpc import DceRpc4, DceRpc4Payload
 from scapy.contrib.rtps.common_types import EField
 from scapy.compat import bytes_hex
@@ -1316,7 +1316,8 @@ class AlarmNotificationPDU(Block):
         XIntField("ModuleIdentNumber", 0),
         XIntField("SubmoduleIdentNUmber", 0),
         XShortField("AlarmSpecifier", 0),
-        PacketListField("AlarmPayload", [], _guess_alarm_payload)
+        PacketListField("AlarmPayload", [], _guess_alarm_payload,
+                        length_from=lambda pkt: max(0, pkt.block_length - 22))
     ]
 
 
@@ -1357,7 +1358,7 @@ class PNIORealTimeAcyclicPDUHeader(Packet):
         BitField("AddFlags", 0, 8),
         XShortField("SendSeqNum", 0),
         XShortField("AckSeqNum", 0),
-        XShortField("VarPartLen", 0),
+        FieldLenField("VarPartLen", None, length_of="RTA_SDU"),
     ]
 
     def __new__(cls, name, bases, dct):
@@ -1367,16 +1368,24 @@ def __new__(cls, name, bases, dct):
 class Alarm_Low(Packet):
     fields_desc = [
         PNIORealTimeAcyclicPDUHeader,
-        PacketField("RTA_SDU", None, AlarmNotification_Low),
+        PacketLenField("RTA_SDU", None, AlarmNotification_Low,
+                       length_from=lambda pkt: pkt.VarPartLen),
     ]
 
+    def extract_padding(self, s):
+        return None, s  # No extra payload
+
 
 class Alarm_High(Packet):
     fields_desc = [
         PNIORealTimeAcyclicPDUHeader,
-        PacketField("RTA_SDU", None, AlarmNotification_High),
+        PacketLenField("RTA_SDU", None, AlarmNotification_High,
+                       length_from=lambda pkt: pkt.VarPartLen),
     ]
 
+    def extract_padding(self, s):
+        return None, s  # No extra payload
+
 
 # PROFINET IO DCE/RPC PDU
 PNIO_RPC_BLOCK_ASSOCIATION = {

test/contrib/pnio_rpc.uts

@@ -838,6 +838,49 @@ assert p[PRAL_AlarmItem].PRAL_Reason == 0x0003
 assert p[PRAL_AlarmItem].PRAL_ReasonAddValue == b'\x00\x01\x00\x00'
 assert raw(p) == raw_pral
 
+= PNIO Alarm decoding with trailing bytes after the block
+p = AlarmNotification_High(raw_pral + b'\x00\x00')
+assert len(p.AlarmPayload) == 1
+assert p[PRAL_AlarmItem].PRAL_ReasonAddValue == b'\x00\x01\x00\x00'
+assert raw(p) == raw_pral + b'\x00\x00'
+
+= PNIO Alarm decoding with a block_length shorter than the fixed fields
+p = AlarmNotification_Low(b'\x00\x02\x00\x04' + b'\x01\x00' * 23)
+assert p.AlarmPayload == []
+
+= PNIO Alarm decoding of a frame padded to the Ethernet minimum
+p = Ether(b'\x00\x11\x22\x33\x44\x55' \
+          b'\x00\x66\x77\x88\x99\xaa' \
+          b'\x88\x92' \
+          b'\xfe\x01' \
+          b'\x00\x03\x00\x03\x11\x00\x00\x00\xff\xff\x00\x1e' \
+          b'\x00\x02\x00\x1a\x01\x00\x00\x02\x00\x00\x00\x00\x00' \
+          b'\x01\x00\x01\x00\x00\x00\x32\x00\x00\x00\x01\x08\x00' \
+          b'\x83\x00\x00\x01' \
+          b'\x00\x00')
+assert p[Alarm_Low].VarPartLen == 30
+assert len(p[AlarmNotification_Low].AlarmPayload) == 1
+assert p[RS_AlarmItem].RS_AlarmInfo == 0x0001
+assert p[Padding].load == b'\x00\x00'
+
+= PNIO Alarm decoding of an acknowledgement without RTA-SDU
+p = Ether(b'\x00\x11\x22\x33\x44\x55' \
+          b'\x00\x66\x77\x88\x99\xaa' \
+          b'\x88\x92' \
+          b'\xfe\x01' \
+          b'\x00\x03\x00\x03\x13\x00\x00\x00\xff\xff\x00\x00' + b'\x00' * 32)
+assert p[Alarm_Low].RTA_SDU is None
+assert AlarmNotification_Low not in p
+assert len(p[Padding].load) == 32
+
+= PNIO Alarm building computes VarPartLen
+p = Alarm_High(RTA_SDU=AlarmNotification_High(AlarmType=3))
+assert raw(p)[10:12] == b'\x00\x1a'
+p = Alarm_High(raw(p) + b'\x00' * 6)
+assert p.VarPartLen == 26
+assert p[AlarmNotification_High].AlarmType == 3
+assert p[AlarmNotification_High].AlarmPayload == []
+
 = PNIO PDPortDataAdjust Decoding
 raw = bytearray.fromhex('0402280000000000dea000006c9711d182710001000305f9dea000016' \
                         'c9711d1827100a02442df7d0777bc51ddaa4d07addb7075183fc28b00' \