Require on-device confirmation for host XPUB/fingerprint requests

[schnuartz-ai]

Summary

• XpubApp.process_host_command() previously returned the master fingerprint or an XPUB at any host-selected derivation path immediately once the device was unlocked, with no trusted-screen confirmation. A malicious or compromised host could silently enumerate the wallet fingerprint and XPUBs for address-monitoring purposes.
• fingerprint stays non-interactive — it's used by companion software (Specter Desktop, Liana) for device discovery/identification and must respond immediately. Still gated by the unchanged locked-device check.
• xpub <path> now requires an on-device Confirm/Cancel before responding, showing the normalized derivation path and the derived XPUB. Reject returns no key material via the existing error: User cancelled protocol — no wire-protocol changes.
• hwidevice.py (this repo's bundled reference HWI client) no longer applies a 3-second response timeout to xpub requests, since a human now has to physically confirm on-device and that can legitimately take longer. get_master_fingerprint keeps its short timeout.

Companion-repo compatibility

The same 3-second-timeout incompatibility exists in Specter Desktop's independent specter_diy.py HWI client, and a related cancellation-mapping bug exists in async-hwi (used by Liana). Both have their own focused fixes in flight:

• Specter Desktop: removes the timeout from get_pubkey_at_path, adds regression tests.
• async-hwi: maps a cancelled xpub request to SpecterError::UserCancelled/HWIError::UserRefused, matching how sign/add_wallet already handle the same cancellation string.
• Liana: no change needed — it never imposes its own timeout on the xpub fetch and only consumes async-hwi's already-parsed error type.

Test plan

• test/run_native_tests.py (CPython + pip embit) — 15/15 pass, including test/tests_native/test_xpubs_host_command.py: fingerprint returns immediately without invoking show_screen; xpub still requires confirmation with the actual requested path and derived xpub shown; reject returns no key material; locked-device check still fires before either branch; malformed/empty paths fail before confirmation.
• Built the real MicroPython Unix simulator (make unix) and ran the frozen-firmware test suite under it (make test → test/run_tests.py on bin/micropython_unix) — 14/14 pass.
• Ran the TCP-driven integration suite (test/integration) against the built simulator end-to-end, driving the real GUI event loop — test_get_xpub (fingerprint non-interactive, xpub confirmed) and test_get_xpub_rejected (xpub rejection) — 4/4 pass.
• Verified on real STM32F469 Discovery hardware over USB-serial (via the CI-built firmware and a small test-host script): fingerprint responds immediately with no on-device screen; xpub shows the confirmation screen with the correct path/value; Cancel returns nothing; Confirm returns the expected value.

[schnuartz-ai]

Manual test scripts for reviewers

Two small standalone helper scripts to exercise this change against real hardware over USB-serial, exactly like the wire protocol in src/hosts/usb.py (write a line, get ACK\r\n, then block until the on-device Confirm/Cancel resolves, then read the final response). Not part of the shippable diff — no pyserial dependency is added to the repo, these are just for manually confirming the behavior on a device. Requires pip install pyserial.

How to test

1. Flash this branch's firmware onto a device (or the CI-built artifact for this PR).
2. On the device: Device settings > Communication > USB communication → enable, confirm, reboot if prompted.
3. Connect via the device's USB-serial port (not the ST-Link/debug port, if your board exposes both) and find it with --list.
4. Run one of the scripts below and use the interactive menu (fingerprint, xpub <path>, or the built-in sweep of valid + malformed paths) to confirm:
   • a locked device rejects both commands outright, no prompt shown,
   • an unlocked device shows a Confirm/Cancel screen naming the actual fingerprint, or the actual derivation path and the derived xpub value,
   • Cancel returns nothing (error: User cancelled), Confirm returns the expected value,
   • malformed paths (empty, garbage, wrong format) fail immediately with no prompt at all.

host_probe.py — single device

python3 host_probe.py --list
python3 host_probe.py --port /dev/ttyACM0        # or COMx on Windows
python3 host_probe.py --port /dev/ttyACM0 --fingerprint
python3 host_probe.py --port /dev/ttyACM0 --xpub "m/84h/0h/0h"
python3 host_probe.py --port /dev/ttyACM0 --sweep

host_probe.py

#!/usr/bin/env python3
"""
host_probe.py - a tiny "malicious/testing host" for Specter DIY

Talks to the device exactly like a real host application would over the
USB-serial (VCP) connection, so you can verify that:
  - requesting the fingerprint or an xpub now shows a confirmation screen
    on the device with the actual request on it,
  - Cancel on the device returns nothing to this script,
  - Confirm returns the expected value,
  - a locked device rejects the request outright (no screen shown).

This mirrors the wire protocol implemented in src/hosts/usb.py:
  - you write a line ending in \\r, \\n or \\r\\n,
  - the device immediately answers "ACK\\r\\n",
  - once you resolve the on-device prompt (or on error) it writes the
    final response, terminated by "\\r\\n".

Requires: pip install pyserial

Usage examples:
  python3 host_probe.py --list                     # show candidate ports
  python3 host_probe.py --port /dev/ttyACM0         # interactive menu
  python3 host_probe.py --port /dev/ttyACM0 --fingerprint
  python3 host_probe.py --port /dev/ttyACM0 --xpub "m/84h/0h/0h"
  python3 host_probe.py --port /dev/ttyACM0 --sweep  # try several paths in a row
"""
import argparse
import sys
import time

try:
    import serial
    from serial.tools import list_ports
except ImportError:
    print("This script needs pyserial: pip install pyserial", file=sys.stderr)
    sys.exit(1)

BAUD = 115200
ACK = b"ACK\r\n"

# A few paths worth trying: standard single-key/multisig, an unusual one,
# and some malformed ones the device must reject *before* showing a prompt.
SWEEP_PATHS = [
    "m/84h/0h/0h",
    "m/48h/0h/0h/2h",
    "m",
    "m/44h/1h/0h",
]
SWEEP_BAD_PATHS = [
    "",
    "garbage",
    "m/84x/0h",
]


def find_ports():
    return list(list_ports.comports())


def open_port(path):
    ser = serial.Serial(path, BAUD, timeout=None)
    # give the device a moment after opening the port
    time.sleep(0.3)
    ser.reset_input_buffer()
    return ser


def send_command(ser, data, prompt_timeout=120):
    """
    Sends one command line and returns the device's final response.
    Blocks (up to prompt_timeout seconds) after the ACK, since that's the
    window where a confirmation screen is waiting for you to physically
    press Cancel/Confirm on the device.
    """
    if isinstance(data, str):
        data = data.encode()
    if not data.endswith(b"\n") and not data.endswith(b"\r"):
        data += b"\r\n"

    ser.write(data)
    ser.flush()

    ack = ser.read(len(ACK))
    if ack != ACK:
        return None, "no ACK received (got %r) - is the device unlocked and USB enabled?" % ack

    print("  -> device ACKed. If a confirmation screen appeared, "
          "go press Cancel or Confirm now (waiting up to %ds)..." % prompt_timeout)

    ser.timeout = prompt_timeout
    line = ser.readline()
    ser.timeout = None
    if not line:
        return None, "timed out waiting for a response - did you leave the prompt unanswered?"
    return line.strip(), None


def do_one(ser, command, label):
    print("\n[%s] sending: %r" % (label, command))
    res, err = send_command(ser, command)
    if err:
        print("  ERROR: %s" % err)
        return
    print("  <- response: %r" % res)
    if res == b"error: User cancelled":
        print("  (rejected on-device: no data was returned - correct fail-closed behavior)")


def interactive(ser):
    while True:
        print("\nWhat do you want to send?")
        print("  1) fingerprint")
        print("  2) xpub m/84h/0h/0h")
        print("  3) xpub <custom path>")
        print("  4) sweep: several valid + malformed xpub requests")
        print("  q) quit")
        choice = input("> ").strip().lower()
        if choice == "1":
            do_one(ser, b"fingerprint", "fingerprint")
        elif choice == "2":
            do_one(ser, b"xpub m/84h/0h/0h", "xpub")
        elif choice == "3":
            path = input("derivation path (e.g. m/48h/0h/7h/2h): ").strip()
            do_one(ser, ("xpub %s" % path).encode(), "xpub custom")
        elif choice == "4":
            sweep(ser)
        elif choice == "q":
            break
        else:
            print("unknown option")


def sweep(ser):
    print("\n--- valid paths: confirm on the device each time and check the ---")
    print("--- path shown on screen matches what's printed here          ---")
    for path in SWEEP_PATHS:
        do_one(ser, ("xpub %s" % path).encode(), "xpub %s" % path)

    print("\n--- malformed paths: these must fail with an error and NEVER ---")
    print("--- show a confirmation screen at all                         ---")
    for path in SWEEP_BAD_PATHS:
        do_one(ser, ("xpub %s" % path).encode(), "xpub %r (malformed)" % path)


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--list", action="store_true", help="list candidate serial ports and exit")
    ap.add_argument("--port", help="serial port, e.g. /dev/ttyACM0, /dev/tty.usbmodemXXXX, COM5")
    ap.add_argument("--fingerprint", action="store_true", help="send a single fingerprint request and exit")
    ap.add_argument("--xpub", metavar="PATH", help="send a single xpub request for PATH and exit")
    ap.add_argument("--sweep", action="store_true", help="run the built-in sweep of valid/malformed paths and exit")
    args = ap.parse_args()

    if args.list or not args.port:
        ports = find_ports()
        if not ports:
            print("No serial ports found.")
        else:
            print("Candidate ports:")
            for p in ports:
                print("  %s  (%s)" % (p.device, p.description))
        if not args.port:
            print("\nPass one with --port, e.g.: python3 host_probe.py --port %s" %
                  (ports[0].device if ports else "/dev/ttyACM0"))
            return

    ser = open_port(args.port)
    try:
        if args.fingerprint:
            do_one(ser, b"fingerprint", "fingerprint")
        elif args.xpub:
            do_one(ser, ("xpub %s" % args.xpub).encode(), "xpub")
        elif args.sweep:
            sweep(ser)
        else:
            interactive(ser)
    finally:
        ser.close()


if __name__ == "__main__":
    main()

host_probe_dual.py — two devices side by side

Sends the same request to two devices at (almost) the same instant, so you can compare what each screen shows.

python3 host_probe_dual.py --list
python3 host_probe_dual.py --port-a COM3 --port-b COM5

host_probe_dual.py

#!/usr/bin/env python3
"""
host_probe_dual.py - send the SAME host request to TWO Specter DIY devices
at (almost) the same time, so you can compare their confirmation screens
side by side.

Same wire protocol as host_probe.py (see src/hosts/usb.py):
  - write a line ending in \\r, \\n or \\r\\n,
  - device answers "ACK\\r\\n" immediately,
  - then blocks until you Confirm/Cancel on its screen, and writes the
    final response terminated by "\\r\\n".

Requires: pip install pyserial

Usage:
  python3 host_probe_dual.py --list
  python3 host_probe_dual.py --port-a COM3 --port-b COM5
  python3 host_probe_dual.py --port-a COM3 --port-b COM5 --fingerprint
  python3 host_probe_dual.py --port-a COM3 --port-b COM5 --xpub "m/84h/0h/0h"
  python3 host_probe_dual.py --port-a COM3 --port-b COM5 --sweep
"""
import argparse
import sys
import threading
import time

try:
    import serial
    from serial.tools import list_ports
except ImportError:
    print("This script needs pyserial: pip install pyserial", file=sys.stderr)
    sys.exit(1)

BAUD = 115200
ACK = b"ACK\r\n"

SWEEP_PATHS = [
    "m/84h/0h/0h",
    "m/48h/0h/0h/2h",
    "m",
    "m/44h/1h/0h",
]
SWEEP_BAD_PATHS = [
    "",
    "garbage",
    "m/84x/0h",
]


def find_ports():
    return list(list_ports.comports())


def open_port(path):
    ser = serial.Serial(path, BAUD, timeout=None)
    time.sleep(0.3)
    ser.reset_input_buffer()
    return ser


def send_command(ser, data, prompt_timeout=180):
    """
    Sends one command and returns the device's final response.
    Blocks (up to prompt_timeout seconds) after the ACK - that's the
    window where a confirmation screen waits for you to physically
    press Cancel/Confirm on the device.
    """
    if isinstance(data, str):
        data = data.encode()
    if not data.endswith(b"\n") and not data.endswith(b"\r"):
        data += b"\r\n"

    ser.write(data)
    ser.flush()

    ack = ser.read(len(ACK))
    if ack != ACK:
        return None, "no ACK received (got %r) - is the device unlocked and USB enabled?" % ack

    ser.timeout = prompt_timeout
    line = ser.readline()
    ser.timeout = None
    if not line:
        return None, "timed out waiting for a response - did you leave a prompt unanswered?"
    return line.strip(), None


def _worker(label, ser, command, results, barrier):
    # both threads wait here so they fire the command at (almost) the
    # same instant, regardless of small per-thread setup differences
    barrier.wait()
    res, err = send_command(ser, command)
    results[label] = (res, err)


def run_on_both(ser_a, ser_b, command, label_a="A", label_b="B"):
    results = {}
    barrier = threading.Barrier(2)
    ta = threading.Thread(target=_worker, args=(label_a, ser_a, command, results, barrier))
    tb = threading.Thread(target=_worker, args=(label_b, ser_b, command, results, barrier))

    print("\nSending %r to BOTH devices now." % command)
    print("Go check both screens and compare what they show, then confirm/cancel each.")
    ta.start()
    tb.start()
    ta.join()
    tb.join()

    for label in (label_a, label_b):
        res, err = results[label]
        if err:
            print("  [%s] ERROR: %s" % (label, err))
        else:
            print("  [%s] response: %r" % (label, res))

    ra, ea = results[label_a]
    rb, eb = results[label_b]
    if ea is None and eb is None:
        if ra == rb:
            print("  -> identical response on both devices")
        else:
            print("  -> responses differ (expected if the two devices use different seeds)")


def do_one(ser_a, ser_b, command, label):
    print("\n=== %s ===" % label)
    run_on_both(ser_a, ser_b, command)


def interactive(ser_a, ser_b):
    while True:
        print("\nWhat do you want to send to BOTH devices?")
        print("  1) fingerprint")
        print("  2) xpub m/84h/0h/0h")
        print("  3) xpub <custom path>")
        print("  4) sweep: several valid + malformed xpub requests")
        print("  q) quit")
        choice = input("> ").strip().lower()
        if choice == "1":
            do_one(ser_a, ser_b, b"fingerprint", "fingerprint")
        elif choice == "2":
            do_one(ser_a, ser_b, b"xpub m/84h/0h/0h", "xpub")
        elif choice == "3":
            path = input("derivation path (e.g. m/48h/0h/7h/2h): ").strip()
            do_one(ser_a, ser_b, ("xpub %s" % path).encode(), "xpub custom")
        elif choice == "4":
            sweep(ser_a, ser_b)
        elif choice == "q":
            break
        else:
            print("unknown option")


def sweep(ser_a, ser_b):
    print("\n--- valid paths: compare path + xpub shown on both screens ---")
    for path in SWEEP_PATHS:
        do_one(ser_a, ser_b, ("xpub %s" % path).encode(), "xpub %s" % path)

    print("\n--- malformed paths: must fail on both, no screen at all ---")
    for path in SWEEP_BAD_PATHS:
        do_one(ser_a, ser_b, ("xpub %s" % path).encode(), "xpub %r (malformed)" % path)


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--list", action="store_true", help="list candidate serial ports and exit")
    ap.add_argument("--port-a", help="serial port for device A, e.g. COM3")
    ap.add_argument("--port-b", help="serial port for device B, e.g. COM5")
    ap.add_argument("--fingerprint", action="store_true", help="send a single fingerprint request to both and exit")
    ap.add_argument("--xpub", metavar="PATH", help="send a single xpub request for PATH to both and exit")
    ap.add_argument("--sweep", action="store_true", help="run the built-in sweep of valid/malformed paths and exit")
    args = ap.parse_args()

    if args.list or not (args.port_a and args.port_b):
        ports = find_ports()
        if not ports:
            print("No serial ports found.")
        else:
            print("Candidate ports:")
            for p in ports:
                print("  %s  (%s)" % (p.device, p.description))
        if not (args.port_a and args.port_b):
            print("\nPass both with --port-a and --port-b, e.g.:")
            print("  python3 host_probe_dual.py --port-a COM3 --port-b COM5")
            return

    ser_a = open_port(args.port_a)
    ser_b = open_port(args.port_b)
    try:
        if args.fingerprint:
            do_one(ser_a, ser_b, b"fingerprint", "fingerprint")
        elif args.xpub:
            do_one(ser_a, ser_b, ("xpub %s" % args.xpub).encode(), "xpub")
        elif args.sweep:
            sweep(ser_a, ser_b)
        else:
            interactive(ser_a, ser_b)
    finally:
        ser_a.close()
        ser_b.close()


if __name__ == "__main__":
    main()

---

Generated by Claude Code

[Schnuartz]

Tested it on Real Hardware and it works.

Here I used the script: host_probe_dual.py

I had 2 Specter devices both connected via Micro USB Cable to the PC both had the usb-communication enabled (Sending XPubs only works if it is enabled, otherwise I was not capable to do so with normal Firmware and Fork). I used the host_probe_dual.py in the terminal to check what data they are sending.

--
Fingerprint: Here I did one try with cancel and one try with confirm:

XPub
Here I did just the canceling, two Runs:

[Schnuartz]

Tested it again. Now it has some better formating.

[schnuartz-ai]

Pushed an update: added scoped multi-XPUB authorization on top of the individual-confirmation flow already in this PR (this addresses the "one confirmation per XPUB is painful for BIP-138-style recovery" concern raised on the upstream PR at cryptoadvance#381).

New: xpubauth begin <scope> / xpubauth end. A host requests permission once for a bounded, explicit set of paths — exact paths and/or paths with one bounded range component (m/84h/0h/{0-9}h), no wildcards, no open ranges. The device shows the exact network, path ranges and max key count on the trusted display and asks for a single Confirm/Cancel. After approval, ordinary xpub <path> requests inside that scope return without further prompts; anything outside the scope still uses this PR's normal per-request confirmation.

Key security properties (full writeup in docs/communication.md):

• Matching is on normalized BIP32 integers only (both h/' accepted), never raw text.
• Overlapping/duplicate scope entries are rejected outright, so the displayed count is always exact.
• Bound to the exact active Specter network (not just BIP44 coin type — testnet/signet/regtest share 1h); any network change destroys it; a scope can never mix networks.
• Each authorized path is consumable once (compact per-entry bitsets, no stored XPUB lists); the authorization self-clears when exhausted.
• Volatile, RAM-only: cleared on lock, on key/seed reload, on explicit end, on reboot. A second begin while one is active replaces it only after a fresh confirmation, never merges.
• Old hosts are unaffected (they never send xpubauth); new hosts against old firmware get an error back from xpubauth begin and fall back to per-path requests.

Also added authorize_xpubs() / end_xpub_authorization() helpers to the HWI client in hwidevice.py for companion software.

New testing scripts added in this update:

• test/tests_native/test_xpub_scope.py — pure grammar/parsing/overlap/network-binding/budget/matching tests for the new apps/xpubs/scope.py module (45 tests).
• test/tests_native/test_xpubauth_host_command.py — host-command integration tests: begin/end lifecycle, in-scope vs out-of-scope behavior, hardened-notation equivalence, budget consumption, lock/network invalidation, replace-on-second-begin, cancellation, and a full BIP-138-style walkthrough (29 tests).
• test/integration/tests/test_xpubauth.py — TCP integration tests run against the real MicroPython Unix simulator binary (10 tests): begin→bulk-lookup→early-stop→end→re-confirm, exact-four-path scope, repeated-request budget behavior, cancelled begin, invalid scope, and the BIP-138-style discovery flow.

Tests run this session:

• Native (CPython harness, test/run_native_tests.py): 89/89 passed (includes the 74 new scope/xpubauth tests above, plus all pre-existing tests, unmodified)
• MicroPython Unix (make unix && make test, built fresh incl. mpy-cross/SDL2): 14/14 passed
• TCP integration suite against the real simulator: 14/14 passed (existing test_basic.py + new test_xpubauth.py)
• test/integration/tests/test_with_rpc.py needs a running bitcoind, unavailable in this sandbox — pre-existing environment limitation unrelated to this change; the rest of the integration suite was run directly against the simulator to route around it.
• Real STM32 hardware was not available for this change, so it hasn't been physically validated on a device — everything above ran against the actual MicroPython Unix build and the real TCP simulator, not just CPython mocks.

---

Generated by Claude Code

[schnuartz-ai]

Per feedback, the new test scripts have been dropped from the committed tree (see the follow-up commit) and are posted here instead. They're not part of the PR diff - paste them locally under the paths noted if you want to run them.

All three passed against this branch before removal:

• native (CPython harness): scope 45/45, host-command integration 29/29
• TCP integration (real MicroPython simulator): 10/10

test/tests_native/test_xpub_scope.py (45 tests - pure grammar/parsing/overlap/network-binding/budget/matching)

import sys

if sys.implementation.name != 'micropython':
    from native_support import setup_native_stubs

    setup_native_stubs()

from unittest import TestCase
from embit import bip32

from apps.xpubs import scope as xpubauth_scope


MAIN = ("main", 0)
TEST = ("test", 1)


class ScopeParsingTest(TestCase):
    def test_single_exact_path(self):
        entries, total = xpubauth_scope.parse_scope("m/84h/0h/0h", *MAIN)
        self.assertEqual(total, 1)
        self.assertEqual(len(entries), 1)
        self.assertEqual(entries[0].format(), "m/84h/0h/0h")

    def test_four_exact_paths(self):
        entries, total = xpubauth_scope.parse_scope(
            "m/84h/0h/0h;m/86h/0h/0h;m/48h/0h/0h/2h;m/48h/0h/1h/2h", *MAIN
        )
        self.assertEqual(total, 4)
        self.assertEqual(len(entries), 4)

    def test_bounded_range_count(self):
        entries, total = xpubauth_scope.parse_scope("m/84h/0h/{0-9}h", *MAIN)
        self.assertEqual(total, 10)

    def test_multiple_ranges_count(self):
        entries, total = xpubauth_scope.parse_scope(
            "m/84h/0h/{0-9}h;m/86h/0h/{0-2}h;m/48h/0h/{0-4}h/2h", *MAIN
        )
        self.assertEqual(total, 10 + 3 + 5)

    def test_bip138_like_scope_count(self):
        # roughly the BIP-0138 common discovery space for one network
        scope = ";".join([
            "m/44h/0h/{0-9}h",
            "m/49h/0h/{0-9}h",
            "m/84h/0h/{0-9}h",
            "m/86h/0h/{0-9}h",
            "m/87h/0h/{0-9}h",
            "m/48h/0h/{0-9}h/1h",
            "m/48h/0h/{0-9}h/2h",
        ])
        entries, total = xpubauth_scope.parse_scope(scope, *MAIN)
        self.assertEqual(total, 70)
        self.assertLessEqual(total, xpubauth_scope.MAX_TOTAL_XPUBS)

    def test_hardened_notation_is_equivalent(self):
        a, _ = xpubauth_scope.parse_scope("m/84h/0h/{0-9}h", *MAIN)
        b, _ = xpubauth_scope.parse_scope("m/84'/0'/{0-9}'", *MAIN)
        self.assertEqual(a[0].components, b[0].components)
        self.assertEqual(a[0].format(), b[0].format())

    def test_zero_width_range_is_valid(self):
        entries, total = xpubauth_scope.parse_scope("m/84h/0h/{0-0}h", *MAIN)
        self.assertEqual(total, 1)

    def test_reversed_range_is_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/{9-0}h", *MAIN)

    def test_negative_range_is_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/{-1-5}h", *MAIN)

    def test_value_at_hardened_boundary_is_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/2147483648", *MAIN)

    def test_huge_decimal_is_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/123456789012h", *MAIN)

    def test_range_too_wide_is_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/{0-500}h", *MAIN)

    def test_huge_range_is_rejected_before_expansion(self):
        # the canonical "don't even try to expand this" example
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/{0-2147483647}h", *MAIN)

    def test_malformed_braces_are_rejected(self):
        for bad in ["m/84h/0h/{0-9", "m/84h/0h/0-9}h", "m/84h/0h/{0-9]h"]:
            with self.assertRaises(xpubauth_scope.ScopeError):
                xpubauth_scope.parse_scope(bad, *MAIN)

    def test_missing_boundary_is_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/{-9}h", *MAIN)

    def test_extra_boundary_is_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/{0-5-9}h", *MAIN)

    def test_unexpected_characters_are_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/{0-9x}h", *MAIN)
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/5x", *MAIN)

    def test_too_many_path_components_is_rejected(self):
        deep = "m/" + "/".join(["0h"] * (xpubauth_scope.MAX_PATH_DEPTH + 1))
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope(deep, *MAIN)

    def test_too_many_scope_entries_is_rejected(self):
        entries = ";".join(
            "m/84h/0h/%dh" % i for i in range(xpubauth_scope.MAX_ENTRIES + 1)
        )
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope(entries, *MAIN)

    def test_total_expansion_over_budget_is_rejected(self):
        scope = ";".join([
            "m/84h/0h/{0-99}h",
            "m/86h/0h/{0-99}h",
            "m/49h/0h/{0-99}h",
        ])
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope(scope, *MAIN)

    def test_wildcards_are_not_a_grammar_construct(self):
        for bad in ["*", "m/*", "m/84h/0h/*"]:
            with self.assertRaises(xpubauth_scope.ScopeError):
                xpubauth_scope.parse_scope(bad, *MAIN)

    def test_multiple_ranges_in_one_entry_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/{0-1}h/{0-9}h", *MAIN)

    def test_empty_scope_is_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("", *MAIN)
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("   ;  ", *MAIN)

    def test_oversized_scope_string_is_rejected(self):
        huge = "m/84h/0h/0h;" * (xpubauth_scope.MAX_SCOPE_LEN // 10)
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope(huge, *MAIN)


class ScopeOverlapTest(TestCase):
    def test_duplicate_exact_paths_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/5h;m/84h/0h/5h", *MAIN)

    def test_exact_path_inside_range_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/{0-9}h;m/84h/0h/5h", *MAIN)

    def test_overlapping_ranges_rejected(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope(
                "m/84h/0h/{0-9}h;m/84h/0h/{5-15}h", *MAIN
            )

    def test_non_overlapping_ranges_allowed(self):
        entries, total = xpubauth_scope.parse_scope(
            "m/84h/0h/{0-9}h;m/84h/0h/{10-15}h", *MAIN
        )
        self.assertEqual(total, 16)

    def test_different_depth_entries_never_overlap(self):
        entries, total = xpubauth_scope.parse_scope(
            "m/84h/0h/0h;m/84h/0h/0h/1h", *MAIN
        )
        self.assertEqual(total, 2)


class ScopeNetworkValidationTest(TestCase):
    def test_mainnet_standard_path_accepted(self):
        xpubauth_scope.parse_scope("m/84h/0h/{0-9}h", *MAIN)

    def test_testnet_style_path_rejected_on_mainnet(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/1h/0h", *MAIN)

    def test_mainnet_style_path_rejected_on_testnet(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope("m/84h/0h/0h", *TEST)

    def test_testnet_style_path_accepted_on_testnet(self):
        xpubauth_scope.parse_scope("m/84h/1h/{0-9}h", *TEST)

    def test_mixed_network_scope_rejected_as_whole(self):
        with self.assertRaises(xpubauth_scope.ScopeError):
            xpubauth_scope.parse_scope(
                "m/84h/0h/{0-9}h;m/84h/1h/{0-9}h", *MAIN
            )

    def test_non_standard_purpose_has_no_coin_policy(self):
        # an unusual/custom path is not a recognized BIP44-style form,
        # so no coin-type policy is enforced on it either way
        xpubauth_scope.parse_scope("m/123456h/0h/0h", *MAIN)
        xpubauth_scope.parse_scope("m/123456h/1h/0h", *MAIN)

    def test_non_hardened_purpose_has_no_coin_policy(self):
        xpubauth_scope.parse_scope("m/84/1h/0h", *MAIN)


class ScopeMatchingTest(TestCase):
    def _entry(self, scope_str, network=MAIN):
        entries, _ = xpubauth_scope.parse_scope(scope_str, *network)
        return entries[0]

    def test_match_boundaries_of_a_range(self):
        entry = self._entry("m/84h/0h/{0-9}h")
        for value in (0, 5, 9):
            path = bip32.parse_path("m/84h/0h/%dh" % value)
            self.assertIsNotNone(entry.match(path))
        for value in (10,):
            path = bip32.parse_path("m/84h/0h/%dh" % value)
            self.assertIsNone(entry.match(path))

    def test_match_uses_normalized_hardened_notation(self):
        entry = self._entry("m/84h/0h/{0-9}h")
        path = bip32.parse_path("m/84'/0'/3'")
        self.assertIsNotNone(entry.match(path))

    def test_out_of_scope_paths_do_not_match(self):
        entry = self._entry("m/84h/0h/{0-9}h")
        for bad in ["m/84h/0h/10h", "m/84h/1h/0h", "m/85h/0h/0h", "m", "m/0h"]:
            try:
                path = bip32.parse_path(bad)
            except Exception:
                continue
            self.assertIsNone(entry.match(path))

    def test_prefix_is_not_authorized(self):
        entry = self._entry("m/84h/0h/1h")
        path = bip32.parse_path("m/84h/0h/1h/999h")
        self.assertIsNone(entry.match(path))


class AuthorizationBudgetTest(TestCase):
    def test_consumable_permission_is_single_use(self):
        entries, _ = xpubauth_scope.parse_scope("m/84h/0h/0h", *MAIN)
        auth = xpubauth_scope.Authorization(entries, "main")
        path = bip32.parse_path("m/84h/0h/0h")
        self.assertTrue(auth.try_consume("main", path))
        self.assertFalse(auth.try_consume("main", path))

    def test_each_value_in_range_consumable_once(self):
        entries, _ = xpubauth_scope.parse_scope("m/84h/0h/{0-2}h", *MAIN)
        auth = xpubauth_scope.Authorization(entries, "main")
        for value in (0, 1, 2):
            path = bip32.parse_path("m/84h/0h/%dh" % value)
            self.assertTrue(auth.try_consume("main", path))
        for value in (0, 1, 2):
            path = bip32.parse_path("m/84h/0h/%dh" % value)
            self.assertFalse(auth.try_consume("main", path))

    def test_remaining_reaches_zero_after_full_consumption(self):
        entries, total = xpubauth_scope.parse_scope("m/84h/0h/{0-2}h", *MAIN)
        auth = xpubauth_scope.Authorization(entries, "main")
        self.assertEqual(auth.remaining, total)
        for value in (0, 1, 2):
            auth.try_consume("main", bip32.parse_path("m/84h/0h/%dh" % value))
        self.assertEqual(auth.remaining, 0)

    def test_network_mismatch_never_consumes(self):
        entries, _ = xpubauth_scope.parse_scope("m/84h/0h/0h", *MAIN)
        auth = xpubauth_scope.Authorization(entries, "main")
        path = bip32.parse_path("m/84h/0h/0h")
        self.assertFalse(auth.try_consume("test", path))
        self.assertEqual(auth.remaining, 1)
        # still consumable on the correct network afterwards
        self.assertTrue(auth.try_consume("main", path))

    def test_unrelated_path_never_consumes(self):
        entries, _ = xpubauth_scope.parse_scope("m/84h/0h/{0-9}h", *MAIN)
        auth = xpubauth_scope.Authorization(entries, "main")
        path = bip32.parse_path("m/85h/0h/0h")
        self.assertFalse(auth.try_consume("main", path))
        self.assertEqual(auth.remaining, 10)

test/tests_native/test_xpubauth_host_command.py (29 tests - host-command integration: begin/end lifecycle, in/out-of-scope, hardened-notation, budget, lock/network invalidation, replace-on-second-begin, cancellation, BIP-138-style walkthrough)

import sys

if sys.implementation.name != 'micropython':
    from native_support import setup_native_stubs

    setup_native_stubs()

from unittest import TestCase
from io import BytesIO
from binascii import hexlify
import gc

from tests.util import get_keystore, get_xpubs_app, clear_testdir
from embit import bip32
from embit.liquid.networks import NETWORKS
from gui.screens import Prompt


class XpubAuthHostCommandTest(TestCase):
    def setUp(self):
        clear_testdir()
        self.keystore = get_keystore()
        self.app = get_xpubs_app(self.keystore, "main")

    def tearDown(self):
        clear_testdir()
        gc.collect()

    # ---- test harness helpers (mirrors test_xpubs_host_command.py) ----

    def _show_screen(self, *responses):
        """Returns (show_screen, seen) where show_screen replies with
        `responses` in order (repeating the last one if exhausted)."""
        seen = []
        responses = list(responses)

        async def show_screen(scr):
            seen.append(scr)
            if len(responses) > 1:
                return responses.pop(0)
            return responses[0]

        return show_screen, seen

    def _run(self, coro):
        try:
            coro.send(None)
        except StopIteration as e:
            return e.value
        raise AssertionError("coroutine did not complete in one step")

    def _xpub(self, path, show_screen):
        stream = BytesIO(b"xpub " + path.encode())
        return self._run(self.app.process_host_command(stream, show_screen))

    def _begin(self, scope, show_screen):
        stream = BytesIO(b"xpubauth begin " + scope.encode())
        return self._run(self.app.process_host_command(stream, show_screen))

    def _end(self, show_screen):
        stream = BytesIO(b"xpubauth end")
        return self._run(self.app.process_host_command(stream, show_screen))

    def _expected_xpub(self, path, network="main"):
        return self.keystore.get_xpub(path).to_base58(NETWORKS[network]["xpub"])

    # ---- begin: happy paths ----

    def test_begin_exact_single_path(self):
        show_screen, seen = self._show_screen(True)
        res = self._begin("m/84h/0h/0h", show_screen)
        self.assertTrue(res)
        self.assertEqual(len(seen), 1)
        self.assertIsInstance(seen[0], Prompt)

    def test_begin_shows_network_paths_and_count(self):
        show_screen, seen = self._show_screen(True)
        self._begin("m/84h/0h/{0-9}h", show_screen)
        shown = seen[0]
        message = shown.args[1] if len(shown.args) > 1 else shown.kwargs.get("message")
        self.assertIn(NETWORKS["main"]["name"], message)
        self.assertIn("m/84h/0h/{0-9}h", message)
        self.assertIn("10", message)

    def test_begin_exactly_one_confirmation_for_bip138_scope(self):
        scope = ";".join([
            "m/44h/0h/{0-9}h", "m/49h/0h/{0-9}h", "m/84h/0h/{0-9}h",
            "m/86h/0h/{0-9}h", "m/87h/0h/{0-9}h",
            "m/48h/0h/{0-9}h/1h", "m/48h/0h/{0-9}h/2h",
        ])
        show_screen, seen = self._show_screen(True)
        res = self._begin(scope, show_screen)
        self.assertTrue(res)
        self.assertEqual(len(seen), 1)

    def test_begin_cancel_leaves_no_authorization(self):
        show_screen, seen = self._show_screen(False)
        res = self._begin("m/84h/0h/{0-9}h", show_screen)
        self.assertFalse(res)
        self.assertIsNone(self.app._authorization)

    def test_begin_invalid_scope_leaves_no_authorization(self):
        show_screen, seen = self._show_screen(True)
        with self.assertRaises(Exception):
            self._begin("garbage", show_screen)
        self.assertEqual(seen, [])
        self.assertIsNone(self.app._authorization)

    def test_begin_locked_device_rejected(self):
        from unittest.mock import patch, PropertyMock
        show_screen, seen = self._show_screen(True)
        with patch.object(
            type(self.keystore), "is_locked", new_callable=PropertyMock, return_value=True
        ):
            with self.assertRaises(Exception):
                self._begin("m/84h/0h/0h", show_screen)
        self.assertEqual(seen, [])

    # ---- in-scope requests bypass confirmation ----

    def test_in_scope_first_boundary_no_prompt(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        show_screen, seen = self._show_screen(True)
        res = self._xpub("m/84h/0h/0h", show_screen)
        stream, meta = res
        self.assertEqual(stream.read().decode(), self._expected_xpub("m/84h/0h/0h"))
        self.assertEqual(seen, [])

    def test_in_scope_middle_no_prompt(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        show_screen, seen = self._show_screen(True)
        res = self._xpub("m/84h/0h/5h", show_screen)
        stream, meta = res
        self.assertEqual(stream.read().decode(), self._expected_xpub("m/84h/0h/5h"))
        self.assertEqual(seen, [])

    def test_in_scope_last_boundary_no_prompt(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        show_screen, seen = self._show_screen(True)
        res = self._xpub("m/84h/0h/9h", show_screen)
        stream, meta = res
        self.assertEqual(stream.read().decode(), self._expected_xpub("m/84h/0h/9h"))
        self.assertEqual(seen, [])

    def test_in_scope_matches_hardened_prime_notation(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        show_screen, seen = self._show_screen(True)
        res = self._xpub("m/84'/0'/3'", show_screen)
        stream, meta = res
        self.assertEqual(stream.read().decode(), self._expected_xpub("m/84h/0h/3h"))
        self.assertEqual(seen, [])

    def test_exact_four_paths_all_retrievable_without_prompt(self):
        paths = ["m/84h/0h/0h", "m/86h/0h/0h", "m/48h/0h/0h/2h", "m/48h/0h/1h/2h"]
        self._begin(";".join(paths), self._show_screen(True)[0])
        for path in paths:
            show_screen, seen = self._show_screen(True)
            res = self._xpub(path, show_screen)
            stream, meta = res
            self.assertEqual(stream.read().decode(), self._expected_xpub(path))
            self.assertEqual(seen, [])

    def test_fifth_unrelated_path_still_prompts(self):
        paths = ["m/84h/0h/0h", "m/86h/0h/0h", "m/48h/0h/0h/2h", "m/48h/0h/1h/2h"]
        self._begin(";".join(paths), self._show_screen(True)[0])
        show_screen, seen = self._show_screen(True)
        self._xpub("m/49h/0h/0h", show_screen)
        self.assertEqual(len(seen), 1)
        self.assertIsInstance(seen[0], Prompt)

    # ---- out-of-scope requests fall back to normal confirmation ----

    def test_out_of_range_index_prompts_normally(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/10h", show_screen)
        self.assertEqual(len(seen), 1)

    def test_out_of_scope_does_not_widen_authorization(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        # approve an unrelated path individually
        self._xpub("m/85h/0h/0h", self._show_screen(True)[0])
        # it must not have joined the scope - still prompts every time
        show_screen, seen = self._show_screen(True)
        self._xpub("m/85h/0h/0h", show_screen)
        self.assertEqual(len(seen), 1)

    def test_different_account_out_of_scope_prompts(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        for bad in ["m/84h/1h/0h", "m/85h/0h/0h", "m", "m/0h"]:
            show_screen, seen = self._show_screen(True)
            self._xpub(bad, show_screen)
            self.assertEqual(len(seen), 1, "expected a prompt for %r" % bad)

    def test_prefix_child_is_not_silently_authorized(self):
        self._begin("m/84h/0h/1h", self._show_screen(True)[0])
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/1h/999h", show_screen)
        self.assertEqual(len(seen), 1)

    # ---- budget: consumable, single-use permissions ----

    def test_repeated_request_for_consumed_path_reprompts(self):
        self._begin("m/84h/0h/0h", self._show_screen(True)[0])
        show_screen1, seen1 = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen1)
        self.assertEqual(seen1, [])
        # second request for the same, already-consumed path
        show_screen2, seen2 = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen2)
        self.assertEqual(len(seen2), 1)

    def test_authorization_ends_after_full_consumption(self):
        self._begin("m/84h/0h/{0-1}h", self._show_screen(True)[0])
        self._xpub("m/84h/0h/0h", self._show_screen(True)[0])
        self.assertIsNotNone(self.app._authorization)
        self._xpub("m/84h/0h/1h", self._show_screen(True)[0])
        self.assertIsNone(self.app._authorization)

    # ---- explicit end ----

    def test_end_clears_authorization(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        self.assertTrue(self._end(self._show_screen(True)[0]))
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen)
        self.assertEqual(len(seen), 1)

    def test_end_without_active_authorization_is_a_noop_success(self):
        self.assertTrue(self._end(self._show_screen(True)[0]))

    # ---- a second begin replaces, never merges ----

    def test_second_begin_replaces_after_fresh_confirmation(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        self._begin("m/86h/0h/{0-9}h", self._show_screen(True)[0])
        # old scope no longer authorized
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen)
        self.assertEqual(len(seen), 1)
        # new scope is
        show_screen2, seen2 = self._show_screen(True)
        self._xpub("m/86h/0h/0h", show_screen2)
        self.assertEqual(seen2, [])

    def test_cancelled_second_begin_keeps_first_authorization(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        res = self._begin("m/86h/0h/{0-9}h", self._show_screen(False)[0])
        self.assertFalse(res)
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen)
        self.assertEqual(seen, [])

    # ---- invalidation ----

    def test_lock_clears_authorization(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        self.app.on_lock()
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen)
        self.assertEqual(len(seen), 1)

    def test_network_switch_clears_authorization(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        self.app.init(self.keystore, "test", self.app.show_loader, self.app.communicate)
        self.app.init(self.keystore, "main", self.app.show_loader, self.app.communicate)
        # back on the original network, but the authorization is gone -
        # the very same path that used to be in-scope must re-prompt
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen)
        self.assertEqual(len(seen), 1)

    def test_network_switch_invalidates_even_when_authorization_object_survives(self):
        # defense in depth: directly exercise the network check inside
        # try_consume, independent of the init()-triggered reset above
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        auth = self.app._authorization
        self.assertIsNotNone(auth)
        auth.network = "test"  # simulate a stale authorization surviving
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen)
        self.assertEqual(len(seen), 1)

    def test_reinit_same_network_clears_authorization(self):
        # covers seed reload / passphrase change / storage switch, which
        # all funnel through init()
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        self.app.init(self.keystore, "main", self.app.show_loader, self.app.communicate)
        show_screen, seen = self._show_screen(True)
        self._xpub("m/84h/0h/0h", show_screen)
        self.assertEqual(len(seen), 1)

    # ---- cancellation never leaks xpubs ----

    def test_out_of_scope_cancel_leaks_nothing(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        show_screen, seen = self._show_screen(False)
        res = self._xpub("m/84h/0h/50h", show_screen)
        self.assertFalse(res)

    # ---- fingerprint is unaffected by any of this ----

    def test_fingerprint_stays_non_interactive_with_active_authorization(self):
        self._begin("m/84h/0h/{0-9}h", self._show_screen(True)[0])
        show_screen, seen = self._show_screen(True)
        stream = BytesIO(b"fingerprint")
        res = self._run(self.app.process_host_command(stream, show_screen))
        result_stream, meta = res
        self.assertEqual(result_stream.read().decode(), hexlify(self.keystore.fingerprint).decode())
        self.assertEqual(seen, [])


class Bip138LikeIntegrationTest(TestCase):
    """End-to-end walk-through of the motivating BIP-0138 use case:
    one authorization confirmation, many silent per-path lookups,
    an early stop, then an explicit end."""

    def setUp(self):
        clear_testdir()
        self.keystore = get_keystore()
        self.app = get_xpubs_app(self.keystore, "main")

    def tearDown(self):
        clear_testdir()
        gc.collect()

    def _show_screen(self, response):
        seen = []

        async def show_screen(scr):
            seen.append(scr)
            return response

        return show_screen, seen

    def _run(self, coro):
        try:
            coro.send(None)
        except StopIteration as e:
            return e.value
        raise AssertionError("coroutine did not complete in one step")

    def test_full_bip138_like_flow(self):
        # 1: fingerprint without UI
        show_screen, seen = self._show_screen(True)
        stream = BytesIO(b"fingerprint")
        self._run(self.app.process_host_command(stream, show_screen))
        self.assertEqual(seen, [])

        # 2-5: one authorization confirmation for the discovery scope
        scope = ";".join([
            "m/44h/0h/{0-9}h", "m/49h/0h/{0-9}h", "m/84h/0h/{0-9}h",
            "m/86h/0h/{0-9}h", "m/87h/0h/{0-9}h",
            "m/48h/0h/{0-9}h/1h", "m/48h/0h/{0-9}h/2h",
        ])
        begin_show, begin_seen = self._show_screen(True)
        stream = BytesIO(b"xpubauth begin " + scope.encode())
        self.assertTrue(self._run(self.app.process_host_command(stream, begin_show)))
        self.assertEqual(len(begin_seen), 1)

        # 6-8: sequential per-path lookups with no further prompts
        for path in ["m/84h/0h/0h", "m/84h/0h/1h", "m/49h/0h/0h"]:
            show_screen, seen = self._show_screen(True)
            stream = BytesIO(b"xpub " + path.encode())
            res = self._run(self.app.process_host_command(stream, show_screen))
            result_stream, meta = res
            expected = self.keystore.get_xpub(path).to_base58(NETWORKS["main"]["xpub"])
            self.assertEqual(result_stream.read().decode(), expected)
            self.assertEqual(seen, [])

        # 9: recovery "succeeds" and stops early - the remaining ~67
        # authorized paths are simply never requested

        # 10: host ends the authorization explicitly
        end_show, _ = self._show_screen(True)
        stream = BytesIO(b"xpubauth end")
        self.assertTrue(self._run(self.app.process_host_command(stream, end_show)))

        # 11: a formerly in-scope path requires normal confirmation again
        show_screen, seen = self._show_screen(True)
        stream = BytesIO(b"xpub m/84h/0h/2h")
        self._run(self.app.process_host_command(stream, show_screen))
        self.assertEqual(len(seen), 1)

test/integration/tests/test_xpubauth.py (10 tests - TCP integration against the real MicroPython Unix simulator)

from unittest import TestCase
from util.controller import sim


class XpubAuthIntegrationTest(TestCase):
    def test_begin_end_and_in_scope_requests(self):
        # one confirmation for the scope...
        res = sim.query(b"xpubauth begin m/84h/1h/{0-2}h", [True])
        self.assertEqual(res, b"success")

        # ...then in-scope requests need no further confirmation at all.
        # sim.query() is called here with no "commands" to send to the
        # GUI socket - if the device were to show a prompt, nothing
        # would confirm it and the USB read below would time out empty,
        # so a real tpub coming back proves no prompt was shown.
        res = sim.query(b"xpub m/84h/1h/0h")
        self.assertTrue(res.startswith(b"tpub"), res)
        res = sim.query(b"xpub m/84h/1h/2h")
        self.assertTrue(res.startswith(b"tpub"), res)

        # a path outside the approved scope still requires confirmation
        res = sim.query(b"xpub m/84h/1h/9h", [True])
        self.assertTrue(res.startswith(b"tpub"), res)

        # explicit end
        res = sim.query(b"xpubauth end")
        self.assertEqual(res, b"success")

        # the same in-scope path now requires confirmation again -
        # without it, the request would time out instead of succeeding
        res = sim.query(b"xpub m/84h/1h/0h", [True])
        self.assertTrue(res.startswith(b"tpub"), res)

    def test_exact_four_paths(self):
        scope = b"m/84h/1h/0h;m/86h/1h/0h;m/48h/1h/0h/2h;m/48h/1h/1h/2h"
        res = sim.query(b"xpubauth begin " + scope, [True])
        self.assertEqual(res, b"success")

        for path in [b"m/84h/1h/0h", b"m/86h/1h/0h", b"m/48h/1h/0h/2h", b"m/48h/1h/1h/2h"]:
            res = sim.query(b"xpub " + path)
            self.assertTrue(res.startswith(b"tpub"), res)

        # a fifth, unrelated path is not covered and needs confirmation
        res = sim.query(b"xpub m/49h/1h/0h", [True])
        self.assertTrue(res.startswith(b"tpub"), res)

        sim.query(b"xpubauth end")

    def test_repeated_request_for_consumed_path_reprompts(self):
        res = sim.query(b"xpubauth begin m/84h/1h/0h", [True])
        self.assertEqual(res, b"success")

        # first silent use consumes the single-path authorization
        res = sim.query(b"xpub m/84h/1h/0h")
        self.assertTrue(res.startswith(b"tpub"), res)

        # authorization is fully consumed and self-clears - a repeat
        # request needs confirmation again
        res = sim.query(b"xpub m/84h/1h/0h", [True])
        self.assertTrue(res.startswith(b"tpub"), res)

    def test_cancelled_begin_grants_nothing(self):
        res = sim.query(b"xpubauth begin m/84h/1h/{0-2}h", [False])
        self.assertEqual(res, b"error: User cancelled")

        # nothing was authorized - a matching path still prompts
        res = sim.query(b"xpub m/84h/1h/0h", [True])
        self.assertTrue(res.startswith(b"tpub"), res)

    def test_invalid_scope_is_rejected_without_prompting(self):
        res = sim.query(b"xpubauth begin m/84h/0h/*")
        self.assertTrue(res.startswith(b"error:"), res)

    def test_bip138_like_discovery_flow(self):
        scope = b";".join([
            b"m/44h/1h/{0-9}h", b"m/49h/1h/{0-9}h", b"m/84h/1h/{0-9}h",
            b"m/86h/1h/{0-9}h", b"m/87h/1h/{0-9}h",
            b"m/48h/1h/{0-9}h/1h", b"m/48h/1h/{0-9}h/2h",
        ])
        res = sim.query(b"xpubauth begin " + scope, [True])
        self.assertEqual(res, b"success")

        # sequential lookups, no further prompts, stopping early
        for path in [b"m/84h/1h/0h", b"m/84h/1h/1h", b"m/49h/1h/0h"]:
            res = sim.query(b"xpub " + path)
            self.assertTrue(res.startswith(b"tpub"), res)

        sim.query(b"xpubauth end")

        res = sim.query(b"xpub m/84h/1h/2h", [True])
        self.assertTrue(res.startswith(b"tpub"), res)

Also note: registering these in test/tests_native/__init__.py required adding two lines there (now reverted) - if you re-add these files locally, re-add

from .test_xpub_scope import *
from .test_xpubauth_host_command import *

to that file's imports to have them picked up by test/run_native_tests.py.

---

Generated by Claude Code

[schnuartz-ai]

Companion script for manually testing this PR against real USB hardware: specter_menu_en.py (repo root, needs hwidevice.py next to it).

Requirements

pip install hwi pyserial

Then:

python3 specter_menu_en.py

The script finds the device automatically (USB serial / simulator TCP ports), connects, and shows a numbered menu. Just enter a number + Enter — no need to type raw protocol commands.

How it's built

It's a thin menu layer over hwidevice.py (the actual USB client, which this PR extends with authorize_xpubs() / end_xpub_authorization()). Each menu option ultimately calls one of the existing protocol commands from docs/communication.md:

Menu option	Calls	Protocol command
1) Show fingerprint	client.get_master_fingerprint()	fingerprint
2) Query a single xpub	client.get_pubkey_at_path(path)	xpub <path>
3) Authorize multiple xpubs	client.authorize_xpubs(scope)	xpubauth begin <scope>
4) Query an xpub inside the authorized scope	client.get_pubkey_at_path(path)	xpub <path>
5) Query every xpub in the authorized scope, in order	loop over get_pubkey_at_path()	multiple xpub <path> in sequence
6) End authorization	client.end_xpub_authorization()	xpubauth end
7) Send a raw command	client.query(cmd)	anything else (fallback)

Option 5 is pure client-side logic (_expand_scope()): it takes the same scope string used for "begin" (e.g. m/84h/0h/{0-9}h), expands it locally in the script (no device communication involved) into a concrete list of paths (m/84h/0h/0h, m/84h/0h/1h, ..., m/84h/0h/9h), then queries each one in turn via xpub <path>. After a successful authorization (option 3), the script remembers that scope as the default for option 5 — just hit Enter to reuse it.

How to test this PR with it

A) Core feature — one authorization, many queries with no further confirmation

1. Option 3, scope e.g. m/84h/0h/{0-9}h → the device shows the confirmation screen "Share multiple XPUBs?" once, with network, path range, count, and the device fingerprint line → confirm.
2. Option 5, Enter (reuses the scope from step 1) → all 10 xpubs come back in sequence, without any further prompt on the device. That's the actual point of this PR.

B) Outside the scope still prompts individually
Option 4 with a path outside the authorized scope (e.g. m/84h/1h/0h or a different account) → the normal individual "Share Xpub?" confirmation appears again on the device (also with the fingerprint line) — the authorization is not widened by this.

C) Regression check — with no active authorization, everything behaves exactly like before this PR
Right after starting (before option 3): option 1 (fingerprint, no confirmation needed) and option 2 (individual query, prompts every time) — the existing behavior from PR cryptoadvance#381, unchanged.

D) Ending an authorization
Option 6, then option 4 again with a previously authorized path → should prompt individually again, since the authorization was explicitly ended.

E) Budget/consumption
Query the same path twice via option 4 after it's been authorized once → the second time, the device prompts individually again (each authorized path can be retrieved silently at most once — see the "Budget" section in docs/communication.md).

