dilly/samplez
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add tunie: read-only KWP2000 diagnostics for Triumph Keihin ECU

Browse Source 
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.
This commit is contained in:
uhryniuk
2026-08-10 14:29:50 -05:00
parent f8904382a3
commit 4c44933b5d
20 changed files with 25326 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -174,3 +174,6 @@ cython_debug/
# PyPI configuration file # PyPI configuration file
.pypirc .pypirc
# macOS
.DS_Store

151
tunie/README.md Normal file
View File

@@ -0,0 +1,151 @@
# tunie
Open-source, **read-only** KWP2000 diagnostics for the Triumph Keihin ECU —
built for a 2010 Bonneville T100 (865cc, mechanical/analog odometer). The long
game is a full open tuning toolchain to replace the closed TuneECU app; this
first piece safely reads the ECU without any risk of bricking it.
> **This tool cannot write to the ECU.** The flash/write services are not
> implemented, and `src/tunie/safety.py` refuses them before any byte leaves the
> program. See [Safety](#safety).
For the full project context, risk register, and pick-up-where-we-left-off
notes, read **[STATUS.md](STATUS.md)**. The seed/key reverse engineering is in
**[research/FINDINGS.md](research/FINDINGS.md)**.
---
## Install
```sh
cd tunie
python3 -m venv .venv
./.venv/bin/pip install -e .
```
Requires Python 3.10+ and `pyserial` (pulled in automatically).
## Use
```sh
tunie ports # list serial devices
tunie info --port /dev/cu.usbserial-XXXX # interrogate ECU (wired KKL cable)
tunie info --port /dev/cu.OBDII --adapter elm327 # or a Bluetooth ELM327
tunie dtc --port /dev/cu.usbserial-XXXX # read fault codes only
tunie -v info --port ... # verbose: log every KWP frame
# Offline map database (1811 TuneECU maps, already extracted to maps.json):
tunie maps Bonneville --ecu 0 --odometer mechanical
tunie extract-maps <path/to/apktool_out/res/values/arrays.xml> # rebuild it
```
Ignition **on**, engine **off**. If fast init times out, try `--init slow`
(5-baud address init).
A first `tunie info` returns the ECU's identity, its currently-flashed map ID,
and any stored DTCs — that's the immediate goal, and it resolves which stock map
is actually on the bike.
## Protocol facts (recovered from the TuneECU APK, not guessed)
| Parameter | Value |
|---|---|
| ECU address | `0xD5` |
| Tester address (K-Line) | `0xF5` (**not** 0xF1) |
| Format byte | `0x80 \| length` |
| Checksum | additive sum mod 256 |
| Init | fast (`ATTP5`) or 5-baud (`ATTP4` + `ATIIAD5`) |
Verified frames (see `tests/verify_protocol.py`):
```
81 d5 f5 81 cc StartCommunication
82 d5 f5 1a 80 e6 ReadEcuIdentification 0x80
84 d5 f5 18 00 ff 00 65 ReadDtcByStatus
```
Run the tests:
```sh
./.venv/bin/python tests/verify_protocol.py
```
## Safety
Bricking an ECU over KWP2000 requires reaching the memory-write services, and
those sit behind Security Access (`0x27`). `safety.assert_read_only()` is called
on **every** outbound request inside `Transport.request()`, before the transport
sees it. It allows only query services and refuses all of: `0x11` EcuReset,
`0x14` Clear, `0x27` SecurityAccess, `0x2E` Write, `0x34`/`0x36` flash,
`0x35` upload, `0x37` — plus every programming diagnostic session. A bug can't
send a write because the write path does not exist.
**The real remaining risk is electrical, not software:** confirm the Triumph
diagnostic connector pinout before plugging any cable in. The 2010 twins do not
use a standard OBD-II socket.
---
## Where we left off (2026-08-10)
Phase 1 (read-only comms) is **built but not yet run against the bike** — nothing
has touched the ECU. Next physical steps, in order:
1. **Identify the wired cable's chip** — `tunie ports`. It's a *VAG* KKL 409.1:
right cable class (K-Line), but wired for a VW OBD-II socket, which this bike
does not have. FTDI shows as `/dev/cu.usbserial-*`; CH340 as
`/dev/cu.wchusbserial*` (needs the CH340 macOS driver).
2. **Confirm the Triumph connector pinout** and build an adapter/re-pin (K-Line,
switched +12V, ground) from the cable to the bike's diagnostic connector.
*Do not plug in until this is confirmed against a wiring diagram.*
3. **First scan:** ignition on / engine off, `tunie info --port …`.
Later phases (all deliberately out of the read-only tool): ROM dump for a
recovery image → firmware checksum patching → the write/flash path.
### Big finding: the seed/key is AES-128
The ECU's Security Access algorithm was fully recovered from the TuneECU APK —
it's **standard AES-128** (verified against FIPS-197 and the app's own T-table),
with three embedded keys selected by an ECU-code index. Not the XOR scheme the
original brief predicted. Working reference and details:
- `research/keihin_seedkey.py` — pure-Python, self-testing
(`python3 research/keihin_seedkey.py`)
- `research/FINDINGS.md` — full writeup
This is **write-path** material, kept in `research/` and never imported by the
`tunie` package. Having the algorithm does not make flashing safe — it's one of
several pieces (checksum patching, a verified stock dump, a recovery plan) that
all have to line up first.
## Layout
```
tunie/
README.md this file
STATUS.md full context, risk register, tuning theory
RESEARCH.md original research brief (partly wrong; STATUS corrects it)
pyproject.toml
maps.json 1811 TuneECU maps, extracted
src/tunie/ the read-only tool
safety.py read-only allowlist, enforced before transmit
kwp2000.py ISO 14230-3 framing / checksums / NRC decoding
triumph.py Triumph constants from the APK
identify.py the read-only interrogation sweep
maps.py map catalogue extraction + search
cli.py command-line entry point
transport/ kline.py (KKL cable) + elm327.py (Bluetooth)
tests/verify_protocol.py
research/ WRITE-PATH work, outside the package
keihin_seedkey.py AES-128 seed/key reference
FINDINGS.md seed/key writeup
```
> The decompiled TuneECU sources used for this research are **not** included here
> (third-party app, bulky). Findings and extracted constants are documented in
> `research/` and `STATUS.md`.
## Legal
Independent interoperability research on a bike I own. TuneECU is a separate
third-party product; this repo contains no TuneECU source, only original code and
documented findings.

1
tunie/RESEARCH.md Normal file
View File

File diff suppressed because one or more lines are too long

245
tunie/STATUS.md Normal file
View File

