canbus: confirm command path live + frame docs as device integration

Command path proven end to end on the bus (node F8 interior lights, on/off/on),
each answering a distinct fresh challenge; bare opcodes without the exchange are
ignored. ids_can_auth.h verified bit-exact against ids_can_auth.py and the
captured/live pairs.

- idscan_cmd.py: stdlib socketcan tool running the full page-42/43 exchange
- esphome/onecontrol-canbus.yaml: correct IDS-CAN read dispatch (was stale RV-C
  DGN code) + command path wired to the auth header
- README/memory: document the read map + command authentication; rename
  sniff/ -> captures/; neutral device-integration framing throughout

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wes
2026-06-12 11:20:12 -04:00
co-authored by Claude Opus 4.8
parent 840cfaf5fc
commit 742ef49c8a
14 changed files with 470 additions and 245 deletions
+105
View File
@@ -0,0 +1,105 @@
#!/usr/bin/env python3
"""Operate an IDS-CAN switched load over socketcan, with the auth exchange.
Acts as controller node 01: request a challenge -> read the module's fresh
challenge -> return remote_control_response(challenge) -> send the opcode x3.
Pure stdlib (raw AF_CAN socket); needs can0 already up (slcand).
Usage: idscan_cmd.py <node_hex> <on|off> e.g. idscan_cmd.py F8 on
"""
import socket
import struct
import sys
import time
from ids_can_auth import remote_control_response
CAN_EFF_FLAG = 0x80000000
FMT = "=IB3x8s" # can_id, can_dlc, pad/res, data[8]
ARM_PAGE, RESP_PAGE, CHAL_PAGE = 0x42, 0x43, 0x42
def eff(i):
return i | CAN_EFF_FLAG
def pack(can_id, data=b""):
return struct.pack(FMT, eff(can_id), len(data), data.ljust(8, b"\x00"))
def unpack(frame):
cid, dlc, data = struct.unpack(FMT, frame)
is_eff = bool(cid & CAN_EFF_FLAG)
return is_eff, cid & 0x1FFFFFFF, dlc, data[:dlc]
def status_id(node): # 11-bit page-3 live-value broadcast
return (3 << 8) | node
def read_status(s, node, timeout=2.0):
want = status_id(node)
end = time.time() + timeout
while time.time() < end:
s.settimeout(max(0.01, end - time.time()))
try:
is_eff, cid, dlc, data = unpack(s.recv(16))
except socket.timeout:
break
if not is_eff and cid == want:
return data
return None
def actuate(node, op):
arm_id = (0x01 << 18) | (0 << 16) | (node << 8) | ARM_PAGE # 01->node p42
resp_id = (0x01 << 18) | (0 << 16) | (node << 8) | RESP_PAGE # 01->node p43
chal_id = (node << 18) | (1 << 16) | (0x01 << 8) | CHAL_PAGE # node->01 p42
opcode_id = (0x0006 << 16) | (node << 8) | op # 0006<node><op>
s = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
s.bind(("can0",))
before = read_status(s, node, 1.5)
print(f" before: page3 = {before.hex(' ') if before else '(none)'}")
s.send(pack(arm_id, b"\x00\x04")) # arm
# await the module's fresh challenge
chal = None
s.settimeout(0.3)
end = time.time() + 0.3
while time.time() < end:
try:
is_eff, cid, dlc, data = unpack(s.recv(16))
except socket.timeout:
break
if is_eff and cid == chal_id and dlc == 6 and data[:2] == b"\x00\x04":
chal = data[2:6]
break
if chal is None:
print(" !! no challenge from module — aborting (load not actuated)")
return False
ch_int = int.from_bytes(chal, "big")
rr = remote_control_response(ch_int)
print(f" challenge {ch_int:08X} -> response {rr:08X}")
s.send(pack(resp_id, b"\x00\x04" + rr.to_bytes(4, "big"))) # response
time.sleep(0.004)
for _ in range(3): # opcode x3
s.send(pack(opcode_id))
time.sleep(0.006)
time.sleep(0.3)
after = read_status(s, node, 1.5)
print(f" after: page3 = {after.hex(' ') if after else '(none)'}")
s.close()
return True
if __name__ == "__main__":
if len(sys.argv) != 3 or sys.argv[2] not in ("on", "off"):
sys.exit(__doc__)
node = int(sys.argv[1], 16)
op = 0x01 if sys.argv[2] == "on" else 0x00
print(f"node {node:02X} -> {sys.argv[2]}")
actuate(node, op)