specter_menu_en.py (click to expand)

#!/usr/bin/env python3
"""
Specter-DIY Test Menu
=====================

Interactive menu for testing the device over USB, in particular the
new "xpubauth" bulk authorization for multiple xpubs at once.

Requirements:  pip install hwi pyserial
Start:         python3 specter_menu_en.py
"""
import re
import sys
from hwidevice import enumerate as find_devices, SpecterClient

# remembers the scope string from the last successful "xpubauth begin",
# so "query all" can default to exactly what was just approved
_last_scope = {"value": None}


def _expand_scope(scope_str):
    """
    Client-side expansion of the same scope grammar the device uses
    (see docs/communication.md / src/apps/xpubs/scope.py): entries
    separated by ';', each an exact path or a path with one bounded
    {lo-hi} range component. Returns the concrete list of derivation
    paths so they can be queried one after another.
    """
    paths = []
    for raw in scope_str.split(";"):
        entry = raw.strip()
        if not entry:
            continue
        m = re.search(r"\{(\d+)-(\d+)\}", entry)
        if not m:
            paths.append(entry)
            continue
        lo, hi = int(m.group(1)), int(m.group(2))
        if lo > hi:
            raise ValueError("Invalid range (start > end) in %r" % entry)
        prefix, suffix = entry[:m.start()], entry[m.end():]
        for i in range(lo, hi + 1):
            paths.append("%s%d%s" % (prefix, i, suffix))
    return paths


