etc

~ scripts and random things
git clone https://github.com/djnnvx/etc.git
uds_protection_map.py (5329B)
      1 #!/usr/bin/env python3
      2 """
      3   0x33 securityAccessDenied          -> SecurityAccess 0x27
      4   0x34 authenticationRequired        -> Authentication 0x29
      5   0x7F serviceNotSupportedInSession  -> session gated
      6   0x7E subFunctionNotSupportedInSess -> session gated (subfunction)
      7   0x13 incorrectMessageLength        -> REACHABLE, no gate hit
      8   0x22 conditionsNotCorrect          -> reached handler, precondition unmet
      9   0x11 serviceNotSupported           -> not implemented
     10   non-standard codes                 -> vendor filter, never reaches the handler
     11 
     12   sudo ./uds_protection_map.py -t 127.0.0.1
     13   sudo ./uds_protection_map.py -t 127.0.0.1 --ta 0x0040 --all-sessions
     14 """
     15 import argparse, socket, struct, sys
     16 
     17 V = 0x03
     18 GATE = {
     19     0x33: ('SecurityAccess 0x27', 'gated'),
     20     0x34: ('Authentication 0x29', 'gated'),
     21     0x7F: ('session', 'gated'),
     22     0x7E: ('session (subfunction)', 'gated'),
     23     0x13: ('none, reachable', 'OPEN'),
     24     0x12: ('none, subfunc invalid', 'OPEN'),
     25     0x31: ('none, out of range', 'OPEN'),
     26     0x22: ('precondition', 'reached'),
     27     0x24: ('request sequence', 'reached'),
     28     0x11: ('not implemented', 'absent'),
     29     0xF0: ('vendor-specific', 'FILTERED'),
     30 }
     31 NAMES = {0x10: 'DiagnosticSessionControl', 0x11: 'ECUReset', 0x14: 'ClearDiagnosticInformation',
     32          0x19: 'ReadDTCInformation', 0x22: 'ReadDataByIdentifier', 0x23: 'ReadMemoryByAddress',
     33          0x24: 'ReadScalingDataByIdentifier', 0x27: 'SecurityAccess', 0x28: 'CommunicationControl',
     34          0x29: 'Authentication', 0x2A: 'ReadDataByPeriodicIdentifier',
     35          0x2C: 'DynamicallyDefineDataIdentifier', 0x2E: 'WriteDataByIdentifier',
     36          0x2F: 'InputOutputControlByIdentifier', 0x31: 'RoutineControl', 0x34: 'RequestDownload',
     37          0x35: 'RequestUpload', 0x36: 'TransferData', 0x37: 'RequestTransferExit',
     38          0x38: 'RequestFileTransfer', 0x3D: 'WriteMemoryByAddress', 0x3E: 'TesterPresent',
     39          0x83: 'AccessTimingParameter', 0x84: 'SecuredDataTransmission',
     40          0x85: 'ControlDTCSetting', 0x86: 'ResponseOnEvent', 0x87: 'LinkControl'}
     41 
     42 
     43 class Session:
     44     def __init__(self, host, sa, ta, timeout):
     45         self.sa, self.ta, self.timeout = sa, ta, timeout
     46         self.s = socket.create_connection((host, 13400), timeout)
     47         self.s.settimeout(timeout)
     48         self.s.sendall(self._h(0x0005, struct.pack('>HB', sa, 0) + b'\0' * 4))
     49         self.s.recv(4096)
     50 
     51     @staticmethod
     52     def _h(pt, pl=b''):
     53         return struct.pack('>BBHI', V, 0xFF ^ V, pt, len(pl)) + pl
     54 
     55     def _rd(self):
     56         h = self.s.recv(8)
     57         if len(h) < 8:
     58             return None, b''
     59         _, _, pt, ln = struct.unpack('>BBHI', h)
     60         b = b''
     61         while len(b) < ln:
     62             c = self.s.recv(ln - len(b))
     63             if not c:
     64                 break
     65             b += c
     66         return pt, b
     67 
     68     def uds(self, payload):
     69         self.s.sendall(self._h(0x8001, struct.pack('>HH', self.sa, self.ta) + payload))
     70         while True:
     71             try:
     72                 pt, b = self._rd()
     73             except socket.timeout:
     74                 return None
     75             if pt == 0x0007:                      # AliveCheckRequest
     76                 self.s.sendall(self._h(0x0008, struct.pack('>H', self.sa)))
     77                 continue
     78             if pt == 0x8002:
     79                 continue
     80             if pt == 0x8003:
     81                 return None
     82             if pt != 0x8001:
     83                 continue
     84             r = b[4:]
     85             if len(r) >= 3 and r[0] == 0x7F and r[2] == 0x78:
     86                 continue
     87             return r
     88 
     89 
     90 def sweep(sess, label):
     91     print(f'\n=== {label}')
     92     print(f'  {"SID":5} {"service":34} {"NRC":5} {"verdict":10} gate')
     93     buckets = {}
     94     for sid in range(0x100):
     95         r = sess.uds(bytes([sid]))
     96         if r is None:
     97             continue
     98         if r[0] == 0x7F and len(r) >= 3:
     99             nrc = r[2]
    100             gate, verdict = GATE.get(nrc, (f'unknown 0x{nrc:02X}', '?'))
    101         else:
    102             nrc, gate, verdict = 0, 'accepted bare SID', 'OPEN'
    103         if verdict == 'absent':
    104             continue
    105         buckets.setdefault(verdict, []).append(sid)
    106         print(f'  0x{sid:02X}  {NAMES.get(sid, ""):34} 0x{nrc:02X}  {verdict:10} {gate}')
    107     print(f'\n  summary for {label}:')
    108     for v in ('OPEN', 'reached', 'gated', 'FILTERED'):
    109         if v in buckets:
    110             print(f'    {v:9} {len(buckets[v]):3}  ' + ' '.join(f'{x:02X}' for x in buckets[v]))
    111     return buckets
    112 
    113 
    114 def main():
    115     a = argparse.ArgumentParser()
    116     a.add_argument('-t', '--target', default='127.0.0.1')
    117     a.add_argument('-s', '--sa', default='0x0EF4')
    118     a.add_argument('-d', '--ta', default='0x0010')
    119     a.add_argument('--timeout', type=float, default=3.0)
    120     a.add_argument('--all-sessions', action='store_true', help='also probe in extended session')
    121     o = a.parse_args()
    122 
    123     sess = Session(o.target, int(o.sa, 0), int(o.ta, 0), o.timeout)
    124     print(f'[*] target {o.target} ta={o.ta}, bare-SID probe (always invalid length, never executes)')
    125     sweep(sess, 'default session (10 01)')
    126 
    127     if o.all_sessions:
    128         r = sess.uds(b'\x10\x03')
    129         if r and r[0] == 0x50:
    130             sweep(sess, 'extended session (10 03)')
    131             sess.uds(b'\x10\x01')
    132         else:
    133             print('\n[-] could not enter extended session')
    134     return 0
    135 
    136 
    137 if __name__ == '__main__':
    138     sys.exit(main())