新版软件后端接口

[version u8=0x03][op u8][seq u8][TLV...]
短 TLV: [tag][len<=0xFE][value]
长 TLV: [tag][0xFF][len u16 little-endian][value]

ownerUid = 2147483646

KDF        = PBKDF2-HMAC-SHA256
password   = 密码的 UTF-8 字节
salt       = 密码学安全随机生成的 16 字节
iterations = 200000
dkLen      = 32 字节
KDF ID     = 1

BIND_START op=0x20
TLV 0x01 = 空；仅云端绑定使用 binding credential
TLV 0x02 = credentialId UTF-8
TLV 0x03 = HELLO_ACK.sn UTF-8
TLV 0x04 = deviceKey raw bytes，正好 32B
TLV 0x05 = bindMode u8 = 1，表示本地绑定请求
TLV 0x06 = ownerUid u64 LE = 2147483646
TLV 0x07 = KDF salt raw bytes，正好 16B
TLV 0x08 = iterations u32 LE = 200000
TLV 0x09 = KDF ID u8 = 1

UNBIND op = 0x23
TLV 0x01  = ts，Unix 秒，u64 little-endian
TLV 0x02  = mac，16 bytes

payload = "unbind|" || UTF8(sn) || "|" || ts_le_8B
mac = HMAC-SHA256(deviceKeyRaw, payload)[0:16]

message = ws|user_id|sn|ts|nonce
sig = Base64(HMAC-SHA256(deviceKeyRaw, UTF8(message)))

user_id=<UID>&client_id=<client-id>&sn=<SN>&ts=<Unix毫秒>&nonce=<随机值>&sig=<URL编码后的Base64签名>

ws://<robot>:8081/?user_id=42&client_id=app&sn=BXI-A1&ts=...&nonce=...&sig=...
ws://<robot>:8081/ws/signaling?user_id=42&client_id=video_v1&sn=BXI-A1&ts=...&nonce=...&sig=...

bodyHash = lowercaseHex(SHA256(exactRawBodyBytes))
message = http.v2|UPPER_METHOD|PATH|bodyHash|user_id|sn|ts|nonce
sig = Base64(HMAC-SHA256(deviceKeyRaw, UTF8(message)))

auth_v=2&user_id=<UID>&sn=<SN>&ts=<Unix毫秒>&nonce=<随机值>&sig=<URL编码后的Base64签名>

22B header: <BBHffffH>
            ver, auth, seq, vx, vy, wz, height, buttons

24B header: <BBHffffHBB>
            基础头 + btn_slot + btn_val

tag = first8Bytes(HMAC-SHA256(deviceKeyRaw, headerBytes))
frame = headerBytes + tag

{
  "v": 1,
  "sn": "BXI-EXAMPLE-0001",
  "role": "co_owner",
  "iat": 1780000000,
  "exp": 1780000120,
  "epoch": 0,
  "jti": "每张分享码的唯一随机值",
  "k": "64位十六进制subkey",
  "sig": "Base64 HMAC-SHA256签名",
  "conn": {"i": "192.168.88.162", "p": 8081}
}

shareKey = HMAC-SHA256(deviceKeyRaw, "bxi-share-key-v1|<epoch>")
tokenSig = Base64(HMAC-SHA256(shareKey,
           "bxishare.v1|<sn>|<role>|<iat>|<exp>|<epoch>|<jti>"))
subkey   = HMAC-SHA256(shareKey, "bxi-share-tx-v1|<jti>")

{"type":"discover"}

{
  "type": "beacon",
  "hostname": "robot_elf3_02",
  "port": 8081,
  "seq": 42,
  "sn": "BXI-EXAMPLE-0001",
  "pairing_protocol": 1
}

{
  "type": "control.cmd_vel",
  "ts": 1780000000000,
  "seq": 1,
  "payload": {}
}

{
  "type": "control.cmd_vel",
  "payload": {
    "vx": 0.2,
    "vy": 0.0,
    "wz": 0.1,
    "height": 1.0,
    "mode": "manual",
    "btn_1": 0,
    "btn_5": 1
  }
}

nav.map                 nav.scan
nav.pose                nav.path.global
nav.costmap.global      nav.costmap.local
nav.footprint           nav.cloud
nav.status              nav.tour.status
nav.mapping.status      nav.runtime.status
nav.reloc_required

current_mode, desired_mode, transition_id, active_map_id,
localized, fitness_score, inlier_ratio, driver_healthy, last_error

HMAC(deviceKey, "wifi-session-v1|" || mode || "|" ||
     client_public_key_bytes || "|" || client_nonce_bytes)

