etc

~ scripts and random things
git clone https://github.com/djnnvx/etc.git
doip_tls.py (5447B)
      1 #!/usr/bin/env python3
      2 """
      3 DoIP over TLS client (ISO 13400 assigns port 3496). Does not verify the server
      4 chain. Use `map` to compare the UDS request filter against plaintext 13400.
      5 
      6   ./doip_tls.py info
      7   ./doip_tls.py map
      8   ./doip_tls.py send 22 F1 86
      9 """
     10 import argparse, socket, ssl, struct, sys
     11 
     12 V = 0x03
     13 NRC = {0x10: 'generalReject', 0x11: 'serviceNotSupported', 0x12: 'subFunctionNotSupported',
     14        0x13: 'incorrectMessageLength', 0x22: 'conditionsNotCorrect', 0x24: 'requestSequenceError',
     15        0x31: 'requestOutOfRange', 0x33: 'securityAccessDenied', 0x34: 'authenticationRequired',
     16        0x35: 'invalidKey', 0x36: 'exceedNumberOfAttempts', 0x37: 'requiredTimeDelayNotExpired',
     17        0x70: 'uploadDownloadNotAccepted', 0x7E: 'subFuncNotSupportedInSession',
     18        0x7F: 'serviceNotSupportedInSession', 0xF0: 'vendor-specific'}
     19 NAMES = {0x10: 'DiagnosticSessionControl', 0x11: 'ECUReset', 0x14: 'ClearDiagnosticInformation',
     20          0x19: 'ReadDTCInformation', 0x22: 'ReadDataByIdentifier', 0x23: 'ReadMemoryByAddress',
     21          0x27: 'SecurityAccess', 0x28: 'CommunicationControl', 0x29: 'Authentication',
     22          0x2E: 'WriteDataByIdentifier', 0x2F: 'InputOutputControlByIdentifier',
     23          0x31: 'RoutineControl', 0x34: 'RequestDownload', 0x35: 'RequestUpload',
     24          0x36: 'TransferData', 0x37: 'RequestTransferExit', 0x3D: 'WriteMemoryByAddress',
     25          0x3E: 'TesterPresent', 0x85: 'ControlDTCSetting', 0x87: 'LinkControl'}
     26 
     27 
     28 def hdr(pt, pl=b''):
     29     return struct.pack('>BBHI', V, 0xFF ^ V, pt, len(pl)) + pl
     30 
     31 
     32 class TlsDoIP:
     33     def __init__(self, host, port, sa, ta, timeout=6.0):
     34         self.sa, self.ta = sa, ta
     35         ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
     36         ctx.check_hostname = False
     37         ctx.verify_mode = ssl.CERT_NONE
     38         raw = socket.create_connection((host, port), timeout)
     39         self.s = ctx.wrap_socket(raw, server_hostname=None)
     40         self.s.settimeout(timeout)
     41         c = self.s.getpeercert(binary_form=False)
     42         print(f'[+] TLS {self.s.version()} {self.s.cipher()[0]}')
     43         self.peercert = c
     44 
     45     def _rd(self):
     46         h = self.s.recv(8)
     47         if len(h) < 8:
     48             return None, b''
     49         _, _, pt, ln = struct.unpack('>BBHI', h)
     50         b = b''
     51         while len(b) < ln:
     52             c = self.s.recv(ln - len(b))
     53             if not c:
     54                 break
     55             b += c
     56         return pt, b
     57 
     58     def activate(self):
     59         self.s.sendall(hdr(0x0005, struct.pack('>HB', self.sa, 0) + b'\0' * 4))
     60         pt, b = self._rd()
     61         if pt == 0x0006 and len(b) >= 5:
     62             ta, ea, code = struct.unpack('>HHB', b[:5])
     63             print(f'[+] routing activation over TLS: entity=0x{ea:04X} code=0x{code:02X}'
     64                   + ('  ACTIVATED' if code == 0x10 else ''))
     65             return code == 0x10
     66         print(f'[-] routing activation failed: ptype={pt} {b.hex()}')
     67         return False
     68 
     69     def uds(self, payload):
     70         self.s.sendall(hdr(0x8001, struct.pack('>HH', self.sa, self.ta) + payload))
     71         while True:
     72             try:
     73                 pt, b = self._rd()
     74             except (socket.timeout, ssl.SSLError):
     75                 return None
     76             if pt == 0x0007:                      # AliveCheckRequest
     77                 self.s.sendall(hdr(0x0008, struct.pack('>H', self.sa)))
     78                 continue
     79             if pt == 0x8002:
     80                 continue
     81             if pt == 0x8003 or pt is None:
     82                 return None
     83             if pt != 0x8001:
     84                 continue
     85             r = b[4:]
     86             if len(r) >= 3 and r[0] == 0x7F and r[2] == 0x78:
     87                 continue
     88             return r
     89 
     90 
     91 def show(r):
     92     if r is None:
     93         print('    (no response)')
     94     elif len(r) >= 3 and r[0] == 0x7F:
     95         print(f'    NEG svc=0x{r[1]:02X} nrc=0x{r[2]:02X} {NRC.get(r[2], "?")}')
     96     else:
     97         t = ''.join(chr(c) if 32 <= c < 127 else '.' for c in r)
     98         print(f'    POS {r.hex()}  "{t}"')
     99 
    100 
    101 def main():
    102     a = argparse.ArgumentParser()
    103     a.add_argument('cmd', choices=['info', 'map', 'send'])
    104     a.add_argument('data', nargs='*')
    105     a.add_argument('-t', '--target', default='127.0.0.1')
    106     a.add_argument('-p', '--port', type=int, default=3496)
    107     a.add_argument('-s', '--sa', default='0x0EF4')
    108     a.add_argument('-d', '--ta', default='0x0010')
    109     o = a.parse_args()
    110 
    111     c = TlsDoIP(o.target, o.port, int(o.sa, 0), int(o.ta, 0))
    112     if o.cmd == 'info':
    113         return 0
    114     if not c.activate():
    115         print('[-] cannot proceed without routing activation')
    116         return 1
    117 
    118     if o.cmd == 'send':
    119         show(c.uds(bytes.fromhex(''.join(o.data))))
    120         return 0
    121 
    122     print('\n[*] bare-SID sweep over TLS, compare against plaintext 13400')
    123     inter = []
    124     for sid in range(0x100):
    125         r = c.uds(bytes([sid]))
    126         if r is None:
    127             continue
    128         if r[0] == 0x7F and len(r) >= 3 and r[2] == 0xF0:
    129             continue                                  # 0xF0: vendor filter
    130         inter.append(sid)
    131         if r[0] == 0x7F:
    132             print(f'  0x{sid:02X} {NAMES.get(sid, ""):30} NEG 0x{r[2]:02X} {NRC.get(r[2], "?")}')
    133         else:
    134             print(f'  0x{sid:02X} {NAMES.get(sid, ""):30} POS {r.hex()[:40]}')
    135     print(f'\n[*] {len(inter)} services bypass the filter over TLS: '
    136           + ' '.join(f'{x:02X}' for x in inter))
    137     print('    plaintext 13400 gave exactly three: 29 36 37')
    138     return 0
    139 
    140 
    141 if __name__ == '__main__':
    142     sys.exit(main())