def pick_device():
    print("Looking for device ...")
    devices = find_devices()
    if len(devices) == 0:
        print("No device found. Is it connected via USB and unlocked?")
        sys.exit(1)
    if len(devices) == 1:
        d = devices[0]
        print("Found: %s (fingerprint %s)" % (d["path"], d["fingerprint"]))
        return SpecterClient(d["path"])
    print("Multiple devices found:")
    for i, d in enumerate(devices):
        print("  [%d] %s (fingerprint %s)" % (i, d["path"], d["fingerprint"]))
    while True:
        choice = input("Which device? Enter number: ").strip()
        if choice.isdigit() and int(choice) < len(devices):
            return SpecterClient(devices[int(choice)]["path"])
        print("Invalid input.")


def ask(prompt, default=None):
    suffix = " [%s]" % default if default is not None else ""
    val = input("%s%s: " % (prompt, suffix)).strip()
    return val if val else default


def show_fingerprint(client):
    try:
        fp = client.get_master_fingerprint().hex()
        print("Fingerprint: %s" % fp)
    except Exception as e:
        print("Error: %s" % e)


def get_single_xpub(client):
    path = ask("Derivation path", "m/84h/0h/0h")
    print("-> Please confirm on the device ...")
    try:
        xpub = client.get_pubkey_at_path(path).to_string()
        print("Xpub: %s" % xpub)
    except Exception as e:
        print("Error/cancelled: %s" % e)