shared = X25519(client_private_key, robot_public_key)
session_key = HKDF-SHA256(
  IKM  = shared || PSK,
  salt = client_nonce || robot_nonce,
  info = "bxi-wifi-pairing-v1|" || hex_decode(session_id),
  len  = 32
)

{"session_id":"32位十六进制","nonce":"Base64 12B","ciphertext":"Base64 AES-GCM 密文与 tag"}

bxi.pairing.v1|<HTTP_METHOD>|<PATH>|<session_id>

HMAC(deviceKey, "wifi-unbind-v1|" || session_id || "|" || sn)

[ver u8=0x03][auth u8][seq u16 LE]
[vx f32 LE][vy f32 LE][wz f32 LE][height f32 LE]
[buttons u16 LE][btn_slot u8][btn_val u8][HMAC tag]

[total_len:u16 little-endian][offset:u16 little-endian][chunk...]

[ver u8=0x03][flags u8][safety u8]

# 女声（默认）
ros2 topic pub --once /tts/say std_msgs/msg/String "{data: '欢迎参观'}"

# 显式指定女声
ros2 topic pub --once /tts/say std_msgs/msg/String \
  "data: '{\"text\":\"欢迎参观\",\"voice\":\"female\",\"speed\":1.0}'"

# 指定男声
ros2 topic pub --once /tts/say std_msgs/msg/String \
  "data: '{\"text\":\"欢迎参观\",\"voice\":\"male\",\"speed\":1.0}'"

{"text":"欢迎参观","success":true,"message":""}

{"id":"say-welcome","type":"voice","content":"欢迎参观","voice":"male"}

sudo /opt/bxi/bxi_rc_ros2/scripts/bxi_maintenance.sh enable
sudo /opt/bxi/bxi_rc_ros2/scripts/bxi_maintenance.sh enable --minutes 120

sudo /opt/bxi/bxi_rc_ros2/scripts/bxi_maintenance.sh status

sudo /opt/bxi/bxi_rc_ros2/scripts/bxi_maintenance.sh disable

BXIM1.<base64url(JSON，无 padding)>

{
  "version": 1,
  "credential_id": "32位小写十六进制字符串",
  "user_id": 2147483645,
  "sn": "BXI-EXAMPLE-0001",
  "key_hex": "编码32字节key的64位十六进制字符串",
  "created_at": 1780000000,
  "expires_at": 1780007200
}

user_id = 2147483645
sn      = token.sn
key     = hexDecode(token.key_hex)

{
  "maintenance": true,
  "user_id": "2147483645",
  "sn": "BXI-EXAMPLE-0001",
  "ts": "1780000000000",
  "nonce": "每次新的随机值",
  "sig": "Base64 HMAC-SHA256 签名"
}

POST /api/app/binding/credential/issue
Authorization: Bearer <app-access-token>
Content-Type: application/json

{"sn":"<BLE HELLO_ACK 返回的 SN>"}

{
  "token": "<payload_b64>.<sig_b64>",
  "credentialId": "<payload.cid>",
  "ownerKeyHex": "<64 位小写 hex>",
  "ownerUid": 42,
  "fingerprint": "<16 位小写 hex>",
  "issuedAt": 1780000000,
  "expiresAt": 1780003600,
  "ttlSeconds": 3600
}

import base64
import hashlib
import hmac
import json
import secrets
import time


def issue_binding_credential(secret: str, uid: int, sn: str,
                             ttl_seconds: int = 3600) -> dict:
    if len(secret.encode("utf-8")) < 32:
        raise ValueError("binding secret must be at least 32 UTF-8 bytes")
    if type(uid) is not int or not 1 <= uid <= 2**63 - 1:
        raise ValueError("uid must be in signed 64-bit positive range")
    if not sn or ttl_seconds <= 120:
        raise ValueError("invalid sn or ttl_seconds")
    now = int(time.time())
    owner_key_hex = secrets.token_hex(32)
    payload = {
        "v": 3, "uid": uid, "sn": sn,
        "cid": secrets.token_urlsafe(24),
        "fp": hashlib.sha256(owner_key_hex.encode()).digest()[:8].hex(),
        "n": secrets.token_urlsafe(24),
        "iat": now, "exp": now + ttl_seconds,
    }
    payload_b64 = base64.b64encode(
        json.dumps(payload, separators=(",", ":")).encode()
    ).decode("ascii")
    sig_b64 = base64.b64encode(hmac.new(
        secret.encode(), payload_b64.encode(), hashlib.sha256
    ).digest()).decode("ascii")
    return {
        "token": f"{payload_b64}.{sig_b64}",
        "credentialId": payload["cid"], "ownerKeyHex": owner_key_hex,
        "ownerUid": uid, "fingerprint": payload["fp"],
        "issuedAt": payload["iat"], "expiresAt": payload["exp"],
        "ttlSeconds": ttl_seconds,
    }

