← Watch feed
Merge remote-tracking branch 'agent/benma-agent/host-passphrase-polling'
Public commit record
50/100 · Thin
What the developer wrote
Authored by Marko Bencun
Merge remote-tracking branch 'agent/benma-agent/host-passphrase-polling'
✓ Specific, descriptive subject✓ Names a concrete action or component! No meaningful explanatory body
What changed, and why it matters
This commit adds a new feature that lets users type their optional BIP39 passphrase on the computer (host) instead of only on the BitBox02 device. The device still asks the user for approval before accepting host input, and the actual passphrase is shown on the device for confirmation. The change also moves the unlock handshake inside the already-encrypted Noise channel, adds protocol messages for polling, and includes many tests. It is a feature addition with security-relevant design, not a fix for a known vulnerability.
Recommended action
Review as a security-relevant feature addition. Verify that host-entry consent cannot be bypassed, that invalid/rejected host input reliably restarts device entry, that the passphrase is cleared from host memory promptly, and that the U2F/UI ownership guard prevents cross-workflow races. No immediate patch or incident response is indicated by the commit alone.
Security signals we found
01
New encrypted unlock workflow inside Noise channel
02
Host-supplied passphrase validated on device for length and allowed character set
03
Device approval and confirmation required before host passphrase is used
04
Passphrase redacted in Python debug logs and zeroized in Rust on drop
05
U2F start blocked while HWW workflow holds UI between requests
06
Extensive Rust and Python tests for race conditions, cancellation, and invalid states
Evidence from the diff
The merge introduces host-side BIP39 passphrase entry during unlock. New protobuf messages (UnlockRequest, UnlockContinueRequest, UnlockHostInfoRequest, UnlockResponse) are added to hww.proto/keystore.proto and generated bindings. Firmware v9.28.0+ performs unlock inside the paired Noise channel; the Python library polls with UnlockContinue and, after device consent, submits the passphrase with UnlockHostInfo. The device validates length (<=149 ASCII bytes) and character set, shows a consent prompt and a final confirmation, and zeroizes the host-provided passphrase on drop. U2F workflow startup now refuses to start if an HWW workflow owns the UI between requests, and USB timeouts are adjusted for intermediate responses.
Changed components
BitBox02 firmware unlock workflowHWW/Noise protocol transportPython bitbox02 libraryProtobuf API (hww.proto, keystore.proto)U2F C API workflow guardUSB HWW transport timeout handlingInspect captured patch +1955 / −107
### CHANGELOG.md
@@ -7,6 +7,7 @@ recorded separately.
## Firmware
### Unreleased
+- Allow entering the optional BIP39 passphrase on the host, with device approval and confirmation
- Reject malformed microSD backups instead of crashing
- Hold the screen reset pin low until firmware is ready initalize it.
- Cardano: limit xpub requests to 20 keypaths per batch
### messages/hww.proto
@@ -60,6 +60,9 @@ message Request {
BluetoothRequest bluetooth = 29;
ChangePasswordRequest change_password = 30;
BitBoxSyncRequest bitbox_sync = 31;
+ UnlockRequest unlock = 32;
+ UnlockContinueRequest unlock_continue = 33;
+ UnlockHostInfoRequest unlock_host_info = 34;
}
}
@@ -84,5 +87,6 @@ message Response {
BIP85Response bip85 = 16;
BluetoothResponse bluetooth = 17;
BitBoxSyncResponse bitbox_sync = 18;
+ UnlockResponse unlock = 19;
}
}
### messages/keystore.proto
@@ -8,6 +8,38 @@ package shiftcrypto.bitbox02;
import "google/protobuf/empty.proto";
+// Unlock inside the paired Noise channel (since v9.28.0). Uninitialized and already unlocked
+// devices return DONE immediately and are left unchanged. Continuations are only valid within
+// this workflow.
+// If the passphrase feature is enabled, device entry returns PASSPHRASE_PENDING. The host
+// polls with UnlockContinueRequest until entry completes or it requests host entry.
+message UnlockRequest {}
+
+message UnlockContinueRequest {
+ // False polls device entry; true interrupts it to ask for host-entry consent.
+ // After PASSPHRASE_ENTERED, continue to await confirmation and unlock;
+ // host entry is ignored.
+ bool request_host_entry = 1;
+}
+
+message UnlockHostInfoRequest {
+ // Only sent after HOST_ENTRY_READY. Absent cancels host input and restarts device entry;
+ // the empty string submits the empty passphrase. The device confirms the actual value.
+ optional string passphrase = 1;
+}
+
+message UnlockResponse {
+ enum State {
+ PASSPHRASE_PENDING = 0;
+ HOST_ENTRY_READY = 1;
+ // Passphrase entry on the device finished; confirmation may still be pending.
+ // Withdraw host entry and send UnlockContinue to await completion.
+ PASSPHRASE_ENTERED = 2;
+ DONE = 3;
+ }
+ State state = 1;
+}
+
message ElectrumEncryptionKeyRequest {
repeated uint32 keypath = 1;
}
### py/bitbox02/README.md
@@ -16,6 +16,52 @@ In this folder is the Python API for communicating with the BitBox02 device.
The folder `generated` contains files generated by `python-protobuf`. Regenerate with `make`.
+### Optional host passphrase entry
+
+Firmware v9.28.0 and later unlock inside the paired Noise channel. Pass a `BitBoxConfig`
+from `bitbox02.communication` as the optional `bitbox_config` argument to `BitBox02`:
+
+- `enter_mnemonic_passphrase() -> Optional[str]` blocks for host input **after device approval**.
+ Return `None` to cancel, or a string to submit. `""` submits the empty passphrase. The device
+ then asks the user to confirm the actual passphrase, including an empty passphrase.
+ With only this callback configured, the library automatically requests host entry once per
+ unlock, as soon as device passphrase entry starts. Rejection, cancellation or invalid input
+ falls back to device entry without requesting host entry again.
+- `on_host_passphrase_available(request_host_entry)` is an optional, nonblocking notification
+ that lets your app choose when to request host entry. Use the supplied callable to request
+ device consent. For example, show a host-entry button and bind its click handler to the callable;
+ on `None`, hide the button. The callable safely queues one request and wakes the library's
+ device-entry polling loop; it does no transport I/O.
+ Availability is announced once per device-entry phase. Device typing completion withdraws
+ availability before the confirmation screens; late requests cannot interrupt confirmation.
+ Rejection or cancellation offers a new callable; old callables cannot affect subsequent phases.
+
+For an app that only needs a host input dialog:
+
+```python
+config = BitBoxConfig(enter_mnemonic_passphrase=show_passphrase_dialog)
+```
+
+Connection is synchronous: for initialized devices, `BitBox02(...)` returns once unlocking
+is complete. Uninitialized devices return from unlock immediately and can proceed with setup.
+
+Host entry requires the input callback and the device's optional passphrase setting.
+GUI clients must run synchronous connection/unlock outside the UI thread
+and marshal notifications to the UI thread. The blocking input callback can request a dialog
+there and wait for its result. Callbacks must not start other device queries while unlock is
+running. The library handles all protocol exchanges, including device-entry polling even without
+host-entry callbacks, and withdraws availability before host consent or device confirmation and
+on completion or errors.
+Host input has no disconnect heartbeat; reconnecting resets an abandoned workflow before pairing again.
+
+`py/send_message.py` demonstrates this with an `h` shortcut and hidden terminal input. It stops
+the shortcut listener and announces that `h` is disabled when the button is withdrawn, then
+prints "Connection ready" when connection setup returns. Host passphrases use the device
+keyboard's characters and at most 149 ASCII bytes, without trimming.
+Unsupported or overlong input shows a status message and restarts device entry, notifying
+availability again if that callback is configured.
+Python strings and protobuf objects are not guaranteed to be erased from host memory.
+
### Development environment
@@ -27,4 +73,4 @@ such as regenerating files have their libraries NOT in requirements.txt as they
Python package.
It is highly advisable to use same version that is used in the [Dockerfile](../../Dockerfile) that is used in dockerized
-setup.
\ No newline at end of file
+setup.
### py/bitbox02/bitbox02/bitbox02/__init__.py
@@ -7,9 +7,9 @@
__version__ = "8.0.0"
-if sys.version_info.major != 3 or sys.version_info.minor < 6:
+if sys.version_info.major != 3 or sys.version_info.minor < 7:
print(
- "Python version is {}.{}, but 3.6+ is required by this script.".format(
+ "Python version is {}.{}, but 3.7+ is required by this script.".format(
sys.version_info.major, sys.version_info.minor
),
file=sys.stderr,
### py/bitbox02/bitbox02/communication/__init__.py
@@ -5,6 +5,7 @@
from .communication import PhysicalLayer, TransportLayer
from .bitbox_api_protocol import (
BitBoxNoiseConfig,
+ BitBoxConfig,
BitBoxCommonAPI,
Bitbox02Exception,
UserAbortException,
### py/bitbox02/bitbox02/communication/bitbox_api_protocol.py
@@ -3,12 +3,14 @@
"""BitBox02"""
from abc import ABC, abstractmethod
+from dataclasses import dataclass
import os
import enum
import sys
import base64
import hashlib
import time
+import threading
from typing import Callable, Optional, Dict, Tuple, Union, Sequence
import ecdsa
@@ -23,6 +25,7 @@
try:
from .generated import hww_pb2 as hww
from .generated import system_pb2 as system
+ from .generated import keystore_pb2 as keystore
except ModuleNotFoundError:
print("Run `make py` to generate the protobuf messages")
sys.exit()
@@ -260,6 +263,39 @@ def __init__(self, need_atleast: semver.VersionInfo):
)
+@dataclass
+class BitBoxConfig:
+ """Callbacks for optional BIP39 passphrase entry during unlock.
+
+ When the device's passphrase setting is enabled, entry starts on the device.
+ Set enter_mnemonic_passphrase to let your app ask for the passphrase after device
+ approval. With this callback alone, the library automatically requests host entry
+ once per unlock. Rejection, cancellation or invalid input falls back to device entry.
+
+ To let your app choose when to request host entry, also set
+ on_host_passphrase_available. For example, your app can offer an
+ "Enter passphrase on host" button and request host entry when it is clicked.
+
+ The library handles device communication and polling. Callbacks run on the thread
+ performing connection/unlock. GUI apps must run this work on a worker thread and
+ schedule UI updates on the UI thread. Callbacks must not make other device queries.
+ """
+
+ # Notifies when host entry can be requested; requires enter_mnemonic_passphrase.
+ # Use the supplied callable to request device consent, e.g. bind it to a button.
+ # None withdraws availability: hide the button if using one. Return immediately.
+ # The supplied callable is thread-safe and wakes the unlock loop without device I/O.
+ # None is sent before host consent or device confirmation, and when unlock ends.
+ # If entry restarts, a fresh callable is supplied; old callables have no effect.
+ on_host_passphrase_available: Optional[Callable[[Optional[Callable[[], None]]], None]] = None
+
+ # Called only after device approval. Wait for host input and return the passphrase
+ # as a string (including "" for an empty passphrase), or None to cancel host input.
+ # GUI apps can open a dialog on the UI thread and wait here for its result.
+ # Without an availability callback, host entry is requested automatically once per unlock.
+ enter_mnemonic_passphrase: Optional[Callable[[], Optional[str]]] = None
+
+
class BitBoxNoiseConfig:
"""Stores Functions required setup a noise connection"""
@@ -554,6 +590,7 @@ def __init__(
transport: TransportLayer,
device_info: Optional[DeviceInfo],
noise_config: BitBoxNoiseConfig,
+ bitbox_config: Optional[BitBoxConfig] = None,
):
"""
Can raise LibraryVersionOutdatedException. check_min_version() should be called following
@@ -604,10 +641,80 @@ def __init__(
if self.version >= semver.VersionInfo(2, 0, 0):
noise_config.attestation_check(self._perform_attestation())
- self._bitbox_protocol.unlock_query()
+ if self.version < semver.VersionInfo(9, 28, 0):
+ self._bitbox_protocol.unlock_query()
self._bitbox_protocol.noise_connect(noise_config)
+ if self.version >= semver.VersionInfo(9, 28, 0):
+ self._unlock(bitbox_config or BitBoxConfig())
+
+ def _unlock(self, config: BitBoxConfig) -> None:
+ """Own all protocol I/O, including polling while device entry is available."""
+ # pylint: disable=no-member
+ available = config.on_host_passphrase_available
+ enter = config.enter_mnemonic_passphrase
+ auto_request_host_entry = enter is not None and available is None
+ host_entry: Optional[threading.Event] = None
+
+ def withdraw() -> None:
+ nonlocal host_entry
+ if host_entry is not None:
+ host_entry = None
+ assert available is not None
+ available(None)
+
+ request = hww.Request(unlock=keystore.UnlockRequest())
+ consent_requested = False
+ try:
+ while True:
+ reply = self._msg_query(request, expected_response="unlock").unlock
+ if reply.state == keystore.UnlockResponse.DONE:
+ return
+ if reply.state == keystore.UnlockResponse.PASSPHRASE_PENDING:
+ consent_requested = False
+ if auto_request_host_entry:
+ # Request only once per unlock so rejection/cancellation can fall back
+ # to device entry without immediately prompting for host entry again.
+ auto_request_host_entry = False
+ consent_requested = True
+ elif available is not None and enter is not None:
+ if host_entry is None:
+ # Each device-entry phase gets its own thread-safe, coalescing event.
+ # Stale callbacks only signal their old, unused event.
+ host_entry = threading.Event()
+ available(host_entry.set)
+ consent_requested = host_entry.wait(0.1)
+ if consent_requested:
+ withdraw()
+ else:
+ time.sleep(0.1)
+ request = hww.Request(
+ unlock_continue=keystore.UnlockContinueRequest(
+ request_host_entry=consent_requested
+ )
+ )
+ elif reply.state == keystore.UnlockResponse.PASSPHRASE_ENTERED:
+ consent_requested = False
+ withdraw()
+ # The next response waits for confirmation. Do not offer host entry or
+ # report successful completion while the user is still confirming.
+ request = hww.Request(unlock_continue=keystore.UnlockContinueRequest())
+ elif reply.state == keystore.UnlockResponse.HOST_ENTRY_READY and consent_requested:
+ assert enter is not None
+ consent_requested = False
+ passphrase = enter()
+ if passphrase is not None and not isinstance(passphrase, str):
+ raise TypeError("Passphrase callback must return a string or None")
+ request = hww.Request(
+ unlock_host_info=keystore.UnlockHostInfoRequest(passphrase=passphrase)
+ )
+ del passphrase
+ else:
+ raise Exception("Unexpected unlock phase")
+ finally:
+ withdraw()
+
# pylint: disable=too-many-return-statements
def _perform_attestation(self) -> bool:
"""Sends a random challenge and verifies that the response can be verified with
@@ -660,7 +767,10 @@ def _msg_query(
"""
# pylint: disable=no-member
if self.debug:
- print(request)
+ if request.WhichOneof("request") == "unlock_host_info":
+ print("unlock_host_info { <redacted> }")
+ else:
+ print(request)
response_bytes = self._bitbox_protocol.encrypted_query(request.SerializeToString())
response = hww.Response()
response.ParseFromString(response_bytes)
### py/bitbox02/bitbox02/communication/generated/hww_pb2.py
@@ -25,7 +25,7 @@
from . import perform_attestation_pb2 as perform__attestation__pb2
-DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\thww.proto\x12\x14shiftcrypto.bitbox02\x1a\x0c\x63ommon.proto\x1a\x15\x62\x61\x63kup_commands.proto\x1a\x15\x62itbox02_system.proto\x1a\x10\x62itboxsync.proto\x1a\x0f\x62luetooth.proto\x1a\tbtc.proto\x1a\rcardano.proto\x1a\teth.proto\x1a\x0ekeystore.proto\x1a\x0emnemonic.proto\x1a\x0csystem.proto\x1a\x19perform_attestation.proto\"&\n\x05\x45rror\x12\x0c\n\x04\x63ode\x18\x01 \x01(\x05\x12\x0f\n\x07message\x18\x02 \x01(\t\"\t\n\x07Success\"\xc2\x0f\n\x07Request\x12\x41\n\x0b\x64\x65vice_name\x18\x02 \x01(\x0b\x32*.shiftcrypto.bitbox02.SetDeviceNameRequestH\x00\x12I\n\x0f\x64\x65vice_language\x18\x03 \x01(\x0b\x32..shiftcrypto.bitbox02.SetDeviceLanguageRequestH\x00\x12>\n\x0b\x64\x65vice_info\x18\x04 \x01(\x0b\x32\'.shiftcrypto.bitbox02.DeviceInfoRequestH\x00\x12@\n\x0cset_password\x18\x05 \x01(\x0b\x32(.shiftcrypto.bitbox02.SetPasswordRequestH\x00\x12\x42\n\rcreate_backup\x18\x06 \x01(\x0b\x32).shiftcrypto.bitbox02.CreateBackupRequestH\x00\x12\x42\n\rshow_mnemonic\x18\x07 \x01(\x0b\x32).shiftcrypto.bitbox02.ShowMnemonicRequestH\x00\x12\x36\n\x07\x62tc_pub\x18\x08 \x01(\x0b\x32#.shiftcrypto.bitbox02.BTCPubRequestH\x00\x12\x41\n\rbtc_sign_init\x18\t \x01(\x0b\x32(.shiftcrypto.bitbox02.BTCSignInitRequestH\x00\x12\x43\n\x0e\x62tc_sign_input\x18\n \x01(\x0b\x32).shiftcrypto.bitbox02.BTCSignInputRequestH\x00\x12\x45\n\x0f\x62tc_sign_output\x18\x0b \x01(\x0b\x32*.shiftcrypto.bitbox02.BTCSignOutputRequestH\x00\x12O\n\x14insert_remove_sdcard\x18\x0c \x01(\x0b\x32/.shiftcrypto.bitbox02.InsertRemoveSDCardRequestH\x00\x12@\n\x0c\x63heck_sdcard\x18\r \x01(\x0b\x32(.shiftcrypto.bitbox02.CheckSDCardRequestH\x00\x12\x64\n\x1fset_mnemonic_passphrase_enabled\x18\x0e \x01(\x0b\x32\x39.shiftcrypto.bitbox02.SetMnemonicPassphraseEnabledRequestH\x00\x12@\n\x0clist_backups\x18\x0f \x01(\x0b\x32(.shiftcrypto.bitbox02.ListBackupsRequestH\x00\x12\x44\n\x0erestore_backup\x18\x10 \x01(\x0b\x32*.shiftcrypto.bitbox02.RestoreBackupRequestH\x00\x12N\n\x13perform_attestation\x18\x11 \x01(\x0b\x32/.shiftcrypto.bitbox02.PerformAttestationRequestH\x00\x12\x35\n\x06reboot\x18\x12 \x01(\x0b\x32#.shiftcrypto.bitbox02.RebootRequestH\x00\x12@\n\x0c\x63heck_backup\x18\x13 \x01(\x0b\x32(.shiftcrypto.bitbox02.CheckBackupRequestH\x00\x12/\n\x03\x65th\x18\x14 \x01(\x0b\x32 .shiftcrypto.bitbox02.ETHRequestH\x00\x12\x33\n\x05reset\x18\x15 \x01(\x0b\x32\".shiftcrypto.bitbox02.ResetRequestH\x00\x12Q\n\x15restore_from_mnemonic\x18\x16 \x01(\x0b\x32\x30.shiftcrypto.bitbox02.RestoreFromMnemonicRequestH\x00\x12\x43\n\x0b\x66ingerprint\x18\x18 \x01(\x0b\x32,.shiftcrypto.bitbox02.RootFingerprintRequestH\x00\x12/\n\x03\x62tc\x18\x19 \x01(\x0b\x32 .shiftcrypto.bitbox02.BTCRequestH\x00\x12U\n\x17\x65lectrum_encryption_key\x18\x1a \x01(\x0b\x32\x32.shiftcrypto.bitbox02.ElectrumEncryptionKeyRequestH\x00\x12\x37\n\x07\x63\x61rdano\x18\x1b \x01(\x0b\x32$.shiftcrypto.bitbox02.CardanoRequestH\x00\x12\x33\n\x05\x62ip85\x18\x1c \x01(\x0b\x32\".shiftcrypto.bitbox02.BIP85RequestH\x00\x12;\n\tbluetooth\x18\x1d \x01(\x0b\x32&.shiftcrypto.bitbox02.BluetoothRequestH\x00\x12\x46\n\x0f\x63hange_password\x18\x1e \x01(\x0b\x32+.shiftcrypto.bitbox02.ChangePasswordRequestH\x00\x12>\n\x0b\x62itbox_sync\x18\x1f \x01(\x0b\x32\'.shiftcrypto.bitbox02.BitBoxSyncRequestH\x00\x42\t\n\x07requestJ\x04\x08\x01\x10\x02J\x04\x08\x17\x10\x18\"\xbe\x08\n\x08Response\x12\x30\n\x07success\x18\x01 \x01(\x0b\x32\x1d.shiftcrypto.bitbox02.SuccessH\x00\x12,\n\x05\x65rror\x18\x02 \x01(\x0b\x32\x1b.shiftcrypto.bitbox02.ErrorH\x00\x12?\n\x0b\x64\x65vice_info\x18\x04 \x01(\x0b\x32(.shiftcrypto.bitbox02.DeviceInfoResponseH\x00\x12\x30\n\x03pub\x18\x05 \x01(\x0b\x32!.shiftcrypto.bitbox02.PubResponseH\x00\x12\x42\n\rbtc_sign_next\x18\x06 \x01(\x0b\x32).shiftcrypto.bitbox02.BTCSignNextResponseH\x00\x12\x41\n\x0clist_backups\x18\x07 \x01(\x0b\x32).shiftcrypto.bitbox02.ListBackupsResponseH\x00\x12\x41\n\x0c\x63heck_backup\x18\x08 \x01(\x0b\x32).shiftcrypto.bitbox02.CheckBackupResponseH\x00\x12O\n\x13perform_attestation\x18\t \x01(\x0b\x32\x30.shiftcrypto.bitbox02.PerformAttestationResponseH\x00\x12\x41\n\x0c\x63heck_sdcard\x18\n \x01(\x0b\x32).shiftcrypto.bitbox02.CheckSDCardResponseH\x00\x12\x30\n\x03\x65th\x18\x0b \x01(\x0b\x32!.shiftcrypto.bitbox02.ETHResponseH\x00\x12\x44\n\x0b\x66ingerprint\x18\x0c \x01(\x0b\x32-.shiftcrypto.bitbox02.RootFingerprintResponseH\x00\x12\x30\n\x03\x62tc\x18\r \x01(\x0b\x32!.shiftcrypto.bitbox02.BTCResponseH\x00\x12V\n\x17\x65lectrum_encryption_key\x18\x0e \x01(\x0b\x32\x33.shiftcrypto.bitbox02.ElectrumEncryptionKeyResponseH\x00\x12\x38\n\x07\x63\x61rdano\x18\x0f \x01(\x0b\x32%.shiftcrypto.bitbox02.CardanoResponseH\x00\x12\x34\n\x05\x62ip85\x18\x10 \x01(\x0b\x32#.shiftcrypto.bitbox02.BIP85ResponseH\x00\x12<\n\tbluetooth\x18\x11 \x01(\x0b\x32\'.shiftcrypto.bitbox02.BluetoothResponseH\x00\x12?\n\x0b\x62itbox_sync\x18\x12 \x01(\x0b\x32(.shiftcrypto.bitbox02.BitBoxSyncResponseH\x00\x42\n\n\x08responseJ\x04\x08\x03\x10\x04\x62\x06proto3')
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\thww.proto\x12\x14shiftcrypto.bitbox02\x1a\x0c\x63ommon.proto\x1a\x15\x62\x61\x63kup_commands.proto\x1a\x15\x62itbox02_system.proto\x1a\x10\x62itboxsync.proto\x1a\x0f\x62luetooth.proto\x1a\tbtc.proto\x1a\rcardano.proto\x1a\teth.proto\x1a\x0ekeystore.proto\x1a\x0emnemonic.proto\x1a\x0csystem.proto\x1a\x19perform_attestation.proto\"&\n\x05\x45rror\x12\x0c\n\x04\x63ode\x18\x01 \x01(\x05\x12\x0f\n\x07message\x18\x02 \x01(\t\"\t\n\x07Success\"\x8a\x11\n\x07Request\x12\x41\n\x0b\x64\x65vice_name\x18\x02 \x01(\x0b\x32*.shiftcrypto.bitbox02.SetDeviceNameRequestH\x00\x12I\n\x0f\x64\x65vice_language\x18\x03 \x01(\x0b\x32..shiftcrypto.bitbox02.SetDeviceLanguageRequestH\x00\x12>\n\x0b\x64\x65vice_info\x18\x04 \x01(\x0b\x32\'.shiftcrypto.bitbox02.DeviceInfoRequestH\x00\x12@\n\x0cset_password\x18\x05 \x01(\x0b\x32(.shiftcrypto.bitbox02.SetPasswordRequestH\x00\x12\x42\n\rcreate_backup\x18\x06 \x01(\x0b\x32).shiftcrypto.bitbox02.CreateBackupRequestH\x00\x12\x42\n\rshow_mnemonic\x18\x07 \x01(\x0b\x32).shiftcrypto.bitbox02.ShowMnemonicRequestH\x00\x12\x36\n\x07\x62tc_pub\x18\x08 \x01(\x0b\x32#.shiftcrypto.bitbox02.BTCPubRequestH\x00\x12\x41\n\rbtc_sign_init\x18\t \x01(\x0b\x32(.shiftcrypto.bitbox02.BTCSignInitRequestH\x00\x12\x43\n\x0e\x62tc_sign_input\x18\n \x01(\x0b\x32).shiftcrypto.bitbox02.BTCSignInputRequestH\x00\x12\x45\n\x0f\x62tc_sign_output\x18\x0b \x01(\x0b\x32*.shiftcrypto.bitbox02.BTCSignOutputRequestH\x00\x12O\n\x14insert_remove_sdcard\x18\x0c \x01(\x0b\x32/.shiftcrypto.bitbox02.InsertRemoveSDCardRequestH\x00\x12@\n\x0c\x63heck_sdcard\x18\r \x01(\x0b\x32(.shiftcrypto.bitbox02.CheckSDCardRequestH\x00\x12\x64\n\x1fset_mnemonic_passphrase_enabled\x18\x0e \x01(\x0b\x32\x39.shiftcrypto.bitbox02.SetMnemonicPassphraseEnabledRequestH\x00\x12@\n\x0clist_backups\x18\x0f \x01(\x0b\x32(.shiftcrypto.bitbox02.ListBackupsRequestH\x00\x12\x44\n\x0erestore_backup\x18\x10 \x01(\x0b\x32*.shiftcrypto.bitbox02.RestoreBackupRequestH\x00\x12N\n\x13perform_attestation\x18\x11 \x01(\x0b\x32/.shiftcrypto.bitbox02.PerformAttestationRequestH\x00\x12\x35\n\x06reboot\x18\x12 \x01(\x0b\x32#.shiftcrypto.bitbox02.RebootRequestH\x00\x12@\n\x0c\x63heck_backup\x18\x13 \x01(\x0b\x32(.shiftcrypto.bitbox02.CheckBackupRequestH\x00\x12/\n\x03\x65th\x18\x14 \x01(\x0b\x32 .shiftcrypto.bitbox02.ETHRequestH\x00\x12\x33\n\x05reset\x18\x15 \x01(\x0b\x32\".shiftcrypto.bitbox02.ResetRequestH\x00\x12Q\n\x15restore_from_mnemonic\x18\x16 \x01(\x0b\x32\x30.shiftcrypto.bitbox02.RestoreFromMnemonicRequestH\x00\x12\x43\n\x0b\x66ingerprint\x18\x18 \x01(\x0b\x32,.shiftcrypto.bitbox02.RootFingerprintRequestH\x00\x12/\n\x03\x62tc\x18\x19 \x01(\x0b\x32 .shiftcrypto.bitbox02.BTCRequestH\x00\x12U\n\x17\x65lectrum_encryption_key\x18\x1a \x01(\x0b\x32\x32.shiftcrypto.bitbox02.ElectrumEncryptionKeyRequestH\x00\x12\x37\n\x07\x63\x61rdano\x18\x1b \x01(\x0b\x32$.shiftcrypto.bitbox02.CardanoRequestH\x00\x12\x33\n\x05\x62ip85\x18\x1c \x01(\x0b\x32\".shiftcrypto.bitbox02.BIP85RequestH\x00\x12;\n\tbluetooth\x18\x1d \x01(\x0b\x32&.shiftcrypto.bitbox02.BluetoothRequestH\x00\x12\x46\n\x0f\x63hange_password\x18\x1e \x01(\x0b\x32+.shiftcrypto.bitbox02.ChangePasswordRequestH\x00\x12>\n\x0b\x62itbox_sync\x18\x1f \x01(\x0b\x32\'.shiftcrypto.bitbox02.BitBoxSyncRequestH\x00\x12\x35\n\x06unlock\x18 \x01(\x0b\x32#.shiftcrypto.bitbox02.UnlockRequestH\x00\x12\x46\n\x0funlock_continue\x18! \x01(\x0b\x32+.shiftcrypto.bitbox02.UnlockContinueRequestH\x00\x12G\n\x10unlock_host_info\x18\" \x01(\x0b\x32+.shiftcrypto.bitbox02.UnlockHostInfoRequestH\x00\x42\t\n\x07requestJ\x04\x08\x01\x10\x02J\x04\x08\x17\x10\x18\"\xf6\x08\n\x08Response\x12\x30\n\x07success\x18\x01 \x01(\x0b\x32\x1d.shiftcrypto.bitbox02.SuccessH\x00\x12,\n\x05\x65rror\x18\x02 \x01(\x0b\x32\x1b.shiftcrypto.bitbox02.ErrorH\x00\x12?\n\x0b\x64\x65vice_info\x18\x04 \x01(\x0b\x32(.shiftcrypto.bitbox02.DeviceInfoResponseH\x00\x12\x30\n\x03pub\x18\x05 \x01(\x0b\x32!.shiftcrypto.bitbox02.PubResponseH\x00\x12\x42\n\rbtc_sign_next\x18\x06 \x01(\x0b\x32).shiftcrypto.bitbox02.BTCSignNextResponseH\x00\x12\x41\n\x0clist_backups\x18\x07 \x01(\x0b\x32).shiftcrypto.bitbox02.ListBackupsResponseH\x00\x12\x41\n\x0c\x63heck_backup\x18\x08 \x01(\x0b\x32).shiftcrypto.bitbox02.CheckBackupResponseH\x00\x12O\n\x13perform_attestation\x18\t \x01(\x0b\x32\x30.shiftcrypto.bitbox02.PerformAttestationResponseH\x00\x12\x41\n\x0c\x63heck_sdcard\x18\n \x01(\x0b\x32).shiftcrypto.bitbox02.CheckSDCardResponseH\x00\x12\x30\n\x03\x65th\x18\x0b \x01(\x0b\x32!.shiftcrypto.bitbox02.ETHResponseH\x00\x12\x44\n\x0b\x66ingerprint\x18\x0c \x01(\x0b\x32-.shiftcrypto.bitbox02.RootFingerprintResponseH\x00\x12\x30\n\x03\x62tc\x18\r \x01(\x0b\x32!.shiftcrypto.bitbox02.BTCResponseH\x00\x12V\n\x17\x65lectrum_encryption_key\x18\x0e \x01(\x0b\x32\x33.shiftcrypto.bitbox02.ElectrumEncryptionKeyResponseH\x00\x12\x38\n\x07\x63\x61rdano\x18\x0f \x01(\x0b\x32%.shiftcrypto.bitbox02.CardanoResponseH\x00\x12\x34\n\x05\x62ip85\x18\x10 \x01(\x0b\x32#.shiftcrypto.bitbox02.BIP85ResponseH\x00\x12<\n\tbluetooth\x18\x11 \x01(\x0b\x32\'.shiftcrypto.bitbox02.BluetoothResponseH\x00\x12?\n\x0b\x62itbox_sync\x18\x12 \x01(\x0b\x32(.shiftcrypto.bitbox02.BitBoxSyncResponseH\x00\x12\x36\n\x06unlock\x18\x13 \x01(\x0b\x32$.shiftcrypto.bitbox02.UnlockResponseH\x00\x42\n\n\x08responseJ\x04\x08\x03\x10\x04\x62\x06proto3')
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals())
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'hww_pb2', globals())
@@ -37,7 +37,7 @@
_SUCCESS._serialized_start=280
_SUCCESS._serialized_end=289
_REQUEST._serialized_start=292
- _REQUEST._serialized_end=2278
- _RESPONSE._serialized_start=2281
- _RESPONSE._serialized_end=3367
+ _REQUEST._serialized_end=2478
+ _RESPONSE._serialized_start=2481
+ _RESPONSE._serialized_end=3623
# @@protoc_insertion_point(module_scope)
### py/bitbox02/bitbox02/communication/generated/hww_pb2.pyi
@@ -83,6 +83,9 @@ class Request(google.protobuf.message.Message):
BLUETOOTH_FIELD_NUMBER: builtins.int
CHANGE_PASSWORD_FIELD_NUMBER: builtins.int
BITBOX_SYNC_FIELD_NUMBER: builtins.int
+ UNLOCK_FIELD_NUMBER: builtins.int
+ UNLOCK_CONTINUE_FIELD_NUMBER: builtins.int
+ UNLOCK_HOST_INFO_FIELD_NUMBER: builtins.int
@property
def device_name(self) -> bitbox02_system_pb2.SetDeviceNameRequest:
"""removed: RandomNumberRequest random_number = 1;"""
@@ -145,6 +148,12 @@ class Request(google.protobuf.message.Message):
def change_password(self) -> bitbox02_system_pb2.ChangePasswordRequest: ...
@property
def bitbox_sync(self) -> bitboxsync_pb2.BitBoxSyncRequest: ...
+ @property
+ def unlock(self) -> keystore_pb2.UnlockRequest: ...
+ @property
+ def unlock_continue(self) -> keystore_pb2.UnlockContinueRequest: ...
+ @property
+ def unlock_host_info(self) -> keystore_pb2.UnlockHostInfoRequest: ...
def __init__(
self,
*,
@@ -177,10 +186,13 @@ class Request(google.protobuf.message.Message):
bluetooth: bluetooth_pb2.BluetoothRequest | None = ...,
change_password: bitbox02_system_pb2.ChangePasswordRequest | None = ...,
bitbox_sync: bitboxsync_pb2.BitBoxSyncRequest | None = ...,
+ unlock: keystore_pb2.UnlockRequest | None = ...,
+ unlock_continue: keystore_pb2.UnlockContinueRequest | None = ...,
+ unlock_host_info: keystore_pb2.UnlockHostInfoRequest | None = ...,
) -> None: ...
- def HasField(self, field_name: typing.Literal["bip85", b"bip85", "bitbox_sync", b"bitbox_sync", "bluetooth", b"bluetooth", "btc", b"btc", "btc_pub", b"btc_pub", "btc_sign_init", b"btc_sign_init", "btc_sign_input", b"btc_sign_input", "btc_sign_output", b"btc_sign_output", "cardano", b"cardano", "change_password", b"change_password", "check_backup", b"check_backup", "check_sdcard", b"check_sdcard", "create_backup", b"create_backup", "device_info", b"device_info", "device_language", b"device_language", "device_name", b"device_name", "electrum_encryption_key", b"electrum_encryption_key", "eth", b"eth", "fingerprint", b"fingerprint", "insert_remove_sdcard", b"insert_remove_sdcard", "list_backups", b"list_backups", "perform_attestation", b"perform_attestation", "reboot", b"reboot", "request", b"request", "reset", b"reset", "restore_backup", b"restore_backup", "restore_from_mnemonic", b"restore_from_mnemonic", "set_mnemonic_passphrase_enabled", b"set_mnemonic_passphrase_enabled", "set_password", b"set_password", "show_mnemonic", b"show_mnemonic"]) -> builtins.bool: ...
- def ClearField(self, field_name: typing.Literal["bip85", b"bip85", "bitbox_sync", b"bitbox_sync", "bluetooth", b"bluetooth", "btc", b"btc", "btc_pub", b"btc_pub", "btc_sign_init", b"btc_sign_init", "btc_sign_input", b"btc_sign_input", "btc_sign_output", b"btc_sign_output", "cardano", b"cardano", "change_password", b"change_password", "check_backup", b"check_backup", "check_sdcard", b"check_sdcard", "create_backup", b"create_backup", "device_info", b"device_info", "device_language", b"device_language", "device_name", b"device_name", "electrum_encryption_key", b"electrum_encryption_key", "eth", b"eth", "fingerprint", b"fingerprint", "insert_remove_sdcard", b"insert_remove_sdcard", "list_backups", b"list_backups", "perform_attestation", b"perform_attestation", "reboot", b"reboot", "request", b"request", "reset", b"reset", "restore_backup", b"restore_backup", "restore_from_mnemonic", b"restore_from_mnemonic", "set_mnemonic_passphrase_enabled", b"set_mnemonic_passphrase_enabled", "set_password", b"set_password", "show_mnemonic", b"show_mnemonic"]) -> None: ...
- def WhichOneof(self, oneof_group: typing.Literal["request", b"request"]) -> typing.Literal["device_name", "device_language", "device_info", "set_password", "create_backup", "show_mnemonic", "btc_pub", "btc_sign_init", "btc_sign_input", "btc_sign_output", "insert_remove_sdcard", "check_sdcard", "set_mnemonic_passphrase_enabled", "list_backups", "restore_backup", "perform_attestation", "reboot", "check_backup", "eth", "reset", "restore_from_mnemonic", "fingerprint", "btc", "electrum_encryption_key", "cardano", "bip85", "bluetooth", "change_password", "bitbox_sync"] | None: ...
+ def HasField(self, field_name: typing.Literal["bip85", b"bip85", "bitbox_sync", b"bitbox_sync", "bluetooth", b"bluetooth", "btc", b"btc", "btc_pub", b"btc_pub", "btc_sign_init", b"btc_sign_init", "btc_sign_input", b"btc_sign_input", "btc_sign_output", b"btc_sign_output", "cardano", b"cardano", "change_password", b"change_password", "check_backup", b"check_backup", "check_sdcard", b"check_sdcard", "create_backup", b"create_backup", "device_info", b"device_info", "device_language", b"device_language", "device_name", b"device_name", "electrum_encryption_key", b"electrum_encryption_key", "eth", b"eth", "fingerprint", b"fingerprint", "insert_remove_sdcard", b"insert_remove_sdcard", "list_backups", b"list_backups", "perform_attestation", b"perform_attestation", "reboot", b"reboot", "request", b"request", "reset", b"reset", "restore_backup", b"restore_backup", "restore_from_mnemonic", b"restore_from_mnemonic", "set_mnemonic_passphrase_enabled", b"set_mnemonic_passphrase_enabled", "set_password", b"set_password", "show_mnemonic", b"show_mnemonic", "unlock", b"unlock", "unlock_continue", b"unlock_continue", "unlock_host_info", b"unlock_host_info"]) -> builtins.bool: ...
+ def ClearField(self, field_name: typing.Literal["bip85", b"bip85", "bitbox_sync", b"bitbox_sync", "bluetooth", b"bluetooth", "btc", b"btc", "btc_pub", b"btc_pub", "btc_sign_init", b"btc_sign_init", "btc_sign_input", b"btc_sign_input", "btc_sign_output", b"btc_sign_output", "cardano", b"cardano", "change_password", b"change_password", "check_backup", b"check_backup", "check_sdcard", b"check_sdcard", "create_backup", b"create_backup", "device_info", b"device_info", "device_language", b"device_language", "device_name", b"device_name", "electrum_encryption_key", b"electrum_encryption_key", "eth", b"eth", "fingerprint", b"fingerprint", "insert_remove_sdcard", b"insert_remove_sdcard", "list_backups", b"list_backups", "perform_attestation", b"perform_attestation", "reboot", b"reboot", "request", b"request", "reset", b"reset", "restore_backup", b"restore_backup", "restore_from_mnemonic", b"restore_from_mnemonic", "set_mnemonic_passphrase_enabled", b"set_mnemonic_passphrase_enabled", "set_password", b"set_password", "show_mnemonic", b"show_mnemonic", "unlock", b"unlock", "unlock_continue", b"unlock_continue", "unlock_host_info", b"unlock_host_info"]) -> None: ...
+ def WhichOneof(self, oneof_group: typing.Literal["request", b"request"]) -> typing.Literal["device_name", "device_language", "device_info", "set_password", "create_backup", "show_mnemonic", "btc_pub", "btc_sign_init", "btc_sign_input", "btc_sign_output", "insert_remove_sdcard", "check_sdcard", "set_mnemonic_passphrase_enabled", "list_backups", "restore_backup", "perform_attestation", "reboot", "check_backup", "eth", "reset", "restore_from_mnemonic", "fingerprint", "btc", "electrum_encryption_key", "cardano", "bip85", "bluetooth", "change_password", "bitbox_sync", "unlock", "unlock_continue", "unlock_host_info"] | None: ...
global___Request = Request
@@ -205,6 +217,7 @@ class Response(google.protobuf.message.Message):
BIP85_FIELD_NUMBER: builtins.int
BLUETOOTH_FIELD_NUMBER: builtins.int
BITBOX_SYNC_FIELD_NUMBER: builtins.int
+ UNLOCK_FIELD_NUMBER: builtins.int
@property
def success(self) -> global___Success: ...
@property
@@ -241,6 +254,8 @@ class Response(google.protobuf.message.Message):
def bluetooth(self) -> bluetooth_pb2.BluetoothResponse: ...
@property
def bitbox_sync(self) -> bitboxsync_pb2.BitBoxSyncResponse: ...
+ @property
+ def unlock(self) -> keystore_pb2.UnlockResponse: ...
def __init__(
self,
*,
@@ -261,9 +276,10 @@ class Response(google.protobuf.message.Message):
bip85: keystore_pb2.BIP85Response | None = ...,
bluetooth: bluetooth_pb2.BluetoothResponse | None = ...,
bitbox_sync: bitboxsync_pb2.BitBoxSyncResponse | None = ...,
+ unlock: keystore_pb2.UnlockResponse | None = ...,
) -> None: ...
- def HasField(self, field_name: typing.Literal["bip85", b"bip85", "bitbox_sync", b"bitbox_sync", "bluetooth", b"bluetooth", "btc", b"btc", "btc_sign_next", b"btc_sign_next", "cardano", b"cardano", "check_backup", b"check_backup", "check_sdcard", b"check_sdcard", "device_info", b"device_info", "electrum_encryption_key", b"electrum_encryption_key", "error", b"error", "eth", b"eth", "fingerprint", b"fingerprint", "list_backups", b"list_backups", "perform_attestation", b"perform_attestation", "pub", b"pub", "response", b"response", "success", b"success"]) -> builtins.bool: ...
- def ClearField(self, field_name: typing.Literal["bip85", b"bip85", "bitbox_sync", b"bitbox_sync", "bluetooth", b"bluetooth", "btc", b"btc", "btc_sign_next", b"btc_sign_next", "cardano", b"cardano", "check_backup", b"check_backup", "check_sdcard", b"check_sdcard", "device_info", b"device_info", "electrum_encryption_key", b"electrum_encryption_key", "error", b"error", "eth", b"eth", "fingerprint", b"fingerprint", "list_backups", b"list_backups", "perform_attestation", b"perform_attestation", "pub", b"pub", "response", b"response", "success", b"success"]) -> None: ...
- def WhichOneof(self, oneof_group: typing.Literal["response", b"response"]) -> typing.Literal["success", "error", "device_info", "pub", "btc_sign_next", "list_backups", "check_backup", "perform_attestation", "check_sdcard", "eth", "fingerprint", "btc", "electrum_encryption_key", "cardano", "bip85", "bluetooth", "bitbox_sync"] | None: ...
+ def HasField(self, field_name: typing.Literal["bip85", b"bip85", "bitbox_sync", b"bitbox_sync", "bluetooth", b"bluetooth", "btc", b"btc", "btc_sign_next", b"btc_sign_next", "cardano", b"cardano", "check_backup", b"check_backup", "check_sdcard", b"check_sdcard", "device_info", b"device_info", "electrum_encryption_key", b"electrum_encryption_key", "error", b"error", "eth", b"eth", "fingerprint", b"fingerprint", "list_backups", b"list_backups", "perform_attestation", b"perform_attestation", "pub", b"pub", "response", b"response", "success", b"success", "unlock", b"unlock"]) -> builtins.bool: ...
+ def ClearField(self, field_name: typing.Literal["bip85", b"bip85", "bitbox_sync", b"bitbox_sync", "bluetooth", b"bluetooth", "btc", b"btc", "btc_sign_next", b"btc_sign_next", "cardano", b"cardano", "check_backup", b"check_backup", "check_sdcard", b"check_sdcard", "device_info", b"device_info", "electrum_encryption_key", b"electrum_encryption_key", "error", b"error", "eth", b"eth", "fingerprint", b"fingerprint", "list_backups", b"list_backups", "perform_attestation", b"perform_attestation", "pub", b"pub", "response", b"response", "success", b"success", "unlock", b"unlock"]) -> None: ...
+ def WhichOneof(self, oneof_group: typing.Literal["response", b"response"]) -> typing.Literal["success", "error", "device_info", "pub", "btc_sign_next", "list_backups", "check_backup", "perform_attestation", "check_sdcard", "eth", "fingerprint", "btc", "electrum_encryption_key", "cardano", "bip85", "bluetooth", "bitbox_sync", "unlock"] | None: ...
global___Response = Response
### py/bitbox02/bitbox02/communication/generated/keystore_pb2.py
@@ -14,21 +14,31 @@
from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2
-DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0ekeystore.proto\x12\x14shiftcrypto.bitbox02\x1a\x1bgoogle/protobuf/empty.proto\"/\n\x1c\x45lectrumEncryptionKeyRequest\x12\x0f\n\x07keypath\x18\x01 \x03(\r\",\n\x1d\x45lectrumEncryptionKeyResponse\x12\x0b\n\x03key\x18\x01 \x01(\t\"\x97\x01\n\x0c\x42IP85Request\x12\'\n\x05\x62ip39\x18\x01 \x01(\x0b\x32\x16.google.protobuf.EmptyH\x00\x12\x36\n\x02ln\x18\x02 \x01(\x0b\x32(.shiftcrypto.bitbox02.BIP85Request.AppLnH\x00\x1a\x1f\n\x05\x41ppLn\x12\x16\n\x0e\x61\x63\x63ount_number\x18\x01 \x01(\rB\x05\n\x03\x61pp\"M\n\rBIP85Response\x12\'\n\x05\x62ip39\x18\x01 \x01(\x0b\x32\x16.google.protobuf.EmptyH\x00\x12\x0c\n\x02ln\x18\x02 \x01(\x0cH\x00\x42\x05\n\x03\x61ppb\x06proto3')
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0ekeystore.proto\x12\x14shiftcrypto.bitbox02\x1a\x1bgoogle/protobuf/empty.proto\"\x0f\n\rUnlockRequest\"3\n\x15UnlockContinueRequest\x12\x1a\n\x12request_host_entry\x18\x01 \x01(\x08\"?\n\x15UnlockHostInfoRequest\x12\x17\n\npassphrase\x18\x01 \x01(\tH\x00\x88\x01\x01\x42\r\n\x0b_passphrase\"\xa4\x01\n\x0eUnlockResponse\x12\x39\n\x05state\x18\x01 \x01(\x0e\x32*.shiftcrypto.bitbox02.UnlockResponse.State\"W\n\x05State\x12\x16\n\x12PASSPHRASE_PENDING\x10\x00\x12\x14\n\x10HOST_ENTRY_READY\x10\x01\x12\x16\n\x12PASSPHRASE_ENTERED\x10\x02\x12\x08\n\x04\x44ONE\x10\x03\"/\n\x1c\x45lectrumEncryptionKeyRequest\x12\x0f\n\x07keypath\x18\x01 \x03(\r\",\n\x1d\x45lectrumEncryptionKeyResponse\x12\x0b\n\x03key\x18\x01 \x01(\t\"\x97\x01\n\x0c\x42IP85Request\x12\'\n\x05\x62ip39\x18\x01 \x01(\x0b\x32\x16.google.protobuf.EmptyH\x00\x12\x36\n\x02ln\x18\x02 \x01(\x0b\x32(.shiftcrypto.bitbox02.BIP85Request.AppLnH\x00\x1a\x1f\n\x05\x41ppLn\x12\x16\n\x0e\x61\x63\x63ount_number\x18\x01 \x01(\rB\x05\n\x03\x61pp\"M\n\rBIP85Response\x12\'\n\x05\x62ip39\x18\x01 \x01(\x0b\x32\x16.google.protobuf.EmptyH\x00\x12\x0c\n\x02ln\x18\x02 \x01(\x0cH\x00\x42\x05\n\x03\x61ppb\x06proto3')
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals())
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'keystore_pb2', globals())
if _descriptor._USE_C_DESCRIPTORS == False:
DESCRIPTOR._options = None
- _ELECTRUMENCRYPTIONKEYREQUEST._serialized_start=69
- _ELECTRUMENCRYPTIONKEYREQUEST._serialized_end=116
- _ELECTRUMENCRYPTIONKEYRESPONSE._serialized_start=118
- _ELECTRUMENCRYPTIONKEYRESPONSE._serialized_end=162
- _BIP85REQUEST._serialized_start=165
- _BIP85REQUEST._serialized_end=316
- _BIP85REQUEST_APPLN._serialized_start=278
- _BIP85REQUEST_APPLN._serialized_end=309
- _BIP85RESPONSE._serialized_start=318
- _BIP85RESPONSE._serialized_end=395
+ _UNLOCKREQUEST._serialized_start=69
+ _UNLOCKREQUEST._serialized_end=84
+ _UNLOCKCONTINUEREQUEST._serialized_start=86
+ _UNLOCKCONTINUEREQUEST._serialized_end=137
+ _UNLOCKHOSTINFOREQUEST._serialized_start=139
+ _UNLOCKHOSTINFOREQUEST._serialized_end=202
+ _UNLOCKRESPONSE._serialized_start=205
+ _UNLOCKRESPONSE._serialized_end=369
+ _UNLOCKRESPONSE_STATE._serialized_start=282
+ _UNLOCKRESPONSE_STATE._serialized_end=369
+ _ELECTRUMENCRYPTIONKEYREQUEST._serialized_start=371
+ _ELECTRUMENCRYPTIONKEYREQUEST._serialized_end=418
+ _ELECTRUMENCRYPTIONKEYRESPONSE._serialized_start=420
+ _ELECTRUMENCRYPTIONKEYRESPONSE._serialized_end=464
+ _BIP85REQUEST._serialized_start=467
+ _BIP85REQUEST._serialized_end=618
+ _BIP85REQUEST_APPLN._serialized_start=580
+ _BIP85REQUEST_APPLN._serialized_end=611
+ _BIP85RESPONSE._serialized_start=620
+ _BIP85RESPONSE._serialized_end=697
# @@protoc_insertion_point(module_scope)
### py/bitbox02/bitbox02/communication/generated/keystore_pb2.pyi
@@ -10,11 +10,112 @@ import collections.abc
import google.protobuf.descriptor
import google.protobuf.empty_pb2
import google.protobuf.internal.containers
+import google.protobuf.internal.enum_type_wrapper
import google.protobuf.message
+import sys
import typing
+if sys.version_info >= (3, 10):
+ import typing as typing_extensions
+else:
+ import typing_extensions
+
DESCRIPTOR: google.protobuf.descriptor.FileDescriptor
+@typing.final
+class UnlockRequest(google.protobuf.message.Message):
+ """Unlock inside the paired Noise channel (since v9.28.0). Uninitialized and already unlocked
+ devices return DONE immediately and are left unchanged. Continuations are only valid within
+ this workflow.
+ If the passphrase feature is enabled, device entry returns PASSPHRASE_PENDING. The host
+ polls with UnlockContinueRequest until entry completes or it requests host entry.
+ """
+
+ DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+ def __init__(
+ self,
+ ) -> None: ...
+
+global___UnlockRequest = UnlockRequest
+
+@typing.final
+class UnlockContinueRequest(google.protobuf.message.Message):
+ DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+ REQUEST_HOST_ENTRY_FIELD_NUMBER: builtins.int
+ request_host_entry: builtins.bool
+ """False polls device entry; true interrupts it to ask for host-entry consent.
+ After PASSPHRASE_ENTERED, continue to await confirmation and unlock;
+ host entry is ignored.
+ """
+ def __init__(
+ self,
+ *,
+ request_host_entry: builtins.bool = ...,
+ ) -> None: ...
+ def ClearField(self, field_name: typing.Literal["request_host_entry", b"request_host_entry"]) -> None: ...
+
+global___UnlockContinueRequest = UnlockContinueRequest
+
+@typing.final
+class UnlockHostInfoRequest(google.protobuf.message.Message):
+ DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+ PASSPHRASE_FIELD_NUMBER: builtins.int
+ passphrase: builtins.str
+ """Only sent after HOST_ENTRY_READY. Absent cancels host input and restarts device entry;
+ the empty string submits the empty passphrase. The device confirms the actual value.
+ """
+ def __init__(
+ self,
+ *,
+ passphrase: builtins.str | None = ...,
+ ) -> None: ...
+ def HasField(self, field_name: typing.Literal["_passphrase", b"_passphrase", "passphrase", b"passphrase"]) -> builtins.bool: ...
+ def ClearField(self, field_name: typing.Literal["_passphrase", b"_passphrase", "passphrase", b"passphrase"]) -> None: ...
+ def WhichOneof(self, oneof_group: typing.Literal["_passphrase", b"_passphrase"]) -> typing.Literal["passphrase"] | None: ...
+
+global___UnlockHostInfoRequest = UnlockHostInfoRequest
+
+@typing.final
+class UnlockResponse(google.protobuf.message.Message):
+ DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+ class _State:
+ ValueType = typing.NewType("ValueType", builtins.int)
+ V: typing_extensions.TypeAlias = ValueType
+
+ class _StateEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[UnlockResponse._State.ValueType], builtins.type):
+ DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor
+ PASSPHRASE_PENDING: UnlockResponse._State.ValueType # 0
+ HOST_ENTRY_READY: UnlockResponse._State.ValueType # 1
+ PASSPHRASE_ENTERED: UnlockResponse._State.ValueType # 2
+ """Passphrase entry on the device finished; confirmation may still be pending.
+ Withdraw host entry and send UnlockContinue to await completion.
+ """
+ DONE: UnlockResponse._State.ValueType # 3
+
+ class State(_State, metaclass=_StateEnumTypeWrapper): ...
+ PASSPHRASE_PENDING: UnlockResponse.State.ValueType # 0
+ HOST_ENTRY_READY: UnlockResponse.State.ValueType # 1
+ PASSPHRASE_ENTERED: UnlockResponse.State.ValueType # 2
+ """Passphrase entry on the device finished; confirmation may still be pending.
+ Withdraw host entry and send UnlockContinue to await completion.
+ """
+ DONE: UnlockResponse.State.ValueType # 3
+
+ STATE_FIELD_NUMBER: builtins.int
+ state: global___UnlockResponse.State.ValueType
+ def __init__(
+ self,
+ *,
+ state: global___UnlockResponse.State.ValueType = ...,
+ ) -> None: ...
+ def ClearField(self, field_name: typing.Literal["state", b"state"]) -> None: ...
+
+global___UnlockResponse = UnlockResponse
+
@typing.final
class ElectrumEncryptionKeyRequest(google.protobuf.message.Message):
DESCRIPTOR: google.protobuf.descriptor.Descriptor
### py/bitbox02/setup.py
@@ -1,6 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
"""BitBox python package"""
+
import os.path
import re
import setuptools
@@ -31,11 +32,11 @@ def find_version() -> str:
long_description=read("README.md"),
long_description_content_type="text/markdown",
url="https://github.com/BitBoxSwiss/bitbox02-firmware",
- python_requires=">=3.6",
+ python_requires=">=3.7",
classifiers=[
"Intended Audience :: Developers",
"License :: OSI Approved :: Apache Software License",
- "Programming Language :: Python :: 3.6",
+ "Programming Language :: Python :: 3.7",
],
keywords="digitalbitbox BitBoxSwiss bitbox bitbox02 bitcoin litecoin ethereum erc20 u2f",
# https://mypy.readthedocs.io/en/stable/installed_packages.html#installed-packages
@@ -67,7 +68,7 @@ def find_version() -> str:
"protobuf>=3.20",
"ecdsa>=0.14",
"semver>=2.8.1",
- # Needed as long as we support python < 3.7
+ # Backports TypedDict and Protocol for Python 3.7.
"typing_extensions>=3.7.4",
"base58>=2.0.0",
],
### py/bitbox02/tests/test_session.py
@@ -10,6 +10,8 @@
from bitbox02.communication import bitbox_api_protocol as protocol
from bitbox02.communication.communication import TransportLayer
from bitbox02.communication.devices import BITBOX02MULTI
+from bitbox02.communication.generated import hww_pb2 as hww
+from bitbox02.communication.generated import keystore_pb2 as keystore
class TestSession(unittest.TestCase):
@@ -33,7 +35,7 @@ def query(data: bytes, endpoint: int, cid: int) -> bytes:
events.append(data)
if data == protocol.HwwRequestCode.REQ_INFO:
encoded = version.encode("ascii")
- return bytes([len(encoded)]) + encoded + bytes(4)
+ return bytes([len(encoded)]) + encoded + b"\x00\x00\x00\x01"
self.assertEqual(data, protocol.HwwRequestCode.REQ_RESET)
return protocol.HwwResponseCode.RSP_ACK
@@ -59,14 +61,48 @@ def query(data: bytes, endpoint: int, cid: int) -> bytes:
protocol.BitBoxProtocolV7,
"noise_connect",
side_effect=lambda _config: events.append("noise"),
+ ), mock.patch.object(
+ protocol.BitBoxCommonAPI,
+ "_unlock",
+ side_effect=lambda _config: events.append("encrypted unlock"),
):
protocol.BitBoxCommonAPI(transport, device_info, protocol.BitBoxNoiseConfig())
expected = [protocol.HwwRequestCode.REQ_INFO] if discover_version else []
if version != "v9.27.2":
expected.append(protocol.HwwRequestCode.REQ_RESET)
- expected.extend(["attestation", "unlock", "noise"])
+ if version == "v9.27.2":
+ expected.extend(["attestation", "unlock", "noise"])
+ else:
+ expected.extend(["attestation", "noise", "encrypted unlock"])
self.assertEqual(events, expected)
+ def test_unlock_without_initialization_query(self) -> None:
+ """An immediate DONE completes connection setup without querying initialization."""
+ # pylint: disable=no-member
+ transport = mock.Mock(spec=TransportLayer)
+ transport.query.side_effect = [protocol.HwwResponseCode.RSP_ACK]
+ enter = mock.Mock()
+ with mock.patch.object(protocol.BitBoxCommonAPI, "_perform_attestation"), mock.patch.object(
+ protocol.BitBoxProtocolV7, "noise_connect"
+ ), mock.patch.object(
+ protocol.BitBoxCommonAPI,
+ "_msg_query",
+ return_value=hww.Response(
+ unlock=keystore.UnlockResponse(state=keystore.UnlockResponse.DONE)
+ ),
+ ) as query:
+ protocol.BitBoxCommonAPI(
+ transport,
+ {"serial_number": "v9.28.0", "product_string": BITBOX02MULTI},
+ protocol.BitBoxNoiseConfig(),
+ protocol.BitBoxConfig(enter_mnemonic_passphrase=enter),
+ )
+ query.assert_called_once_with(
+ hww.Request(unlock=keystore.UnlockRequest()), expected_response="unlock"
+ )
+ self.assertEqual(transport.query.call_count, 1)
+ enter.assert_not_called()
+
def test_reset_session_busy(self) -> None:
"""Wait for another owner of the UI before continuing startup."""
transport = mock.Mock(spec=TransportLayer)
### py/bitbox02/tests/test_unlock.py
@@ -0,0 +1,432 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unlock protocol and callback lifetimes, without a physical device."""
+
+# pylint: disable=protected-access,no-member
+
+import io
+import queue
+import threading
+import unittest
+from contextlib import redirect_stdout
+from unittest import mock
+
+from bitbox02.communication import bitbox_api_protocol as protocol
+from bitbox02.communication.generated import hww_pb2 as hww
+from bitbox02.communication.generated import keystore_pb2 as keystore
+
+
+def reply(state):
+ return hww.Response(unlock=keystore.UnlockResponse(state=state))
+
+
+PENDING = reply(keystore.UnlockResponse.PASSPHRASE_PENDING)
+READY = reply(keystore.UnlockResponse.HOST_ENTRY_READY)
+ENTERED = reply(keystore.UnlockResponse.PASSPHRASE_ENTERED)
+DONE = reply(keystore.UnlockResponse.DONE)
+
+
+class TestUnlock(unittest.TestCase):
+ """The synchronous worker owns protocol I/O; UI clicks only enqueue a request."""
+
+ def setUp(self):
+ self.api = object.__new__(protocol.BitBoxCommonAPI)
+ self.api.debug = False
+ self.sent = []
+
+ def responses(self, responses):
+ sequence = iter(responses)
+
+ def query(request, expected_response):
+ self.assertEqual(expected_response, "unlock")
+ self.sent.append(request)
+ response = next(sequence)
+ if isinstance(response, Exception):
+ raise response
+ return response
+
+ self.api._msg_query = mock.Mock(side_effect=query)
+
+ def test_device_entry_notifies_availability_once(self):
+ self.responses([PENDING, PENDING, PENDING, ENTERED, DONE])
+ events = []
+ input_callback = mock.Mock()
+ config = protocol.BitBoxConfig(
+ on_host_passphrase_available=events.append,
+ enter_mnemonic_passphrase=input_callback,
+ )
+ self.api._unlock(config)
+ self.assertEqual(len(events), 2)
+ self.assertTrue(callable(events[0]))
+ self.assertEqual(events[1:], [None])
+ input_callback.assert_not_called()
+ self.assertTrue(self.sent[0].HasField("unlock"))
+ self.assertTrue(
+ all(not request.unlock_continue.request_host_entry for request in self.sent[1:])
+ )
+ sent_count = len(self.sent)
+ events[0]()
+ self.assertEqual(len(self.sent), sent_count)
+
+ def test_device_entry_withdraws_before_confirmation(self):
+ # Callbacks are consumed synchronously within each subtest.
+ # pylint: disable=cell-var-from-loop
+ for click in [False, True]:
+ with self.subTest(click=click):
+ events = []
+ entered = mock.Mock()
+ self.sent.clear()
+
+ def available(button):
+ events.append(button)
+ if button is not None and click:
+ button()
+
+ def query(request, expected_response):
+ self.assertEqual(expected_response, "unlock")
+ self.sent.append(request)
+ if len(self.sent) == 1:
+ return PENDING
+ if len(self.sent) == 2:
+ self.assertEqual(request.unlock_continue.request_host_entry, click)
+ return ENTERED
+ self.assertEqual(len(self.sent), 3)
+ self.assertEqual(events[1:], [None])
+ # A delayed UI event cannot switch to host entry during confirmation.
+ events[0]()
+ self.assertFalse(request.unlock_continue.request_host_entry)
+ entered.assert_not_called()
+ return DONE
+
+ self.api._msg_query = mock.Mock(side_effect=query)
+ self.api._unlock(protocol.BitBoxConfig(available, entered))
+ self.assertEqual(len(self.sent), 3)
+ self.assertEqual(len(events), 2)
+
+ def test_device_confirmation_rejection_offers_fresh_button(self):
+ self.responses([PENDING, ENTERED, PENDING, ENTERED, DONE])
+ buttons = []
+ events = []
+
+ def available(button):
+ events.append(button)
+ if button is not None:
+ buttons.append(button)
+ if len(buttons) == 2:
+ buttons[0]()
+
+ entered = mock.Mock()
+ self.api._unlock(protocol.BitBoxConfig(available, entered))
+ self.assertIsNot(buttons[0], buttons[1])
+ self.assertEqual(events, [buttons[0], None, buttons[1], None])
+ self.assertTrue(
+ all(not request.unlock_continue.request_host_entry for request in self.sent[1:])
+ )
+ entered.assert_not_called()
+
+ def test_host_input_after_consent_and_empty_is_submission(self):
+ # Callbacks are consumed synchronously within each subtest.
+ # pylint: disable=cell-var-from-loop
+ for value in [" exact value ", ""]:
+ with self.subTest(value=value):
+ self.sent.clear()
+ self.responses([PENDING, READY, DONE])
+ events = []
+
+ def available(button):
+ events.append(button)
+ if button is not None:
+ count = len(self.sent)
+ button()
+ button() # Coalesce double clicks without I/O here.
+ self.assertEqual(len(self.sent), count)
+
+ def enter():
+ self.assertEqual(events[-1], None)
+ self.assertTrue(self.sent[-1].unlock_continue.request_host_entry)
+ return value
+
+ self.api._unlock(protocol.BitBoxConfig(available, enter))
+ self.assertEqual(events[1:], [None])
+ request = self.sent[-1].unlock_host_info
+ self.assertTrue(request.HasField("passphrase"))
+ self.assertEqual(request.passphrase, value)
+ self.assertEqual(len(self.sent), 3)
+
+ def test_automatic_host_entry_once_per_unlock(self):
+ def enter():
+ self.assertEqual(len(self.sent), 2)
+ self.assertTrue(self.sent[-1].unlock_continue.request_host_entry)
+ return "value"
+
+ input_callback = mock.Mock(side_effect=enter)
+ config = protocol.BitBoxConfig(enter_mnemonic_passphrase=input_callback)
+ # Reusing the same config must still request host entry on the next unlock.
+ for _ in range(2):
+ self.sent.clear()
+ input_callback.reset_mock()
+ self.responses([PENDING, READY, DONE])
+ with mock.patch.object(protocol.time, "sleep") as sleep:
+ self.api._unlock(config)
+ sleep.assert_not_called()
+ input_callback.assert_called_once_with()
+ self.assertEqual(
+ [request.WhichOneof("request") for request in self.sent],
+ ["unlock", "unlock_continue", "unlock_host_info"],
+ )
+ self.assertEqual(self.sent[-1].unlock_host_info.passphrase, "value")
+
+ def test_automatic_host_entry_falls_back_to_device(self):
+ for after_request, value in [
+ ([PENDING], None), # Device consent rejected.
+ ([READY, PENDING], None), # Host input cancelled.
+ ([READY, PENDING], "value"), # Passphrase confirmation rejected.
+ ([READY, PENDING], "~"), # Unsupported characters.
+ ([READY, PENDING], "a" * 150), # Passphrase too long.
+ ]:
+ with self.subTest(after_request=after_request, value=value):
+ self.sent.clear()
+ self.responses([PENDING] + after_request + [PENDING, ENTERED, DONE])
+ enter = mock.Mock(return_value=value)
+ with mock.patch.object(protocol.time, "sleep") as sleep:
+ self.api._unlock(protocol.BitBoxConfig(enter_mnemonic_passphrase=enter))
+ self.assertEqual(sleep.call_count, 2)
+ self.assertEqual(
+ [
+ request.unlock_continue.request_host_entry
+ for request in self.sent
+ if request.HasField("unlock_continue")
+ ],
+ [True, False, False, False],
+ )
+ if READY in after_request:
+ enter.assert_called_once_with()
+ self.assertEqual(
+ self.sent[2].unlock_host_info.HasField("passphrase"), value is not None
+ )
+ if value is not None:
+ self.assertEqual(self.sent[2].unlock_host_info.passphrase, value)
+ else:
+ enter.assert_not_called()
+
+ def test_automatic_host_entry_loses_to_device_completion(self):
+ # Device completion wins the automatic request; rejecting confirmation must not
+ # trigger another automatic request when device entry restarts.
+ self.responses([PENDING, ENTERED, PENDING, ENTERED, DONE])
+ enter = mock.Mock()
+ with mock.patch.object(protocol.time, "sleep"):
+ self.api._unlock(protocol.BitBoxConfig(enter_mnemonic_passphrase=enter))
+ enter.assert_not_called()
+ self.assertEqual(
+ [request.unlock_continue.request_host_entry for request in self.sent[1:]],
+ [True, False, False, False],
+ )
+
+ def test_automatic_host_entry_requires_consent_response(self):
+ for responses in [[READY], [PENDING, PENDING, READY]]:
+ with self.subTest(responses=responses):
+ self.responses(responses)
+ enter = mock.Mock()
+ with mock.patch.object(protocol.time, "sleep"):
+ with self.assertRaisesRegex(Exception, "Unexpected unlock phase"):
+ self.api._unlock(protocol.BitBoxConfig(enter_mnemonic_passphrase=enter))
+ enter.assert_not_called()
+
+ def test_cancel_and_rejection_offer_fresh_button(self):
+ # Callbacks are consumed synchronously within each subtest.
+ # pylint: disable=cell-var-from-loop
+ for after_click, value in [
+ ([PENDING], None),
+ ([READY, PENDING], None),
+ ([READY, PENDING], "~"), # Unsupported host input also restarts device entry.
+ ]:
+ with self.subTest(after_click=after_click, value=value):
+ self.sent.clear()
+ self.responses([PENDING] + after_click + [ENTERED, DONE])
+ buttons = []
+ events = []
+
+ def available(button):
+ events.append(button)
+ if button is not None:
+ buttons.append(button)
+ if len(buttons) == 1:
+ button()
+ else:
+ buttons[0]() # Old click must not request consent in the new phase.
+
+ enter = mock.Mock(return_value=value)
+ self.api._unlock(protocol.BitBoxConfig(available, enter))
+ self.assertEqual(len(buttons), 2)
+ self.assertIsNot(buttons[0], buttons[1])
+ self.assertEqual(events, [buttons[0], None, buttons[1], None])
+ self.assertTrue(
+ all(
+ not request.unlock_continue.request_host_entry for request in self.sent[-2:]
+ )
+ )
+ if READY in after_click:
+ enter.assert_called_once_with()
+ self.assertEqual(
+ self.sent[2].unlock_host_info.HasField("passphrase"), value is not None
+ )
+ if value is not None:
+ self.assertEqual(self.sent[2].unlock_host_info.passphrase, value)
+ else:
+ enter.assert_not_called()
+
+ def test_stale_button_cannot_affect_later_attempt(self):
+ old = []
+ self.responses([PENDING, ENTERED, DONE])
+ self.api._unlock(protocol.BitBoxConfig(old.append, lambda: "unused"))
+ self.sent.clear()
+ self.responses([PENDING, ENTERED, DONE])
+ fresh = []
+
+ def available(button):
+ fresh.append(button)
+ old[0]()
+
+ self.api._unlock(protocol.BitBoxConfig(available, lambda: "unused"))
+ self.assertTrue(
+ all(not request.unlock_continue.request_host_entry for request in self.sent[1:])
+ )
+ self.assertIsNot(old[0], fresh[0])
+
+ def test_errors_withdraw_and_invalidate(self):
+ for failure in [
+ OSError("disconnected"),
+ hww.Response(unlock=keystore.UnlockResponse(state=99)),
+ ]:
+ self.responses([PENDING, failure])
+ notifications = []
+ with self.assertRaises(Exception):
+ self.api._unlock(protocol.BitBoxConfig(notifications.append, lambda: "unused"))
+ self.assertEqual(len(notifications), 2)
+ self.assertIsNone(notifications[1])
+ sent_count = len(self.sent)
+ notifications[0]()
+ self.assertEqual(len(self.sent), sent_count)
+
+ def test_callback_failure_invalidates_button(self):
+ self.responses([PENDING])
+ buttons = []
+
+ def available(button):
+ buttons.append(button)
+ if button is not None:
+ raise RuntimeError("notification failed")
+
+ with self.assertRaisesRegex(RuntimeError, "notification failed"):
+ self.api._unlock(protocol.BitBoxConfig(available, lambda: "unused"))
+ self.assertEqual(len(buttons), 2)
+ self.assertIsNone(buttons[-1])
+ sent_count = len(self.sent)
+ buttons[0]()
+ self.assertEqual(len(self.sent), sent_count)
+
+ def test_input_exception_stops_host_submission(self):
+ self.responses([PENDING, READY])
+ notifications = []
+
+ def available(button):
+ notifications.append(button)
+ if button is not None:
+ button()
+
+ with self.assertRaisesRegex(RuntimeError, "input failed"):
+ self.api._unlock(
+ protocol.BitBoxConfig(
+ available, mock.Mock(side_effect=RuntimeError("input failed"))
+ )
+ )
+ self.assertIsNone(notifications[-1])
+ self.assertEqual(len(self.sent), 2)
+
+ def test_immediate_done_does_not_request_passphrase(self):
+ for available, enter in [
+ (None, None),
+ (mock.Mock(), None),
+ (None, mock.Mock()),
+ (mock.Mock(), mock.Mock()),
+ ]:
+ self.responses([DONE])
+ self.api._unlock(protocol.BitBoxConfig(available, enter))
+ if available is not None:
+ available.assert_not_called()
+ if enter is not None:
+ enter.assert_not_called()
+
+ def test_device_entry_polls_without_host_callbacks(self):
+ for available in [None, mock.Mock()]:
+ self.sent.clear()
+ self.responses([PENDING, PENDING, PENDING, ENTERED, DONE])
+ with mock.patch.object(protocol.time, "sleep") as sleep:
+ self.api._unlock(protocol.BitBoxConfig(on_host_passphrase_available=available))
+ self.assertEqual(sleep.call_count, 3)
+ self.assertEqual(
+ [request.WhichOneof("request") for request in self.sent],
+ ["unlock"] + ["unlock_continue"] * 4,
+ )
+ self.assertTrue(
+ all(not request.unlock_continue.request_host_entry for request in self.sent[1:])
+ )
+ if available is not None:
+ available.assert_not_called()
+
+ def test_click_from_ui_thread_wakes_worker(self):
+ buttons = queue.Queue()
+ events = []
+ errors = []
+ worker_id = []
+
+ def query(request, expected_response):
+ del expected_response
+ self.assertEqual(threading.get_ident(), worker_id[0])
+ events.append(request.WhichOneof("request"))
+ if request.HasField("unlock"):
+ return PENDING
+ if request.HasField("unlock_continue"):
+ return READY if request.unlock_continue.request_host_entry else PENDING
+ return DONE
+
+ self.api._msg_query = mock.Mock(side_effect=query)
+
+ def available(button):
+ self.assertEqual(threading.get_ident(), worker_id[0])
+ if button is not None:
+ buttons.put(button)
+
+ def run():
+ worker_id.append(threading.get_ident())
+ try:
+ self.api._unlock(protocol.BitBoxConfig(available, lambda: "value"))
+ except Exception as exc: # pylint: disable=broad-exception-caught
+ # Relay thread assertion failures to the test runner.
+ errors.append(exc)
+
+ thread = threading.Thread(target=run, daemon=True)
+ thread.start()
+ button = buttons.get(timeout=2)
+ button()
+ thread.join(timeout=2)
+ self.assertFalse(thread.is_alive())
+ self.assertFalse(errors)
+ self.assertEqual(events[-1], "unlock_host_info")
+
+ def test_debug_redacts_passphrase(self):
+ self.api.debug = True
+ self.api._bitbox_protocol = mock.Mock()
+ self.api._bitbox_protocol.encrypted_query.return_value = DONE.SerializeToString()
+ request = hww.Request(
+ unlock_host_info=keystore.UnlockHostInfoRequest(passphrase="never-log-this")
+ )
+ output = io.StringIO()
+ with redirect_stdout(output):
+ self.api._msg_query(request)
+ self.assertNotIn("never-log-this", output.getvalue())
+ self.assertIn("redacted", output.getvalue())
+
+
+if __name__ == "__main__":
+ unittest.main()
### py/send_message.py
@@ -17,6 +17,9 @@
import json
from pathlib import Path
import os
+import getpass
+import select
+import threading
import requests
import base58
@@ -85,6 +88,79 @@ def ask_user(
return choices[ans - 1][1]
+def passphrase_demo() -> bitbox_api_protocol.BitBoxConfig:
+ """Terminal equivalent of the host-entry button, with no competing stdin readers.
+
+ The library withdraws availability on completion and errors, stopping the watcher
+ and restoring the terminal.
+ GUI apps instead run connection on a worker and marshal these notifications to their
+ UI thread. The click callable itself never communicates with the device.
+ """
+ stop = threading.Event()
+ watcher: Optional[threading.Thread] = None
+
+ def withdraw() -> None:
+ nonlocal watcher
+ stop.set()
+ if watcher is not None:
+ watcher.join()
+ watcher = None
+
+ def available(request_host_entry: Optional[Callable[[], None]]) -> None:
+ nonlocal watcher
+ withdraw()
+ if request_host_entry is None:
+ print("Host passphrase entry is unavailable; h is disabled.")
+ return
+ import termios # pylint: disable=import-outside-toplevel
+
+ # A new device-entry phase gets a new button. Discard keys typed while the previous
+ # button was hidden, before announcing that h is available again.
+ termios.tcflush(sys.stdin.fileno(), termios.TCIFLUSH)
+ stop.clear()
+ print("Enter the passphrase on the device, or press h to request host entry.")
+
+ def watch() -> None:
+ # Only used with a POSIX terminal. Read one key without leaving a blocked
+ # input() behind when device entry finishes or the connection fails.
+ import tty # pylint: disable=import-outside-toplevel
+
+ fd = sys.stdin.fileno()
+ previous = termios.tcgetattr(fd)
+ try:
+ # Preserve a shortcut typed immediately after the notification, before this
+ # thread starts. The default TCSAFLUSH would discard that queued key.
+ tty.setcbreak(fd, termios.TCSANOW)
+ while not stop.is_set():
+ readable, _, _ = select.select([fd], [], [], 0.05)
+ if readable and not stop.is_set():
+ key = os.read(fd, 1)
+ if key == b"h":
+ request_host_entry()
+ return
+ if not key:
+ return
+ finally:
+ termios.tcsetattr(fd, termios.TCSADRAIN, previous)
+
+ watcher = threading.Thread(target=watch, daemon=True)
+ watcher.start()
+
+ def enter() -> Optional[str]:
+ try:
+ # Ctrl-D only raises EOFError before any input; after typing it can submit the text.
+ return getpass.getpass("Host passphrase (Ctrl-C cancels): ")
+ except (EOFError, KeyboardInterrupt):
+ print("\nHost input cancelled; continue on the device.")
+ return None
+
+ interactive = sys.stdin.isatty() and os.name == "posix"
+ return bitbox_api_protocol.BitBoxConfig(
+ on_host_passphrase_available=available if interactive else None,
+ enter_mnemonic_passphrase=enter if interactive else None,
+ )
+
+
BITBOXSYNC_CHALLENGE = bytes([0x10]) * 32
BITBOXSYNC_NAMESPACE_ID = bytes(range(0x20, 0x30))
BITBOXSYNC_INVITE_ID = bytes(range(0x30, 0x40))
@@ -1973,7 +2049,7 @@ def _register(self) -> None:
try:
res = self._device.u2f_register(self.APPID)
if res is not None:
- (self._dev_pubkey, self._dev_keyhandle) = res
+ self._dev_pubkey, self._dev_keyhandle = res
except u2f.ConditionsNotSatisfiedException:
print("Not registered")
@@ -2065,7 +2141,9 @@ def __del__(self) -> None:
transport=u2fhid.U2FHid(simulator),
device_info=None,
noise_config=noise_config,
+ bitbox_config=passphrase_demo(),
)
+ print("Connection ready.")
try:
bitbox_connection.check_min_version()
except FirmwareVersionOutdatedException as exc:
@@ -2155,8 +2233,12 @@ def attestation_check(self, result: bool) -> None:
print("Could not connect to the BitBox, device may be already connected to another app.")
return 1
bitbox_connection = bitbox02.BitBox02(
- transport=u2fhid.U2FHid(hid_device), device_info=bitbox, noise_config=config
+ transport=u2fhid.U2FHid(hid_device),
+ device_info=bitbox,
+ noise_config=config,
+ bitbox_config=passphrase_demo(),
)
+ print("Connection ready.")
try:
bitbox_connection.check_min_version()
except FirmwareVersionOutdatedException as exc:
### src/rust/Cargo.lock
@@ -277,6 +277,7 @@ name = "bitbox-proto"
version = "0.1.0"
dependencies = [
"prost",
+ "zeroize",
]
[[package]]
### src/rust/bitbox-hal/src/ui.rs
@@ -119,6 +119,10 @@ pub trait Ui {
async fn status(&mut self, title: &str, status_success: bool);
+ /// Display a message without controls until this future is dropped. Dropping it removes
+ /// the screen, including when the enclosing workflow is cancelled.
+ async fn waiting(&mut self, message: &str);
+
/// Demo/testing only: show a screen with all navigation icon buttons. Defaults to a no-op so
/// only platforms that implement it (BitBox03) do anything.
async fn show_demo_nav_buttons(&mut self) {}
### src/rust/bitbox-proto/Cargo.toml
@@ -13,3 +13,8 @@ license = "Apache-2.0"
version = "0.13.1"
default-features = false
features = ["derive"]
+
+[dependencies.zeroize]
+version = "1.7.0"
+default-features = false
+features = ["alloc"]
### src/rust/bitbox-proto/src/generated/shiftcrypto.bitbox02.rs
@@ -1924,6 +1924,74 @@ impl EthAddressCase {
}
}
}
+/// Unlock inside the paired Noise channel (since v9.28.0). Uninitialized and already unlocked
+/// devices return DONE immediately and are left unchanged. Continuations are only valid within
+/// this workflow.
+/// If the passphrase feature is enabled, device entry returns PASSPHRASE_PENDING. The host
+/// polls with UnlockContinueRequest until entry completes or it requests host entry.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct UnlockRequest {}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct UnlockContinueRequest {
+ /// False polls device entry; true interrupts it to ask for host-entry consent.
+ /// After PASSPHRASE_ENTERED, continue to await confirmation and unlock;
+ /// host entry is ignored.
+ #[prost(bool, tag = "1")]
+ pub request_host_entry: bool,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct UnlockHostInfoRequest {
+ /// Only sent after HOST_ENTRY_READY. Absent cancels host input and restarts device entry;
+ /// the empty string submits the empty passphrase. The device confirms the actual value.
+ #[prost(string, optional, tag = "1")]
+ pub passphrase: ::core::option::Option<::prost::alloc::string::String>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct UnlockResponse {
+ #[prost(enumeration = "unlock_response::State", tag = "1")]
+ pub state: i32,
+}
+/// Nested message and enum types in `UnlockResponse`.
+pub mod unlock_response {
+ #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+ #[repr(i32)]
+ pub enum State {
+ PassphrasePending = 0,
+ HostEntryReady = 1,
+ /// Passphrase entry on the device finished; confirmation may still be pending.
+ /// Withdraw host entry and send UnlockContinue to await completion.
+ PassphraseEntered = 2,
+ Done = 3,
+ }
+ impl State {
+ /// String value of the enum field names used in the ProtoBuf definition.
+ ///
+ /// The values are not transformed in any way and thus are considered stable
+ /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+ pub fn as_str_name(&self) -> &'static str {
+ match self {
+ State::PassphrasePending => "PASSPHRASE_PENDING",
+ State::HostEntryReady => "HOST_ENTRY_READY",
+ State::PassphraseEntered => "PASSPHRASE_ENTERED",
+ State::Done => "DONE",
+ }
+ }
+ /// Creates an enum from field names used in the ProtoBuf definition.
+ pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+ match value {
+ "PASSPHRASE_PENDING" => Some(Self::PassphrasePending),
+ "HOST_ENTRY_READY" => Some(Self::HostEntryReady),
+ "PASSPHRASE_ENTERED" => Some(Self::PassphraseEntered),
+ "DONE" => Some(Self::Done),
+ _ => None,
+ }
+ }
+ }
+}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ElectrumEncryptionKeyRequest {
@@ -2067,7 +2135,7 @@ pub struct Success {}
pub struct Request {
#[prost(
oneof = "request::Request",
- tags = "2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 24, 25, 26, 27, 28, 29, 30, 31"
+ tags = "2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34"
)]
pub request: ::core::option::Option<request::Request>,
}
@@ -2136,14 +2204,20 @@ pub mod request {
ChangePassword(super::ChangePasswordRequest),
#[prost(message, tag = "31")]
BitboxSync(super::BitBoxSyncRequest),
+ #[prost(message, tag = "32")]
+ Unlock(super::UnlockRequest),
+ #[prost(message, tag = "33")]
+ UnlockContinue(super::UnlockContinueRequest),
+ #[prost(message, tag = "34")]
+ UnlockHostInfo(super::UnlockHostInfoRequest),
}
}
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Response {
#[prost(
oneof = "response::Response",
- tags = "1, 2, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18"
+ tags = "1, 2, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19"
)]
pub response: ::core::option::Option<response::Response>,
}
@@ -2187,5 +2261,7 @@ pub mod response {
Bluetooth(super::BluetoothResponse),
#[prost(message, tag = "18")]
BitboxSync(super::BitBoxSyncResponse),
+ #[prost(message, tag = "19")]
+ Unlock(super::UnlockResponse),
}
}
### src/rust/bitbox-proto/src/lib.rs
@@ -9,3 +9,11 @@ pub mod pb {
pub mod pb_backup {
include!("./generated/shiftcrypto.bitbox02.backups.rs");
}
+
+// Also clear passphrases discarded during decoding, invalid-state handling or cancellation.
+impl Drop for pb::UnlockHostInfoRequest {
+ fn drop(&mut self) {
+ use zeroize::Zeroize;
+ self.passphrase.zeroize();
+ }
+}
### src/rust/bitbox02-rust-c/src/u2f_c_api.rs
@@ -62,6 +62,11 @@ fn next_task_token() -> u32 {
/// Must be called from the same single-threaded, non-reentrant execution context as all other
/// U2F workflow C API calls.
unsafe fn try_start_workflow() -> Option<ActiveWorkflowGuard> {
+ // A multi-request HWW workflow still owns the UI between requests, even though the
+ // C transport lock is released while waiting for the next request.
+ if !bitbox02_rust::async_usb::is_idle() {
+ return None;
+ }
let guard = ActiveWorkflowGuard::try_new()?;
unsafe {
if !matches!(UNLOCK_STATE.get().as_ref().unwrap(), TaskState::Nothing)
@@ -262,5 +267,21 @@ mod tests {
unsafe {
UNLOCK_STATE.get().write(TaskState::Nothing);
}
+
+ // The HWW task owns the UI even after an intermediate response has been read.
+ async fn host_input(_request: alloc::vec::Vec<u8>) -> alloc::vec::Vec<u8> {
+ bitbox02_rust::async_usb::next_request(alloc::vec![1]).await
+ }
+ bitbox02_rust::async_usb::spawn(host_input, &[]);
+ bitbox02_rust::async_usb::spin();
+ assert!(unsafe { try_start_workflow() }.is_none());
+ assert_eq!(
+ bitbox02_rust::async_usb::take_response().unwrap(),
+ alloc::vec![1]
+ );
+ assert!(bitbox02_rust::async_usb::waiting_for_next_request());
+ assert!(unsafe { try_start_workflow() }.is_none());
+ bitbox02_rust::async_usb::cancel();
+ assert!(unsafe { try_start_workflow() }.is_some());
}
}
### src/rust/bitbox02-rust/src/hal/testing/ui.rs
@@ -63,6 +63,13 @@ pub struct ProgressScreen {
}
type EnterStringCb<'a> = Box<dyn FnMut(&EnterStringParams<'_>) -> Result<String, UserAbort> + 'a>;
+type EnterStringAsyncCb<'a> = Box<
+ dyn FnMut(
+ &EnterStringParams<'_>,
+ ) -> core::pin::Pin<
+ Box<dyn core::future::Future<Output = Result<String, UserAbort>> + 'a>,
+ > + 'a,
+>;
type EnterWordlistWordCb<'a> =
Box<dyn FnMut(&EnterStringParams<'_>) -> Result<String, WordlistEntryAbort> + 'a>;
type MenuCb<'a> = Box<dyn FnMut(&[&str], Option<&str>) -> Result<u8, UserAbort> + 'a>;
@@ -77,7 +84,7 @@ pub struct TestingUi<'a> {
pub confirm_display_sizes: Vec<usize>,
pub confirm_scrollable: Vec<bool>,
progress_screens: Rc<RefCell<Vec<ProgressScreen>>>,
- _enter_string: Option<EnterStringCb<'a>>,
+ _enter_string_async: Option<EnterStringAsyncCb<'a>>,
_enter_wordlist_word: Option<EnterWordlistWordCb<'a>>,
_menu: Option<MenuCb<'a>>,
_trinary_choice: Option<TrinaryChoiceCb<'a>>,
@@ -227,6 +234,14 @@ impl Ui for TestingUi<'_> {
});
}
+ async fn waiting(&mut self, message: &str) {
+ self.screens.push(Screen::PrintScreen {
+ message: message.into(),
+ duration: Duration::ZERO,
+ });
+ core::future::pending::<()>().await;
+ }
+
fn switch_to_logo(&mut self) {}
fn reset(&mut self) {}
@@ -237,7 +252,9 @@ impl Ui for TestingUi<'_> {
_can_cancel: CanCancel,
_preset: &str,
) -> Result<zeroize::Zeroizing<String>, UserAbort> {
- self._enter_string.as_mut().unwrap()(params).map(zeroize::Zeroizing::new)
+ self._enter_string_async.as_mut().unwrap()(params)
+ .await
+ .map(zeroize::Zeroizing::new)
}
async fn enter_wordlist_word(
@@ -331,7 +348,7 @@ impl<'a> TestingUi<'a> {
confirm_scrollable: vec![],
progress_screens: Rc::new(RefCell::new(vec![])),
_abort_nth: None,
- _enter_string: None,
+ _enter_string_async: None,
_enter_wordlist_word: None,
_menu: None,
_trinary_choice: None,
@@ -356,16 +373,22 @@ impl<'a> TestingUi<'a> {
})
}
- pub fn set_enter_string(&mut self, cb: EnterStringCb<'a>) {
- self._enter_string = Some(cb);
+ pub fn set_enter_string(&mut self, mut cb: EnterStringCb<'a>) {
+ self.set_enter_string_async(Box::new(move |params| {
+ Box::pin(core::future::ready(cb(params)))
+ }));
+ }
+
+ pub fn set_enter_string_async(&mut self, cb: EnterStringAsyncCb<'a>) {
+ self._enter_string_async = Some(cb);
}
pub fn set_enter_wordlist_word(&mut self, cb: EnterWordlistWordCb<'a>) {
self._enter_wordlist_word = Some(cb);
}
pub fn remove_enter_string(&mut self) {
- self._enter_string = None;
+ self._enter_string_async = None;
}
pub fn set_menu(&mut self, cb: MenuCb<'a>) {
@@ -438,7 +461,7 @@ impl<'a> TestingUi<'a> {
let words: Vec<String> = words.iter().map(|word| (*word).into()).collect();
let mut first_words: VecDeque<String> = words[..23].iter().cloned().collect();
let last_word = words[23].clone();
- let mut fallback_enter_string = self._enter_string.take();
+ let mut fallback_enter_string = self._enter_string_async.take();
self.set_trinary_choice(Box::new(
|message, label_left, label_middle, label_right| {
@@ -460,11 +483,12 @@ impl<'a> TestingUi<'a> {
.unwrap())
}));
- self.set_enter_string(Box::new(move |params| {
+ self.set_enter_string_async(Box::new(move |params| {
if params.wordlist.is_some() && params.title.ends_with(" of 24") {
- return Ok(first_words
+ let word = first_words
.pop_front()
- .expect("too many mnemonic word entries"));
+ .expect("too many mnemonic word entries");
+ return Box::pin(core::future::ready(Ok(word)));
}
if let Some(ref mut fallback) = fallback_enter_string {
return fallback(params);
### src/rust/bitbox02-rust/src/hww.rs
@@ -27,17 +27,22 @@ pub fn reset_session(hal: &mut impl crate::hal::Hal) {
/// Must be called during the execution of a usb task. This sends out the response to the host and
/// awaits the next request. If the request is not a valid noise encrypted protofbuf api request
/// message, `Err(Error::InvalidInput)` is returned.
-#[cfg(not(any(test, feature = "testing")))]
+/// In tests, set `MOCK_NEXT_REQUEST` to replace the encrypted transport.
pub async fn next_request(
response: crate::pb::response::Response,
) -> Result<crate::pb::request::Request, api::error::Error> {
+ #[cfg(any(test, feature = "testing"))]
+ if let Some(func) = MOCK_NEXT_REQUEST.0.borrow().as_ref() {
+ return func(response);
+ }
let mut out = [OP_STATUS_SUCCESS].to_vec();
noise::encrypt(&api::encode(response), &mut out).or(Err(api::error::Error::NoiseEncrypt))?;
let request = crate::async_usb::next_request(out).await;
match request.split_first() {
Some((&noise::OP_NOISE_MSG, encrypted_request)) => {
- let decrypted_request =
- noise::decrypt(encrypted_request).or(Err(api::error::Error::NoiseDecrypt))?;
+ let decrypted_request = zeroize::Zeroizing::new(
+ noise::decrypt(encrypted_request).or(Err(api::error::Error::NoiseDecrypt))?,
+ );
api::decode(&decrypted_request[..])
}
_ => Err(api::error::Error::InvalidInput),
@@ -63,15 +68,6 @@ pub static MOCK_NEXT_REQUEST: SafeData<
>,
> = SafeData(core::cell::RefCell::new(None));
-/// Set `MOCK_NEXT_REQUEST` to mock requests from the host.
-#[cfg(any(test, feature = "testing"))]
-pub async fn next_request(
- response: crate::pb::response::Response,
-) -> Result<crate::pb::request::Request, api::error::Error> {
- let func = MOCK_NEXT_REQUEST.0.borrow();
- func.as_ref().unwrap()(response)
-}
-
/// Process OP_UNLOCK.
async fn api_unlock(hal: &mut impl crate::hal::Hal) -> Vec<u8> {
match crate::workflow::unlock::unlock(hal).await {
@@ -150,7 +146,8 @@ mod tests {
/// Make a new noise channel by invoking the noise handshake. Returns a request function which
/// encrypts the message going in and decrypts the message coming out.
- fn init_noise<H: crate::hal::Hal>() -> Box<dyn FnMut(&mut H, &[u8]) -> Result<Vec<u8>, ()>> {
+ pub(super) fn init_noise_ciphers()
+ -> (impl FnMut(&[u8]) -> Vec<u8>, impl FnMut(&[u8]) -> Vec<u8>) {
assert_eq!(
block_on(process_packet(&mut TestingHal::new(), b"h".to_vec())),
[OP_STATUS_SUCCESS].to_vec()
@@ -209,15 +206,22 @@ mod tests {
}
let (mut host_send, mut host_recv) = host_noise.get_ciphers();
+ (
+ move |msg| {
+ let mut packet = b"n".to_vec(); // message opcode
+ packet.extend_from_slice(&host_send.encrypt_vec(msg));
+ packet
+ },
+ move |msg| host_recv.decrypt_vec(msg).unwrap(),
+ )
+ }
+
+ fn init_noise<H: crate::hal::Hal>() -> Box<dyn FnMut(&mut H, &[u8]) -> Result<Vec<u8>, ()>> {
+ let (mut encrypt, mut decrypt) = init_noise_ciphers();
Box::new(move |hal, msg| -> Result<Vec<u8>, ()> {
- let msg_encrypted = host_send.encrypt_vec(msg);
- let response_encrypted = block_on(process_packet(hal, {
- let mut m = b"n".to_vec(); // message opcode
- m.extend_from_slice(&msg_encrypted);
- m
- }));
+ let response_encrypted = block_on(process_packet(hal, encrypt(msg)));
match response_encrypted.split_first() {
- Some((&OP_STATUS_SUCCESS, rest)) => Ok(host_recv.decrypt_vec(rest).unwrap()),
+ Some((&OP_STATUS_SUCCESS, rest)) => Ok(decrypt(rest)),
_ => Err(()),
}
})
### src/rust/bitbox02-rust/src/hww/api.rs
@@ -32,6 +32,7 @@ mod set_mnemonic_passphrase_enabled;
mod set_password;
mod show_mnemonic;
mod system;
+mod unlock;
use alloc::vec::Vec;
@@ -115,6 +116,8 @@ fn can_call(hal: &mut impl crate::hal::Hal, request: &Request) -> bool {
};
match request {
+ Request::Unlock(_) => true,
+ Request::UnlockContinue(_) | Request::UnlockHostInfo(_) => false,
// Deprecated call, last used in v1.0.0.
Request::PerformAttestation(_) => false,
Request::DeviceInfo(_)
@@ -158,6 +161,7 @@ fn can_call(hal: &mut impl crate::hal::Hal, request: &Request) -> bool {
/// Handle a protobuf api call.
async fn process_api(hal: &mut impl crate::hal::Hal, request: &Request) -> Result<Response, Error> {
match request {
+ Request::Unlock(_) => unlock::process(hal).await,
Request::Reboot(request) => system::reboot_to_bootloader(hal, request).await,
Request::DeviceInfo(_) => device_info::process(hal).await,
Request::DeviceName(request) => set_device_name::process(hal, request).await,
@@ -220,6 +224,8 @@ async fn process_api(hal: &mut impl crate::hal::Hal, request: &Request) -> Resul
/// `input` is a hww.proto Request message, protobuf encoded.
/// Returns a protobuf encoded hww.proto Response message.
pub async fn process(hal: &mut impl crate::hal::Hal, input: Vec<u8>) -> Vec<u8> {
+ // An invalid standalone request can also contain a host passphrase.
+ let input = zeroize::Zeroizing::new(input);
let request = match decode(&input[..]) {
Ok(request) => request,
Err(err) => return encode(make_error(err)),
### src/rust/bitbox02-rust/src/hww/api/unlock.rs
@@ -0,0 +1,163 @@
+// SPDX-License-Identifier: Apache-2.0
+
+use super::error::Error;
+use crate::hal::ui::ConfirmParams;
+use crate::hal::{Memory, Ui};
+use crate::pb;
+use crate::pb::request::Request;
+use crate::pb::response::Response;
+use crate::pb::unlock_response::State;
+use crate::workflow::unlock;
+use alloc::string::String;
+use core::pin::pin;
+use zeroize::Zeroizing;
+
+// Match the BitBox02 keyboard buffer, excluding its terminating NUL.
+// Keep in sync with INPUT_STRING_MAX_SIZE - 1 in src/ui/components/trinary_input_string.h.
+const MAX_PASSPHRASE_LEN: usize = 149;
+
+// Keep in sync with _special_chars in src/ui/components/trinary_input_string.c.
+const SPECIAL: &[u8] = b" !\"#$%&'()*+,-./:;<=>?^[\\]@_{|}";
+
+fn response(state: State) -> Response {
+ Response::Unlock(pb::UnlockResponse { state: state as _ })
+}
+
+/// Returns None when the host requested entry. The caller can then borrow the UI again
+/// after the pinned device-entry future has been dropped. Confirmation starts only after
+/// device entry completes and host entry has been withdrawn.
+async fn device_entry(hal: &mut impl crate::hal::Hal) -> Result<Option<Zeroizing<String>>, Error> {
+ enum Completed {
+ Device(Zeroizing<String>),
+ Request(Result<pb::UnlockContinueRequest, Error>),
+ }
+
+ // Keep the same UI future across continuations; restarting it per poll would erase partial input.
+ let mut entry = pin!(unlock::enter_mnemonic_passphrase(hal));
+ loop {
+ let mut next = pin!(async {
+ match crate::hww::next_request(response(State::PassphrasePending)).await? {
+ Request::UnlockContinue(request) => Ok(request),
+ _ => Err(Error::InvalidState),
+ }
+ });
+ // Device input completion wins a simultaneous click. Borrow both futures so the
+ // unfinished one stays alive after the race.
+ let completed =
+ futures_lite::future::or(async { Completed::Device(entry.as_mut().await) }, async {
+ Completed::Request(next.as_mut().await)
+ })
+ .await;
+ match completed {
+ Completed::Device(passphrase) => {
+ // Never cancel next_request when the UI finishes: its intermediate response must
+ // be consumed and the next request received before sending the final response.
+ next.await?;
+ return Ok(Some(passphrase));
+ }
+ Completed::Request(request) => {
+ if request?.request_host_entry {
+ return Ok(None);
+ }
+ }
+ }
+ }
+}
+
+fn validate_host_passphrase(passphrase: &str) -> Result<(), &'static str> {
+ // Restrict host input to values that can also be entered again on the device.
+ if passphrase.len() > MAX_PASSPHRASE_LEN {
+ return Err("Passphrase too\nlong");
+ }
+ if !passphrase
+ .bytes()
+ .all(|c| c.is_ascii_alphanumeric() || SPECIAL.contains(&c))
+ {
+ return Err("Unsupported\ncharacters");
+ }
+ Ok(())
+}
+
+async fn passphrase(hal: &mut impl crate::hal::Hal) -> Result<Zeroizing<String>, Error> {
+ loop {
+ let (passphrase, from_host) = match device_entry(hal).await? {
+ Some(passphrase) => {
+ // Withdraw host entry before showing either confirmation screen. A queued host
+ // click loses to completed input, including on this continuation; it must not
+ // interrupt confirmation or reopen the host-entry consent flow.
+ let request = crate::hww::next_request(response(State::PassphraseEntered)).await?;
+ if !matches!(request, Request::UnlockContinue(_)) {
+ return Err(Error::InvalidState);
+ }
+ (passphrase, false)
+ }
+ None => {
+ // Entry was interrupted by the host button. Rejection/cancellation always starts a
+ // fresh device-entry screen, without preserving a partially entered passphrase.
+ if hal
+ .ui()
+ .confirm(&ConfirmParams {
+ title: "",
+ body: "Enter passphrase\non host?",
+ ..Default::default()
+ })
+ .await
+ .is_err()
+ {
+ continue;
+ }
+
+ let request = {
+ let waiting = async {
+ hal.ui().waiting("Enter passphrase\non host").await;
+ unreachable!()
+ };
+ // Poll the waiting screen first so it is visible before HOST_ENTRY_READY. This
+ // race only cancels the screen: next_request is always consumed to completion.
+ futures_lite::future::or(
+ waiting,
+ crate::hww::next_request(response(State::HostEntryReady)),
+ )
+ .await?
+ };
+ let Request::UnlockHostInfo(mut request) = request else {
+ return Err(Error::InvalidState);
+ };
+ let Some(passphrase) = request.passphrase.take().map(Zeroizing::new) else {
+ continue;
+ };
+ if let Err(message) = validate_host_passphrase(&passphrase) {
+ // Invalid user input restarts entry just like rejection; malformed protocol
+ // messages still fail in next_request.
+ hal.ui().status(message, false).await;
+ continue;
+ }
+ (passphrase, true)
+ }
+ };
+ if unlock::confirm_mnemonic_passphrase(hal, &passphrase, from_host)
+ .await
+ .is_ok()
+ {
+ return Ok(passphrase);
+ }
+ hal.ui().status("Please try again", false).await;
+ }
+}
+
+pub async fn process(hal: &mut impl crate::hal::Hal) -> Result<Response, Error> {
+ // Let clients call unlock after pairing without first checking initialization. Return DONE
+ // without touching setup in progress.
+ if !hal.memory().is_initialized() {
+ return Ok(response(State::Done));
+ }
+ // Switch the waiting screen from "See the BitBoxApp" to the logo immediately
+ // before unlock, avoiding a logo flash before entry or the waiting text
+ // reappearing afterward. Also do this if U2F already unlocked the device.
+ hal.ui().switch_to_logo();
+ unlock::unlock_with_passphrase(hal, passphrase).await?;
+ Ok(response(State::Done))
+}
+
+#[cfg(test)]
+mod tests;
### src/rust/bitbox02-rust/src/hww/api/unlock/tests.rs
@@ -0,0 +1,533 @@
+// SPDX-License-Identifier: Apache-2.0
+
+use super::*;
+use crate::async_usb;
+use crate::hal::testing::TestingHal;
+use crate::hal::testing::ui::Screen;
+use crate::hww::SafeData;
+use alloc::boxed::Box;
+use alloc::rc::Rc;
+use alloc::vec::Vec;
+use core::cell::{Cell, RefCell};
+use core::future::poll_fn;
+use core::task::Poll;
+use hex_lit::hex;
+use prost::Message;
+
+// These tests exercise real Noise and async_usb continuations, including time spent between
+// requests, rather than the synchronous MOCK_NEXT_REQUEST shortcut. Run with --test-threads 1.
+static HAL: SafeData<RefCell<Option<TestingHal<'static>>>> = SafeData(RefCell::new(None));
+const SEED: [u8; 32] = hex!("c7940c13479b8d9a6498f4e50d5a42e0d617bc8e8ac9f2b8cecf97e94c2b035c");
+
+async fn task(packet: Vec<u8>) -> Vec<u8> {
+ let mut hal = HAL.0.borrow_mut().take().unwrap();
+ let response = crate::hww::process_packet(&mut hal, packet).await;
+ HAL.0.borrow_mut().replace(hal);
+ response
+}
+
+struct EntryGuard(Rc<Cell<usize>>);
+impl Drop for EntryGuard {
+ fn drop(&mut self) {
+ self.0.set(self.0.get() - 1);
+ }
+}
+
+struct Host {
+ encrypt: Box<dyn FnMut(&[u8]) -> Vec<u8>>,
+ decrypt: Box<dyn FnMut(&[u8]) -> Vec<u8>>,
+ input: Rc<RefCell<Option<String>>>,
+ entries: Rc<Cell<usize>>,
+ active: Rc<Cell<usize>>,
+}
+
+impl Host {
+ async fn new(enabled: bool, abort_screen: Option<usize>) -> Self {
+ async_usb::cancel();
+ crate::hww::MOCK_NEXT_REQUEST.0.borrow_mut().take();
+ crate::keystore::lock();
+ let (encrypt, decrypt) = crate::hww::tests::init_noise_ciphers();
+ let mut hal = TestingHal::new();
+ crate::keystore::encrypt_and_store_seed(&mut hal, &SEED, "password")
+ .await
+ .unwrap();
+ hal.memory.set_initialized().unwrap();
+ hal.memory.set_mnemonic_passphrase_enabled(enabled).unwrap();
+ crate::keystore::lock();
+ if let Some(screen) = abort_screen {
+ hal.ui.abort_nth(screen);
+ }
+ let host = Self {
+ encrypt: Box::new(encrypt),
+ decrypt: Box::new(decrypt),
+ input: Rc::new(RefCell::new(None)),
+ entries: Rc::new(Cell::new(0)),
+ active: Rc::new(Cell::new(0)),
+ };
+ let input = host.input.clone();
+ let entries = host.entries.clone();
+ let active = host.active.clone();
+ hal.ui.set_enter_string_async(Box::new(move |params| {
+ if !params.passphrase {
+ return Box::pin(async { Ok("password".into()) });
+ }
+ entries.set(entries.get() + 1);
+ active.set(active.get() + 1);
+ let guard = EntryGuard(active.clone());
+ let input = input.clone();
+ Box::pin(async move {
+ let _guard = guard;
+ poll_fn(|_| match input.borrow_mut().take() {
+ Some(input) => Poll::Ready(Ok(input)),
+ None => Poll::Pending,
+ })
+ .await
+ })
+ }));
+ HAL.0.borrow_mut().replace(hal);
+ host
+ }
+
+ fn send(&mut self, request: Request) {
+ let encoded = pb::Request {
+ request: Some(request),
+ }
+ .encode_to_vec();
+ let packet = (self.encrypt)(&encoded);
+ if async_usb::is_idle() {
+ async_usb::spawn(task, &packet);
+ } else {
+ assert!(async_usb::waiting_for_next_request());
+ async_usb::on_next_request(&packet);
+ }
+ }
+
+ fn receive(&mut self) -> Response {
+ for _ in 0..10_000 {
+ async_usb::spin();
+ if let Ok(packet) = async_usb::take_response() {
+ assert_eq!(packet[0], crate::hww::OP_STATUS_SUCCESS);
+ let plain = (self.decrypt)(&packet[1..]);
+ return pb::Response::decode(plain.as_slice())
+ .unwrap()
+ .response
+ .unwrap();
+ }
+ }
+ panic!("no response");
+ }
+
+ fn query(&mut self, request: Request) -> Response {
+ self.send(request);
+ self.receive()
+ }
+
+ fn start(&mut self) -> Response {
+ self.query(Request::Unlock(pb::UnlockRequest {}))
+ }
+
+ fn poll(&mut self, request_host_entry: bool) -> Response {
+ self.query(Request::UnlockContinue(pb::UnlockContinueRequest {
+ request_host_entry,
+ }))
+ }
+
+ fn submit(&mut self, passphrase: Option<&str>) -> Response {
+ self.query(Request::UnlockHostInfo(pb::UnlockHostInfoRequest {
+ passphrase: passphrase.map(Into::into),
+ }))
+ }
+
+ async fn check_wallet(&self, passphrase: &str) {
+ assert!(!crate::keystore::is_locked());
+ let mut hal = HAL.0.borrow_mut().take().unwrap();
+ assert_eq!(
+ crate::keystore::copy_bip39_seed(&mut hal)
+ .await
+ .unwrap()
+ .as_slice(),
+ bip39::Mnemonic::from_entropy(&SEED)
+ .unwrap()
+ .to_seed_normalized(passphrase)
+ .as_slice()
+ );
+ HAL.0.borrow_mut().replace(hal);
+ }
+
+ fn screens(&self) -> Vec<Screen> {
+ HAL.0.borrow().as_ref().unwrap().ui.screens.clone()
+ }
+}
+
+impl Drop for Host {
+ fn drop(&mut self) {
+ async_usb::cancel();
+ HAL.0.borrow_mut().take();
+ crate::keystore::lock();
+ }
+}
+
+fn pending() -> Response {
+ response(State::PassphrasePending)
+}
+fn ready() -> Response {
+ response(State::HostEntryReady)
+}
+fn entered() -> Response {
+ response(State::PassphraseEntered)
+}
+fn done() -> Response {
+ response(State::Done)
+}
+
+#[async_test::test]
+async fn test_process_device_completion_wins_click() {
+ let mut host = Host::new(true, None).await;
+ assert_eq!(host.start(), pending());
+ for _ in 0..3 {
+ assert_eq!(host.poll(false), pending());
+ assert_eq!(host.entries.get(), 1);
+ assert_eq!(host.active.get(), 1);
+ }
+ host.input.borrow_mut().replace("device value".into());
+ async_usb::spin();
+ assert!(async_usb::waiting_for_next_request());
+ assert!(matches!(
+ async_usb::take_response(),
+ Err(async_usb::CopyResponseErr::NotReady)
+ ));
+ assert_eq!(host.poll(true), entered());
+ assert_eq!(host.active.get(), 0);
+ assert!(crate::keystore::is_locked());
+ // A second late host-entry request cannot interrupt device confirmation either.
+ assert_eq!(host.poll(true), done());
+ host.check_wallet("device value").await;
+ // Late input cannot change the chosen wallet; a repeated Unlock is a no-op.
+ assert_eq!(
+ host.submit(Some("late")),
+ super::super::error::make_error(Error::InvalidState)
+ );
+ assert_eq!(host.start(), done());
+ host.check_wallet("device value").await;
+ assert!(!host.screens().iter().any(
+ |s| matches!(s, Screen::Confirm { body, .. } if body == "Enter passphrase\non host?")
+ ));
+}
+
+#[async_test::test]
+async fn test_process_completion_before_pending_ack() {
+ let mut host = Host::new(true, None).await;
+ host.send(Request::Unlock(pb::UnlockRequest {}));
+ for _ in 0..100 {
+ async_usb::spin();
+ }
+ host.input.borrow_mut().replace(String::new());
+ for _ in 0..5 {
+ async_usb::spin();
+ }
+ // Completing the UI does not overwrite the intermediate response or abandon next_request.
+ assert_eq!(host.receive(), pending());
+ assert_eq!(host.poll(false), entered());
+ assert_eq!(host.poll(false), done());
+ host.check_wallet("").await;
+}
+
+#[async_test::test]
+async fn test_process_host_requires_actual_confirmation() {
+ for value in ["host passphrase", "", " spaces "] {
+ let mut host = Host::new(true, None).await;
+ assert_eq!(host.start(), pending());
+ assert_eq!(host.poll(true), ready());
+ assert_eq!(host.active.get(), 0);
+ for _ in 0..100 {
+ async_usb::spin();
+ }
+ assert!(crate::keystore::is_locked());
+ assert_eq!(host.submit(Some(value)), done());
+ let screens = host.screens();
+ assert!(screens.iter().any(
+ |s| matches!(s, Screen::Confirm { body, .. } if body == "Enter passphrase\non host?")
+ ));
+ assert!(screens.iter().any(|s| matches!(s, Screen::PrintScreen { message, .. } if message == "Enter passphrase\non host")));
+ let expected = if value.is_empty() {
+ "Use empty passphrase?"
+ } else {
+ value
+ };
+ assert!(screens.iter().any(
+ |s| matches!(s, Screen::Confirm { body, longtouch: true, .. } if body == expected)
+ ));
+ host.check_wallet(value).await;
+ }
+}
+
+#[async_test::test]
+async fn test_process_consent_rejection_and_host_cancellation_restart_entry() {
+ // Screen 0 is the paused unlock animation; screen 1 is host-entry consent.
+ for reject in [true, false] {
+ let mut host = Host::new(true, if reject { Some(1) } else { None }).await;
+ assert_eq!(host.start(), pending());
+ if reject {
+ assert_eq!(host.poll(true), pending());
+ } else {
+ assert_eq!(host.poll(true), ready());
+ assert_eq!(host.submit(None), pending());
+ }
+ assert_eq!(host.entries.get(), 2);
+ assert_eq!(host.active.get(), 1);
+ host.input.borrow_mut().replace(String::new());
+ assert_eq!(host.poll(false), entered());
+ assert_eq!(host.poll(false), done());
+ host.check_wallet("").await;
+ }
+}
+
+#[async_test::test]
+async fn test_process_actual_passphrase_rejection_restarts_entry() {
+ for (value, abort_screen) in [("secret", 4), ("", 3)] {
+ let mut host = Host::new(true, Some(abort_screen)).await;
+ assert_eq!(host.start(), pending());
+ assert_eq!(host.poll(true), ready());
+ assert_eq!(host.submit(Some(value)), pending());
+ assert_eq!(host.entries.get(), 2);
+ host.input.borrow_mut().replace("device".into());
+ assert_eq!(host.poll(false), entered());
+ assert_eq!(host.poll(false), done());
+ host.check_wallet("device").await;
+ }
+}
+
+#[async_test::test]
+async fn test_process_device_confirmation_rejection_restarts_entry() {
+ // Screen 0 is the paused animation, followed by the introduction and actual value.
+ for abort_screen in [1, 2] {
+ let mut host = Host::new(true, Some(abort_screen)).await;
+ assert_eq!(host.start(), pending());
+ host.input.borrow_mut().replace("device".into());
+ assert_eq!(host.poll(false), entered());
+ assert_eq!(host.active.get(), 0);
+ assert!(crate::keystore::is_locked());
+ assert_eq!(host.poll(false), pending());
+ assert_eq!(host.entries.get(), 2);
+ assert_eq!(host.active.get(), 1);
+ // Only the new entry phase allows the host to request consent again.
+ assert_eq!(host.poll(true), ready());
+ assert_eq!(host.submit(Some("host")), done());
+ host.check_wallet("host").await;
+ }
+}
+
+#[async_test::test]
+async fn test_process_invalid_continuations_clear_partial_unlock() {
+ for phase in 0..3 {
+ for request in [
+ Request::Unlock(pb::UnlockRequest {}),
+ Request::UnlockContinue(pb::UnlockContinueRequest {
+ request_host_entry: true,
+ }),
+ Request::UnlockHostInfo(pb::UnlockHostInfoRequest {
+ passphrase: Some("early".into()),
+ }),
+ ] {
+ if matches!(
+ (&request, phase),
+ (Request::UnlockContinue(_), 0 | 2) | (Request::UnlockHostInfo(_), 1)
+ ) {
+ continue;
+ }
+ let mut host = Host::new(true, None).await;
+ assert_eq!(host.start(), pending());
+ if phase == 1 {
+ assert_eq!(host.poll(true), ready());
+ } else if phase == 2 {
+ host.input.borrow_mut().replace("device".into());
+ assert_eq!(host.poll(false), entered());
+ }
+ assert_eq!(
+ host.query(request),
+ super::super::error::make_error(Error::InvalidState)
+ );
+ assert!(async_usb::is_idle());
+ assert_eq!(host.active.get(), 0);
+ let mut hal = HAL.0.borrow_mut().take().unwrap();
+ assert!(crate::keystore::copy_seed(&mut hal).await.is_err());
+ HAL.0.borrow_mut().replace(hal);
+ }
+ }
+}
+
+#[async_test::test]
+async fn test_process_reset_clears_partial_unlock() {
+ for phase in 0..5 {
+ let mut host = Host::new(true, None).await;
+ if phase == 0 {
+ host.send(Request::Unlock(pb::UnlockRequest {}));
+ for _ in 0..100 {
+ async_usb::spin();
+ }
+ } else {
+ assert_eq!(host.start(), pending());
+ if phase == 2 {
+ assert_eq!(host.poll(true), ready());
+ } else if phase >= 3 {
+ host.input.borrow_mut().replace("device".into());
+ if phase == 3 {
+ assert_eq!(host.poll(false), entered());
+ } else {
+ // Reset with PASSPHRASE_ENTERED still unread, before its continuation.
+ host.send(Request::UnlockContinue(pb::UnlockContinueRequest::default()));
+ for _ in 0..100 {
+ async_usb::spin();
+ }
+ }
+ }
+ }
+ let mut hal = TestingHal::new();
+ crate::hww::reset_session(&mut hal);
+ assert!(async_usb::is_idle());
+ assert_eq!(host.active.get(), 0);
+ assert!(crate::keystore::copy_seed(&mut hal).await.is_err());
+ assert!(crate::hww::noise::encrypt(b"old session", &mut Vec::new()).is_err());
+ // A fresh handshake after reset works without an old continuation.
+ let _ = crate::hww::tests::init_noise_ciphers();
+ }
+}
+
+#[async_test::test]
+async fn test_process_passphrase_disabled() {
+ let mut host = Host::new(false, None).await;
+ assert_eq!(host.start(), done());
+ assert_eq!(host.entries.get(), 0);
+ host.check_wallet("").await;
+}
+
+#[async_test::test]
+async fn test_process_invalid_host_input_restarts_entry() {
+ for (value, expected_status) in [
+ ("x".repeat(MAX_PASSPHRASE_LEN + 1), "Passphrase too\nlong"),
+ ("\0".into(), "Unsupported\ncharacters"),
+ ("é".into(), "Unsupported\ncharacters"),
+ ("line\nbreak".into(), "Unsupported\ncharacters"),
+ ("~".into(), "Unsupported\ncharacters"),
+ ("`".into(), "Unsupported\ncharacters"),
+ ] {
+ let mut host = Host::new(true, None).await;
+ assert_eq!(host.start(), pending());
+ assert_eq!(host.poll(true), ready());
+ assert_eq!(host.submit(Some(&value)), pending());
+ assert!(crate::keystore::is_locked());
+ assert_eq!(host.entries.get(), 2);
+ assert_eq!(host.active.get(), 1);
+ host.input.borrow_mut().replace("device".into());
+ assert_eq!(host.poll(false), entered());
+ assert_eq!(host.poll(false), done());
+ host.check_wallet("device").await;
+ let screens = host.screens();
+ assert!(screens.iter().any(
+ |s| matches!(s, Screen::Status { title, success: false } if title == expected_status)
+ ));
+ assert!(
+ !screens
+ .iter()
+ .any(|s| matches!(s, Screen::Confirm { body, .. } if body == &value))
+ );
+ }
+ assert!(validate_host_passphrase(&"x".repeat(MAX_PASSPHRASE_LEN)).is_ok());
+ assert!(validate_host_passphrase("AZaz09 !\"#$%&'()*+,-./:;<=>?^[\\]@_{|}").is_ok());
+}
+
+#[async_test::test]
+async fn test_process_invalid_host_input_can_retry_on_host() {
+ let mut host = Host::new(true, None).await;
+ assert_eq!(host.start(), pending());
+ for value in ["x".repeat(MAX_PASSPHRASE_LEN + 1), "~".into()] {
+ assert_eq!(host.poll(true), ready());
+ assert_eq!(host.submit(Some(&value)), pending());
+ assert!(crate::keystore::is_locked());
+ assert_eq!(host.active.get(), 1);
+ }
+ assert_eq!(host.entries.get(), 3);
+ assert_eq!(host.poll(true), ready());
+ let valid = "x".repeat(MAX_PASSPHRASE_LEN);
+ assert_eq!(host.submit(Some(&valid)), done());
+ host.check_wallet(&valid).await;
+}
+
+#[async_test::test]
+async fn test_process_uninitialized_returns_done() {
+ crate::keystore::lock();
+ let mut hal = TestingHal::new();
+ for seeded in [false, true] {
+ if seeded {
+ crate::keystore::encrypt_and_store_seed(&mut hal, &SEED, "password")
+ .await
+ .unwrap();
+ }
+ // Repeated unlock requests must not change an uninitialized device or setup in progress.
+ for _ in 0..2 {
+ let request = pb::Request {
+ request: Some(Request::Unlock(pb::UnlockRequest {})),
+ }
+ .encode_to_vec();
+ let reply = super::super::process(&mut hal, request).await;
+ assert_eq!(
+ pb::Response::decode(reply.as_slice())
+ .unwrap()
+ .response
+ .unwrap(),
+ done()
+ );
+ assert!(hal.ui.screens.is_empty());
+ assert!(!hal.memory.is_initialized());
+ assert_eq!(hal.memory.is_seeded(), seeded);
+ assert!(crate::keystore::is_locked());
+ if seeded {
+ assert_eq!(
+ crate::keystore::copy_seed(&mut hal)
+ .await
+ .unwrap()
+ .as_slice(),
+ SEED.as_slice()
+ );
+ } else {
+ assert!(crate::keystore::copy_seed(&mut hal).await.is_err());
+ }
+ }
+ }
+ crate::keystore::lock();
+}
+
+#[test]
+fn test_decode_host_passphrase_fields() {
+ use super::super::decode;
+ // Length and character validation belongs to the unlock workflow, so it can show a
+ // status and restart entry. Decoding only checks the protobuf encoding.
+ let too_long = "x".repeat(MAX_PASSPHRASE_LEN + 1);
+ for value in [
+ None,
+ Some(""),
+ Some("value"),
+ Some(too_long.as_str()),
+ Some("~"),
+ Some("é"),
+ Some("\0"),
+ ] {
+ let encoded = pb::Request {
+ request: Some(Request::UnlockHostInfo(pb::UnlockHostInfoRequest {
+ passphrase: value.map(Into::into),
+ })),
+ }
+ .encode_to_vec();
+ let Request::UnlockHostInfo(request) = decode(&encoded).unwrap() else {
+ panic!("expected host passphrase request");
+ };
+ assert_eq!(request.passphrase.as_deref(), value);
+ }
+ // Invalid UTF-8 and truncated strings are rejected as malformed protobuf.
+ for bytes in [
+ hex!("9202030a01ff").as_slice(),
+ hex!("9202030a0561").as_slice(),
+ ] {
+ assert!(matches!(decode(bytes), Err(Error::InvalidInput)));
+ }
+}
### src/rust/bitbox02-rust/src/hww/transport.rs
@@ -83,11 +83,12 @@ impl<H> HwwVendorHandler<H> {
}
fn refresh_timeout(&mut self, now_ms: u64) {
- self.deadline_ms = if crate::async_usb::is_idle() {
- None
- } else {
- Some(now_ms.saturating_add(USB_OUTSTANDING_OP_TIMEOUT_MS))
- };
+ self.deadline_ms =
+ if crate::async_usb::is_idle() || crate::async_usb::waiting_for_next_request() {
+ None
+ } else {
+ Some(now_ms.saturating_add(USB_OUTSTANDING_OP_TIMEOUT_MS))
+ };
}
}
@@ -144,12 +145,14 @@ where
crate::async_usb::spin();
// Respond with NOT_READY if the async task needs more time, or ACK with the payload
// if the task already completed.
+ let response = encode_hww_response();
self.refresh_timeout(now_ms);
- encode_hww_response()
+ response
}
HWW_REQ_RETRY => {
+ let response = encode_hww_response();
self.refresh_timeout(now_ms);
- encode_hww_response()
+ response
}
HWW_REQ_CANCEL => {
// TODO: cancel async usb task.
@@ -160,7 +163,9 @@ where
}
fn tick(&mut self, now_ms: u64) {
- if crate::async_usb::is_idle() {
+ // Match the C transport: after an intermediate ACK there is no outstanding
+ // request to time out. A new connection recovers abandoned workflows with RESET.
+ if crate::async_usb::is_idle() || crate::async_usb::waiting_for_next_request() {
self.deadline_ms = None;
return;
}
@@ -375,8 +380,8 @@ mod tests {
crate::version::FIRMWARE_VERSION_SHORT.len()
);
assert_eq!(handler.deadline_ms, Some(USB_OUTSTANDING_OP_TIMEOUT_MS));
- assert_eq!(crate::async_usb::take_response().unwrap(), hex!("bb"));
- assert!(crate::async_usb::waiting_for_next_request());
+ // Leave the intermediate response unread: INFO must neither consume it nor keep
+ // an outstanding request alive. After ACK, the next request has no deadline.
handler.tick(USB_OUTSTANDING_OP_TIMEOUT_MS + 1);
assert!(crate::async_usb::is_idle());
}
@@ -408,8 +413,13 @@ mod tests {
assert_eq!(response, vec![HWW_RSP_ACK, 0xbb]);
assert!(crate::async_usb::waiting_for_next_request());
+ // There is no outstanding transport request between intermediate ACKs and the
+ // next host request. In particular, host passphrase input may take arbitrarily long.
+ handler.tick(10_000);
+ assert!(crate::async_usb::waiting_for_next_request());
+
let response = handler
- .handle_vendor_command(1, HWW_CMD, &[HWW_REQ_NEW, 0xcc], 0)
+ .handle_vendor_command(1, HWW_CMD, &[HWW_REQ_NEW, 0xcc], 10_001)
.unwrap();
assert_eq!(response, vec![HWW_RSP_ACK, 0xdd]);
assert!(crate::async_usb::is_idle());
### src/rust/bitbox02-rust/src/workflow/unlock.rs
@@ -7,14 +7,27 @@ use crate::workflow::password;
use alloc::vec::Vec;
-/// Confirm the entered mnemonic passphrase with the user. Returns true if the user confirmed it,
-/// false if the user rejected it.
-async fn confirm_mnemonic_passphrase(
+/// Confirm the entered mnemonic passphrase with the user. Returns Ok if the user confirmed it,
+/// Err if the user rejected it.
+pub(crate) async fn confirm_mnemonic_passphrase(
hal: &mut impl crate::hal::Hal,
passphrase: &str,
+ from_host: bool,
) -> Result<(), crate::hal::ui::UserAbort> {
- // Accept empty passphrase without confirmation.
+ // Accept empty passphrase without confirmation when entered on the device. Host input
+ // must always be explicitly confirmed so the host cannot silently select another wallet.
if passphrase.is_empty() {
+ if from_host {
+ return hal
+ .ui()
+ .confirm(&ConfirmParams {
+ title: "Confirm",
+ body: "Use empty passphrase?",
+ longtouch: true,
+ ..Default::default()
+ })
+ .await;
+ }
return Ok(());
}
@@ -40,6 +53,35 @@ async fn confirm_mnemonic_passphrase(
hal.ui().confirm(¶ms).await
}
+/// Enter the passphrase without confirmation. Dropping this future discards partial input.
+pub(crate) async fn enter_mnemonic_passphrase(
+ hal: &mut impl crate::hal::Hal,
+) -> zeroize::Zeroizing<alloc::string::String> {
+ password::enter(
+ hal,
+ "Optional passphrase",
+ password::PasswordType::Bip39Passphrase,
+ CanCancel::No,
+ )
+ .await
+ .expect("not cancelable and does not call memory functions")
+}
+
+/// Enter and visually confirm the passphrase. Dropping this future discards partial input.
+async fn enter_and_confirm_mnemonic_passphrase(
+ hal: &mut impl crate::hal::Hal,
+) -> zeroize::Zeroizing<alloc::string::String> {
+ // Loop until the user confirms.
+ loop {
+ let passphrase = enter_mnemonic_passphrase(hal).await;
+
+ if let Ok(()) = confirm_mnemonic_passphrase(hal, passphrase.as_str(), false).await {
+ return passphrase;
+ }
+ hal.ui().status("Please try again", false).await;
+ }
+}
+
#[derive(Debug)]
pub enum UnlockError {
UserAbort,
@@ -149,25 +191,19 @@ pub async fn unlock_bip39<H: crate::hal::Hal>(
// If setting activated, get the passphrase from the user.
if hal.memory().is_mnemonic_passphrase_enabled() {
- // Loop until the user confirms.
- loop {
- mnemonic_passphrase = password::enter(
- hal,
- "Optional passphrase",
- password::PasswordType::Bip39Passphrase,
- CanCancel::No,
- )
- .await
- .expect("not cancelable and does not call memory functions");
-
- if let Ok(()) = confirm_mnemonic_passphrase(hal, mnemonic_passphrase.as_str()).await {
- break;
- }
-
- hal.ui().status("Please try again", false).await;
- }
+ mnemonic_passphrase = enter_and_confirm_mnemonic_passphrase(hal).await;
}
+ unlock_bip39_with_passphrase(hal, seed, &mnemonic_passphrase, unlock_animation).await;
+}
+
+/// Derive the selected wallet and play the already-created unlock animation.
+async fn unlock_bip39_with_passphrase<H: crate::hal::Hal>(
+ hal: &mut H,
+ seed: &[u8],
+ mnemonic_passphrase: &str,
+ unlock_animation: <H::Ui as crate::hal::ui::Ui>::UnlockAnimation,
+) {
let result = {
let crate::hal::HalSubsystems {
ui,
@@ -185,7 +221,7 @@ pub async fn unlock_bip39<H: crate::hal::Hal>(
crate::keystore::unlock_bip39(
&mut keystore_hal,
seed,
- &mnemonic_passphrase,
+ mnemonic_passphrase,
// for the simulator, we don't yield at all, otherwise unlock becomes very slow in the
// simulator.
#[cfg(any(feature = "c-unit-testing", feature = "simulator-graphical"))]
@@ -215,19 +251,54 @@ pub async fn unlock(hal: &mut impl crate::hal::Hal) -> Result<(), ()> {
if !hal.memory().is_initialized() {
return Err(());
}
+
+ unlock_with_passphrase(hal, async |hal| {
+ Ok(enter_and_confirm_mnemonic_passphrase(hal).await)
+ })
+ .await
+}
+
+/// A cancelled or failed attempt must not leave the password-unlocked seed behind. The
+/// guard is only created for a locked device and is disarmed after successful BIP39 unlock.
+struct UnlockGuard(bool);
+
+impl Drop for UnlockGuard {
+ fn drop(&mut self) {
+ if self.0 {
+ crate::keystore::lock();
+ }
+ }
+}
+
+/// Unlock an initialized device, leaving an already unlocked wallet unchanged.
+/// `enter_passphrase` must return a confirmed passphrase. It is called only when the optional
+/// passphrase feature is enabled. Errors and cancellation relock the keystore.
+pub(crate) async fn unlock_with_passphrase<H: crate::hal::Hal, E>(
+ hal: &mut H,
+ enter_passphrase: impl AsyncFnOnce(&mut H) -> Result<zeroize::Zeroizing<alloc::string::String>, E>,
+) -> Result<(), E> {
if !crate::keystore::is_locked() {
return Ok(());
}
+ let mut guard = UnlockGuard(true);
let unlock_animation = hal.ui().unlock_animation_create();
// Loop unlock until the password is correct or the device resets.
- loop {
+ let seed = loop {
if let Ok(seed) = unlock_keystore(hal, "Enter password", CanCancel::No).await {
- unlock_bip39(hal, &seed, unlock_animation).await;
- return Ok(());
+ break seed;
}
- }
+ };
+
+ let passphrase = if hal.memory().is_mnemonic_passphrase_enabled() {
+ enter_passphrase(hal).await?
+ } else {
+ zeroize::Zeroizing::new(alloc::string::String::new())
+ };
+ unlock_bip39_with_passphrase(hal, &seed, &passphrase, unlock_animation).await;
+ guard.0 = false;
+ Ok(())
}
#[cfg(test)]
### src/rust/bitbox02-sys/build.rs
@@ -87,6 +87,7 @@ const ALLOWLIST_FNS: &[&str] = &[
"keystore_bip39_mnemonic_to_seed",
"keystore_get_bip39_word",
"label_create",
+ "info_centered_create",
"label_fits_width",
"memory_add_noise_remote_static_pubkey",
"memory_ble_enable",
### src/rust/bitbox02-sys/wrapper.h
@@ -29,6 +29,7 @@
#include <ui/components/confirm_swap.h>
#include <ui/components/confirm_transaction.h>
#include <ui/components/empty.h>
+#include <ui/components/info_centered.h>
#include <ui/components/label.h>
#include <ui/components/menu.h>
#include <ui/components/orientation_arrows.h>
### src/rust/bitbox02/src/hal/ui.rs
@@ -195,6 +195,13 @@ impl<Timer: bitbox_hal::timer::Timer> Ui for BitBox02Ui<Timer> {
Timer::delay_for(Duration::from_millis(2000)).await;
}
+ async fn waiting(&mut self, message: &str) {
+ let _no_screensaver = crate::screen_saver::ScreensaverInhibitor::new();
+ let mut component = crate::ui::info_centered_create(message);
+ component.screen_stack_push();
+ core::future::pending::<()>().await;
+ }
+
fn print_screen(&mut self, duration: Duration, msg: &str) {
crate::screen_clear();
crate::ug_font_select_9x9();
### src/rust/bitbox02/src/ui/ui.rs
@@ -14,21 +14,23 @@ use alloc::string::String;
use alloc::vec::Vec;
use core::cell::RefCell;
use core::task::{Poll, Waker};
+use zeroize::Zeroizing;
// Keep enough bytes beyond the C label limit to prove that truncation is needed even when the
// Rust-side cut moves back to a UTF-8 boundary.
const LABEL_TRUNCATE_SIZE: usize = super::types::MAX_LABEL_SIZE + 4;
/// BitBox02 fonts contain glyphs for printable ASCII only. Keep this check at the common UI
/// boundary so unsupported text cannot be silently omitted by the renderer.
-fn display_str_to_cstr_vec(text: &str) -> Vec<c_char> {
+/// The returned buffer is wiped on drop.
+fn display_str_to_cstr_vec(text: &str) -> Zeroizing<Vec<c_char>> {
assert!(
util::ascii::is_printable_ascii(text, util::ascii::Charset::AllNewline),
"BitBox02 UI text contains unsupported characters"
);
- let mut result: Vec<c_char> = text.bytes().map(|byte| byte as c_char).collect();
- result.push(0);
- result
+ // UI strings can contain passphrases or mnemonic words. Include the NUL in the initial
+ // allocation so growing the buffer cannot leave a discarded copy of the secret.
+ util::strings::str_to_cstr_vec_zeroizing(text).unwrap()
}
fn label_fits_width(text: &str, font: *const bitbox02_sys::UG_FONT) -> bool {
@@ -327,6 +329,18 @@ pub fn status_create(text: &str, status_success: bool) -> Component {
}
}
+pub fn info_centered_create(text: &str) -> Component {
+ Component {
+ component: unsafe {
+ bitbox02_sys::info_centered_create(
+ display_str_to_cstr_vec(text).as_ptr(), // copied in C
+ None,
+ )
+ },
+ is_pushed: false,
+ }
+}
+
pub async fn sdcard() -> SdcardResponse {
let _no_screensaver = crate::screen_saver::ScreensaverInhibitor::new();
@@ -439,7 +453,7 @@ pub async fn menu(params: MenuParams<'_>) -> MenuResponse {
//
// Step 1: create the C strings. This var has to be alive until after menu() finishes,
// otherwise the pointers we send to menu_create() will be invalid.
- let words: Vec<Vec<core::ffi::c_char>> = params
+ let words: Vec<_> = params
.words
.iter()
.map(|word| display_str_to_cstr_vec(word))
### src/rust/bitbox02/src/ui/ui_stub.rs
@@ -54,6 +54,10 @@ pub fn status_create(_text: &str, _status_success: bool) -> Component {
Component { is_pushed: false }
}
+pub fn info_centered_create(_text: &str) -> Component {
+ Component { is_pushed: false }
+}
+
pub async fn sdcard() -> SdcardResponse {
panic!("not used");
}
### src/rust/bitbox02/src/ui/ui_stub_c_unit_tests.rs
@@ -57,6 +57,10 @@ pub async fn confirm(params: &ConfirmParams<'_>) -> ConfirmResponse {
pub fn screen_process() {}
+pub fn info_centered_create(_text: &str) -> Component {
+ Component { is_pushed: false }
+}
+
pub fn status_create(text: &str, _status_success: bool) -> Component {
crate::print_stdout(&format!(
"STATUS SCREEN START\nTITLE: {}\nSTATUS SCREEN END\n",
### src/rust/bitbox03/src/ui.rs
@@ -123,6 +123,20 @@ impl<Timer: bitbox_hal::timer::Timer> hal::ui::Ui for BitBox03Ui<Timer> {
Timer::delay_for(Duration::from_millis(2000)).await;
}
+ async fn waiting(&mut self, message: &str) {
+ let screen = LvObj::new().unwrap();
+ screen.set_style_bg_color(lvgl::color::black(), 0);
+ screen.set_style_text_color(lvgl::color::white(), 0);
+ let label = LvLabel::new(&screen).unwrap();
+ label.set_width(380);
+ label.set_text(message).unwrap();
+ label.set_style_text_align(lvgl::LvTextAlign::LV_TEXT_ALIGN_CENTER, 0);
+ label.set_style_text_font(lvgl::fonts::INTER_REGULAR_32, 0);
+ label.align(LvAlign::LV_ALIGN_CENTER, 0, 0);
+ let _screen = self.push_guard(screen);
+ core::future::pending::<()>().await;
+ }
+
fn print_screen(&mut self, _duration: core::time::Duration, _msg: &str) {
todo!()
}
### src/ui/components/trinary_input_string.c
@@ -45,6 +45,7 @@ static char _alphabet_lowercase[] = "abcdefghijklmnopqrstuvwxyz";
static char _digits[] = "0123456789";
// ` and ~ are missing here as they don't legible on the device with arial 9x9. Can add them back
// after tuning the font.
+// Keep in sync with SPECIAL in src/rust/bitbox02-rust/src/hww/api/unlock.rs.
static char _special_chars[] = " !\"#$%&'()*+,-./:;<=>?^[\\]@_{|}";
static const UG_FONT* _font = &font_password_11X12;
### src/ui/components/trinary_input_string.h
@@ -8,6 +8,8 @@
#include <stddef.h>
// including null terminator
+// Keep INPUT_STRING_MAX_SIZE - 1 in sync with MAX_PASSPHRASE_LEN in
+// src/rust/bitbox02-rust/src/hww/api/unlock.rs.
#define INPUT_STRING_MAX_SIZE 150
typedef struct {
### test/simulator-graphical-bb03/Cargo.lock
@@ -420,6 +420,7 @@ name = "bitbox-proto"
version = "0.1.0"
dependencies = [
"prost",
+ "zeroize",
]
[[package]]
### test/simulator-graphical/Cargo.lock
@@ -365,6 +365,7 @@ name = "bitbox-proto"
version = "0.1.0"
dependencies = [
"prost",
+ "zeroize",
]
[[package]]Risk score
Our methodology →Why this scored 47/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.