storage: remove pbuf, add pin lock

src/apps/common/storage.py

@@ -1,115 +1,166 @@
-import protobuf as p
 from micropython import const
+import ustruct
+import utime
+
 from trezor import config
+from trezor import utils
+
+_APP = const(1)
+
+_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
+        if failure_callback:
+            try:
+                failure_callback(delay_ms)
+            except:
+                pass
+        utime.sleep_ms(delay_ms)
+        return False
+
+
+def lock():
+    global _locked
+    _locked = True
+
+
+def const_equal(a: bytes, b: bytes) -> bool:
+    return a == b
+
 
-_APP_COMMON = const(1)
-
-_CFG_ID = const(0)
-_CFG_VERSION = const(1)
-_CFG_MNEMONIC = const(2)
-_CFG_LANGUAGE = const(3)
-_CFG_LABEL = const(4)
-_CFG_PIN = const(5)
-_CFG_PIN_ATTEMPTS = const(6)
-_CFG_PASSPHRASE_PROTECTION = const(7)
-
-_types = {
-    _CFG_ID: p.UnicodeType,
-    _CFG_VERSION: p.UVarintType,
-    _CFG_MNEMONIC: p.UnicodeType,
-    _CFG_LANGUAGE: p.UnicodeType,
-    _CFG_LABEL: p.UnicodeType,
-    _CFG_PIN: p.UnicodeType,
-    _CFG_PIN_ATTEMPTS: p.UVarintType,
-    _CFG_PASSPHRASE_PROTECTION: p.BoolType,
-}
+# settings
+# ===
 
 
 def get_device_id() -> str:
-    devid = _get(_CFG_ID)
-    if devid is None:
-        devid = _new_device_id()
-        _set(_CFG_ID, devid)
-    return devid
+    dev_id = config_get(_DEVICE_ID).decode()
+    if dev_id is None:
+        dev_id = new_device_id()
+        config_set(_DEVICE_ID, dev_id.encode())
+    return dev_id
 
 
 def is_initialized() -> bool:
-    return _get(_CFG_VERSION) is not None
+    return bool(config_get(_VERSION))
 
 
 def is_protected_by_pin() -> bool:
-    return _get(_CFG_PIN) is not None
+    return bool(config_get(_PIN))
 
 
 def is_protected_by_passphrase() -> bool:
-    return _get(_CFG_PASSPHRASE_PROTECTION) is True
-
-
-def check_pin(pin: str) -> bool:
-    return _get(_CFG_PIN) == pin
+    return bool(bytes_to_int(config_get(_PASSPHRASE_PROTECTION)))
 
 
 def get_pin() -> str:
-    return _get(_CFG_PIN)
+    return config_get(_PIN).decode()
 
 
 def get_label() -> str:
-    return _get(_CFG_LABEL)
+    return config_get(_LABEL).decode()
 
 
 def get_mnemonic() -> str:
-    return _get(_CFG_MNEMONIC)
+    utils.ensure(is_initialized())
+    utils.ensure(not is_locked())
+
+    return config_get(_MNEMONIC).decode()
+
+
+# settings configuration
+# ===
 
 
 def load_mnemonic(mnemonic: str):
-    if is_initialized():
-        raise Exception('Device is already initialized')
-    _set(_CFG_VERSION, 1)
-    _set(_CFG_MNEMONIC, mnemonic)
+    utils.ensure(not is_initialized())
 
+    config_set(_VERSION, int_to_bytes(1))
+    config_set(_MNEMONIC, mnemonic.encode())
 
-def load_settings(language: str,
-                  label: str,
-                  pin: str,
-                  passphrase_protection: bool):
-    if not is_initialized():
-        raise Exception('Device is not initialized')
-    _set(_CFG_LANGUAGE, language or None)
-    _set(_CFG_LABEL, label or None)
-    _set(_CFG_PIN, pin or None)
-    _set(_CFG_PIN_ATTEMPTS, None)
-    _set(_CFG_PASSPHRASE_PROTECTION, passphrase_protection)
 
+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())
 
-def wipe():
-    _set(_CFG_ID, _new_device_id())
-    _set(_CFG_VERSION, None)
-    _set(_CFG_MNEMONIC, None)
-    _set(_CFG_LANGUAGE, None)
-    _set(_CFG_LABEL, None)
-    _set(_CFG_PIN, None)
-    _set(_CFG_PIN_ATTEMPTS, None)
-    _set(_CFG_PASSPHRASE_PROTECTION, None)
-
-
-def _get(key: int):
-    buf = config.get(_APP_COMMON, key)
-    if buf:
-        val = _types[key].loads(buf)
-    else:
-        val = None
-    return val
+    if language is not None:
+        config_set(_LANGUAGE, language.encode())
+    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))
 
 
-def _set(key: int, val):
-    if val is not None:
-        buf = _types[key].dumps(val)
-    else:
-        buf = b''
-    config.set(_APP_COMMON, key, buf)
+def wipe():
+    config.wipe()
 
 
-def _new_device_id() -> str:
+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')
+
+
+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) if b else 0

src/trezor/utils.py

@@ -46,3 +46,8 @@ def unimport(func):
 def chunks(l, n):
     for i in range(0, len(l), n):
         yield l[i:i + n]
+
+
+def ensure(cond):
+    if not cond:
+        raise AssertionError()