storage: use new api, remove pin
This commit is contained in:
Jan Pochyla
Pavol Rusnak
parent
1f6cc70480
commit
8288255048
|
|
@ -22,7 +22,7 @@ async def request_passphrase(ctx):
|
|||
async def protect_by_passphrase(ctx):
|
||||
from apps.common import storage
|
||||
|
||||
if storage.is_protected_by_passphrase():
|
||||
if storage.has_passphrase():
|
||||
return await request_passphrase(ctx)
|
||||
else:
|
||||
return ''
|
||||
|
|
|
|||
|
|
@ -1,37 +1,24 @@
|
|||
from trezor import ui, res
|
||||
from trezor import wire
|
||||
from trezor.utils import unimport
|
||||
from trezor import res
|
||||
from trezor import ui
|
||||
|
||||
if __debug__:
|
||||
matrix = None
|
||||
|
||||
DEFAULT_CANCEL = res.load(ui.ICON_CLEAR)
|
||||
DEFAULT_LOCK = res.load(ui.ICON_LOCK)
|
||||
class PinCancelled(Exception):
|
||||
pass
|
||||
|
||||
|
||||
@ui.layout
|
||||
async def request_pin_on_display(ctx: wire.Context, code: int=None) -> str:
|
||||
from trezor.messages.ButtonRequest import ButtonRequest
|
||||
from trezor.messages.ButtonRequestType import ProtectCall
|
||||
from trezor.messages.FailureType import PinCancelled
|
||||
from trezor.messages.wire_types import ButtonAck
|
||||
async def request_pin(code: int = None) -> str:
|
||||
from trezor.ui.confirm import ConfirmDialog, CONFIRMED
|
||||
from trezor.ui.pin import PinMatrix
|
||||
|
||||
if __debug__:
|
||||
global matrix
|
||||
|
||||
_, label = _get_code_and_label(code)
|
||||
|
||||
await ctx.call(ButtonRequest(code=ProtectCall),
|
||||
ButtonAck)
|
||||
label = _get_label(code)
|
||||
|
||||
def onchange():
|
||||
c = dialog.cancel
|
||||
if matrix.pin:
|
||||
c.content = DEFAULT_CANCEL
|
||||
c.content = res.load(ui.ICON_CLEAR)
|
||||
else:
|
||||
c.content = DEFAULT_LOCK
|
||||
c.content = res.load(ui.ICON_LOCK)
|
||||
c.taint()
|
||||
c.render()
|
||||
|
||||
|
|
@ -44,94 +31,18 @@ async def request_pin_on_display(ctx: wire.Context, code: int=None) -> str:
|
|||
matrix.onchange()
|
||||
|
||||
while True:
|
||||
res = await dialog
|
||||
pin = matrix.pin
|
||||
result = await dialog
|
||||
|
||||
if res == CONFIRMED:
|
||||
matrix = None
|
||||
return pin
|
||||
elif res != CONFIRMED and pin:
|
||||
if result == CONFIRMED:
|
||||
return matrix.pin
|
||||
elif result != CONFIRMED and matrix.pin:
|
||||
matrix.change('')
|
||||
continue
|
||||
else:
|
||||
matrix = None
|
||||
raise wire.FailureError(PinCancelled, 'PIN cancelled')
|
||||
raise PinCancelled()
|
||||
|
||||
|
||||
@ui.layout
|
||||
@unimport
|
||||
async def request_pin_on_client(ctx: wire.Context, code: int=None) -> str:
|
||||
from trezor.messages.FailureType import PinCancelled
|
||||
from trezor.messages.PinMatrixRequest import PinMatrixRequest
|
||||
from trezor.messages.wire_types import PinMatrixAck, Cancel
|
||||
from trezor.ui.pin import PinMatrix
|
||||
|
||||
if __debug__:
|
||||
global matrix
|
||||
|
||||
code, label = _get_code_and_label(code)
|
||||
|
||||
ui.display.clear()
|
||||
matrix = PinMatrix(label)
|
||||
matrix.render()
|
||||
|
||||
ack = await ctx.call(PinMatrixRequest(type=code),
|
||||
PinMatrixAck, Cancel)
|
||||
digits = matrix.digits
|
||||
matrix = None
|
||||
|
||||
if ack.MESSAGE_WIRE_TYPE == Cancel:
|
||||
raise wire.FailureError(PinCancelled, 'PIN cancelled')
|
||||
return _decode_pin(ack.pin, digits)
|
||||
|
||||
|
||||
request_pin = request_pin_on_client
|
||||
|
||||
|
||||
@unimport
|
||||
async def request_pin_twice(ctx: wire.Context) -> str:
|
||||
from trezor.messages.FailureType import ActionCancelled
|
||||
from trezor.messages import PinMatrixRequestType
|
||||
|
||||
pin_first = await request_pin(ctx, PinMatrixRequestType.NewFirst)
|
||||
pin_again = await request_pin(ctx, PinMatrixRequestType.NewSecond)
|
||||
if pin_first != pin_again:
|
||||
# changed message due to consistency with T1 msgs
|
||||
raise wire.FailureError(ActionCancelled, 'PIN change failed')
|
||||
|
||||
return pin_first
|
||||
|
||||
|
||||
async def protect_by_pin_repeatedly(ctx: wire.Context, at_least_once: bool=False):
|
||||
from . import storage
|
||||
|
||||
locked = storage.is_locked() or at_least_once
|
||||
while locked:
|
||||
pin = await request_pin(ctx)
|
||||
locked = not storage.unlock(pin, _render_pin_failure)
|
||||
|
||||
|
||||
async def protect_by_pin_or_fail(ctx: wire.Context, at_least_once: bool=False):
|
||||
from trezor.messages.FailureType import PinInvalid
|
||||
from . import storage
|
||||
|
||||
locked = storage.is_locked() or at_least_once
|
||||
if locked:
|
||||
pin = await request_pin(ctx)
|
||||
if not storage.unlock(pin, _render_pin_failure):
|
||||
raise wire.FailureError(PinInvalid, 'PIN invalid')
|
||||
|
||||
|
||||
protect_by_pin = protect_by_pin_or_fail
|
||||
|
||||
|
||||
def _render_pin_failure(sleep_ms: int):
|
||||
ui.display.clear()
|
||||
ui.display.text_center(240, 240, 'Sleeping for %d seconds' % (sleep_ms / 1000),
|
||||
ui.BOLD, ui.RED, ui.BG)
|
||||
|
||||
|
||||
def _get_code_and_label(code: int):
|
||||
def _get_label(code: int):
|
||||
from trezor.messages import PinMatrixRequestType
|
||||
if code is None:
|
||||
code = PinMatrixRequestType.Current
|
||||
|
|
@ -141,8 +52,4 @@ def _get_code_and_label(code: int):
|
|||
label = 'Enter PIN again'
|
||||
else: # PinMatrixRequestType.Current
|
||||
label = 'Enter PIN'
|
||||
return code, label
|
||||
|
||||
|
||||
def _decode_pin(pin: str, secret: list) -> str:
|
||||
return ''.join([str(secret[int(d) - 1]) for d in pin])
|
||||
return label
|
||||
|
|
|
|||
|
|
@ -21,14 +21,11 @@ async def get_seed(ctx: wire.Context) -> bytes:
|
|||
async def compute_seed(ctx: wire.Context) -> bytes:
|
||||
from trezor.messages.FailureType import ProcessError
|
||||
from .request_passphrase import protect_by_passphrase
|
||||
from .request_pin import protect_by_pin
|
||||
from . import storage
|
||||
|
||||
if not storage.is_initialized():
|
||||
raise wire.FailureError(ProcessError, 'Device is not initialized')
|
||||
|
||||
await protect_by_pin(ctx)
|
||||
|
||||
passphrase = await protect_by_passphrase(ctx)
|
||||
return bip39.seed(storage.get_mnemonic(), passphrase)
|
||||
|
||||
|
|
@ -37,8 +34,6 @@ def get_root_without_passphrase(curve_name=_DEFAULT_CURVE):
|
|||
from . import storage
|
||||
if not storage.is_initialized():
|
||||
raise Exception('Device is not initialized')
|
||||
if storage.is_locked():
|
||||
raise Exception('Unlock first')
|
||||
seed = bip39.seed(storage.get_mnemonic(), '')
|
||||
root = bip32.from_seed(seed, curve_name)
|
||||
return root
|
||||
|
|
|
|||
|
|
@ -1,152 +1,62 @@
|
|||
from micropython import const
|
||||
|
||||
import ustruct
|
||||
import utime
|
||||
|
||||
from trezor import config
|
||||
from trezor import utils
|
||||
|
||||
_APP = const(1)
|
||||
_STORAGE_VERSION = b'\x01'
|
||||
|
||||
DEVICE_ID = const(0) # str
|
||||
VERSION = const(1) # varint
|
||||
MNEMONIC = const(2) # str
|
||||
LANGUAGE = const(3) # str
|
||||
LABEL = const(4) # str
|
||||
PIN = const(5) # bytes
|
||||
PIN_FAILS = const(6) # varint
|
||||
PASSPHRASE_PROTECTION = const(7) # varint
|
||||
|
||||
|
||||
# pin lock
|
||||
# ===
|
||||
|
||||
_locked = True
|
||||
|
||||
|
||||
def is_locked() -> bool:
|
||||
return is_protected_by_pin() and _locked
|
||||
|
||||
|
||||
def unlock(user_pin: str, failure_callback=None) -> bool:
|
||||
global _locked
|
||||
|
||||
if not is_protected_by_pin():
|
||||
return True
|
||||
|
||||
# increment the pin fail counter before checking the pin
|
||||
fails = bytes_to_int(config_get(PIN_FAILS)) + 1
|
||||
config_set_checked(PIN_FAILS, int_to_bytes(fails))
|
||||
|
||||
if const_equal(config_get(PIN), user_pin.encode()):
|
||||
# unlock and reset the counter
|
||||
_locked = False
|
||||
config_set(PIN_FAILS, int_to_bytes(0))
|
||||
return True
|
||||
|
||||
else:
|
||||
# lock, run the callback (ie for ui) and sleep for a quadratic delay
|
||||
_locked = True
|
||||
delay_ms = fails * fails * 1000
|
||||
try:
|
||||
if failure_callback:
|
||||
failure_callback(delay_ms)
|
||||
finally:
|
||||
utime.sleep_ms(delay_ms)
|
||||
return False
|
||||
|
||||
|
||||
def lock():
|
||||
global _locked
|
||||
_locked = True
|
||||
|
||||
|
||||
def const_equal(a: bytes, b: bytes) -> bool:
|
||||
return a == b # TODO: proper const equal
|
||||
|
||||
|
||||
# settings
|
||||
# ===
|
||||
_APP = const(0x0001) # app namespace
|
||||
_DEVICE_ID = const(0x0000) # bytes
|
||||
_VERSION = const(0x0001) # int
|
||||
_MNEMONIC = const(0x0002) # str
|
||||
_LANGUAGE = const(0x0003) # str
|
||||
_LABEL = const(0x0004) # str
|
||||
_USE_PASSPHRASE = const(0x0005) # 0x01 or empty
|
||||
|
||||
|
||||
def get_device_id() -> str:
|
||||
dev_id = config_get(DEVICE_ID).decode()
|
||||
dev_id = config.get(_APP, _DEVICE_ID).decode()
|
||||
if not dev_id:
|
||||
dev_id = new_device_id()
|
||||
config_set(DEVICE_ID, dev_id.encode())
|
||||
config.set(_APP, _DEVICE_ID, dev_id.encode())
|
||||
return dev_id
|
||||
|
||||
|
||||
def is_initialized() -> bool:
|
||||
return bool(config_get(VERSION))
|
||||
|
||||
|
||||
def is_protected_by_pin() -> bool:
|
||||
return bool(config_get(PIN))
|
||||
|
||||
|
||||
def is_protected_by_passphrase() -> bool:
|
||||
return bool(bytes_to_int(config_get(PASSPHRASE_PROTECTION)))
|
||||
|
||||
|
||||
def get_pin() -> str:
|
||||
return config_get(PIN).decode()
|
||||
return bool(config.get(_APP, _VERSION))
|
||||
|
||||
|
||||
def get_label() -> str:
|
||||
return config_get(LABEL).decode()
|
||||
|
||||
|
||||
def get_language() -> str:
|
||||
return config_get(LANGUAGE).decode() or _DEFAULT_LANGUAGE
|
||||
return config.get(_APP, _LABEL).decode()
|
||||
|
||||
|
||||
def get_mnemonic() -> str:
|
||||
utils.ensure(is_initialized())
|
||||
utils.ensure(not is_locked())
|
||||
|
||||
return config_get(MNEMONIC).decode()
|
||||
return config.get(_APP, _MNEMONIC).decode()
|
||||
|
||||
|
||||
# settings configuration
|
||||
# ===
|
||||
def has_passphrase() -> bool:
|
||||
return bool(config.get(_APP, _USE_PASSPHRASE))
|
||||
|
||||
|
||||
def load_mnemonic(mnemonic: str):
|
||||
utils.ensure(not is_initialized())
|
||||
|
||||
config_set(VERSION, int_to_bytes(1))
|
||||
config_set(MNEMONIC, mnemonic.encode())
|
||||
config.set(_APP, _VERSION, _STORAGE_VERSION)
|
||||
config.set(_APP, _MNEMONIC, mnemonic.encode())
|
||||
|
||||
|
||||
_ALLOWED_LANGUAGES = ('english')
|
||||
_DEFAULT_LANGUAGE = 'english'
|
||||
|
||||
|
||||
def load_settings(language: str=None,
|
||||
label: str=None,
|
||||
pin: str=None,
|
||||
passphrase_protection: bool=None):
|
||||
utils.ensure(is_initialized())
|
||||
utils.ensure(not is_locked())
|
||||
|
||||
if language is not None and language in _ALLOWED_LANGUAGES:
|
||||
if language is _DEFAULT_LANGUAGE:
|
||||
config_set(LANGUAGE, b'')
|
||||
else:
|
||||
config_set(LANGUAGE, language.encode())
|
||||
def load_settings(label: str = None, use_passphrase: bool = None):
|
||||
if label is not None:
|
||||
config_set(LABEL, label.encode())
|
||||
if pin is not None:
|
||||
config_set(PIN, pin.encode())
|
||||
if passphrase_protection is not None:
|
||||
config_set(PASSPHRASE_PROTECTION,
|
||||
int_to_bytes(passphrase_protection))
|
||||
config.set(_APP, _LABEL, label.encode())
|
||||
if use_passphrase is True:
|
||||
config.set(_APP, _USE_PASSPHRASE, b'\x01')
|
||||
if use_passphrase is False:
|
||||
config.set(_APP, _USE_PASSPHRASE, b'')
|
||||
|
||||
|
||||
def change_pin(pin: str, newpin: str):
|
||||
return config.change_pin(pin, newpin)
|
||||
|
||||
|
||||
def wipe():
|
||||
from . import cache
|
||||
lock()
|
||||
config.wipe()
|
||||
cache.clear()
|
||||
|
||||
|
|
@ -155,28 +65,3 @@ def new_device_id() -> str:
|
|||
from ubinascii import hexlify
|
||||
from trezor.crypto import random
|
||||
return hexlify(random.bytes(12)).decode('ascii').upper()
|
||||
|
||||
|
||||
def config_get(key: int) -> bytes:
|
||||
return config.get(_APP, key)
|
||||
|
||||
|
||||
def config_set(key: int, value: bytes):
|
||||
config.set(_APP, key, value)
|
||||
|
||||
|
||||
def config_set_checked(key, value: bytes):
|
||||
config_set(key, value)
|
||||
check = config_get(key)
|
||||
if check != value:
|
||||
utils.halt('config.set failed')
|
||||
|
||||
|
||||
# TODO: store ints as varints
|
||||
|
||||
def int_to_bytes(i: int) -> bytes:
|
||||
return ustruct.pack('>L', i) if i else bytes()
|
||||
|
||||
|
||||
def bytes_to_int(b: bytes) -> int:
|
||||
return ustruct.unpack('>L', b)[0] if b else 0
|
||||
|
|
|
|||
|
|
@ -19,10 +19,10 @@ async def respond_Features(ctx, msg):
|
|||
|
||||
f.device_id = storage.get_device_id()
|
||||
f.label = storage.get_label()
|
||||
f.language = storage.get_language()
|
||||
f.initialized = storage.is_initialized()
|
||||
f.pin_protection = storage.is_protected_by_pin()
|
||||
f.passphrase_protection = storage.is_protected_by_passphrase()
|
||||
f.passphrase_protection = storage.has_passphrase()
|
||||
f.pin_protection = False
|
||||
f.language = 'english'
|
||||
|
||||
return f
|
||||
|
||||
|
|
@ -41,10 +41,6 @@ async def respond_Pong(ctx, msg):
|
|||
from trezor import ui
|
||||
await require_confirm(ctx, Text('Confirm', ui.ICON_RESET), ProtectCall)
|
||||
|
||||
if msg.pin_protection:
|
||||
from apps.common.request_pin import protect_by_pin
|
||||
await protect_by_pin(ctx)
|
||||
|
||||
if msg.passphrase_protection:
|
||||
from apps.common.request_passphrase import protect_by_passphrase
|
||||
await protect_by_passphrase(ctx)
|
||||
|
|
|
|||
|
|
@ -8,11 +8,8 @@ async def layout_apply_settings(ctx, msg):
|
|||
from trezor.messages.FailureType import ProcessError
|
||||
from trezor.ui.text import Text
|
||||
from ..common.confirm import require_confirm
|
||||
from ..common.request_pin import protect_by_pin
|
||||
from ..common import storage
|
||||
|
||||
await protect_by_pin(ctx)
|
||||
|
||||
if msg.homescreen is not None:
|
||||
raise wire.FailureError(
|
||||
ProcessError, 'ApplySettings.homescreen is not supported')
|
||||
|
|
@ -42,7 +39,6 @@ async def layout_apply_settings(ctx, msg):
|
|||
'encryption?'))
|
||||
|
||||
storage.load_settings(label=msg.label,
|
||||
language=msg.language,
|
||||
passphrase_protection=msg.use_passphrase)
|
||||
use_passphrase=msg.use_passphrase)
|
||||
|
||||
return Success(message='Settings applied')
|
||||
|
|
|
|||
|
|
@ -26,9 +26,9 @@ async def layout_load_device(ctx, msg):
|
|||
ui.NORMAL, 'Continue only if you', 'know what you are doing!'))
|
||||
|
||||
storage.load_mnemonic(msg.mnemonic)
|
||||
storage.load_settings(pin=msg.pin,
|
||||
passphrase_protection=msg.passphrase_protection,
|
||||
language=msg.language,
|
||||
storage.load_settings(use_passphrase=msg.passphrase_protection,
|
||||
label=msg.label)
|
||||
if msg.pin:
|
||||
storage.change_pin('', msg.pin)
|
||||
|
||||
return Success(message='Device loaded')
|
||||
|
|
|
|||
Loading…
Reference in New Issue