uhryniuk
4c44933b5d
A Python tool to safely read the Keihin ECU on a 2010 Bonneville T100 over K-Line (KKL cable) or a Bluetooth ELM327, plus the reverse-engineering research behind it. Phase 1 (read-only comms) of an open tuning toolchain to replace the closed TuneECU app. Read-only by construction: safety.assert_read_only() runs on every outbound request before it hits the wire and refuses all write/flash services (0x27, 0x31, 0x34/0x36, 0x35, 0x37, 0x14, 0x11, 0x2E) and programming sessions, so a bug cannot brick the ECU. Verified frames match TuneECU byte-for-byte in tests/verify_protocol.py. Protocol constants recovered from the TuneECU APK (not guessed): ECU address 0xD5, K-Line tester 0xF5, format byte 0x80|len, additive mod-256 checksum. Includes the full TuneECU map catalogue (1811 entries) extracted to maps.json, searchable and filterable by ECU type and mechanical-vs-LCD odometer. research/ documents the Security Access seed/key algorithm, recovered as standard AES-128 (three embedded keys), with a self-testing reference impl verified against FIPS-197. This is write-path material, kept outside the read-only package. STATUS.md and README.md capture full context, the risk register, and where we left off: comms built but not yet run against the bike; next step is wiring the VAG KKL cable to the Triumph connector and running the first scan.
75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
import sys
|
|
from tunie import kwp2000 as k, triumph, safety
|
|
|
|
ok = True
|
|
def check(label, got, want):
|
|
global ok
|
|
good = got == want
|
|
ok &= good
|
|
print(f" [{'PASS' if good else 'FAIL'}] {label}: {got}" + ("" if good else f" (expected {want})"))
|
|
|
|
print("== framing vs TuneECU Ld() / ATSH81D5F5 ==")
|
|
# TuneECU a3: ATSH81D5F5 then payload "81" -> wire bytes 81 D5 F5 81 + additive checksum
|
|
f = k.build_frame(bytes([0x81]), triumph.ECU_ADDRESS, triumph.TESTER_ADDRESS_KLINE)
|
|
check("StartCommunication frame", f.hex(" "), "81 d5 f5 81 cc")
|
|
check(" checksum is additive sum", hex(sum(f[:-1]) & 0xFF), hex(f[-1]))
|
|
|
|
# m.java:1106 -> Ld({26}) = ReadEcuIdentification, no record byte
|
|
f = k.build_frame(bytes([0x1A, 0x80]), triumph.ECU_ADDRESS, triumph.TESTER_ADDRESS_KLINE)
|
|
check("ReadEcuIdentification 0x80", f.hex(" "), "82 d5 f5 1a 80 e6")
|
|
|
|
# m.java:737 -> Ld({24, 0, -1, 0}) = ReadDtcByStatus
|
|
f = k.build_frame(bytes([0x18, 0x00, 0xFF, 0x00]), triumph.ECU_ADDRESS, triumph.TESTER_ADDRESS_KLINE)
|
|
check("ReadDtcByStatus", f.hex(" "), "84 d5 f5 18 00 ff 00 65")
|
|
|
|
print("\n== round trip ==")
|
|
frame = k.build_frame(bytes([0x1A, 0x80]), 0xD5, 0xF5)
|
|
p = k.parse_frame(frame)
|
|
check("parse target", hex(p.target), "0xd5")
|
|
check("parse source", hex(p.source), "0xf5")
|
|
check("parse payload", p.payload.hex(" "), "1a 80")
|
|
|
|
print("\n== negative response decoding ==")
|
|
# 7F 1A 33 = securityAccessDenied
|
|
resp = k.build_frame(bytes([0x7F, 0x1A, 0x33]), 0xF5, 0xD5)
|
|
try:
|
|
k.expect_positive(k.parse_frame(resp), 0x1A); check("NRC raised", False, True)
|
|
except k.NegativeResponse as e:
|
|
check("NRC decoded", "securityAccessDenied" in str(e), True)
|
|
|
|
print("\n== checksum rejection ==")
|
|
bad = bytearray(k.build_frame(bytes([0x1A, 0x80]), 0xD5, 0xF5)); bad[-1] ^= 0xFF
|
|
try:
|
|
k.parse_frame(bytes(bad)); check("bad checksum rejected", False, True)
|
|
except k.ChecksumError: check("bad checksum rejected", True, True)
|
|
|
|
print("\n== SAFETY GUARD: every write service must be refused ==")
|
|
for sid, name in sorted(triumph.OBSERVED_WRITE_SERVICES.items()):
|
|
try:
|
|
safety.assert_read_only(bytes([sid, 0x01, 0x02]))
|
|
check(f"0x{sid:02X} {name.split()[0]}", "ALLOWED", "BLOCKED")
|
|
except safety.WriteAttemptBlocked:
|
|
check(f"0x{sid:02X} {name.split()[0]}", "BLOCKED", "BLOCKED")
|
|
|
|
print("\n== programming sessions must be refused ==")
|
|
for s in (0x02, 0x85, 0x86):
|
|
try:
|
|
safety.assert_read_only(bytes([0x10, s])); check(f"session 0x{s:02X}", "ALLOWED", "BLOCKED")
|
|
except safety.WriteAttemptBlocked: check(f"session 0x{s:02X}", "BLOCKED", "BLOCKED")
|
|
|
|
print("\n== read services must be permitted ==")
|
|
for payload, label in [
|
|
(bytes([0x1A, 0x80]), "ReadEcuIdentification"),
|
|
(bytes([0x18, 0x00, 0xFF, 0x00]), "ReadDtcByStatus"),
|
|
(bytes([0x21, 0x80]), "ReadDataByLocalId"),
|
|
(bytes([0x3E, 0x00]), "TesterPresent"),
|
|
(bytes([0x81]), "StartCommunication"),
|
|
(bytes([0x10, 0x81]), "StartDiagnosticSession(default)"),
|
|
]:
|
|
try:
|
|
safety.assert_read_only(payload); check(label, "ALLOWED", "ALLOWED")
|
|
except safety.WriteAttemptBlocked as e: check(label, f"BLOCKED ({e})", "ALLOWED")
|
|
|
|
print("\n" + ("ALL CHECKS PASSED" if ok else "SOME CHECKS FAILED"))
|
|
sys.exit(0 if ok else 1)
|