204 lines
8.4 KiB
Python
204 lines
8.4 KiB
Python
#!/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)
|