UDS: more precise parsing of DTCs (#4094)

scapy/contrib/automotive/uds.py

@@ -16,7 +16,8 @@
     BitEnumField, BitField, XByteField, FieldListField, \
     XShortField, X3BytesField, XIntField, ByteField, \
     ShortField, ObservableDict, XShortEnumField, XByteEnumField, StrLenField, \
-    FieldLenField, XStrFixedLenField, XStrLenField
+    FieldLenField, XStrFixedLenField, XStrLenField, FlagsField, PacketListField, \
+    PacketField
 from scapy.packet import Packet, bind_layers, NoPayload
 from scapy.config import conf
 from scapy.error import log_loading
@@ -956,12 +957,39 @@ class UDS_RDTCI(Packet):
         20: 'reportDTCFaultDetectionCounter',
         21: 'reportDTCWithPermanentStatus'
     }
+    dtcStatus = {
+        1: 'TestFailed',
+        2: 'TestFailedThisOperationCycle',
+        4: 'PendingDTC',
+        8: 'ConfirmedDTC',
+        16: 'TestNotCompletedSinceLastClear',
+        32: 'TestFailedSinceLastClear',
+        64: 'TestNotCompletedThisOperationCycle',
+        128: 'WarningIndicatorRequested'
+    }
+    dtcStatusMask = {
+        1: 'ActiveDTCs',
+        4: 'PendingDTCs',
+        8: 'ConfirmedOrStoredDTCs',
+        255: 'AllRecordDTCs'
+    }
+    dtcSeverityMask = {
+        # 0: 'NoSeverityInformation',
+        1: 'NoClassInformation',
+        2: 'WWH-OBDClassA',
+        4: 'WWH-OBDClassB1',
+        8: 'WWH-OBDClassB2',
+        16: 'WWH-OBDClassC',
+        32: 'MaintenanceRequired',
+        64: 'CheckAtNextHalt',
+        128: 'CheckImmediately'
+    }
     name = 'ReadDTCInformation'
     fields_desc = [
         ByteEnumField('reportType', 0, reportTypes),
-        ConditionalField(ByteField('DTCSeverityMask', 0),
+        ConditionalField(FlagsField('DTCSeverityMask', 0, 8, dtcSeverityMask),
                          lambda pkt: pkt.reportType in [0x07, 0x08]),
-        ConditionalField(XByteField('DTCStatusMask', 0),
+        ConditionalField(FlagsField('DTCStatusMask', 0, 8, dtcStatusMask),
                          lambda pkt: pkt.reportType in [
                              0x01, 0x02, 0x07, 0x08, 0x0f, 0x11, 0x12, 0x13]),
         ConditionalField(ByteField('DTCHighByte', 0),
@@ -983,16 +1011,47 @@ class UDS_RDTCI(Packet):
 bind_layers(UDS, UDS_RDTCI, service=0x19)
 
 
+class DTC(Packet):
+    name = 'Diagnostic Trouble Code'
+    fields_desc = [
+        BitEnumField("system", 0, 2, {
+            0: "Powertrain",
+            1: "Chassis",
+            2: "Body",
+            3: "Network"}),
+        BitEnumField("type", 0, 2, {
+            0: "Generic",
+            1: "ManufacturerSpecific",
+            2: "Generic",
+            3: "Generic"}),
+        BitField("numeric_value_code", 0, 12),
+        ByteField("additional_information_code", 0),
+    ]
+
+    def extract_padding(self, s):
+        return '', s
+
+
+class DTC_Status(Packet):
+    name = 'DTC and status record'
+    fields_desc = [
+        PacketField("dtc", None, pkt_cls=DTC),
+        FlagsField("status", 0, 8, UDS_RDTCI.dtcStatus)
+    ]
+
+    def extract_padding(self, s):
+        return '', s
+
+
 class UDS_RDTCIPR(Packet):
     name = 'ReadDTCInformationPositiveResponse'
     fields_desc = [
         ByteEnumField('reportType', 0, UDS_RDTCI.reportTypes),
-        ConditionalField(XByteField('DTCStatusAvailabilityMask', 0),
-                         lambda pkt: pkt.reportType in [0x01, 0x07, 0x11,
-                                                        0x12, 0x02, 0x0A,
-                                                        0x0B, 0x0C, 0x0D,
-                                                        0x0E, 0x0F, 0x13,
-                                                        0x15]),
+        ConditionalField(
+            FlagsField('DTCStatusAvailabilityMask', 0, 8, UDS_RDTCI.dtcStatus),
+            lambda pkt: pkt.reportType in [0x01, 0x07, 0x11, 0x12, 0x02, 0x0A,
+                                           0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x13,
+                                           0x15]),
         ConditionalField(ByteEnumField('DTCFormatIdentifier', 0,
                                        {0: 'ISO15031-6DTCFormat',
                                         1: 'UDS-1DTCFormat',
@@ -1003,7 +1062,8 @@ class UDS_RDTCIPR(Packet):
         ConditionalField(ShortField('DTCCount', 0),
                          lambda pkt: pkt.reportType in [0x01, 0x07,
                                                         0x11, 0x12]),
-        ConditionalField(StrField('DTCAndStatusRecord', b""),
+        ConditionalField(PacketListField('DTCAndStatusRecord', None,
+                                         pkt_cls=DTC_Status),
                          lambda pkt: pkt.reportType in [0x02, 0x0A, 0x0B,
                                                         0x0C, 0x0D, 0x0E,
                                                         0x0F, 0x13, 0x15]),
@@ -1259,24 +1319,24 @@ def _contains_data_format_identifier(packet):
                                        fmt='B'),
                          lambda p: p.modeOfOperation != 2),
         ConditionalField(StrLenField('maxNumberOfBlockLength', b"",
-                         length_from=lambda p: p.lengthFormatIdentifier),
+                                     length_from=lambda p: p.lengthFormatIdentifier),
                          lambda p: p.modeOfOperation != 2),
         ConditionalField(BitField('compressionMethod', 0, 4),
                          lambda p: p.modeOfOperation != 0x02),
         ConditionalField(BitField('encryptingMethod', 0, 4),
                          lambda p: p.modeOfOperation != 0x02),
         ConditionalField(FieldLenField('fileSizeOrDirInfoParameterLength',
-                         None,
-                         length_of='fileSizeUncompressedOrDirInfoLength'),
+                                       None,
+                                       length_of='fileSizeUncompressedOrDirInfoLength'),
                          lambda p: p.modeOfOperation not in [1, 2, 3]),
         ConditionalField(StrLenField('fileSizeUncompressedOrDirInfoLength',
                                      b"",
                                      length_from=lambda p:
                                      p.fileSizeOrDirInfoParameterLength),
                          lambda p: p.modeOfOperation not in [1, 2, 3]),
         ConditionalField(StrLenField('fileSizeCompressed', b"",
-                         length_from=lambda p:
-                         p.fileSizeOrDirInfoParameterLength),
+                                     length_from=lambda p:
+                                     p.fileSizeOrDirInfoParameterLength),
                          lambda p: p.modeOfOperation not in [1, 2, 3, 5]),
     ]
 

test/contrib/automotive/uds.uts

@@ -1046,6 +1046,20 @@ assert rdtcipr.DTCCount == 0xddaa
 
 assert rdtcipr.answers(rdtci)
 
+rdtcipr1 = UDS(b'\x59\x02\xff\x11\x07\x11\'\x022\x12\'\x01\x07\x11\'\x01\x18\x12\'\x01\x13\x12\'\x01"\x11\'\x06C\x00\'\x06S\x00\'\x161\x00\'\x14\x03\x12\'')
+
+assert len(rdtcipr1.DTCAndStatusRecord) == 10
+assert rdtcipr1.DTCAndStatusRecord[0].dtc.system == 0
+assert rdtcipr1.DTCAndStatusRecord[0].dtc.type == 1
+assert rdtcipr1.DTCAndStatusRecord[0].dtc.numeric_value_code == 263
+assert rdtcipr1.DTCAndStatusRecord[0].dtc.additional_information_code == 17
+assert rdtcipr1.DTCAndStatusRecord[0].status == 0x27
+assert rdtcipr1.DTCAndStatusRecord[-1].dtc.system == 0
+assert rdtcipr1.DTCAndStatusRecord[-1].dtc.type == 1
+assert rdtcipr1.DTCAndStatusRecord[-1].dtc.numeric_value_code == 1027
+assert rdtcipr1.DTCAndStatusRecord[-1].dtc.additional_information_code == 18
+assert rdtcipr1.DTCAndStatusRecord[-1].status == 0x27
+
 = Check UDS_RDTCI
 
 rdtci = UDS(b'\x19\x02\xff')
@@ -1156,11 +1170,16 @@ assert rdtci.DTCExtendedDataRecordNumber == 0xaa
 
 = Check UDS_RDTCIPR
 
-rdtcipr = UDS(b'\x59\x02\xff\xee\xdd\xaa')
+rdtcipr = UDS(b'\x59\x02\xff\xee\xdd\xaa\x02')
+rdtcipr.show()
 assert rdtcipr.service == 0x59
 assert rdtcipr.reportType == 2
 assert rdtcipr.DTCStatusAvailabilityMask == 0xff
-assert rdtcipr.DTCAndStatusRecord == b'\xee\xdd\xaa'
+assert rdtcipr.DTCAndStatusRecord[0].dtc.system == 3
+assert rdtcipr.DTCAndStatusRecord[0].dtc.type == 2
+assert rdtcipr.DTCAndStatusRecord[0].dtc.numeric_value_code == 3805
+assert rdtcipr.DTCAndStatusRecord[0].dtc.additional_information_code == 170
+assert rdtcipr.DTCAndStatusRecord[0].status == 2
 
 assert not rdtcipr.answers(rdtci)