archy/reticulum-daemon/archy_rnodeconf.py

#!/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, *args[2:]]
    return args


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"]:
        import esptool
        sys.argv = ["archy-esptool", *args[1:]]
        esptool._main()
        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 esptool.loader import StubFlasher
        stub = StubFlasher("ESP32-S3")
        if not stub.text or not stub.text_start or not stub.entry:
            raise ValueError("Packaged ESP32-S3 flashing stub is incomplete")
        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 b'4.8.1' 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)