wesandClaude Opus 4.8
742ef49c8a
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>
106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
#!/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)
|