def begin_authorization(client):
    print("Scope examples (separate multiple entries with ';'):")
    print("  m/84h/0h/0h                      (exactly one path)")
    print("  m/84h/0h/{0-9}h                   (10 accounts at once)")
    print("  m/84h/0h/{0-9}h;m/86h/0h/{0-9}h    (multiple ranges)")
    scope = ask("Scope to authorize", "m/84h/0h/{0-9}h")
    print("-> Please confirm ONCE on the device ...")
    try:
        ok = client.authorize_xpubs(scope)
        if ok:
            print("Authorized!")
            _last_scope["value"] = scope
        else:
            print("Not confirmed.")
    except Exception as e:
        print("Error/cancelled: %s" % e)


def get_xpub_in_scope(client):
    path = ask("Derivation path (inside the authorized scope)", "m/84h/0h/0h")
    print("-> Should come back WITHOUT another confirmation on the device, if in scope ...")
    try:
        xpub = client.get_pubkey_at_path(path).to_string()
        print("Xpub: %s" % xpub)
    except Exception as e:
        print("Error/cancelled (outside the scope? then confirm on the device): %s" % e)


def query_all_in_scope(client):
    default = _last_scope["value"] or "m/84h/0h/{0-9}h"
    scope = ask("Scope to fully query (as used for the authorization)", default)
    try:
        paths = _expand_scope(scope)
    except ValueError as e:
        print("Error parsing the scope: %s" % e)
        return
    print("Querying %d xpub(s) one after another ..." % len(paths))
    ok_count = 0
    for path in paths:
        try:
            xpub = client.get_pubkey_at_path(path).to_string()
            print("  %-24s -> %s" % (path, xpub))
            ok_count += 1
        except Exception as e:
            print("  %-24s -> Error/cancelled: %s" % (path, e))
    print("Done: %d of %d succeeded." % (ok_count, len(paths)))