@@ -0,0 +1,245 @@
# Triumph Bonneville tuning project — status & context
**Bike:** 2010 Triumph Bonneville T100, 865cc air-cooled parallel twin, Keihin
ECU (Renesas SH7054), **mechanical/analog odometer**.
**Goal:** an open-source Python toolchain to read and (eventually) tune the ECU,
replacing the closed TuneECU app. Ultimate hardware path: SAI removal, O2 delete,
airbox removal, full exhaust — each needs a matching recalibration.
**Last updated:** 2026-08-10. Not getting to actual tuning this week — this doc
is the cold-start reference to pick it back up.
---
## TL;DR — where we are
- **Phase 1 (read-only comms) is built, not yet run against the bike.** Nothing
has touched the ECU. No hardware connected yet.
- **Next real step:** wire the cable to the Triumph connector correctly, then run
`tunie info` to read the ECU's identity, current map ID, and fault codes.
- **Blocker:** the wired cable is a *VAG* KKL — right cable type, wrong plug for
this bike. Needs an adapter/re-pin to the Triumph diagnostic connector. This is
the one genuine (electrical) risk and must be confirmed against a wiring
diagram before plugging in.
- **Big win:** the ECU's security-access seed/key algorithm was fully recovered
from the TuneECU APK — it's **AES-128** with three embedded keys. Documented,
reference-implemented, and verified. This is write-path work, quarantined
outside the read-only tool.
---
## Directory map
```
/Users/dylan/dojo/tuner/
tunie/ the read-only Python tool (installed, working)
src/tunie/ package source
tests/ protocol verification (passes)
maps.json 1811 extracted TuneECU maps
README.md tool-level docs
research/ WRITE-PATH reverse engineering (kept OUT of the tool)
keihin_seedkey.py AES-128 seed/key reference impl (self-testing)
FINDINGS.md seed/key writeup
work/ decompiled TuneECU
jadx_out/ Java decompile (read-only, best for reading logic)
apktool_out/ smali decompile (rebuildable)
TuneECU.apk original
TuneECU-logging.apk a clean rebuild (no real instrumentation yet)
samplez/ pristine original TuneECU.apk (git repo)
STATUS.md this file
RESEARCH.md the original (LLM-written, partly wrong) research brief
```
---
## What has been built
### `tunie` — read-only KWP2000 diagnostic tool
Installed and working. Commands:
```sh
tunie ports # list serial devices
tunie info --port /dev/cu.usbserial-XXXX # interrogate ECU (KKL cable)
tunie info --port /dev/cu.OBDII --adapter elm327
tunie dtc --port ... # fault codes only
tunie maps Bonneville --ecu 0 --odometer mechanical
tunie extract-maps <arrays.xml> # rebuild maps.json
tunie -v info --port ... # verbose: log every frame
```
Supports both a wired K-Line KKL cable (`--adapter kline`) and a Bluetooth
ELM327 (`--adapter elm327`) behind one interface.
**Read-only by construction.** `src/tunie/safety.py::assert_read_only()` runs on
every outbound request inside `Transport.request()`, before any byte reaches the
serial port. It allows only an allowlist of query services and refuses all
write/flash services (`0x27` SecurityAccess, `0x31` erase, `0x34`/`0x36` write,
`0x35` upload, `0x37`, `0x14`, `0x11`, `0x2E`) and all programming diagnostic
sessions. A bug cannot brick the ECU because the write path is not implemented.
Module layout:
```
safety.py read-only allowlist, enforced before transmit
kwp2000.py ISO 14230-3 framing, checksums, negative-response decoding
triumph.py Triumph constants recovered from the APK
identify.py the read-only interrogation sweep
maps.py TuneECU map catalogue extraction + search
cli.py command-line entry point
transport/
base.py Transport ABC; routes every request through safety
kline.py FTDI/KKL cable, raw serial, break-condition fast init
elm327.py ELM327 adapter, using TuneECU's own AT init sequence
```
`tests/verify_protocol.py` passes: it asserts our frames are byte-identical to
TuneECU's and that every write service / programming session is blocked.
---
## What was learned (all recovered from the APK, not guessed)
### Protocol facts (source: `com/tuneecu/m.java` + `MainActivity.java`)
| Parameter | Value | Notes |
|---|---|---|
| ECU address | `0xD5` | from `Ld()` framing + `ATSH81D5F5` |
| Tester address (K-Line) | `0xF5` | **not** 0xF1 — that's another brand |
| Format byte | `0x80 \| length` | long form (0x80 + length byte) when >120 B |
| Checksum | additive sum mod 256 | `Ub()` in m.java |
| Init | fast (`ATTP5`) or 5-baud (`ATTP4`+`ATIIAD5`) | address 0xD5 |
| ECU type code | `0`=Keihin, `1`=Sagem, `P`=Bosch | `ecu` string-array |
Reference frames (verified in tests):
```
81 d5 f5 81 cc StartCommunication
82 d5 f5 1a 80 e6 ReadEcuIdentification 0x80
84 d5 f5 18 00 ff 00 65 ReadDtcByStatus
```
> RESEARCH.md guessed target `0x10` / source `0xF1` and a 16-pin OBD-II port.
> Both wrong for this bike.
### Map database
Extracted TuneECU's full catalogue: **1811 entries** in `tunie/maps.json`.
TuneECU's own database distinguishes **"Mechanical odometer"** from **"LCD
odometer"** Bonnevilles — they are not interchangeable, and this is the ECU
generation split. For a 2010 mechanical-odo 865:
- **Stock baselines (`:0:` Keihin, mechanical odo):**
- `20187` production silencers, `20188` aftermarket silencers
- `20191`/`20192` same, up to VIN 739050, E25 fuel
- **Exhaust maps:** `20262`–`20265`, `20313`–`20316` (Arrow 2-in-1 / 2-in-2)
> RESEARCH.md recommended `20498` as an "OEM Arrow, safe rich baseline." It is
> actually a **Thruxton, LCD-odometer** map — wrong model AND wrong ECU
> generation. Do not use it.
### Security Access seed/key — RECOVERED, it's AES-128
Source: `com/tuneecu/m.java` method `Vb()`. Full writeup in
`research/FINDINGS.md`; working reference in `research/keihin_seedkey.py`
(`python3 research/keihin_seedkey.py` → all self-tests pass).
- The algorithm is **standard AES-128** (verified: all 256 T-table entries match
AES Te0; reference passes the FIPS-197 known-answer vector). It is **not** the
Honda XOR/bit-shift scheme RESEARCH.md predicted.
- `Vb()` AES-encrypts a 128-bit seed block with one of **three** embedded keys
(`iArr4`, m.java:5351), selected by `MainActivity.h7 ∈ {0,1,2}`, and returns
the first ciphertext word as the 4-byte key (`Gd()` sends `27 02 <key>`).
- The three keys (little-endian):
```
h7=0: ef704ca051b800cc9287df6a3511a978
h7=1: d4b15ff4c92ab7f098316e7a5b11ac39
h7=2: dc9fdba46f2fad18a4b8e1123c7183c2
```
- **Still open (framing, not crypto):** which h7 applies to the mechanical-odo
865, and the exact seed-block padding (`Ie[1..3]` from the response + fixed
`0x03`/`0x01`; `Ie[0]` set on an earlier path). **One captured (seed, key) pair
resolves both.**
### The logging APK
`work/TuneECU-logging.apk` is a clean apktool rebuild of the original (~17 KB
delta = re-signing/recompression). No custom instrumentation; all `Log` calls are
TuneECU's own, gated behind its debug boolean; no `debuggable` flag. It does not
currently capture seed/key at runtime — but it's the right vehicle for the
validation step (see next steps).
---
## The cable (VAG KKL) — what to do
A "VAG KKL" 409.1 cable is the **correct cable class** (K-Line), but wired for the
**VW/Audi OBD-II socket**, which this bike does not have.
**Step 1 — identify the chip.** Plug into the Mac, run `tunie ports`:
- `/dev/cu.usbserial-XXXX` → FTDI, ideal, no driver.
- `/dev/cu.wchusbserialXXXX` → CH340; works but install the CH340 macOS driver.
- nothing → driver missing.
**Step 2 — adapt the connector (the risk step).** The VAG plug puts K-Line on
OBD-II pin 7, +12V on 16, ground on 4/5. The 2010 Bonneville uses Triumph's
proprietary diagnostic connector (under seat/side panel), not OBD-II. You must
adapt/re-pin K-Line + switched-12V + ground to the correct three Triumph pins.
> **Do not plug in until the Triumph connector pinout is confirmed** against a
> wiring diagram or the known TuneECU-cable wiring. Wrong pin damages the
> transceiver. This is the one risk software can't remove.
---
## Next steps (in order)
1. **Identify the cable chip** — `tunie ports`, install CH340 driver if needed.
2. **Confirm the Triumph diagnostic connector pinout** — service manual / wiring
diagram. Build the adapter (K-Line, switched +12V, ground).
3. **First scan (safe):** ignition on / engine off, `tunie info --port …`.
Yields ECU ID, currently-flashed map ID, DTCs. If fast init times out, try
`--init slow`. This resolves which stock map is actually on the bike.
4. *(Later, Phase 2)* ROM dump for a recovery image — requires deliberately
enabling `0x35` upload; blocked in the read-only tool by design.
5. *(Later, Phase 3)* Checksum patching — not yet investigated; needed before any
modified map can boot.
6. *(Later, Phase 4)* Write path — validate the AES seed/key against a captured
pair first; ideally against a spare ECU, not the bike's only one.
### Optional now: capture a real seed/key pair
Write a minimal smali patch to `work/apktool_out` that logs every KWP frame
(patch the BT send/receive in `d.smali`) to logcat, rebuild, run one real TuneECU
Security Access against the bike, and capture the `27 01` seed + `27 02` key.
Then check `compute_key(seed, h7)` reproduces it — validates the whole AES port
and pins down h7 + padding.
---
## Risk register
| Area | Status |
|---|---|
| Software bricking | **Eliminated** — write services structurally unreachable in `tunie`. |
| Electrical / wiring | **LIVE** — VAG cable ≠ Triumph connector; confirm pinout before plugging in. |
| Seed/key correctness | Algorithm recovered + AES-verified; which-key + padding need one captured pair. Phase 4 only. |
| Firmware checksum | Not yet investigated. Needed before flashing a modified map. Phase 3. |
| Single ECU, no spare | Read-only-first mandatory; validate any write path against TuneECU before trusting ours. |
---
## Tuning theory (from RESEARCH.md §9 — this part is sound)
- **SAI removal:** flip the software SAI flag or the ECU throws a DTC; SAI air
also corrupts AFR readings on a dyno, so disable it before any fuel tuning.
- **O2 delete / open loop:** removing narrowband O2 + disabling closed-loop lets
you command a richer light-load AFR (~13.5–13.8:1 vs 14.7) — cools the
air-cooled top end and fixes the snatchy off-idle response.
- **Airbox removal & full exhaust:** both raise cylinder fill (VE / scavenging);
fuel tables must be enriched or it runs lean under load. Community airbox maps
derived from `20188` provide pre-calculated enrichment.
XDF definition files (to edit the .bin in TunerPro) are sold by OldSkullTuning
and Tuniverse (~€70) for the SH7054 — a later purchase, only once we're editing.
```

23173
tunie/maps.json Normal file
View File

File diff suppressed because it is too large Load Diff

16
tunie/pyproject.toml Normal file
View File

@@ -0,0 +1,16 @@
[project]
name = "tunie"
version = "0.1.0"
description = "Read-only KWP2000 diagnostics for Triumph Keihin ECUs"
requires-python = ">=3.10"
dependencies = ["pyserial>=3.5"]
[project.scripts]
tunie = "tunie.cli:main"
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]

82
tunie/research/FINDINGS.md Normal file
View File

@@ -0,0 +1,82 @@
# Seed/key recovery — findings
## Question
Is the Triumph Keihin KWP2000 Security Access (service `0x27`) seed/key algorithm
recoverable, and does the logging APK help?
## Answer
**Yes, fully — and from the *static* decompile, so the logging build isn't
needed for it.** The algorithm is **AES-128**, and all key material is embedded
in the app. RESEARCH.md's prediction ("Honda-style XOR / bit-shift with a factory
constant, possibly brute-forced") was wrong about the primitive.
## Where it lives
`com/tuneecu/m.java`, method `Vb()` (jadx), = `Lcom/tuneecu/m;->Vb()I` (smali).
| Evidence in `Vb()` | What it proves |
|---|---|
| `sArr = {1,2,4,8,16,32,64,128,27,54}` | AES Rcon (0x01,02,04,08,10,20,40,80,**1B,36**) |
| 4 table lookups/column at `+0,+256,+512,+768` | AES T-tables (Te0–Te3) |
| 10-iteration loop building 44 words | AES-128 key expansion |
| 9 rounds + 1 S-box-only final round | AES-128 encryption |
| `iArr3 = w.f` first entry `0xA56363C6` | `(3·S0, S0, S0, 2·S0)` with S0=0x63 → standard S-box + MixColumns |
**Verified:** all 256 entries of `w.f` match the standard AES Te0 table exactly
(`keihin_seedkey.py` cross-checks them). Our reference AES also passes the
FIPS-197 Appendix-B known-answer test. This is unmodified AES-128.
## What `Vb()` does
1. Loads a 128-bit seed block into `Ie[0..3]` from the ECU's `0x27` seed response.
2. AES-128-encrypts it with one of **three** embedded keys (`iArr4`, m.java:5351),
selected by `MainActivity.h7 ∈ {0,1,2}`.
3. Returns the first ciphertext word; `Gd()` sends it as `27 02 <4 bytes>`.
The three keys (little-endian, as the cipher consumes them):
```
h7=0: ef704ca051b800cc9287df6a3511a978
h7=1: d4b15ff4c92ab7f098316e7a5b11ac39
h7=2: dc9fdba46f2fad18a4b8e1123c7183c2
```
`h7` is looked up from an ECU calibration-code string (`MainActivity.java:6221`).
## Still open (framing, not crypto — one captured pair resolves both)
1. **Which key** applies to a mechanical-odometer 865 twin. `h7` defaults toward
0 for the older no-code Keihin maps, but confirm rather than assume.
2. **Exact seed-block layout.** `Vb()` encrypts `Ie[0..3]`; m.java fills `Ie[1..3]`
from the seed response and pads two bytes with fixed constants (`0x03`, `0x01`),
with `Ie[0]` set on an earlier path. A single real (seed, key) pair pins this
down exactly.
## The logging APK
`work/TuneECU-logging.apk` is a clean apktool rebuild of the original (the ~17 KB
size delta is re-signing/recompression). No custom instrumentation was found —
every `Log` call is TuneECU's own, gated behind its debug boolean, and no
`debuggable` flag is set. So it does not currently capture seed/key at runtime.
It's still the right tool for the **validation** step above: make the app perform
one real Security Access against the bike (or a bench ECU) and capture the `27 01`
seed and `27 02` key off the wire, then check `compute_key(seed, h7)` reproduces
it. That either confirms the port outright or tells us the exact `h7`/padding.
## Reference implementation
`research/keihin_seedkey.py` — pure-Python, self-testing. Run it:
```sh
python3 research/keihin_seedkey.py
```
## Boundary
This is **write-path** material. It is kept in `research/`, outside the read-only
`tunie` package, and `tunie` does not import it. Reproducing the key does not make
flashing safe — it's one of several pieces (checksum patching, a verified stock
dump, a recovery plan, ideally a spare ECU) that all have to line up first.

223
tunie/research/keihin_seedkey.py Normal file
View File

@@ -0,0 +1,223 @@
"""Reference implementation of the Triumph Keihin KWP2000 seed/key algorithm.
RECOVERED FROM: TuneECU, com/tuneecu/m.java, method Vb() (jadx decompile).
Headline: the security-access algorithm is **AES-128**, not the Honda-style
XOR/bit-shift scheme RESEARCH.md predicted. The proof is in m.java's Vb():
* round constants sArr = {1,2,4,8,16,32,64,128,27,54} -> AES Rcon
(0x01,0x02,0x04,0x08,0x10,0x20,0x40,0x80,0x1B,0x36)
* four table lookups per column at offsets +0,+256,+512,+768 -> AES T-tables
* 10-iteration key schedule producing 44 words -> AES-128 expand
* 9 full rounds + 1 final (S-box only) round -> AES-128 encrypt
The lookup table iArr3 = w.f (field `f` in smali, shown as `f2739f` in jadx).
Its first entry is 0xA56363C6, which decomposes as (0x63*3, 0x63, 0x63, 0x63*2)
in GF(2^8): the standard AES S-box (S[0]=0x63) and MixColumns coefficients.
So this is unmodified AES-128 -- we do not need to extract the 2048-entry table;
the standard S-box reproduces it exactly (verified below against FIPS-197).
>>> IMPORTANT <<<
This is WRITE-PATH material. It lives here in research/, deliberately OUTSIDE the
read-only `tunie` package, and nothing in `tunie` imports it. Computing a valid
key is one of the several things that must line up before a flash is possible;
having the algorithm does not make writing to the ECU safe. See README notes.
Two things still need a single real seed/key capture (from the bike or the
logging build) to pin down, because they are framing details rather than crypto:
1. Which of the three embedded keys applies to a mechanical-odometer 865 twin.
The key is chosen by MainActivity.h7 in {0,1,2}; h7 is looked up from an
ECU calibration-code string. For the older no-code Keihin maps this appears
to default to 0, but that should be confirmed, not assumed.
2. The exact seed-block layout. Vb() encrypts Ie[0..3] (a 128-bit block).
m.java loads Ie[1..3] from the received 0x27 response and pads two bytes
with fixed constants (0x03 into Ie[1] low byte, 0x01 into Ie[3] high byte);
Ie[0] is set on an earlier path. One captured pair resolves this instantly.
"""
from __future__ import annotations
# The three 128-bit keys, verbatim from m.java:5351 (iArr4), as signed 32-bit
# ints exactly as Java stores them. Four consecutive words == one AES-128 key;
# MainActivity.h7 selects the key as iArr4[h7*4 : h7*4+4].
_IARR4_SIGNED = [
-1605603089, -872368047, 1793034130, 2024345909, # key index 0
-195055148, -256431415, 2054042008, 967577947, # key index 1
-1529110564, 414003055, 316782756, -1031573188, # key index 2
]
def _u32(x: int) -> int:
return x & 0xFFFFFFFF
def key_words(h7: int) -> list[int]:
"""Return the four unsigned 32-bit key words for key selector h7 in {0,1,2}."""
if not 0 <= h7 <= 2:
raise ValueError(f"h7 must be 0, 1 or 2 (got {h7})")
return [_u32(w) for w in _IARR4_SIGNED[h7 * 4 : h7 * 4 + 4]]
def key_bytes(h7: int) -> bytes:
"""The selected AES-128 key as 16 bytes.
m.java packs each key word little-endian when it feeds the cipher (Ie/iArr2
words are consumed low-byte first via the T-table indices), so we emit
little-endian to match Vb() exactly.
"""
return b"".join(w.to_bytes(4, "little") for w in key_words(h7))
# --- textbook AES-128, byte-oriented (matches Vb()'s T-table math) -----------
def _gmul(a: int, b: int) -> int:
p = 0
for _ in range(8):
if b & 1:
p ^= a
hi = a & 0x80
a = (a << 1) & 0xFF
if hi:
a ^= 0x1B
b >>= 1
return p
def _build_sbox() -> list[int]:
# Multiplicative inverse in GF(2^8) followed by the AES affine transform.
inv = [0] * 256
p = q = 1
for _ in range(255):
p = _gmul(p, 3)
# q = p^-1 via the standard log/antilog trick
q = _gmul(q, 0xF6) # 3^-1
inv[p] = q
inv[1] = 1
sbox = [0] * 256
for i in range(256):
x = inv[i]
s = x ^ ((x << 1) | (x >> 7)) ^ ((x << 2) | (x >> 6)) ^ \
((x << 3) | (x >> 5)) ^ ((x << 4) | (x >> 4))
sbox[i] = (s ^ 0x63) & 0xFF
return sbox
SBOX = _build_sbox()
RCON = [0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1B, 0x36]
def _expand_key(key: bytes) -> list[list[int]]:
assert len(key) == 16
words = [list(key[i : i + 4]) for i in range(0, 16, 4)]
for i in range(4, 44):
temp = list(words[i - 1])
if i % 4 == 0:
temp = temp[1:] + temp[:1] # RotWord
temp = [SBOX[b] for b in temp] # SubWord
temp[0] ^= RCON[i // 4 - 1]
words.append([words[i - 4][j] ^ temp[j] for j in range(4)])
return words
def _add_round_key(state: list[int], words: list[list[int]], rnd: int) -> None:
for c in range(4):
for r in range(4):
state[r * 4 + c] ^= words[rnd * 4 + c][r]
def _sub_shift_mix(state: list[int], final: bool) -> None:
s = [SBOX[b] for b in state]
# ShiftRows
s = [
s[0], s[1], s[2], s[3],
s[5], s[6], s[7], s[4],
s[10], s[11], s[8], s[9],
s[15], s[12], s[13], s[14],
]
if final:
state[:] = s
return
for c in range(4):
col = [s[r * 4 + c] for r in range(4)]
state[0 * 4 + c] = _gmul(col[0], 2) ^ _gmul(col[1], 3) ^ col[2] ^ col[3]
state[1 * 4 + c] = col[0] ^ _gmul(col[1], 2) ^ _gmul(col[2], 3) ^ col[3]
state[2 * 4 + c] = col[0] ^ col[1] ^ _gmul(col[2], 2) ^ _gmul(col[3], 3)
state[3 * 4 + c] = _gmul(col[0], 3) ^ col[1] ^ col[2] ^ _gmul(col[3], 2)
def aes128_encrypt_block(block: bytes, key: bytes) -> bytes:
"""Standard AES-128 ECB single block. Column-major state, per FIPS-197."""
assert len(block) == 16 and len(key) == 16
words = _expand_key(key)
state = [block[r + 4 * c] for r in range(4) for c in range(4)]
_add_round_key(state, words, 0)
for rnd in range(1, 10):
_sub_shift_mix(state, final=False)
_add_round_key(state, words, rnd)
_sub_shift_mix(state, final=True)
_add_round_key(state, words, 10)
return bytes(state[r + 4 * c] for r in range(4) for c in range(4))
def compute_key(seed_block: bytes, h7: int = 0) -> bytes:
"""Reproduce Vb(): AES-128 encrypt the seed block, take the first 4 bytes.
m.java's Gd() sends `27 02` followed by these four bytes, low byte first,
which is the natural byte order of the first ciphertext word.
"""
if len(seed_block) != 16:
raise ValueError("seed block must be 16 bytes")
cipher = aes128_encrypt_block(seed_block, key_bytes(h7))
return cipher[:4]
# --- self test ---------------------------------------------------------------
def _selftest() -> bool:
ok = True
# 1. S-box against the well-known first row of the AES S-box.
expected = [0x63, 0x7C, 0x77, 0x7B, 0xF2, 0x6B, 0x6F, 0xC5]
got = SBOX[:8]
good = got == expected
ok &= good
print(f" [{'PASS' if good else 'FAIL'}] S-box row 0: {[hex(x) for x in got]}")
# 2. FIPS-197 Appendix B known-answer test.
pt = bytes.fromhex("3243f6a8885a308d313198a2e0370734")
kk = bytes.fromhex("2b7e151628aed2a6abf7158809cf4f3c")
ct = aes128_encrypt_block(pt, kk).hex()
want = "3925841d02dc09fbdc118597196a0b32"
good = ct == want
ok &= good
print(f" [{'PASS' if good else 'FAIL'}] FIPS-197 vector: {ct}")
# 3. The T-table's first entry confirms S-box+MixColumns: 0xA56363C6.
s0 = SBOX[0]
good = (
s0 == 0x63
and _gmul(s0, 2) == 0xC6
and _gmul(s0, 3) == 0xA5
)
ok &= good
print(f" [{'PASS' if good else 'FAIL'}] Te[0] decode: "
f"3*S0=0x{_gmul(s0,3):02X} S0=0x{s0:02X} 2*S0=0x{_gmul(s0,2):02X} "
"-> A5 63 63 C6")
# 4. The three embedded keys, for the record.
print(" embedded AES-128 keys (little-endian, from iArr4):")
for h in range(3):
print(f" h7={h}: {key_bytes(h).hex()}")
# 5. Demonstrate a full seed->key with a placeholder seed.
demo_seed = bytes(range(16))
print(f" demo compute_key(00..0f, h7=0) = {compute_key(demo_seed, 0).hex()}")
return ok
if __name__ == "__main__":
import sys
print("Keihin seed/key reference self-test:")
sys.exit(0 if _selftest() else 1)

7
tunie/src/tunie/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
"""tunie -- read-only KWP2000 diagnostics for Triumph Keihin ECUs.
This package deliberately does not implement the flash write path. See
:mod:`tunie.safety`.
"""
__version__ = "0.1.0"

208
tunie/src/tunie/cli.py Normal file
View File

@@ -0,0 +1,208 @@
"""Command line entry point.
tunie ports list candidate serial devices
tunie info --port ... interrogate the ECU (read-only)
tunie dtc --port ... read diagnostic trouble codes only
tunie maps --search ... query the offline TuneECU map database
"""
from __future__ import annotations
import argparse
import logging
import sys
from . import triumph
from .identify import interrogate
from .maps import find_maps, load_map_database
from .safety import WriteAttemptBlocked
from .transport.base import Transport, TransportError
from .transport.elm327 import Elm327Transport
from .transport.kline import KLineTransport
def _build_transport(args: argparse.Namespace) -> Transport:
common = dict(target=args.target, source=args.source, init=args.init)
if args.adapter == "elm327":
return Elm327Transport(args.port, baud=args.baud or 38400, **common)
return KLineTransport(args.port, baud=args.baud or 10400, **common)
def _cmd_ports(_: argparse.Namespace) -> int:
try:
from serial.tools import list_ports
except ImportError:
print("pyserial is not installed: pip install pyserial", file=sys.stderr)
return 1
found = list(list_ports.comports())
if not found:
print("No serial ports found.")
return 1
for port in found:
hint = ""
low = f"{port.description} {port.manufacturer or ''}".lower()
if "ftdi" in low or "usbserial" in low:
hint = " <- likely KKL cable (use --adapter kline)"
elif "obd" in low or "bluetooth" in low or "linker" in low:
hint = " <- likely ELM327 (use --adapter elm327)"
print(f"{port.device:<30} {port.description}{hint}")
return 0
def _cmd_info(args: argparse.Namespace) -> int:
transport = _build_transport(args)
print(
f"Connecting on {args.port} via {args.adapter}, "
f"target 0x{args.target:02X} source 0x{args.source:02X}, {args.init} init..."
)
try:
with transport:
report = interrogate(transport)
except TransportError as exc:
print(f"\nConnection failed: {exc}", file=sys.stderr)
print(
"\nThings to check, in order:\n"
" 1. Ignition on, engine off. The ECU sleeps otherwise.\n"
" 2. Diagnostic connector pinout -- confirm before suspecting anything else.\n"
" 3. Try --init slow (5-baud) if fast init times out.\n"
" 4. On an ELM327 clone, KWP support is often broken; try the KKL cable.",
file=sys.stderr,
)
return 1
print()
print(report.render())
return 0
def _cmd_dtc(args: argparse.Namespace) -> int:
from .identify import EcuReport, _read_dtcs
transport = _build_transport(args)
try:
with transport:
report = EcuReport(key_bytes=transport.key_bytes)
report.dtcs = _read_dtcs(transport, report)
except TransportError as exc:
print(f"Connection failed: {exc}", file=sys.stderr)
return 1
print(report.render())
return 0
def _cmd_maps(args: argparse.Namespace) -> int:
try:
database = load_map_database(args.database)
except FileNotFoundError as exc:
print(f"{exc}\n\nRun: tunie extract-maps --apk-res <apktool_out/res/values/arrays.xml>",
file=sys.stderr)
return 1
matches = find_maps(database, args.search, ecu=args.ecu, odometer=args.odometer)
if not matches:
print("No maps matched.")
return 1
for entry in matches:
print(f"{entry['id']} (ecu type {entry['ecu']})")
for line in entry["description"]:
print(f" {line}")
print()
print(f"{len(matches)} map(s).")
return 0
def _cmd_extract_maps(args: argparse.Namespace) -> int:
from .maps import extract_from_arrays_xml
count = extract_from_arrays_xml(args.arrays_xml, args.database)
print(f"Wrote {count} maps to {args.database}")
return 0
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="tunie",
description="Read-only KWP2000 diagnostics for Triumph Keihin ECUs.",
epilog="This tool cannot write to the ECU. The flash services are not implemented.",
)
parser.add_argument("-v", "--verbose", action="store_true", help="log every frame")
sub = parser.add_subparsers(dest="command", required=True)
def add_connection_args(p: argparse.ArgumentParser) -> None:
p.add_argument("--port", required=True, help="serial device (see: tunie ports)")
p.add_argument(
"--adapter",
choices=("kline", "elm327"),
default="kline",
help="kline for an FTDI KKL cable, elm327 for a Bluetooth OBD adapter",
)
p.add_argument(
"--init",
choices=("fast", "slow"),
default="fast",
help="fast init (ISO 14230-2) or 5-baud address init",
)
p.add_argument(
"--target",
type=lambda s: int(s, 0),
default=triumph.ECU_ADDRESS,
help=f"ECU address (default 0x{triumph.ECU_ADDRESS:02X}, from TuneECU)",
)
p.add_argument(
"--source",
type=lambda s: int(s, 0),
default=triumph.TESTER_ADDRESS_KLINE,
help=f"tester address (default 0x{triumph.TESTER_ADDRESS_KLINE:02X})",
)
p.add_argument("--baud", type=int, default=None, help="override the serial baud rate")
sub.add_parser("ports", help="list serial ports").set_defaults(func=_cmd_ports)
p_info = sub.add_parser("info", help="interrogate the ECU (read-only)")
add_connection_args(p_info)
p_info.set_defaults(func=_cmd_info)
p_dtc = sub.add_parser("dtc", help="read diagnostic trouble codes")
add_connection_args(p_dtc)
p_dtc.set_defaults(func=_cmd_dtc)
p_maps = sub.add_parser("maps", help="search the offline map database")
p_maps.add_argument("search", nargs="?", default="", help="text to match, e.g. Bonneville")
p_maps.add_argument("--ecu", default=None, help="filter by ECU type code, e.g. 0 for Keihin")
p_maps.add_argument(
"--odometer",
choices=("mechanical", "lcd"),
default=None,
help="filter by instrument type -- this distinguishes ECU generations",
)
p_maps.add_argument("--database", default="maps.json", help="path to the extracted database")
p_maps.set_defaults(func=_cmd_maps)
p_ex = sub.add_parser("extract-maps", help="build maps.json from a decompiled APK")
p_ex.add_argument("arrays_xml", help="path to apktool_out/res/values/arrays.xml")
p_ex.add_argument("--database", default="maps.json", help="output path")
p_ex.set_defaults(func=_cmd_extract_maps)
args = parser.parse_args(argv)
logging.basicConfig(
level=logging.DEBUG if args.verbose else logging.INFO,
format="%(levelname)-7s %(name)s: %(message)s",
)
try:
return args.func(args)
except WriteAttemptBlocked as exc:
print(f"\nBLOCKED: {exc}", file=sys.stderr)
return 2
except KeyboardInterrupt:
print("\nInterrupted.", file=sys.stderr)
return 130
if __name__ == "__main__":
raise SystemExit(main())

177
tunie/src/tunie/identify.py Normal file
View File

@@ -0,0 +1,177 @@
"""Read-only interrogation of the ECU.
Everything here is a query. Nothing in this module can alter ECU state, and the
transport's allowlist would refuse it anyway.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from . import kwp2000 as k
from .transport.base import Transport, TransportError
log = logging.getLogger(__name__)
#: ReadEcuIdentification record numbers worth probing. KWP2000 leaves most of
#: this range manufacturer-defined, so the strategy is to sweep and keep what
#: answers rather than assume a layout.
ECU_ID_RECORDS: tuple[int, ...] = (
0x80, # manufacturer spare
0x81, # ECU identification data table
0x86,
0x87,
0x88, # VIN (ISO-defined)
0x89,
0x8A,
0x90, # VIN (alternate, ISO-defined)
0x91,
0x92, # system supplier ECU hardware number
0x93,
0x94, # system supplier ECU software number
0x95,
0x96,
0x97,
0x98,
0x99,
0x9A,
0x9B,
0x9C,
)
#: ReadDataByLocalIdentifier records. 0x80 is the one TuneECU reads at
#: ``m.java:763``; the rest of the low range is a conservative sweep.
LOCAL_ID_RECORDS: tuple[int, ...] = (0x80, 0x81, 0x00, 0x01, 0x02, 0x03, 0x10, 0x11)
@dataclass
class EcuReport:
"""Everything a read-only pass could establish about the ECU."""
key_bytes: tuple[int, int] | None = None
identification: dict[int, bytes] = field(default_factory=dict)
local_records: dict[int, bytes] = field(default_factory=dict)
dtcs: list[tuple[int, int]] = field(default_factory=list)
errors: dict[str, str] = field(default_factory=dict)
def render(self) -> str:
lines: list[str] = []
if self.key_bytes:
lines.append(
f"Key bytes : 0x{self.key_bytes[0]:02X} 0x{self.key_bytes[1]:02X}"
)
if self.identification:
lines.append("")
lines.append("ReadEcuIdentification (0x1A)")
for rec, data in sorted(self.identification.items()):
lines.append(f" 0x{rec:02X} {_pretty(data)}")
if self.local_records:
lines.append("")
lines.append("ReadDataByLocalIdentifier (0x21)")
for rec, data in sorted(self.local_records.items()):
lines.append(f" 0x{rec:02X} {_pretty(data)}")
if self.dtcs:
lines.append("")
lines.append("Diagnostic trouble codes")
for code, status in self.dtcs:
lines.append(f" {_dtc_name(code)} (raw 0x{code:04X}, status 0x{status:02X})")
else:
lines.append("")
lines.append("Diagnostic trouble codes: none reported")
if self.errors:
lines.append("")
lines.append("Queries that did not answer")
for label, why in self.errors.items():
lines.append(f" {label}: {why}")
return "\n".join(lines) if lines else "ECU returned nothing."
def _pretty(data: bytes) -> str:
"""Show hex, plus ASCII when the record looks like text."""
text = "".join(chr(b) if 0x20 <= b < 0x7F else "." for b in data)
printable = sum(1 for b in data if 0x20 <= b < 0x7F)
hexpart = data.hex(" ")
if data and printable / len(data) > 0.6:
return f"{hexpart} |{text}|"
return hexpart
def _dtc_name(code: int) -> str:
"""Render a 2-byte DTC as the usual Pxxxx/Cxxxx/Bxxxx/Uxxxx form."""
letter = "PCBU"[(code >> 14) & 0x03]
return f"{letter}{(code >> 8) & 0x3F:02X}{code & 0xFF:02X}"
def interrogate(
transport: Transport,
*,
ecu_id_records: tuple[int, ...] = ECU_ID_RECORDS,
local_id_records: tuple[int, ...] = LOCAL_ID_RECORDS,
) -> EcuReport:
"""Sweep every read-only query we know of and collect what answers."""
report = EcuReport(key_bytes=transport.key_bytes)
for record in ecu_id_records:
try:
data = transport.request(bytes([k.Sid.READ_ECU_IDENTIFICATION, record]))
except k.NegativeResponse as exc:
log.debug("0x1A 0x%02X rejected: %s", record, exc)
continue
except (TransportError, k.Kwp2000Error) as exc:
report.errors[f"ReadEcuIdentification 0x{record:02X}"] = str(exc)
continue
# The ECU echoes the record number before its data.
report.identification[record] = data[1:] if data[:1] == bytes([record]) else data
for record in local_id_records:
try:
data = transport.request(bytes([k.Sid.READ_DATA_BY_LOCAL_ID, record]))
except k.NegativeResponse as exc:
log.debug("0x21 0x%02X rejected: %s", record, exc)
continue
except (TransportError, k.Kwp2000Error) as exc:
report.errors[f"ReadDataByLocalIdentifier 0x{record:02X}"] = str(exc)
continue
report.local_records[record] = data[1:] if data[:1] == bytes([record]) else data
report.dtcs = _read_dtcs(transport, report)
return report
def _read_dtcs(transport: Transport, report: EcuReport) -> list[tuple[int, int]]:
"""Try both DTC services TuneECU uses; keep whichever the ECU answers."""
attempts = (
("ReadDtcByStatus 0x18", bytes([k.Sid.READ_DTC_BY_STATUS, 0x00, 0xFF, 0x00])),
("ReadDtc 0x13", bytes([k.Sid.READ_DTC, 0x40, 0xFF])),
)
for label, payload in attempts:
try:
data = transport.request(payload, timeout_ms=2000)
except k.NegativeResponse as exc:
log.debug("%s rejected: %s", label, exc)
continue
except (TransportError, k.Kwp2000Error) as exc:
report.errors[label] = str(exc)
continue
return _parse_dtcs(data)
return []
def _parse_dtcs(data: bytes) -> list[tuple[int, int]]:
"""Decode a count byte followed by 3-byte (code high, code low, status) tuples."""
if not data:
return []
body = data[1:]
return [
((body[i] << 8) | body[i + 1], body[i + 2])
for i in range(0, len(body) - 2, 3)
]

222
tunie/src/tunie/kwp2000.py Normal file
View File

@@ -0,0 +1,222 @@
"""ISO 14230-3 (KWP2000) frame construction, parsing and response decoding.
This module is transport-agnostic: it turns service payloads into wire frames
and back, and knows nothing about serial ports or ELM327 adapters.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import IntEnum
POSITIVE_RESPONSE_OFFSET = 0x40
NEGATIVE_RESPONSE = 0x7F
# ISO 14230-2 timing, milliseconds. P2 is stretched when the ECU sends the
# "response pending" negative code 0x78.
P2_MIN_MS = 25
P2_MAX_MS = 50
P2_EXTENDED_MS = 5000
P3_MIN_MS = 55
P4_MIN_MS = 5
# Fast-init line timing, milliseconds.
W5_IDLE_MS = 300
TINIL_MS = 25
TINIH_MS = 25
class Sid(IntEnum):
"""Service identifiers this tool is able to send (all read-only)."""
START_DIAGNOSTIC_SESSION = 0x10
READ_DTC = 0x13
READ_STATUS_OF_DTC = 0x17
READ_DTC_BY_STATUS = 0x18
READ_ECU_IDENTIFICATION = 0x1A
READ_DATA_BY_LOCAL_ID = 0x21
READ_DATA_BY_COMMON_ID = 0x22
READ_MEMORY_BY_ADDRESS = 0x23
TESTER_PRESENT = 0x3E
START_COMMUNICATION = 0x81
STOP_COMMUNICATION = 0x82
ACCESS_TIMING_PARAMETERS = 0x83
NEGATIVE_RESPONSE_CODES: dict[int, str] = {
0x10: "generalReject",
0x11: "serviceNotSupported",
0x12: "subFunctionNotSupported / invalidFormat",
0x21: "busyRepeatRequest",
0x22: "conditionsNotCorrect / requestSequenceError",
0x23: "routineNotComplete",
0x31: "requestOutOfRange",
0x33: "securityAccessDenied",
0x35: "invalidKey",
0x36: "exceedNumberOfAttempts",
0x37: "requiredTimeDelayNotExpired",
0x40: "downloadNotAccepted",
0x50: "uploadNotAccepted",
0x71: "transferAborted",
0x78: "requestCorrectlyReceived-ResponsePending",
0x80: "serviceNotSupportedInActiveSession",
0x9A: "dataDecompressionFailed",
0x9B: "dataDecryptionFailed",
0xA0: "ecuNotResponding",
0xA1: "ecuAddressUnknown",
}
RESPONSE_PENDING = 0x78
class Kwp2000Error(Exception):
"""Base class for protocol-level failures."""
class ChecksumError(Kwp2000Error):
"""A received frame did not satisfy its trailing additive checksum."""
class NegativeResponse(Kwp2000Error):
"""The ECU answered a request with 0x7F."""
def __init__(self, service: int, code: int) -> None:
self.service = service
self.code = code
name = NEGATIVE_RESPONSE_CODES.get(code, "unknown")
super().__init__(
f"service 0x{service:02X} rejected: 0x{code:02X} ({name})"
)
def checksum(data: bytes) -> int:
"""Additive modulo-256 checksum over every preceding byte of the frame."""
return sum(data) & 0xFF
def build_frame(
payload: bytes,
target: int,
source: int,
*,
force_length_byte: bool = False,
) -> bytes:
"""Wrap a service payload in an addressed ISO 14230 frame.
Uses the compact form (length packed into the low six bits of the format
byte) when the payload fits and ``force_length_byte`` is not set; otherwise
emits the separate-length-byte form. Some ECUs only accept one of the two,
which is why the caller can force it.
"""
length = len(payload)
if length == 0:
raise ValueError("payload must not be empty")
if length <= 0x3F and not force_length_byte:
header = bytes([0x80 | length, target, source])
else:
if length > 0xFF:
raise ValueError(f"payload of {length} bytes exceeds the 255-byte frame limit")
header = bytes([0x80, target, source, length])
frame = header + payload
return frame + bytes([checksum(frame)])
@dataclass(frozen=True)
class Frame:
"""A parsed inbound frame."""
format_byte: int
target: int | None
source: int | None
payload: bytes
@property
def sid(self) -> int:
return self.payload[0]
@property
def is_negative(self) -> bool:
return self.payload[0] == NEGATIVE_RESPONSE
@property
def is_response_pending(self) -> bool:
return self.is_negative and len(self.payload) >= 3 and self.payload[2] == RESPONSE_PENDING
def frame_length(buffer: bytes) -> int | None:
"""Total on-wire length of the frame starting at ``buffer[0]``.
Returns ``None`` when more bytes are needed to determine it. Used to know
when a read from a half-duplex line is complete.
"""
if not buffer:
return None
fmt = buffer[0]
addressed = bool(fmt & 0x80)
header_len = 3 if addressed else 1
packed_len = fmt & 0x3F
if packed_len:
return header_len + packed_len + 1
# Length lives in its own byte immediately after the header.
if len(buffer) < header_len + 1:
return None
return header_len + 1 + buffer[header_len] + 1
def parse_frame(buffer: bytes) -> Frame:
"""Validate and decode a complete inbound frame."""
if len(buffer) < 3:
raise Kwp2000Error(f"frame too short: {buffer.hex(' ')}")
expected = frame_length(buffer)
if expected is None or len(buffer) < expected:
raise Kwp2000Error(f"truncated frame: {buffer.hex(' ')}")
frame = buffer[:expected]
received = frame[-1]
computed = checksum(frame[:-1])
if received != computed:
raise ChecksumError(
f"checksum mismatch: got 0x{received:02X}, computed 0x{computed:02X} "
f"over {frame[:-1].hex(' ')}"
)
fmt = frame[0]
addressed = bool(fmt & 0x80)
header_len = 3 if addressed else 1
target = frame[1] if addressed else None
source = frame[2] if addressed else None
body_start = header_len if (fmt & 0x3F) else header_len + 1
payload = frame[body_start:-1]
if not payload:
raise Kwp2000Error(f"frame carries no payload: {frame.hex(' ')}")
return Frame(format_byte=fmt, target=target, source=source, payload=payload)
def expect_positive(frame: Frame, service: int) -> bytes:
"""Return the response data, raising :class:`NegativeResponse` on 0x7F.
The leading positive-response SID is stripped; what comes back is just the
service's answer.
"""
payload = frame.payload
if payload[0] == NEGATIVE_RESPONSE:
rejected = payload[1] if len(payload) > 1 else service
code = payload[2] if len(payload) > 2 else 0x00
raise NegativeResponse(rejected, code)
if payload[0] != service + POSITIVE_RESPONSE_OFFSET:
raise Kwp2000Error(
f"expected positive response 0x{service + POSITIVE_RESPONSE_OFFSET:02X} "
f"to service 0x{service:02X}, got 0x{payload[0]:02X}"
)
return payload[1:]

86
tunie/src/tunie/maps.py Normal file
View File

@@ -0,0 +1,86 @@
"""Offline Triumph map database, extracted from the TuneECU APK resources.
TuneECU ships its map catalogue as ``<string-array name="t2xx">`` entries in
``res/values/arrays.xml``. Each item has the form::
<map id>:<ecu type>:<description line>\n<description line>\n...
The ECU type character matches the ``ecu`` string-array in the same file, where
``0`` is Triumph (Keihin). The description lines carry the details that actually
decide map compatibility -- VIN range, silencer type, market, and critically
"Mechanical odometer" vs "LCD odometer", which separates ECU generations.
"""
from __future__ import annotations
import json
import re
from pathlib import Path
_ITEM = re.compile(r"<item>(.*?)</item>", re.S)
_ARRAY = re.compile(r'<string-array name="(t\d+)"[^>]*>(.*?)</string-array>', re.S)
def extract_from_arrays_xml(arrays_xml: str | Path, output: str | Path) -> int:
"""Parse ``arrays.xml`` into a JSON map database. Returns the entry count."""
text = Path(arrays_xml).read_text(encoding="utf-8")
entries: list[dict] = []
for array_match in _ARRAY.finditer(text):
group = array_match.group(1)
for raw in _ITEM.findall(array_match.group(2)):
entry = _parse_item(raw, group)
if entry:
entries.append(entry)
entries.sort(key=lambda e: e["id"])
Path(output).write_text(json.dumps(entries, indent=2), encoding="utf-8")
return len(entries)
def _parse_item(raw: str, group: str) -> dict | None:
cleaned = raw.strip().strip('"').replace("\\n", "\n")
parts = cleaned.split(":", 2)
if len(parts) < 3 or not parts[0].strip().isdigit():
return None
description = [line.strip() for line in parts[2].splitlines() if line.strip()]
return {
"id": parts[0].strip(),
"ecu": parts[1].strip(),
"group": group,
"description": description,
}
def load_map_database(path: str | Path) -> list[dict]:
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"map database not found at {p}")
return json.loads(p.read_text(encoding="utf-8"))
def find_maps(
database: list[dict],
search: str = "",
*,
ecu: str | None = None,
odometer: str | None = None,
) -> list[dict]:
"""Filter the database by free text, ECU type code and instrument type."""
needle = search.lower()
wanted_odo = {"mechanical": "mechanical odometer", "lcd": "lcd odometer"}.get(
odometer or ""
)
results = []
for entry in database:
blob = " ".join(entry["description"]).lower()
if needle and needle not in blob and needle not in entry["id"]:
continue
if ecu is not None and entry["ecu"] != ecu:
continue
if wanted_odo is not None and wanted_odo not in blob:
continue
results.append(entry)
return results

109
tunie/src/tunie/safety.py Normal file
View File

@@ -0,0 +1,109 @@
"""Structural enforcement of read-only operation.
The only way to damage a Keihin ECU over KWP2000 is to reach the memory-write
services, and those are gated behind Security Access (0x27). This module makes
that unreachable by construction rather than by discipline: every outbound
service passes through :func:`assert_read_only`, which raises before any byte is
handed to a transport.
If you are extending this tool and find yourself wanting to relax this file,
stop. Get a full ROM dump and a spare ECU first.
"""
from __future__ import annotations
# Services that cannot alter ECU state. Anything not listed here is refused.
READ_ONLY_SERVICES: frozenset[int] = frozenset(
{
0x10, # StartDiagnosticSession (session type additionally restricted below)
0x13, # ReadDiagnosticTroubleCodes
0x17, # ReadStatusOfDiagnosticTroubleCodes
0x18, # ReadDiagnosticTroubleCodesByStatus
0x1A, # ReadEcuIdentification
0x21, # ReadDataByLocalIdentifier
0x22, # ReadDataByCommonIdentifier
0x23, # ReadMemoryByAddress
0x3E, # TesterPresent
0x81, # StartCommunication
0x82, # StopCommunication
0x83, # AccessTimingParameters (read/report subfunctions only, see below)
}
)
# Named for the error messages. These are the services that can brick the bike.
_DESTRUCTIVE = {
0x14: "ClearDiagnosticInformation",
0x27: "SecurityAccess",
0x28: "CommunicationControl",
0x2C: "DynamicallyDefineLocalIdentifier",
0x2E: "WriteDataByCommonIdentifier",
0x2F: "InputOutputControlByCommonIdentifier",
0x30: "InputOutputControlByLocalIdentifier",
0x31: "StartRoutineByLocalIdentifier (FLASH ERASE)",
0x34: "RequestDownload (FLASH WRITE)",
0x35: "RequestUpload",
0x36: "TransferData (FLASH WRITE)",
0x37: "RequestTransferExit",
0x38: "StartRoutineByAddress",
0x3B: "WriteDataByLocalIdentifier",
0x3D: "WriteMemoryByAddress",
0x11: "EcuReset",
}
# Diagnostic sessions that suspend engine control or arm the bootloader.
# 0x81 is the plain default session and is all a read-only tool needs.
_ALLOWED_SESSIONS = {0x81, 0x89}
_PROGRAMMING_SESSIONS = {0x02, 0x85, 0x86, 0x02}
# AccessTimingParameters subfunctions that only report values.
_ALLOWED_TIMING_SUBFUNCTIONS = {0x00, 0x01, 0x02}
class WriteAttemptBlocked(RuntimeError):
"""Raised when something tried to send a state-modifying service."""
def assert_read_only(payload: bytes) -> None:
"""Refuse any request that could modify ECU state.
``payload`` is the service identifier followed by its arguments, i.e. the
KWP2000 frame with the header and checksum stripped.
"""
if not payload:
raise WriteAttemptBlocked("refusing to send an empty request")
sid = payload[0]
if sid in _DESTRUCTIVE:
raise WriteAttemptBlocked(
f"blocked service 0x{sid:02X} ({_DESTRUCTIVE[sid]}): this tool is "
"read-only and does not implement the flash write path"
)
if sid not in READ_ONLY_SERVICES:
raise WriteAttemptBlocked(
f"blocked unrecognised service 0x{sid:02X}: only services on the "
"read-only allowlist may be sent"
)
if sid == 0x10:
session = payload[1] if len(payload) > 1 else None
if session in _PROGRAMMING_SESSIONS:
raise WriteAttemptBlocked(
f"blocked StartDiagnosticSession 0x{session:02X}: programming "
"sessions suspend engine control and arm the bootloader"
)
if session not in _ALLOWED_SESSIONS:
raise WriteAttemptBlocked(
f"blocked StartDiagnosticSession 0x{session:02X}: only the "
f"default session ({', '.join(f'0x{s:02X}' for s in sorted(_ALLOWED_SESSIONS))}) "
"is permitted"
)
if sid == 0x83:
sub = payload[1] if len(payload) > 1 else None
if sub not in _ALLOWED_TIMING_SUBFUNCTIONS:
raise WriteAttemptBlocked(
f"blocked AccessTimingParameters 0x{sub:02X}: only the read and "
"report subfunctions are permitted"
)

0
tunie/src/tunie/transport/__init__.py Normal file
View File

55
tunie/src/tunie/transport/base.py Normal file
View File

@@ -0,0 +1,55 @@
"""Transport interface shared by the K-Line and ELM327 backends.
A transport is responsible for everything below the service layer: line
initialisation, framing, checksums and echo handling. Callers hand it a service
payload (SID + arguments) and get back the ECU's payload.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from ..safety import assert_read_only
class TransportError(Exception):
"""Physical or adapter-level failure."""
class NoResponse(TransportError):
"""The ECU did not answer within the protocol timeout."""
class Transport(ABC):
"""Base class enforcing the read-only guard on every outbound request."""
#: Filled in by :meth:`connect` when the ECU reports them during init.
key_bytes: tuple[int, int] | None = None
def request(self, payload: bytes, *, timeout_ms: int | None = None) -> bytes:
"""Send a service payload and return the ECU's response payload.
Every request passes through the read-only allowlist first. Subclasses
implement :meth:`_exchange` and must not be called directly.
"""
assert_read_only(payload)
return self._exchange(payload, timeout_ms=timeout_ms)
@abstractmethod
def _exchange(self, payload: bytes, *, timeout_ms: int | None = None) -> bytes:
"""Perform the actual send/receive. Assume the payload is already vetted."""
@abstractmethod
def connect(self) -> None:
"""Open the port and bring up a diagnostic link with the ECU."""
@abstractmethod
def close(self) -> None:
"""Tear the link down and release the port."""
def __enter__(self) -> "Transport":
self.connect()
return self
def __exit__(self, *exc_info: object) -> None:
self.close()

189
tunie/src/tunie/transport/elm327.py Normal file
View File

@@ -0,0 +1,189 @@
"""ELM327 transport (Bluetooth or USB), for adapters like the OBDLink LX.
The ELM327 handles ISO 14230 timing, framing and checksums in firmware, so this
backend hands it hex payloads and reads hex back. That makes it the safest way
to interrogate an ECU -- the adapter physically will not emit a raw flash
sequence -- but it comes with two real caveats:
1. The ELM's built-in KWP modes (``ATSP4``/``ATSP5``) assume OBD-II addressing
(target 0x33). Triumph does not use OBD-II addressing, so we override the
header with ``ATSH`` and disable automatic formatting where possible.
2. Clone adapters frequently have broken or absent KWP support. If init fails
here but the protocol is otherwise right, suspect the adapter before the code.
"""
from __future__ import annotations
import logging
import time
try:
import serial
except ImportError as exc: # pragma: no cover
raise ImportError("tunie requires pyserial: pip install pyserial") from exc
from .. import kwp2000 as k
from .. import triumph
from .base import NoResponse, Transport, TransportError
log = logging.getLogger(__name__)
PROMPT = b">"
class Elm327Transport(Transport):
"""KWP2000 over an ELM327-compatible adapter.
Args:
port: Serial device. A paired Bluetooth OBD adapter shows up on macOS as
something like ``/dev/cu.OBDII`` or ``/dev/cu.vLinker-SPPDev``.
target: ECU address used in the ``ATSH`` header.
source: Tester address used in the ``ATSH`` header.
protocol: ``5`` for ISO 14230-4 KWP fast init, ``4`` for 5-baud init.
"""
def __init__(
self,
port: str,
*,
target: int = triumph.ECU_ADDRESS,
source: int = triumph.TESTER_ADDRESS_KLINE,
init: str = "fast",
baud: int = 38400,
) -> None:
self.port = port
self.target = target
self.source = source
self.init = init
self.baud = baud
self._serial: serial.Serial | None = None
self._current_header: str | None = None
# -- lifecycle ---------------------------------------------------------
def connect(self) -> None:
"""Bring up the link using TuneECU's own Triumph init sequence.
The command list is lifted verbatim from ``MainActivity.java`` in the
decompiled APK: ``a3`` for fast init, ``W2`` for 5-baud init. Both end
by setting header ``D5 F5``.
"""
self._serial = serial.Serial(self.port, baudrate=self.baud, timeout=1.0)
time.sleep(0.5)
self._serial.reset_input_buffer()
identity = self._at("Z", timeout=5.0) # reset
log.info("adapter identifies as: %s", identity)
if "ELM" not in identity.upper() and "OBD" not in identity.upper():
log.warning("adapter did not report an ELM327 identity string")
sequence = (
triumph.ELM_INIT_FAST if self.init == "fast" else triumph.ELM_INIT_SLOW
)
for command in sequence:
response = self._command(command, timeout=5.0)
if "OK" not in response.upper() and command not in ("ATZ",):
log.warning("%s returned %r", command, response)
# TuneECU then sends the bare StartCommunication service, "81", which is
# what actually triggers the bus init. Route it through the normal
# request path so the read-only guard sees it.
data = self.request(bytes([k.Sid.START_COMMUNICATION]), timeout_ms=5000)
if len(data) >= 2:
self.key_bytes = (data[0], data[1])
log.info("ECU key bytes: 0x%02X 0x%02X", data[0], data[1])
def close(self) -> None:
if self._serial is None:
return
try:
self._at("PC") # protocol close
except Exception:
log.debug("protocol close failed, releasing port anyway")
finally:
self._serial.close()
self._serial = None
# -- AT command layer --------------------------------------------------
def _at(self, command: str, *, timeout: float = 2.0) -> str:
return self._command(f"AT{command}", timeout=timeout)
def _command(self, command: str, *, timeout: float) -> str:
if self._serial is None:
raise TransportError("transport is not connected")
log.debug(">> %s", command)
self._serial.reset_input_buffer()
self._serial.write(command.encode("ascii") + b"\r")
self._serial.flush()
buf = bytearray()
deadline = time.monotonic() + timeout
while PROMPT not in buf:
if time.monotonic() > deadline:
raise NoResponse(
f"adapter did not return a prompt for {command!r}; "
f"partial: {bytes(buf).decode('ascii', 'replace')!r}"
)
chunk = self._serial.read(64)
if chunk:
buf.extend(chunk)
text = bytes(buf).replace(PROMPT, b"").decode("ascii", "replace").strip()
log.debug("<< %s", text)
return text
# -- exchange ----------------------------------------------------------
def _exchange(self, payload: bytes, *, timeout_ms: int | None = None) -> bytes:
timeout = (timeout_ms or k.P2_MAX_MS * 4) / 1000
# The format byte encodes the payload length, so the header has to be
# re-set whenever the length changes. TuneECU hardcodes 0x81/0x82 for
# its fixed-size init messages; we compute it instead.
fmt = 0x80 | len(payload) if len(payload) <= 0x3F else 0x80
header = f"ATSH{fmt:02X}{self.target:02X}{self.source:02X}"
if header != self._current_header:
self._command(header, timeout=2.0)
self._current_header = header
# The ELM appends the checksum and header itself; send the payload only.
text = self._command(payload.hex().upper(), timeout=max(timeout, 2.0))
for marker in ("NO DATA", "UNABLE TO CONNECT", "BUS ERROR", "CAN ERROR", "?"):
if marker in text.upper():
raise NoResponse(f"adapter reported {text!r} for {payload.hex(' ')}")
raw = self._parse_hex_response(text)
frame = k.parse_frame(raw) if len(raw) > 3 and raw[0] & 0x80 else None
if frame is None:
# ATH1 was set, so we normally get headers. If not, treat the whole
# response as a bare payload.
return k.expect_positive(
k.Frame(format_byte=0, target=None, source=None, payload=raw),
payload[0],
)
while frame.is_response_pending:
text = self._command("", timeout=k.P2_EXTENDED_MS / 1000)
frame = k.parse_frame(self._parse_hex_response(text))
return k.expect_positive(frame, payload[0])
@staticmethod
def _parse_hex_response(text: str) -> bytes:
"""Flatten the adapter's hex lines into bytes, dropping status noise."""
cleaned = []
for line in text.splitlines():
line = line.strip().replace(" ", "")
if not line or not all(c in "0123456789ABCDEFabcdef" for c in line):
continue
cleaned.append(line)
joined = "".join(cleaned)
if not joined:
raise TransportError(f"no hex payload in adapter response: {text!r}")
if len(joined) % 2:
raise TransportError(f"odd-length hex payload: {joined!r}")
return bytes.fromhex(joined)

233
tunie/src/tunie/transport/kline.py Normal file
View File

@@ -0,0 +1,233 @@
"""Raw K-Line transport for FTDI-based KKL 409.1 cables.
The K-Line is a single-wire half-duplex bus, which has one consequence that
dominates the implementation: **everything we transmit is echoed back to us**.
Each write must be followed by reading and discarding exactly as many bytes as
were sent, or the echo will be mistaken for the ECU's reply.
Line initialisation uses the serial break condition to hold TX low, which is how
fast init's 25 ms low / 25 ms high pulse is produced without bit-banging GPIO.
"""
from __future__ import annotations
import logging
import time
try:
import serial
except ImportError as exc: # pragma: no cover - dependency is declared in pyproject
raise ImportError("tunie requires pyserial: pip install pyserial") from exc
from .. import kwp2000 as k
from .. import triumph
from .base import NoResponse, Transport, TransportError
log = logging.getLogger(__name__)
KWP_BAUD = 10400
class KLineTransport(Transport):
"""KWP2000 over a raw serial K-Line interface.
Args:
port: Serial device, e.g. ``/dev/cu.usbserial-A9007UX1`` on macOS.
target: ECU address. Defaults to Triumph's 0xD5, recovered from the
TuneECU APK (see :mod:`tunie.triumph`).
source: Tester address. Triumph uses 0xF5 on K-Line, not the more
common 0xF1.
init: ``"fast"`` (ISO 14230-2 fast init) or ``"slow"`` (5-baud address).
force_length_byte: Emit the separate-length-byte header form.
"""
def __init__(
self,
port: str,
*,
target: int = triumph.ECU_ADDRESS,
source: int = triumph.TESTER_ADDRESS_KLINE,
init: str = "fast",
force_length_byte: bool = False,
baud: int = KWP_BAUD,
) -> None:
self.port = port
self.target = target
self.source = source
self.init = init
self.force_length_byte = force_length_byte
self.baud = baud
self._serial: serial.Serial | None = None
self._last_activity = 0.0
# -- lifecycle ---------------------------------------------------------
def connect(self) -> None:
self._serial = serial.Serial(
port=self.port,
baudrate=self.baud,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=0.1,
)
self._serial.reset_input_buffer()
self._serial.reset_output_buffer()
if self.init == "fast":
self._fast_init()
elif self.init == "slow":
self._slow_init()
else:
raise ValueError(f"unknown init mode {self.init!r}, expected 'fast' or 'slow'")
def close(self) -> None:
if self._serial is None:
return
try:
# StopCommunication is courtesy; the ECU times out on its own anyway.
self._exchange(bytes([k.Sid.STOP_COMMUNICATION]), timeout_ms=200)
except Exception:
log.debug("StopCommunication failed on close, releasing port anyway")
finally:
self._serial.close()
self._serial = None
# -- initialisation ----------------------------------------------------
def _fast_init(self) -> None:
"""ISO 14230-2 fast init: idle, 25 ms low, 25 ms high, then StartComms."""
assert self._serial is not None
s = self._serial
s.break_condition = False
time.sleep(k.W5_IDLE_MS / 1000)
s.break_condition = True
time.sleep(k.TINIL_MS / 1000)
s.break_condition = False
time.sleep(k.TINIH_MS / 1000)
s.reset_input_buffer()
data = self._exchange(bytes([k.Sid.START_COMMUNICATION]), timeout_ms=1000)
self._record_key_bytes(data)
def _slow_init(self, address: int = triumph.SLOW_INIT_ADDRESS) -> None:
"""ISO 9141 style 5-baud address init, as a fallback if fast init fails."""
assert self._serial is not None
s = self._serial
s.break_condition = False
time.sleep(k.W5_IDLE_MS / 1000)
# 5 baud == 200 ms per bit: start bit, 8 data bits LSB first, stop bit.
bit_time = 0.2
s.break_condition = True # start bit (low)
time.sleep(bit_time)
for i in range(8):
s.break_condition = not (address >> i) & 1
time.sleep(bit_time)
s.break_condition = False # stop bit (high)
time.sleep(bit_time)
sync = self._read_exact(1, deadline=time.monotonic() + 0.5)
if sync != b"\x55":
raise TransportError(
f"5-baud init: expected sync byte 0x55, got {sync.hex() or '<nothing>'}"
)
kb = self._read_exact(2, deadline=time.monotonic() + 0.5)
self.key_bytes = (kb[0], kb[1])
# Tester acknowledges by returning the inverse of key byte 2.
time.sleep(0.030)
self._write(bytes([(~kb[1]) & 0xFF]))
self._read_exact(1, deadline=time.monotonic() + 0.5) # our own echo
inv_addr = self._read_exact(1, deadline=time.monotonic() + 0.5)
expected = (~address) & 0xFF
if inv_addr[0] != expected:
raise TransportError(
f"5-baud init: expected inverted address 0x{expected:02X}, "
f"got 0x{inv_addr[0]:02X}"
)
def _record_key_bytes(self, data: bytes) -> None:
if len(data) >= 2:
self.key_bytes = (data[0], data[1])
log.info("ECU key bytes: 0x%02X 0x%02X", data[0], data[1])
# -- exchange ----------------------------------------------------------
def _exchange(self, payload: bytes, *, timeout_ms: int | None = None) -> bytes:
if self._serial is None:
raise TransportError("transport is not connected")
frame = k.build_frame(
payload,
self.target,
self.source,
force_length_byte=self.force_length_byte,
)
log.debug("TX %s", frame.hex(" "))
# P3: minimum gap between the previous response and a new request.
elapsed = time.monotonic() - self._last_activity
if elapsed < k.P3_MIN_MS / 1000:
time.sleep(k.P3_MIN_MS / 1000 - elapsed)
self._serial.reset_input_buffer()
self._write(frame)
self._consume_echo(frame)
budget = (timeout_ms if timeout_ms is not None else k.P2_MAX_MS * 4) / 1000
response = self._read_frame(budget)
# The ECU may stall us with 0x78 while it works; keep waiting.
while response.is_response_pending:
log.debug("response pending, extending timeout")
response = self._read_frame(k.P2_EXTENDED_MS / 1000)
self._last_activity = time.monotonic()
return k.expect_positive(response, payload[0])
def _write(self, data: bytes) -> None:
assert self._serial is not None
self._serial.write(data)
self._serial.flush()
def _consume_echo(self, sent: bytes) -> None:
"""Read back and verify the half-duplex echo of our own transmission."""
echo = self._read_exact(len(sent), deadline=time.monotonic() + 1.0)
if echo != sent:
raise TransportError(
f"K-Line echo mismatch: sent {sent.hex(' ')}, echoed {echo.hex(' ')}. "
"Check the transceiver wiring and that nothing else is driving the bus."
)
def _read_exact(self, count: int, *, deadline: float) -> bytes:
assert self._serial is not None
buf = bytearray()
while len(buf) < count:
if time.monotonic() > deadline:
raise NoResponse(
f"timed out waiting for {count} bytes, got {len(buf)} "
f"({bytes(buf).hex(' ') or 'none'})"
)
chunk = self._serial.read(count - len(buf))
if chunk:
buf.extend(chunk)
return bytes(buf)
def _read_frame(self, budget_s: float) -> k.Frame:
"""Read one complete frame, using the header to know when to stop."""
deadline = time.monotonic() + budget_s
buf = bytearray()
while True:
expected = k.frame_length(bytes(buf))
if expected is not None and len(buf) >= expected:
break
need = 1 if expected is None else expected - len(buf)
buf.extend(self._read_exact(need, deadline=deadline))
frame_bytes = bytes(buf)
log.debug("RX %s", frame_bytes.hex(" "))
return k.parse_frame(frame_bytes)

72
tunie/src/tunie/triumph.py Normal file
View File

@@ -0,0 +1,72 @@
"""Triumph-specific protocol constants.
Everything here was recovered from the TuneECU APK rather than guessed, and the
two independent sources agree:
* ``com/tuneecu/m.java:2732`` -- the ``Ld()`` frame builder. For the Triumph
Keihin ECU (flag ``Hf``, set when the ECU-type character is ``"0"``), none of
the ``bf``/``hg``/``ig``/``Rf`` branches are taken, so the target resolves to
``(Rf ? Be : 0) + 213`` = ``0xD5`` and the source to ``0xF5``.
* ``com/tuneecu/MainActivity.java:922`` -- the ELM327 init string list ``a3``,
which sets ``ATSH81D5F5``: format 0x81, target 0xD5, source 0xF5.
The newer CAN-based Triumphs use ``ATSHDAD5F1`` / ``ATSH18DAD5F1``, i.e. 0xD5
again -- the ECU address is stable across Triumph's protocol generations.
"""
from __future__ import annotations
#: Triumph ECU diagnostic address.
ECU_ADDRESS = 0xD5
#: Tester address Triumph expects on K-Line. Note this is 0xF5, not the more
#: common 0xF1 -- the CAN-era sequences do use 0xF1, K-Line does not.
TESTER_ADDRESS_KLINE = 0xF5
#: Tester address used by the CAN-era Triumphs, for reference.
TESTER_ADDRESS_CAN = 0xF1
#: ELM327 init for K-Line Triumph, fast init (TuneECU list ``a3``).
ELM_INIT_FAST = ("ATE0", "ATL0", "ATAL", "ATS0", "ATH1", "ATTP5", "ATSH81D5F5")
#: ELM327 init for K-Line Triumph, 5-baud init (TuneECU list ``W2``).
#: ``ATIIAD5`` sets the slow-init address to 0xD5.
ELM_INIT_SLOW = ("ATE0", "ATL0", "ATAL", "ATS0", "ATH1", "ATIIAD5", "ATTP4", "ATSH82D5F5")
#: Slow-init address byte, from ``ATIIAD5``.
SLOW_INIT_ADDRESS = 0xD5
#: ECU-type characters from the ``ecu`` string-array in the APK resources.
ECU_TYPES = {
"0": "Triumph (Keihin)",
"1": "Triumph (Sagem)",
"P": "Triumph 400 (Bosch)",
}
#: Read-only KWP2000 services observed in TuneECU's ``m.java``, with the source
#: line each was recovered from. Useful as a work list for the identify pass.
OBSERVED_READ_SERVICES = {
0x81: "StartCommunication (m.java:1335, payload {0x81})",
0x1A: "ReadEcuIdentification (m.java:1106, payload {0x1A})",
0x21: "ReadDataByLocalIdentifier (m.java:763, payload {0x21,0x80})",
0x22: "ReadDataByCommonIdentifier (m.java:4917, payload {0x22,0x03,0x00})",
0x18: "ReadDtcByStatus (m.java:737, payload {0x18,0x00,0xFF,0x00})",
0x13: "ReadDtc (m.java:7941, payload {0x13,0x40,0xFF})",
0x3E: "TesterPresent (m.java:8909, payload {0x3E,0x00})",
0x83: "AccessTimingParameters (m.java:6345, payload {0x83,0x03,0x1E,...})",
}
#: Services TuneECU uses that this tool deliberately does not implement. 0x35
#: RequestUpload is how a ROM dump is performed and will be needed in phase 2,
#: but it shares the 0x36 TransferData machinery with the write path, so it
#: stays out of the read-only tool.
OBSERVED_WRITE_SERVICES = {
0x27: "SecurityAccess (m.java:1110/1491/4005/4054/9300/9304)",
0x34: "RequestDownload (m.java:9116)",
0x35: "RequestUpload (m.java:8958, {0x35,0x00,0x5F,0xE0,0,0,0,0x20})",
0x36: "TransferData (m.java:6319/10710)",
0x37: "RequestTransferExit (m.java:881)",
0x14: "ClearDiagnosticInformation (m.java:6341)",
0x11: "EcuReset (m.java:9206)",
0x2E: "WriteDataByCommonIdentifier (m.java:2236/9824)",
}

74
tunie/tests/verify_protocol.py Normal file
View File

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