mirror of
https://github.com/trezor/trezor-firmware.git
synced 2024-11-13 19:18:56 +00:00
src/apps/debug: simplify debuglink, add more decision/state fields
- move data exported over debuglink into apps.debug - move debug signals into apps.debug - make pin/mnemonic dialogs testable - streamline code style of apps.management.reset_device - check __debug__ when debug app starts
This commit is contained in:
parent
f9c51af32f
commit
01bc12ec27
@ -1,13 +1,7 @@
|
||||
from trezor import loop, ui, wire
|
||||
from trezor import ui, wire
|
||||
from trezor.messages import ButtonRequestType, FailureType, wire_types
|
||||
from trezor.messages.ButtonRequest import ButtonRequest
|
||||
from trezor.ui.confirm import CONFIRMED, ConfirmDialog, HoldToConfirmDialog
|
||||
from apps.common import cache
|
||||
|
||||
# used to confirm/cancel the dialogs from outside of this module (i.e.
|
||||
# through debug link)
|
||||
if __debug__:
|
||||
signal = cache.memory.setdefault('confirm_signal', loop.signal())
|
||||
|
||||
|
||||
@ui.layout
|
||||
@ -18,11 +12,7 @@ async def confirm(ctx, content, code=None, *args, **kwargs):
|
||||
|
||||
dialog = ConfirmDialog(content, *args, **kwargs)
|
||||
|
||||
if __debug__:
|
||||
waiter = ctx.wait(signal, dialog)
|
||||
else:
|
||||
waiter = ctx.wait(dialog)
|
||||
return await waiter == CONFIRMED
|
||||
return await ctx.wait(dialog) == CONFIRMED
|
||||
|
||||
|
||||
@ui.layout
|
||||
@ -33,11 +23,7 @@ async def hold_to_confirm(ctx, content, code=None, *args, **kwargs):
|
||||
|
||||
dialog = HoldToConfirmDialog(content, 'Hold to confirm', *args, **kwargs)
|
||||
|
||||
if __debug__:
|
||||
waiter = ctx.wait(signal, dialog)
|
||||
else:
|
||||
waiter = ctx.wait(dialog)
|
||||
return await waiter == CONFIRMED
|
||||
return await ctx.wait(dialog) == CONFIRMED
|
||||
|
||||
|
||||
async def require_confirm(*args, **kwargs):
|
||||
|
@ -1,8 +1,11 @@
|
||||
from trezor import res, ui
|
||||
from trezor import loop, res, ui
|
||||
from trezor.messages import PinMatrixRequestType
|
||||
from trezor.ui.confirm import CONFIRMED, ConfirmDialog
|
||||
from trezor.ui.pin import PinMatrix
|
||||
|
||||
if __debug__:
|
||||
from apps.debug import input_signal
|
||||
|
||||
|
||||
class PinCancelled(Exception):
|
||||
pass
|
||||
@ -42,13 +45,18 @@ async def request_pin(code=None, cancellable: bool=True) -> str:
|
||||
matrix.onchange()
|
||||
|
||||
while True:
|
||||
result = await dialog
|
||||
if __debug__:
|
||||
result = await loop.wait(dialog, input_signal)
|
||||
if isinstance(result, str):
|
||||
return result
|
||||
else:
|
||||
result = await dialog
|
||||
if result == CONFIRMED:
|
||||
return matrix.pin
|
||||
elif result != CONFIRMED and matrix.pin:
|
||||
elif matrix.pin: # reset
|
||||
matrix.change('')
|
||||
continue
|
||||
else:
|
||||
else: # cancel
|
||||
raise PinCancelled()
|
||||
|
||||
|
||||
|
@ -1,69 +1,42 @@
|
||||
import micropython
|
||||
import gc
|
||||
from uctypes import bytes_at, bytearray_at
|
||||
|
||||
from trezor import loop
|
||||
from trezor.wire import register, protobuf_workflow
|
||||
from trezor.messages.wire_types import \
|
||||
DebugLinkDecision, DebugLinkGetState, DebugLinkStop, \
|
||||
DebugLinkMemoryRead, DebugLinkMemoryWrite, DebugLinkFlashErase
|
||||
from trezor.messages.DebugLinkMemory import DebugLinkMemory
|
||||
from trezor import loop, utils
|
||||
from trezor.messages import wire_types
|
||||
from trezor.messages.DebugLinkState import DebugLinkState
|
||||
from trezor.ui.confirm import CONFIRMED, CANCELLED
|
||||
|
||||
from apps.common.confirm import signal
|
||||
from trezor.ui import confirm, swipe
|
||||
from trezor.wire import register, protobuf_workflow
|
||||
from apps.common import storage
|
||||
from apps.management import reset_device
|
||||
|
||||
if not __debug__:
|
||||
utils.halt("debug mode inactive")
|
||||
|
||||
reset_internal_entropy = None
|
||||
reset_current_words = None
|
||||
reset_word_index = None
|
||||
|
||||
confirm_signal = loop.signal()
|
||||
swipe_signal = loop.signal()
|
||||
input_signal = loop.signal()
|
||||
|
||||
|
||||
async def dispatch_DebugLinkDecision(ctx, msg):
|
||||
signal.send(CONFIRMED if msg.yes_no else CANCELLED)
|
||||
if msg.yes_no is not None:
|
||||
confirm_signal.send(confirm.CONFIRMED if msg.yes_no else confirm.CANCELLED)
|
||||
if msg.up_down is not None:
|
||||
swipe_signal.send(swipe.SWIPE_DOWN if msg.up_down else swipe.SWIPE_UP)
|
||||
if msg.input is not None:
|
||||
input_signal.send(msg.input)
|
||||
|
||||
|
||||
async def dispatch_DebugLinkGetState(ctx, msg):
|
||||
m = DebugLinkState()
|
||||
m.mnemonic = storage.get_mnemonic()
|
||||
m.passphrase_protection = storage.has_passphrase()
|
||||
m.reset_entropy = reset_device.internal_entropy
|
||||
m.reset_word = reset_device.current_word
|
||||
m.reset_word_pos = reset_word_index
|
||||
m.reset_entropy = reset_internal_entropy
|
||||
if reset_current_words:
|
||||
m.reset_word = ' '.join(reset_current_words)
|
||||
return m
|
||||
|
||||
|
||||
async def dispatch_DebugLinkStop(ctx, msg):
|
||||
pass
|
||||
|
||||
|
||||
async def dispatch_DebugLinkMemoryRead(ctx, msg):
|
||||
m = DebugLinkMemory()
|
||||
m.memory = bytes_at(msg.address, msg.length)
|
||||
return m
|
||||
|
||||
|
||||
async def dispatch_DebugLinkMemoryWrite(ctx, msg):
|
||||
l = len(msg.memory)
|
||||
data = bytearray_at(msg.address, l)
|
||||
data[0:l] = msg.memory
|
||||
|
||||
|
||||
async def dispatch_DebugLinkFlashErase(ctx, msg):
|
||||
# TODO: erase(msg.sector)
|
||||
pass
|
||||
|
||||
|
||||
async def memory_stats(interval):
|
||||
sleep = loop.sleep(interval * 1000 * 1000)
|
||||
while True:
|
||||
micropython.mem_info()
|
||||
gc.collect()
|
||||
await sleep
|
||||
|
||||
|
||||
def boot():
|
||||
register(DebugLinkDecision, protobuf_workflow, dispatch_DebugLinkDecision)
|
||||
register(DebugLinkGetState, protobuf_workflow, dispatch_DebugLinkGetState)
|
||||
register(DebugLinkStop, protobuf_workflow, dispatch_DebugLinkStop)
|
||||
register(DebugLinkMemoryRead, protobuf_workflow, dispatch_DebugLinkMemoryRead)
|
||||
register(DebugLinkMemoryWrite, protobuf_workflow, dispatch_DebugLinkMemoryWrite)
|
||||
register(DebugLinkFlashErase, protobuf_workflow, dispatch_DebugLinkFlashErase)
|
||||
|
||||
# loop.schedule(memory_stats(10))
|
||||
register(wire_types.DebugLinkDecision, protobuf_workflow, dispatch_DebugLinkDecision)
|
||||
register(wire_types.DebugLinkGetState, protobuf_workflow, dispatch_DebugLinkGetState)
|
||||
|
@ -18,14 +18,10 @@ from apps.common.confirm import require_confirm
|
||||
from apps.management.change_pin import request_pin_confirm
|
||||
|
||||
if __debug__:
|
||||
internal_entropy = None
|
||||
current_word = None
|
||||
from apps import debug
|
||||
|
||||
|
||||
async def reset_device(ctx, msg):
|
||||
if __debug__:
|
||||
global internal_entropy
|
||||
|
||||
# validate parameters and device state
|
||||
if msg.strength not in (128, 192, 256):
|
||||
raise wire.FailureError(
|
||||
@ -36,31 +32,29 @@ async def reset_device(ctx, msg):
|
||||
FailureType.UnexpectedMessage,
|
||||
'Already initialized')
|
||||
|
||||
# request new PIN
|
||||
if msg.pin_protection:
|
||||
# request new PIN
|
||||
newpin = await request_pin_confirm(ctx)
|
||||
else:
|
||||
# new PIN is empty
|
||||
newpin = ''
|
||||
|
||||
# generate and display internal entropy
|
||||
internal_entropy = random.bytes(32)
|
||||
internal_ent = random.bytes(32)
|
||||
if __debug__:
|
||||
debug.reset_internal_entropy = internal_ent
|
||||
if msg.display_random:
|
||||
await show_entropy(ctx, internal_entropy)
|
||||
await show_entropy(ctx, internal_ent)
|
||||
|
||||
# request external entropy and compute mnemonic
|
||||
ack = await ctx.call(EntropyRequest(), wire_types.EntropyAck)
|
||||
mnemonic = generate_mnemonic(
|
||||
msg.strength, internal_entropy, ack.entropy)
|
||||
ent_ack = await ctx.call(EntropyRequest(), wire_types.EntropyAck)
|
||||
mnemonic = generate_mnemonic(msg.strength, internal_ent, ent_ack.entropy)
|
||||
|
||||
if msg.skip_backup:
|
||||
# let user backup the mnemonic later
|
||||
pass
|
||||
else:
|
||||
# warn user about mnemonic safety
|
||||
if not msg.skip_backup:
|
||||
# require confirmation of the mnemonic safety
|
||||
await show_warning(ctx)
|
||||
|
||||
# show mnemonic and require confirmation of a random word
|
||||
while True:
|
||||
# show mnemonic and require confirmation of a random word
|
||||
await show_mnemonic(ctx, mnemonic)
|
||||
if await check_mnemonic(ctx, mnemonic):
|
||||
break
|
||||
@ -77,11 +71,11 @@ async def reset_device(ctx, msg):
|
||||
storage.load_mnemonic(
|
||||
mnemonic=mnemonic, needs_backup=msg.skip_backup)
|
||||
|
||||
# show success message
|
||||
# show success message. if we skipped backup, it's possible that homescreen
|
||||
# is still running, uninterrupted. restart it to pick up new label.
|
||||
if not msg.skip_backup:
|
||||
await show_success(ctx)
|
||||
else:
|
||||
# trigger reload of homescreen
|
||||
workflow.restartdefault()
|
||||
|
||||
return Success(message='Initialized')
|
||||
@ -142,7 +136,7 @@ async def show_success(ctx):
|
||||
cancel=None)
|
||||
|
||||
|
||||
async def show_entropy(ctx, entropy: int):
|
||||
async def show_entropy(ctx, entropy: bytes):
|
||||
estr = hexlify(entropy).decode()
|
||||
lines = chunks(estr, 16)
|
||||
content = Text('Internal entropy', ui.ICON_RESET, ui.MONO, *lines)
|
||||
@ -165,6 +159,9 @@ async def show_mnemonic(ctx, mnemonic: str):
|
||||
|
||||
@ui.layout
|
||||
async def show_mnemonic_page(page: int, page_count: int, pages: list):
|
||||
if __debug__:
|
||||
debug.reset_current_words = [word for _, word in pages[page]]
|
||||
|
||||
lines = ['%2d. %s' % (wi + 1, word) for wi, word in pages[page]]
|
||||
content = Text('Recovery seed', ui.ICON_RESET, ui.MONO, *lines)
|
||||
content = Scrollpage(content, page, page_count)
|
||||
@ -194,6 +191,9 @@ async def check_mnemonic(ctx, mnemonic: str) -> bool:
|
||||
|
||||
@ui.layout
|
||||
async def check_word(ctx, words: list, index: int):
|
||||
if __debug__:
|
||||
debug.reset_word_index = index
|
||||
|
||||
keyboard = MnemonicKeyboard('Type the %s word:' % format_ordinal(index + 1))
|
||||
result = await ctx.wait(keyboard)
|
||||
return result == words[index]
|
||||
|
@ -4,14 +4,20 @@ import protobuf as p
|
||||
|
||||
class DebugLinkDecision(p.MessageType):
|
||||
FIELDS = {
|
||||
1: ('yes_no', p.BoolType, 0), # required
|
||||
1: ('yes_no', p.BoolType, 0),
|
||||
2: ('up_down', p.BoolType, 0),
|
||||
3: ('input', p.UnicodeType, 0),
|
||||
}
|
||||
MESSAGE_WIRE_TYPE = 100
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
yes_no: bool = None,
|
||||
up_down: bool = None,
|
||||
input: str = None,
|
||||
**kwargs,
|
||||
):
|
||||
self.yes_no = yes_no
|
||||
self.up_down = up_down
|
||||
self.input = input
|
||||
p.MessageType.__init__(self, **kwargs)
|
||||
|
@ -15,6 +15,7 @@ class DebugLinkState(p.MessageType):
|
||||
8: ('reset_entropy', p.BytesType, 0),
|
||||
9: ('recovery_fake_word', p.UnicodeType, 0),
|
||||
10: ('recovery_word_pos', p.UVarintType, 0),
|
||||
11: ('reset_word_pos', p.UVarintType, 0),
|
||||
}
|
||||
MESSAGE_WIRE_TYPE = 102
|
||||
|
||||
@ -30,6 +31,7 @@ class DebugLinkState(p.MessageType):
|
||||
reset_entropy: bytes = None,
|
||||
recovery_fake_word: str = None,
|
||||
recovery_word_pos: int = None,
|
||||
reset_word_pos: int = None,
|
||||
**kwargs,
|
||||
):
|
||||
self.layout = layout
|
||||
@ -42,4 +44,5 @@ class DebugLinkState(p.MessageType):
|
||||
self.reset_entropy = reset_entropy
|
||||
self.recovery_fake_word = recovery_fake_word
|
||||
self.recovery_word_pos = recovery_word_pos
|
||||
self.reset_word_pos = reset_word_pos
|
||||
p.MessageType.__init__(self, **kwargs)
|
||||
|
@ -4,6 +4,9 @@ from trezor.ui import Widget
|
||||
from trezor.ui.button import BTN_ACTIVE, BTN_CLICKED, BTN_STARTED, Button
|
||||
from trezor.ui.loader import Loader
|
||||
|
||||
if __debug__:
|
||||
from apps.debug import confirm_signal
|
||||
|
||||
CONFIRMED = const(1)
|
||||
CANCELLED = const(2)
|
||||
DEFAULT_CONFIRM = res.load(ui.ICON_CONFIRM)
|
||||
@ -11,8 +14,12 @@ DEFAULT_CANCEL = res.load(ui.ICON_CANCEL)
|
||||
|
||||
|
||||
class ConfirmDialog(Widget):
|
||||
|
||||
def __init__(self, content, confirm=DEFAULT_CONFIRM, cancel=DEFAULT_CANCEL, confirm_style=ui.BTN_CONFIRM, cancel_style=ui.BTN_CANCEL):
|
||||
def __init__(self,
|
||||
content,
|
||||
confirm=DEFAULT_CONFIRM,
|
||||
cancel=DEFAULT_CANCEL,
|
||||
confirm_style=ui.BTN_CONFIRM,
|
||||
cancel_style=ui.BTN_CANCEL):
|
||||
self.content = content
|
||||
if cancel is not None:
|
||||
self.confirm = Button(
|
||||
@ -37,7 +44,10 @@ class ConfirmDialog(Widget):
|
||||
return CANCELLED
|
||||
|
||||
async def __iter__(self):
|
||||
return await loop.wait(super().__iter__(), self.content)
|
||||
if __debug__:
|
||||
return await loop.wait(super().__iter__(), self.content, confirm_signal)
|
||||
else:
|
||||
return await loop.wait(super().__iter__(), self.content)
|
||||
|
||||
|
||||
_STARTED = const(-1)
|
||||
@ -81,5 +91,8 @@ class HoldToConfirmDialog(Widget):
|
||||
else:
|
||||
content_loop = self.content
|
||||
confirm_loop = super().__iter__() # default loop (render on touch)
|
||||
result = await loop.wait(content_loop, confirm_loop)
|
||||
if __debug__:
|
||||
result = await loop.wait(content_loop, confirm_loop, confirm_signal)
|
||||
else:
|
||||
result = await loop.wait(content_loop, confirm_loop)
|
||||
return result
|
||||
|
@ -3,6 +3,9 @@ from trezor.crypto import bip39
|
||||
from trezor.ui import display
|
||||
from trezor.ui.button import BTN_CLICKED, ICON, Button
|
||||
|
||||
if __debug__:
|
||||
from apps.debug import input_signal
|
||||
|
||||
MNEMONIC_KEYS = ('abc', 'def', 'ghi', 'jkl', 'mno', 'pqr', 'stu', 'vwx', 'yz')
|
||||
|
||||
|
||||
@ -145,6 +148,12 @@ class MnemonicKeyboard(ui.Widget):
|
||||
btn.disable()
|
||||
|
||||
async def __iter__(self):
|
||||
if __debug__:
|
||||
return await loop.wait(self.edit_loop(), input_signal)
|
||||
else:
|
||||
return await self.edit_loop()
|
||||
|
||||
async def edit_loop(self):
|
||||
timeout = loop.sleep(1000 * 1000 * 1)
|
||||
touch = loop.select(io.TOUCH)
|
||||
wait_timeout = loop.wait(touch, timeout)
|
||||
|
@ -1,6 +1,9 @@
|
||||
from micropython import const
|
||||
from trezor import loop, ui, res
|
||||
from .swipe import Swipe, SWIPE_UP, SWIPE_DOWN, SWIPE_VERTICAL
|
||||
from trezor.ui.swipe import Swipe, SWIPE_UP, SWIPE_DOWN, SWIPE_VERTICAL
|
||||
|
||||
if __debug__:
|
||||
from apps.debug import swipe_signal
|
||||
|
||||
|
||||
async def change_page(page, page_count):
|
||||
@ -11,7 +14,11 @@ async def change_page(page, page_count):
|
||||
d = SWIPE_DOWN
|
||||
else:
|
||||
d = SWIPE_VERTICAL
|
||||
s = await Swipe(directions=d)
|
||||
swipe = Swipe(directions=d)
|
||||
if __debug__:
|
||||
s = await loop.wait(swipe, swipe_signal)
|
||||
else:
|
||||
s = await swipe
|
||||
if s == SWIPE_UP:
|
||||
return page + 1 # scroll down
|
||||
elif s == SWIPE_DOWN:
|
||||
|
Loading…
Reference in New Issue
Block a user