# device_key_from_password
import hashlib


def device_key_from_password(password: str, salt_hex: str,
                             iterations: int = 200_000) -> bytes:
    raw = password.encode("utf-8")
    if not 8 <= len(raw) <= 64 or any(b < 0x20 or b > 0x7E for b in raw):
        raise ValueError("password must be 8..64 printable ASCII bytes")
    salt = bytes.fromhex(salt_hex)
    if len(salt) != 16 or iterations != 200_000:
        raise ValueError("invalid local KDF parameters")
    return hashlib.pbkdf2_hmac("sha256", raw, salt, iterations, 32)

# wifi_pairing.py
import base64
import hashlib
import hmac
import json
import secrets
from urllib.parse import urlencode
from urllib.request import Request, urlopen

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric.x25519 import (
    X25519PrivateKey, X25519PublicKey,
)
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF

PATH_SESSION = "/api/v1/pairing/session"
PATH_BIND = "/api/v1/pairing/bind"
PATH_UNBIND = "/api/v1/pairing/binding"
PAIRING_HOST = "192.168.88.162"


def _b64(raw: bytes) -> str:
    return base64.b64encode(raw).decode("ascii")


def _unb64(value: str, size: int | None = None) -> bytes:
    raw = base64.b64decode(value, validate=True)
    if size is not None and len(raw) != size:
        raise ValueError("invalid byte length")
    return raw


def _aad(method: str, path: str, session_id: str) -> bytes:
    return f"bxi.pairing.v1|{method}|{path}|{session_id}".encode("ascii")


def _json(value: dict) -> bytes:
    return json.dumps(value, ensure_ascii=False,
                      separators=(",", ":"), sort_keys=True).encode()


def post(path: str, body: dict) -> dict:
    request = Request(f"http://{PAIRING_HOST}:8082{path}", data=_json(body),
                      headers={"Content-Type": "application/json"},
                      method="POST")
    with urlopen(request, timeout=10) as response:
        return json.loads(response.read())


def open_session(mode: str, device_key: bytes | None = None,
                 token: str | None = None,
                 recover: bool = False):
    private = X25519PrivateKey.generate()
    public = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
    client_nonce = secrets.token_bytes(16)
    request = {
        "protocol": 1, "mode": mode,
        "client_public_key": _b64(public),
        "client_nonce": _b64(client_nonce),
    }
    if recover:
        request["recover"] = True
    if token:
        request["capability_payload"] = token.split(".", 1)[0]
    if device_key is not None:
        request["auth"] = hmac.new(
            device_key,
            b"wifi-session-v1|" + mode.encode() + b"|" + public + b"|" + client_nonce,
            hashlib.sha256,
        ).hexdigest()
    response = post(PATH_SESSION, request)
    robot_public = _unb64(response["robot_public_key"], 32)
    robot_nonce = _unb64(response["robot_nonce"], 16)
    if device_key is not None:
        psk = device_key
    elif token:
        psk = base64.b64decode(token.rsplit(".", 1)[-1])
    else:
        psk = b""
    key = HKDF(algorithm=hashes.SHA256(), length=32,
               salt=client_nonce + robot_nonce,
               info=b"bxi-wifi-pairing-v1|" +
               bytes.fromhex(response["session_id"])).derive(
                   private.exchange(X25519PublicKey.from_public_bytes(robot_public)) + psk
               )
    status = AESGCM(key).decrypt(
        _unb64(response["nonce"], 12), _unb64(response["ciphertext"]),
        _aad("POST", PATH_SESSION, response["session_id"]),
    )
    return key, response["session_id"], json.loads(status)


def seal(key: bytes, method: str, path: str, session_id: str,
         payload: dict) -> dict:
    nonce = secrets.token_bytes(12)
    return {
        "session_id": session_id, "nonce": _b64(nonce),
        "ciphertext": _b64(AESGCM(key).encrypt(
            nonce, _json(payload), _aad(method, path, session_id)))
    }


def unseal(key: bytes, method: str, path: str, session_id: str,
           response: dict) -> dict:
    return json.loads(AESGCM(key).decrypt(
        _unb64(response["nonce"], 12), _unb64(response["ciphertext"]),
        _aad(method, path, session_id)).decode())

import secrets

SN = "BXI-EXAMPLE-0001"  # 必须来自 BLE HELLO_ACK，不要信任广播名
PASSWORD = "my-robot-password"
SALT = secrets.token_bytes(16)
CREDENTIAL_ID = secrets.token_urlsafe(24)