def end_authorization(client):
    try:
        ok = client.end_xpub_authorization()
        print("Authorization ended." if ok else "Error ending authorization.")
    except Exception as e:
        print("Error: %s" % e)


def raw_command(client):
    cmd = ask("Raw command (e.g. 'xpub m/84h/0h/0h')")
    if not cmd:
        return
    try:
        print(client.query(cmd))
    except Exception as e:
        print("Error/cancelled: %s" % e)


MENU = [
    ("Show fingerprint (no confirmation needed)", show_fingerprint),
    ("Query a single xpub (asks on the device)", get_single_xpub),
    ("Authorize multiple xpubs at once (xpubauth begin)", begin_authorization),
    ("Query an xpub inside the authorized scope (no extra confirmation)", get_xpub_in_scope),
    ("Query every xpub in the authorized scope, in order", query_all_in_scope),
    ("End authorization (xpubauth end)", end_authorization),
    ("Send a raw command (for anything else)", raw_command),
]


def main():
    client = pick_device()
    print()
    while True:
        print("=" * 50)
        for i, (label, _) in enumerate(MENU, start=1):
            print("  %d) %s" % (i, label))
        print("  0) Quit")
        choice = input("Choice: ").strip()
        print()
        if choice == "0":
            break
        if choice.isdigit() and 1 <= int(choice) <= len(MENU):
            MENU[int(choice) - 1][1](client)
        else:
            print("Invalid choice.")
        print()


if __name__ == "__main__":
    main()

---

Generated by Claude Code

[Schnuartz]

Tested it and now everything works as expected.