#!/usr/bin/env python3 """Fail-closed automation contract around upstream's interactive RNode tool. Never infer success from prose or graceful_exit(0). Verification runs in a fresh process and inspects the EEPROM signature and firmware hashes actually read by upstream. Legacy human-facing arguments remain available. """ import contextlib import json import os import re import stat import subprocess import sys OWNED_PATH = None def serial_target(path): if OWNED_PATH is not None: if path != OWNED_PATH: raise ValueError("Serial target differs from appointment") return path target = os.path.realpath(path) if not re.fullmatch(r"/dev/tty(?:USB|ACM)[0-9]+", target): raise ValueError("Expected a USB serial tty") if not stat.S_ISCHR(os.stat(target).st_mode): raise ValueError("Serial target is not a character device") return target def verified_state(node, board): if node is None or node.detected is not True or node.provisioned is not True: raise ValueError("RNode did not respond or EEPROM is not provisioned; reset/reconnect and retry") if node.signature_valid is not True: raise ValueError("RNode EEPROM signature is not valid") if node.product != {"heltecv3": 0xC1, "heltecv4": 0xC3}[board]: raise ValueError("Provisioned board does not match selected model") actual, target = node.firmware_hash, node.firmware_hash_target if not isinstance(actual, bytes) or len(actual) != 32 or not any(actual): raise ValueError("RNode did not report a valid firmware hash") if actual != target: raise ValueError("Running firmware does not match provisioned firmware hash") if not isinstance(node.version, str) or not re.fullmatch(r"[0-9]+\.[0-9]+", node.version): raise ValueError("Missing firmware version") return {"firmware": node.version, "sha256": actual.hex(), "signature_valid": True} def invoke(upstream, args): sys.argv = ["archy-rnodeconf", *args] original = subprocess.call def call(argv, *pos, **kw): if OWNED_PATH is not None: kw['pass_fds'] = (3, 4, 8) return original(packaged_flasher(argv), *pos, **kw) subprocess.call = call try: upstream.main() return 0 except SystemExit as exc: return 0 if exc.code is None else exc.code finally: subprocess.call = original def packaged_flasher(args): """Replace upstream's downloaded Python runner with our frozen flasher. Keep upstream's board-specific offsets and reset flags. Never execute the downloaded esptool.py or an interpreter selected from HOME/PATH/environment. The fresh child remains inside the runtime owner's existing job cgroup. """ if (isinstance(args, (list, tuple)) and len(args) > 2 and args[0] == sys.executable and isinstance(args[1], str) and os.path.basename(args[1]) == "esptool.py"): prefix = [sys.executable] if not getattr(sys, "frozen", False): prefix.append(os.path.abspath(__file__)) mode = ["--archy-owned-esptool", OWNED_PATH] if OWNED_PATH is not None else ["--archy-esptool"] return [*prefix, *mode, *esptool_v5_args(args[2:])] return args def esptool_v5_args(args): """Translate RNS 1.3.5's CLI spellings without changing operands/offsets.""" options = {"--flash_size": "--flash-size", "--flash_mode": "--flash-mode", "--flash_freq": "--flash-freq"} global_values = {"--chip", "--port", "--baud", "--before", "--after", "--connect-attempts", "--port-filter"} commands = {"write_flash": "write-flash", "read_flash": "read-flash", "erase_flash": "erase-flash", "flash_id": "flash-id", "read_mac": "read-mac", "chip_id": "chip-id"} result, command_seen, value_for = [], False, None for arg in args: if value_for is not None: result.append(arg.replace("_", "-") if value_for in ("--before", "--after") else arg) value_for = None elif not command_seen and arg in global_values: result.append(arg) value_for = arg elif not command_seen and not arg.startswith("-"): result.append(commands.get(arg, arg)) command_seen = True else: result.append(options.get(arg, arg)) return result def install(upstream, path): # Upstream imports subprocess.call locally. Observe the actual writer's # return, not a stale success line from its initial device inspection. original = subprocess.call writes = [] def call(args, *pos, **kw): result = original(args, *pos, **kw) if isinstance(args, (list, tuple)) and any(a in ("write_flash", "write-flash") for a in args): writes.append(result) return result subprocess.call = call try: code = invoke(upstream, ["--autoinstall", path]) finally: subprocess.call = original if code != 0 or not writes or any(result != 0 for result in writes): raise ValueError("RNode installation did not complete a successful firmware write") def main(): global OWNED_PATH args = sys.argv[1:] owned_keys = False if args[:1] in (["--archy-owned"], ["--archy-owned-esptool"]): from archy_owned_serial import install as adopt_serial if args[0] == "--archy-owned": if len(args) != 5 or args[1] not in ("install", "verify", "capabilities"): raise ValueError("Invalid owned radio invocation") _, action, path, board, nonce = args owned_keys = action != 'capabilities' OWNED_PATH = path adopt_serial(path) args = (["--archy-install", path] if action == "install" else ["--archy-verify", path, board, nonce] if action == "verify" else ["--archy-capabilities"]) else: OWNED_PATH = args[1] adopt_serial(OWNED_PATH) args = ["--archy-esptool", *args[2:]] if args[:1] == ["--archy-esptool"]: from archy_esptool import run_esptool run_esptool(args[1:]) return if args == ["--archy-capabilities"]: # Exercise packaged resources as well as the JSON dispatch. A binary # missing its ESP32-S3 stub must fail before listener shutdown. from archy_esptool import check_stub, ESPTOOL_VERSION check_stub() if OWNED_PATH is not None: # Qualify the real frozen descendant and its inherited TTY/HOME/ # extraction FDs without sending any bytes or modem controls. output = subprocess.run(packaged_flasher([sys.executable, 'esptool.py', 'version']), pass_fds=(3, 4, 8), capture_output=True, timeout=20, check=True) if ESPTOOL_VERSION.encode() not in output.stdout: raise ValueError('Owned packaged flasher version differs') print('{"contract":1}') return from RNS.Utilities import rnodeconf as upstream if owned_keys: from archy_owned_keys import install as bind_keys bind_keys(upstream) if len(args) == 2 and args[0] == "--archy-install": install(upstream, serial_target(args[1])) elif len(args) == 4 and args[0] == "--archy-verify": _, path, board, nonce = args path = serial_target(path) if board not in ("heltecv3", "heltecv4") or not re.fullmatch(r"[a-f0-9]{32}", nonce): raise ValueError("Invalid verification request") # Only the contract goes to stdout. Upstream output cannot manufacture # evidence, and this process has no previous install/probe state. nodes = [] original = upstream.RNode def new_node(*pos, **kw): node = original(*pos, **kw) nodes.append(node) return node upstream.RNode = new_node with contextlib.redirect_stdout(sys.stderr): code = invoke(upstream, ["--info", path]) if code != 0: raise ValueError("RNode verification tool failed") result = verified_state(nodes[-1] if nodes else None, board) print(json.dumps({"contract": 1, "nonce": nonce, "path": path, "board": board, **result})) else: raise SystemExit(invoke(upstream, args)) if __name__ == "__main__": try: main() except Exception as exc: print(f"RNode operation failed: {exc}", file=sys.stderr) sys.exit(1)