# 1. 本地密码 -> 32B owner key
owner_key = device_key_from_password(PASSWORD, SALT.hex())

# 2. 创建未绑定 local session；status 应为 bound=false
session_key, session_id, status = open_session(mode="local")
assert status == {
    "bound": False, "sn": SN, "binding_mode": "unbound",
}, status

# 3. 将 local claim 放进 AES-GCM 信封并提交
request = seal(session_key, "POST", PATH_BIND, session_id, {
    "sn": SN,
    "owner_uid": 2147483646,
    "credential_id": CREDENTIAL_ID,
    "owner_key": owner_key.hex(),
    "kdf_salt": SALT.hex(),
    "kdf_iterations": 200000,
    "kdf_id": 1,
})
response = post(PATH_BIND, request)
result = unseal(session_key, "POST", PATH_BIND, session_id, response)
assert result["status"] == "bound"
assert result["sn"] == SN

# 4. 只有校验 result 后，才把 owner_key 写入系统安全存储
print(result["credential_id"], result["fingerprint"])

# ws_control.py
import asyncio
import base64
import hashlib
import hmac
import json
import secrets
import time
from urllib.parse import urlencode
import urllib.request

import websockets

HOST = "192.168.88.162"
UID = "2147483646"
SN = "BXI-EXAMPLE-0001"
DEVICE_KEY_HEX = "11" * 32
DEVICE_KEY = bytes.fromhex(DEVICE_KEY_HEX)


def _hmac_b64(message: str) -> str:
    return base64.b64encode(
        hmac.new(DEVICE_KEY, message.encode(), hashlib.sha256).digest()
    ).decode("ascii")


def ws_url(path: str = "/", client_id: str = "py_demo") -> str:
    ts = str(int(time.time() * 1000))
    nonce = secrets.token_urlsafe(18)
    query = {
        "user_id": UID, "client_id": client_id, "sn": SN,
        "ts": ts, "nonce": nonce,
        "sig": _hmac_b64(f"ws|{UID}|{SN}|{ts}|{nonce}"),
    }
    return f"ws://{HOST}:8081{path}?{urlencode(query)}"


def http_json(method: str, path: str, payload):
    method = method.upper()
    body = (b"" if payload is None else
            json.dumps(payload, ensure_ascii=False,
                       separators=(",", ":")).encode())
    ts = str(int(time.time() * 1000))
    nonce = secrets.token_urlsafe(18)
    body_hash = hashlib.sha256(body).hexdigest()
    message = f"http.v2|{method}|{path}|{body_hash}|{UID}|{SN}|{ts}|{nonce}"
    query = {
        "auth_v": "2", "user_id": UID, "sn": SN, "ts": ts,
        "nonce": nonce, "sig": _hmac_b64(message),
    }
    request = urllib.request.Request(
        f"http://{HOST}:8082{path}?{urlencode(query)}",
        data=body if payload is not None else None,
        headers={"Content-Type": "application/json"}, method=method,
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        return json.loads(response.read())


async def control() -> None:
    async with websockets.connect(ws_url()) as ws:
        await ws.send(json.dumps({
            "type": "control.cmd_vel",
            "ts": int(time.time() * 1000), "seq": 1,
            "payload": {
                "vx": 0.2, "vy": 0.0, "wz": 0.1,
                "height": 1.0, "mode": "manual",
                "btn_1": 0, "btn_5": 1,
            },
        }))
        await ws.send(json.dumps({
            "type": "control.heartbeat",
            "ts": int(time.time() * 1000), "seq": 2, "payload": {},
        }))
        print(await ws.recv())


if __name__ == "__main__":
    asyncio.run(control())
    print(http_json("GET", "/api/v1/maps", None))

maps = http_json("GET", "/api/v1/maps", None)
http_json("POST", "/api/v1/maps/demo_map/activate", None)

# 继续消费控制 WS，不要固定 sleep 20 秒：
# 1. 等 nav.runtime.status.payload.active_map_id == "demo_map"
# 2. 等 current_mode in {"localizing", "navigation"}
# 3. 确认 driver_healthy == true；last_error 为空

http_json("POST", "/api/v1/nav/initial_pose",
          {"x": 0.0, "y": 0.0, "yaw": 0.0, "frame_id": "map"})

# 等 nav.reloc_required.payload.required == false，且最新
# nav.runtime.status.payload.localized == true 后再发送目标。
http_json("POST", "/api/v1/nav/goal",
          {"x": 2.0, "y": 1.0, "yaw": 0.0, "frame_id": "map"})