Add BYD vehicle telemetry exporter
Test / test (push) Failing after 25m47s

Go port of the read path of pybyd 0.0.75 (the client behind
jkaberg/hass-byd-vehicle): the bangcle white-box AES envelope, the
signing/checkcode layer, login, vehicle list, and realtime readings.

Scraping runs on a cron schedule rather than per Prometheus request
because each reading wakes the car's T-Box and drains the traction
battery; /metrics serves the last result.

Every numeric field of the realtime payload becomes a gauge, so the
exporter does not need a hand-maintained field list that goes stale
when BYD changes the payload.

Tests pin the wire format against pybyd byte-for-byte and drive the
client end to end against a fake cloud that speaks the real protocol.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
thistineandClaude Opus 5 committed 2026-09-05 22:03:16 +07:00
commit 52e1aaa0c0
30 files changed
+3501

No files matched your search

+155
View File
@@ -0,0 +1,155 @@
package bangcle
// Expected values come from pybyd 0.0.75 — regenerate with testdata/gen_fixtures.py.
import (
"strings"
"testing"
)
func TestRoundTrip(t *testing.T) {
inputs := []string{
"",
"a",
"0123456789abcdef", // exactly one block: must gain a whole pad block
strings.Repeat("x", 4096),
`{"request":"hello world"}`,
"ünïcodé ✓",
}
for _, in := range inputs {
out, err := Decode(Encode([]byte(in)))
if err != nil {
t.Fatalf("Decode(Encode(%.20q)): %v", in, err)
}
if string(out) != in {
t.Errorf("round trip %.20q -> %.20q", in, out)
}
}
}
// TestVectors pins the white-box AES port against ciphertext from pybyd. If the
// block functions drift by a single index this fails; round-tripping alone
// would not, since it would drift symmetrically.
func TestVectors(t *testing.T) {
vectors := []struct{ plain, want string }{
{"", "FlpybH7xiohrkjagBOJY5Mg=="},
{"hello", "FsFbBBn9/qw4cYEF9S1Jk6w=="},
{"ünïcodé ✓", "FRqewj1Ypq1V2n8VqsiZPUA=="},
{`{"functionType":"pwdLogin"}`, "FCHhJ73+HviiIT3dTDBeITnzHp5caNe/C50GlE8P0dqQ="},
}
for _, v := range vectors {
if got := Encode([]byte(v.plain)); got != v.want {
t.Errorf("Encode(%q)\n got %s\nwant %s", v.plain, got, v.want)
}
}
}
func TestDecodeRejectsMalformed(t *testing.T) {
cases := map[string]string{
"empty": "",
"missing prefix": "aGVsbG8=",
"bad base64": "F!!!!",
"short block": "F" + "AAAA", // 3 bytes: not a multiple of 16
}
for name, in := range cases {
if _, err := Decode(in); err == nil {
t.Errorf("%s: expected an error for %q", name, in)
}
}
}
// TestDecodeNormalises covers the url-safe alphabet, stray whitespace and the
// missing padding that BYD responses sometimes arrive with.
func TestDecodeNormalises(t *testing.T) {
want := "the quick brown fox"
env := Encode([]byte(want))
variants := []string{
env,
strings.TrimRight(env, "="),
strings.NewReplacer("+", "-", "/", "_").Replace(env),
" " + env + "\n",
}
for _, v := range variants {
got, err := Decode(v)
if err != nil {
t.Errorf("Decode(%q): %v", v, err)
continue
}
if string(got) != want {
t.Errorf("Decode(%q) = %q, want %q", v, got, want)
}
}
}
func TestStripPKCS7LeavesUnpaddedData(t *testing.T) {
// BYD sometimes returns payloads without padding; pybyd passes them
// through rather than erroring, and so must this.
for _, in := range []string{"", "no padding here", "\x05\x05\x05"} {
if got := StripPKCS7([]byte(in)); string(got) != in {
t.Errorf("StripPKCS7(%q) = %q, want it unchanged", in, got)
}
}
padded := append([]byte("abc"), 13, 13, 13, 13, 13, 13, 13, 13, 13, 13, 13, 13, 13)
if got := StripPKCS7(padded); string(got) != "abc" {
t.Errorf("StripPKCS7(valid) = %q, want %q", got, "abc")
}
}
func TestPadPKCS7AlwaysAddsAtLeastOneByte(t *testing.T) {
for _, n := range []int{0, 1, 15, 16, 17, 32} {
got := PadPKCS7(make([]byte, n))
if len(got)%16 != 0 || len(got) <= n {
t.Errorf("PadPKCS7(%d bytes) produced %d bytes", n, len(got))
}
}
}
func TestTablesLoaded(t *testing.T) {
for _, spec := range tableSpecs {
var got int
switch spec.name {
case "inv_round":
got = len(tables.invRound)
case "inv_xor":
got = len(tables.invXor)
case "inv_first":
got = len(tables.invFirst)
case "round":
got = len(tables.round)
case "xor":
got = len(tables.xor)
case "final":
got = len(tables.final)
case "perm_decrypt":
got = len(tables.permDecrypt)
case "perm_encrypt":
got = len(tables.permEncrypt)
}
if got != spec.size {
t.Errorf("table %s: %d bytes, want %d", spec.name, got, spec.size)
}
}
}
func TestLoadTablesRejectsBadContainer(t *testing.T) {
cases := map[string][]byte{
"too short": []byte("BGTB"),
"bad magic": append([]byte("XXXX"), make([]byte, 100)...),
"bad version": append([]byte("BGTB\x02\x00\x08\x00"), make([]byte, 100)...),
"bad count": append([]byte("BGTB\x01\x00\x03\x00"), make([]byte, 100)...),
}
for name, data := range cases {
if _, err := loadTables(data); err == nil {
t.Errorf("%s: expected an error", name)
}
}
}
func BenchmarkEncode(b *testing.B) {
payload := []byte(strings.Repeat(`{"field":"value"},`, 100))
b.SetBytes(int64(len(payload)))
for i := 0; i < b.N; i++ {
Encode(payload)
}
}