Tomas Susanka
4 years ago
committed by
GitHub
commit
2c0504ad1c
@ -0,0 +1,78 @@
|
||||
from micropython import const
|
||||
|
||||
import storage.device
|
||||
from trezor import wire
|
||||
from trezor.messages import ButtonRequestType
|
||||
from trezor.messages.ButtonAck import ButtonAck
|
||||
from trezor.messages.ButtonRequest import ButtonRequest
|
||||
from trezor.messages.PassphraseAck import PassphraseAck
|
||||
from trezor.messages.PassphraseRequest import PassphraseRequest
|
||||
from trezor.ui import ICON_CONFIG, draw_simple
|
||||
from trezor.ui.passphrase import CANCELLED, PassphraseKeyboard
|
||||
from trezor.ui.text import Text
|
||||
|
||||
if __debug__:
|
||||
from apps.debug import input_signal
|
||||
|
||||
_MAX_PASSPHRASE_LEN = const(50)
|
||||
|
||||
|
||||
def is_enabled() -> bool:
|
||||
return storage.device.is_passphrase_enabled()
|
||||
|
||||
|
||||
async def get(ctx: wire.Context) -> str:
|
||||
if is_enabled():
|
||||
return await _request_from_user(ctx)
|
||||
else:
|
||||
return ""
|
||||
|
||||
|
||||
async def _request_from_user(ctx: wire.Context) -> str:
|
||||
if storage.device.get_passphrase_always_on_device():
|
||||
passphrase = await _request_on_device(ctx)
|
||||
else:
|
||||
passphrase = await _request_on_host(ctx)
|
||||
if len(passphrase) > _MAX_PASSPHRASE_LEN:
|
||||
raise wire.DataError("Maximum passphrase length is %d" % _MAX_PASSPHRASE_LEN)
|
||||
|
||||
return passphrase
|
||||
|
||||
|
||||
async def _request_on_host(ctx: wire.Context) -> str:
|
||||
_entry_dialog()
|
||||
|
||||
request = PassphraseRequest()
|
||||
ack = await ctx.call(request, PassphraseAck)
|
||||
if ack.on_device:
|
||||
if ack.passphrase is not None:
|
||||
raise wire.DataError("Passphrase provided when it should not be")
|
||||
return await _request_on_device(ctx)
|
||||
|
||||
if ack.passphrase is None:
|
||||
raise wire.DataError(
|
||||
"Passphrase not provided and on_device is False. Use empty string to set an empty passphrase."
|
||||
)
|
||||
return ack.passphrase
|
||||
|
||||
|
||||
async def _request_on_device(ctx: wire.Context) -> str:
|
||||
await ctx.call(ButtonRequest(code=ButtonRequestType.PassphraseEntry), ButtonAck)
|
||||
|
||||
keyboard = PassphraseKeyboard("Enter passphrase", _MAX_PASSPHRASE_LEN)
|
||||
if __debug__:
|
||||
passphrase = await ctx.wait(keyboard, input_signal())
|
||||
else:
|
||||
passphrase = await ctx.wait(keyboard)
|
||||
if passphrase is CANCELLED:
|
||||
raise wire.ActionCancelled("Passphrase entry cancelled")
|
||||
|
||||
assert isinstance(passphrase, str)
|
||||
|
||||
return passphrase
|
||||
|
||||
|
||||
def _entry_dialog() -> None:
|
||||
text = Text("Passphrase entry", ICON_CONFIG)
|
||||
text.normal("Please type your", "passphrase on the", "connected host.")
|
||||
draw_simple(text)
|
||||
@ -1,86 +0,0 @@
|
||||
from micropython import const
|
||||
|
||||
import storage.device
|
||||
from storage import cache
|
||||
from trezor import ui, wire
|
||||
from trezor.messages import ButtonRequestType, PassphraseSourceType
|
||||
from trezor.messages.ButtonAck import ButtonAck
|
||||
from trezor.messages.ButtonRequest import ButtonRequest
|
||||
from trezor.messages.PassphraseAck import PassphraseAck
|
||||
from trezor.messages.PassphraseRequest import PassphraseRequest
|
||||
from trezor.messages.PassphraseStateAck import PassphraseStateAck
|
||||
from trezor.messages.PassphraseStateRequest import PassphraseStateRequest
|
||||
from trezor.ui.passphrase import CANCELLED, PassphraseKeyboard, PassphraseSource
|
||||
from trezor.ui.popup import Popup
|
||||
from trezor.ui.text import Text
|
||||
|
||||
if __debug__:
|
||||
from apps.debug import input_signal
|
||||
|
||||
_MAX_PASSPHRASE_LEN = const(50)
|
||||
|
||||
|
||||
async def protect_by_passphrase(ctx: wire.Context) -> str:
|
||||
if storage.device.has_passphrase():
|
||||
return await request_passphrase(ctx)
|
||||
else:
|
||||
return ""
|
||||
|
||||
|
||||
async def request_passphrase(ctx: wire.Context) -> str:
|
||||
source = storage.device.get_passphrase_source()
|
||||
if source == PassphraseSourceType.ASK:
|
||||
source = await request_passphrase_source(ctx)
|
||||
passphrase = await request_passphrase_ack(
|
||||
ctx, source == PassphraseSourceType.DEVICE
|
||||
)
|
||||
if len(passphrase) > _MAX_PASSPHRASE_LEN:
|
||||
raise wire.DataError("Maximum passphrase length is %d" % _MAX_PASSPHRASE_LEN)
|
||||
return passphrase
|
||||
|
||||
|
||||
async def request_passphrase_source(ctx: wire.Context) -> int:
|
||||
req = ButtonRequest(code=ButtonRequestType.PassphraseType)
|
||||
await ctx.call(req, ButtonAck)
|
||||
|
||||
text = Text("Enter passphrase", ui.ICON_CONFIG)
|
||||
text.normal("Where do you want to", "enter your passphrase?")
|
||||
source = PassphraseSource(text)
|
||||
|
||||
response = await ctx.wait(source)
|
||||
assert isinstance(response, int)
|
||||
return response
|
||||
|
||||
|
||||
async def request_passphrase_ack(ctx: wire.Context, on_device: bool) -> str:
|
||||
if not on_device:
|
||||
text = Text("Passphrase entry", ui.ICON_CONFIG)
|
||||
text.normal("Please type your", "passphrase on the", "connected host.")
|
||||
await Popup(text)
|
||||
|
||||
passphrase_request = PassphraseRequest(on_device=on_device)
|
||||
ack = await ctx.call(passphrase_request, PassphraseAck)
|
||||
|
||||
if on_device:
|
||||
if ack.passphrase is not None:
|
||||
raise wire.ProcessError("Passphrase provided when it should not be")
|
||||
|
||||
keyboard = PassphraseKeyboard("Enter passphrase", _MAX_PASSPHRASE_LEN)
|
||||
if __debug__:
|
||||
passphrase = await ctx.wait(keyboard, input_signal())
|
||||
else:
|
||||
passphrase = await ctx.wait(keyboard)
|
||||
if passphrase is CANCELLED:
|
||||
raise wire.ActionCancelled("Passphrase cancelled")
|
||||
else:
|
||||
if ack.passphrase is None:
|
||||
raise wire.ProcessError("Passphrase not provided")
|
||||
passphrase = ack.passphrase
|
||||
|
||||
assert isinstance(passphrase, str)
|
||||
|
||||
state = cache.get_state(prev_state=ack.state, passphrase=passphrase)
|
||||
state_request = PassphraseStateRequest(state=state)
|
||||
await ctx.call(state_request, PassphraseStateAck)
|
||||
|
||||
return passphrase
|
||||
@ -1,75 +1,36 @@
|
||||
from storage.device import get_device_id
|
||||
from trezor.crypto import hashlib, hmac, random
|
||||
from trezor.crypto import random
|
||||
|
||||
if False:
|
||||
from typing import Optional
|
||||
|
||||
_cached_seed = None # type: Optional[bytes]
|
||||
_cached_seed_without_passphrase = None # type: Optional[bytes]
|
||||
_cached_passphrase = None # type: Optional[str]
|
||||
_cached_passphrase_fprint = b"\x00\x00\x00\x00" # type: bytes
|
||||
APP_COMMON_SEED = 0
|
||||
APP_COMMON_SEED_WITHOUT_PASSPHRASE = 1
|
||||
APP_CARDANO_ROOT = 2
|
||||
APP_MONERO_LIVE_REFRESH = 3
|
||||
|
||||
_cache_session_id = None # type: Optional[bytes]
|
||||
_cache = {}
|
||||
|
||||
def get_state(prev_state: bytes = None, passphrase: str = None) -> Optional[bytes]:
|
||||
if prev_state is None:
|
||||
salt = random.bytes(32) # generate a random salt if no state provided
|
||||
else:
|
||||
salt = prev_state[:32] # use salt from provided state
|
||||
if len(salt) != 32:
|
||||
return None # invalid state
|
||||
if passphrase is None:
|
||||
if _cached_passphrase is None:
|
||||
return None # we don't have any passphrase to compute the state
|
||||
else:
|
||||
passphrase = _cached_passphrase # use cached passphrase
|
||||
return _compute_state(salt, passphrase)
|
||||
|
||||
|
||||
def _compute_state(salt: bytes, passphrase: str) -> bytes:
|
||||
# state = HMAC(passphrase, salt || device_id)
|
||||
message = salt + get_device_id().encode()
|
||||
state = hmac.new(passphrase.encode(), message, hashlib.sha256).digest()
|
||||
return salt + state
|
||||
|
||||
|
||||
def get_seed() -> Optional[bytes]:
|
||||
return _cached_seed
|
||||
|
||||
|
||||
def get_seed_without_passphrase() -> Optional[bytes]:
|
||||
return _cached_seed_without_passphrase
|
||||
|
||||
|
||||
def get_passphrase() -> Optional[str]:
|
||||
return _cached_passphrase
|
||||
|
||||
|
||||
def get_passphrase_fprint() -> bytes:
|
||||
return _cached_passphrase_fprint
|
||||
|
||||
|
||||
def has_passphrase() -> bool:
|
||||
return _cached_passphrase is not None
|
||||
if False:
|
||||
from typing import Any
|
||||
|
||||
|
||||
def set_seed(seed: Optional[bytes]) -> None:
|
||||
global _cached_seed
|
||||
_cached_seed = seed
|
||||
def get_session_id() -> bytes:
|
||||
global _cache_session_id
|
||||
if not _cache_session_id:
|
||||
_cache_session_id = random.bytes(32)
|
||||
return _cache_session_id
|
||||
|
||||
|
||||
def set_seed_without_passphrase(seed: Optional[bytes]) -> None:
|
||||
global _cached_seed_without_passphrase
|
||||
_cached_seed_without_passphrase = seed
|
||||
def set(key: int, value: Any) -> None:
|
||||
_cache[key] = value
|
||||
|
||||
|
||||
def set_passphrase(passphrase: Optional[str]) -> None:
|
||||
global _cached_passphrase, _cached_passphrase_fprint
|
||||
_cached_passphrase = passphrase
|
||||
_cached_passphrase_fprint = _compute_state(b"FPRINT", passphrase or "")[:4]
|
||||
def get(key: int) -> Any:
|
||||
return _cache.get(key)
|
||||
|
||||
|
||||
def clear(keep_passphrase: bool = False) -> None:
|
||||
set_seed(None)
|
||||
set_seed_without_passphrase(None)
|
||||
if not keep_passphrase:
|
||||
set_passphrase(None)
|
||||
def clear() -> None:
|
||||
global _cache_session_id
|
||||
_cache_session_id = None
|
||||
_cache.clear()
|
||||
|
||||
@ -1,8 +0,0 @@
|
||||
# Automatically generated by pb2py
|
||||
# fmt: off
|
||||
if False:
|
||||
from typing_extensions import Literal
|
||||
|
||||
ASK = 0 # type: Literal[0]
|
||||
DEVICE = 1 # type: Literal[1]
|
||||
HOST = 2 # type: Literal[2]
|
||||
@ -1,9 +1,23 @@
|
||||
import sys
|
||||
|
||||
sys.path.append('../src')
|
||||
sys.path.append("../src")
|
||||
|
||||
from ubinascii import hexlify, unhexlify # noqa: F401
|
||||
|
||||
import unittest # noqa: F401
|
||||
|
||||
from trezor import utils # noqa: F401
|
||||
|
||||
|
||||
def await_result(task: Awaitable) -> Any:
|
||||
value = None
|
||||
while True:
|
||||
try:
|
||||
result = task.send(value)
|
||||
except StopIteration as e:
|
||||
return e.value
|
||||
|
||||
if result:
|
||||
value = await_result(result)
|
||||
else:
|
||||
value = None
|
||||
|
||||
@ -0,0 +1,81 @@
|
||||
from common import *
|
||||
from mock import patch
|
||||
from mock_storage import mock_storage
|
||||
|
||||
import storage
|
||||
from storage import cache
|
||||
from trezor.messages.Initialize import Initialize
|
||||
from trezor.messages.ClearSession import ClearSession
|
||||
from trezor.wire import DUMMY_CONTEXT
|
||||
|
||||
from apps.homescreen import handle_Initialize, handle_ClearSession
|
||||
|
||||
KEY = 99
|
||||
|
||||
|
||||
class TestStorageCache(unittest.TestCase):
|
||||
def test_session_id(self):
|
||||
session_id_a = cache.get_session_id()
|
||||
self.assertIsNotNone(session_id_a)
|
||||
session_id_b = cache.get_session_id()
|
||||
self.assertEqual(session_id_a, session_id_b)
|
||||
|
||||
cache.clear()
|
||||
session_id_c = cache.get_session_id()
|
||||
self.assertIsNotNone(session_id_c)
|
||||
self.assertNotEqual(session_id_a, session_id_c)
|
||||
|
||||
def test_get_set(self):
|
||||
value = cache.get(KEY)
|
||||
self.assertIsNone(value)
|
||||
|
||||
cache.set(KEY, "hello")
|
||||
value = cache.get(KEY)
|
||||
self.assertEqual(value, "hello")
|
||||
|
||||
cache.clear()
|
||||
value = cache.get(KEY)
|
||||
self.assertIsNone(value)
|
||||
|
||||
@mock_storage
|
||||
def test_Initialize(self):
|
||||
def call_Initialize(**kwargs):
|
||||
msg = Initialize(**kwargs)
|
||||
return await_result(handle_Initialize(DUMMY_CONTEXT, msg))
|
||||
|
||||
# calling Initialize without an ID allocates a new one
|
||||
session_id = cache.get_session_id()
|
||||
features = call_Initialize()
|
||||
new_session_id = cache.get_session_id()
|
||||
self.assertNotEqual(session_id, new_session_id)
|
||||
self.assertEqual(new_session_id, features.session_id)
|
||||
|
||||
# calling Initialize with the current ID does not allocate a new one
|
||||
features = call_Initialize(session_id=new_session_id)
|
||||
same_session_id = cache.get_session_id()
|
||||
self.assertEqual(new_session_id, same_session_id)
|
||||
self.assertEqual(same_session_id, features.session_id)
|
||||
|
||||
call_Initialize()
|
||||
# calling Initialize with a non-current ID returns a different one
|
||||
features = call_Initialize(session_id=new_session_id)
|
||||
self.assertNotEqual(new_session_id, features.session_id)
|
||||
|
||||
# allocating a new session ID clears the cache
|
||||
cache.set(KEY, "hello")
|
||||
features = call_Initialize()
|
||||
self.assertIsNone(cache.get(KEY))
|
||||
|
||||
# resuming a session does not clear the cache
|
||||
cache.set(KEY, "hello")
|
||||
call_Initialize(session_id=features.session_id)
|
||||
self.assertEqual(cache.get(KEY), "hello")
|
||||
|
||||
# supplying a different session ID clears the cache
|
||||
self.assertNotEqual(new_session_id, features.session_id)
|
||||
call_Initialize(session_id=new_session_id)
|
||||
self.assertIsNone(cache.get(KEY))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -1,5 +0,0 @@
|
||||
# Communication
|
||||
|
||||
We use [Protobuf v2](https://developers.google.com/protocol-buffers/) for host-device communication. The communication cycle is very simple, Trezor receives a message, acts on it and responds with another message. Trezor on its own is incapable of initiating the communication.
|
||||
|
||||
The Protobuf definitions can be found in `common/protob`.
|
||||
@ -0,0 +1,9 @@
|
||||
# Communication
|
||||
|
||||
_Note: In this section we describe the internal functioning of the communication protocol. If you wish to implement Trezor support you should use [Connect](https://github.com/trezor/connect/) or [python-trezor](https://pypi.org/project/trezor/), which will do all this hard work for you._
|
||||
|
||||
We use [Protobuf v2](https://developers.google.com/protocol-buffers/) for host-device communication. The communication cycle is very simple, Trezor receives a message (request), acts on it and responds with another one (response). Trezor on its own is incapable of initiating the communication.
|
||||
|
||||
## Definitions
|
||||
|
||||
Protobuf messages are defined in the [Common](https://github.com/trezor/trezor-firmware/tree/master/common) project, which is part of this monorepo. This repository is also exported to [trezor/trezor-common](https://github.com/trezor/trezor-common) to be used by third parties, which prefer not to include the whole monorepo. That copy is read-only mirror and all changes are happening in this monorepo.
|
||||
@ -0,0 +1,64 @@
|
||||
# Passphrase
|
||||
|
||||
As of \[versions TBD\] we have changed how [passphrase](https://wiki.trezor.io/Passphrase) is communicated between the Host and the Device.
|
||||
|
||||
Passphrase is very tightly coupled with _sessions_. The reader is encouraged to read on that topic first in the [sessions.md](sessions.md) section.
|
||||
|
||||
## Scheme
|
||||
|
||||
As soon as Trezor needs the passphrase to do BIP-39/SLIP-39 derivations it prompts the user for passphrase.
|
||||
|
||||
```
|
||||
GetAddress(...)
|
||||
---------> PassphraseRequest()
|
||||
<---------
|
||||
PassphraseAck
|
||||
(str passphrase, bool on_device)
|
||||
---------> Address(...)
|
||||
<---------
|
||||
```
|
||||
|
||||
In the default Trezor setting, the passphrase is obtained from the Host. Trezor sends a PassphraseRequest message and awaits PassphraseAck as a response. This message contains field `passphrase` to transmit it or it has `on_device` boolean flag to indicate that the user wishes to enter the passphrase on Trezor instead. Setting both `passphrase` and `on_device` to true is forbidden.
|
||||
|
||||
Note that this has changed as of TBD. In previous firmware versions the `on_device` flag was in the PassphraseRequest message, since this decision has been made on Trezor. We also had two additional messages PassphraseStateRequest and PassphraseStateAck which were removed.
|
||||
|
||||
## Example
|
||||
|
||||
On an initialized device with passphrase enabled a common communication starts like this:
|
||||
|
||||
```
|
||||
Initialize()
|
||||
---------> Features(..., session_id)
|
||||
<---------
|
||||
GetAddress(...)
|
||||
---------> PassphraseRequest()
|
||||
<---------
|
||||
PassphraseAck(...)
|
||||
---------> Address(...)
|
||||
<---------
|
||||
```
|
||||
|
||||
The device requested the passphrase since the BIP-39/SLIP-39 seed is not yet cached. After this workflow the seed is cached and the passphrase will therefore never be requested again unless the session is cleared*.
|
||||
|
||||
Since we do not have sessions, the Host can not be sure that someone else has not used the device and applied another session id (e.g. changed the Passphrase). To work around this we send the session id again on every subsequent message. See more on that in [session.md]().
|
||||
|
||||
```
|
||||
Initialize(session_id)
|
||||
---------> Features(..., session_id)
|
||||
<---------
|
||||
GetPublicKey(...)
|
||||
---------> PublicKey(...)
|
||||
<---------
|
||||
```
|
||||
|
||||
As long as the session_id in `Initialize` is the same as the one Trezor stores internally, Trezor guarantees the same passphrase is being used.
|
||||
|
||||
----
|
||||
|
||||
\* There is one exception and that is Cardano. Because Cardano has a different BIP-39/SLIP-39 derivation scheme for passphrase we can not use the cached seed. As a workaround we prompt for the passphrase again in such case and cache the cardano seed in the cardano app directly.
|
||||
|
||||
## Passphrase always on device
|
||||
|
||||
User might want to enforce the passphrase entry on the device every time without the hassle of instructing the Host to do so.
|
||||
|
||||
For such cases the user may apply the *Passphrase always on device* setting. As the name suggests, with this setting the passphrase is prompted on the device right away and no PassphraseRequest/PassphraseAck messages are exchanged. Note that the passphrase is prompted only once for given session id. If the user wishes to enter another passphrase they need to either send Initialize(session_id=None) or replug the device.
|
||||
@ -0,0 +1,36 @@
|
||||
# Sessions (the lack of them)
|
||||
|
||||
Currently the communication protocol lacks sessions, which are planned to be introduced in the near future (see [#79](https://github.com/trezor/trezor-firmware/issues/79)).
|
||||
|
||||
To ensure the device is in the expected state we use something called _session_id_. Session id is a 32 bytes long random blob which identifies the internal device state (mainly its caches). This is primarily useful for passphrase to make sure the same passphrase is cached in the device as the one the user entered a few minutes ago. See [passphrase.md](passphrase.md) for more on that.
|
||||
|
||||
On first initialization the Host does not have a session id and starts the communication with an empty Initialize:
|
||||
|
||||
```
|
||||
Initialize()
|
||||
---------> Features(..., session_id)
|
||||
<---------
|
||||
```
|
||||
|
||||
After the first Features message is received the Host might store the session_id. To ensure the device state Host must send the Initialize message again including that particular session_id:
|
||||
|
||||
```
|
||||
Initialize(session_id)
|
||||
---------> Features(..., session_id)
|
||||
<---------
|
||||
Request
|
||||
---------> Response
|
||||
<---------
|
||||
```
|
||||
|
||||
So to make sure the device state has not changed, the Host must send the Initialize message with the correctly stored session_id before each request. Yes, this is stupid.
|
||||
|
||||
As mentioned, sessions will be introduced soon™ and fix that. We will probably take the first few bytes of the session_id, declare it a session id, and the rest will remain, without the annoying requirement of sending Initialize before every message.
|
||||
|
||||
----
|
||||
|
||||
The session is terminated and therefore the caches are cleared if:
|
||||
- Initialize.session_id is empty.
|
||||
- Initialize.session_id is different then the one cached in Trezor.
|
||||
- Trezor is replugged (session is not persistent).
|
||||
- ClearSession is received.
|
||||
@ -1,7 +1,7 @@
|
||||
#define VERSION_MAJOR 1
|
||||
#define VERSION_MINOR 8
|
||||
#define VERSION_PATCH 4
|
||||
#define VERSION_MINOR 9
|
||||
#define VERSION_PATCH 0
|
||||
|
||||
#define FIX_VERSION_MAJOR 1
|
||||
#define FIX_VERSION_MINOR 8
|
||||
#define FIX_VERSION_MINOR 9
|
||||
#define FIX_VERSION_PATCH 0
|
||||
|
||||
@ -1,8 +0,0 @@
|
||||
# Automatically generated by pb2py
|
||||
# fmt: off
|
||||
if False:
|
||||
from typing_extensions import Literal
|
||||
|
||||
ASK = 0 # type: Literal[0]
|
||||
DEVICE = 1 # type: Literal[1]
|
||||
HOST = 2 # type: Literal[2]
|
||||
@ -1 +1,2 @@
|
||||
junit.xml
|
||||
trezor.log
|
||||
|
||||
@ -0,0 +1,280 @@
|
||||
# This file is part of the Trezor project.
|
||||
#
|
||||
# Copyright (C) 2012-2019 SatoshiLabs and contributors
|
||||
#
|
||||
# This library is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Lesser General Public License version 3
|
||||
# as published by the Free Software Foundation.
|
||||
#
|
||||
# This library is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Lesser General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the License along with this library.
|
||||
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
|
||||
|
||||
import pytest
|
||||
|
||||
from trezorlib import messages
|
||||
from trezorlib.messages import FailureType
|
||||
from trezorlib.tools import parse_path
|
||||
|
||||
XPUB_PASSPHRASE_A = "xpub6CekxGcnqnJ6osfY4Rrq7W5ogFtR54KUvz4H16XzaQuukMFZCGebEpVznfq4yFcKEmYyShwj2UKjL7CazuNSuhdkofF4mHabHkLxCMVvsqG"
|
||||
XPUB_PASSPHRASE_NONE = "xpub6BiVtCpG9fQPxnPmHXG8PhtzQdWC2Su4qWu6XW9tpWFYhxydCLJGrWBJZ5H6qTAHdPQ7pQhtpjiYZVZARo14qHiay2fvrX996oEP42u8wZy"
|
||||
XPUB_CARDANO_PASSPHRASE_B = "d80e770f6dfc3edb58eaab68aa091b2c27b08a47583471e93437ac5f8baa61880c7af4938a941c084c19731e6e57a5710e6ad1196263291aea297ce0eec0f177"
|
||||
|
||||
ADDRESS_N = parse_path("44h/0h/0h")
|
||||
XPUB_REQUEST = messages.GetPublicKey(address_n=ADDRESS_N, coin_name="Bitcoin")
|
||||
|
||||
|
||||
def _init_session(client, session_id=None):
|
||||
"""Call Initialize, check and return the session ID."""
|
||||
response = client.call(messages.Initialize(session_id=session_id))
|
||||
assert isinstance(response, messages.Features)
|
||||
assert len(response.session_id) == 32
|
||||
return response.session_id
|
||||
|
||||
|
||||
def _get_xpub(client, passphrase=None):
|
||||
"""Get XPUB and check that the appropriate passphrase flow has happened."""
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
if passphrase is not None:
|
||||
assert isinstance(response, messages.PassphraseRequest)
|
||||
response = client.call_raw(messages.PassphraseAck(passphrase=passphrase))
|
||||
assert isinstance(response, messages.PublicKey)
|
||||
return response.xpub
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
@pytest.mark.setup_client(passphrase=True)
|
||||
def test_session_with_passphrase(client):
|
||||
# Let's start the communication by calling Initialize.
|
||||
session_id = _init_session(client)
|
||||
|
||||
# GetPublicKey requires passphrase and since it is not cached,
|
||||
# Trezor will prompt for it.
|
||||
assert _get_xpub(client, passphrase="A") == XPUB_PASSPHRASE_A
|
||||
|
||||
# Call Initialize again, this time with the received session id and then call
|
||||
# GetPublicKey. The passphrase should be cached now so Trezor must
|
||||
# not ask for it again, whilst returning the same xpub.
|
||||
new_session_id = _init_session(client, session_id=session_id)
|
||||
assert new_session_id == session_id
|
||||
assert _get_xpub(client, passphrase=None) == XPUB_PASSPHRASE_A
|
||||
|
||||
# If we set session id in Initialize to None, the cache will be cleared
|
||||
# and Trezor will ask for the passphrase again.
|
||||
new_session_id = _init_session(client)
|
||||
assert new_session_id != session_id
|
||||
assert _get_xpub(client, passphrase="A") == XPUB_PASSPHRASE_A
|
||||
|
||||
# Unknown session id is the same as setting it to None.
|
||||
_init_session(client, session_id=b"X" * 32)
|
||||
assert _get_xpub(client, passphrase="A") == XPUB_PASSPHRASE_A
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
def test_session_enable_passphrase(client):
|
||||
# Let's start the communication by calling Initialize.
|
||||
session_id = _init_session(client)
|
||||
|
||||
# Trezor will not prompt for passphrase because it is turned off.
|
||||
assert _get_xpub(client, passphrase=None) == XPUB_PASSPHRASE_NONE
|
||||
|
||||
# Turn on passphrase.
|
||||
# Emit the call explicitly to avoid ClearSession done by the library function
|
||||
response = client.call(messages.ApplySettings(use_passphrase=True))
|
||||
assert isinstance(response, messages.Success)
|
||||
|
||||
# The session id is unchanged, therefore we do not prompt for the passphrase.
|
||||
new_session_id = _init_session(client, session_id=session_id)
|
||||
assert session_id == new_session_id
|
||||
assert _get_xpub(client, passphrase=None) == XPUB_PASSPHRASE_NONE
|
||||
|
||||
# We clear the session id now, so the passphrase should be asked.
|
||||
new_session_id = _init_session(client)
|
||||
assert session_id != new_session_id
|
||||
assert _get_xpub(client, passphrase="A") == XPUB_PASSPHRASE_A
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
@pytest.mark.setup_client(passphrase=True)
|
||||
def test_clear_session_passphrase(client):
|
||||
# at first attempt, we are prompted for passphrase
|
||||
assert _get_xpub(client, passphrase="A") == XPUB_PASSPHRASE_A
|
||||
|
||||
# now the passphrase is cached
|
||||
assert _get_xpub(client, passphrase=None) == XPUB_PASSPHRASE_A
|
||||
|
||||
# Erase the cached passphrase
|
||||
response = client.call(messages.Initialize())
|
||||
assert isinstance(response, messages.Features)
|
||||
|
||||
# we have to enter passphrase again
|
||||
assert _get_xpub(client, passphrase="A") == XPUB_PASSPHRASE_A
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
@pytest.mark.skip_t1
|
||||
@pytest.mark.setup_client(passphrase=True)
|
||||
def test_passphrase_on_device(client):
|
||||
_init_session(client)
|
||||
|
||||
# try to get xpub with passphrase on host:
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PassphraseRequest)
|
||||
response = client.call_raw(messages.PassphraseAck(passphrase="A", on_device=False))
|
||||
assert isinstance(response, messages.PublicKey)
|
||||
assert response.xpub == XPUB_PASSPHRASE_A
|
||||
|
||||
# try to get xpub again, passphrase should be cached
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PublicKey)
|
||||
assert response.xpub == XPUB_PASSPHRASE_A
|
||||
|
||||
# make a new session
|
||||
_init_session(client)
|
||||
|
||||
# try to get xpub with passphrase on device:
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PassphraseRequest)
|
||||
response = client.call_raw(messages.PassphraseAck(on_device=True))
|
||||
assert isinstance(response, messages.ButtonRequest)
|
||||
client.debug.input("A")
|
||||
response = client.call_raw(messages.ButtonAck())
|
||||
assert isinstance(response, messages.PublicKey)
|
||||
assert response.xpub == XPUB_PASSPHRASE_A
|
||||
|
||||
# try to get xpub again, passphrase should be cached
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PublicKey)
|
||||
assert response.xpub == XPUB_PASSPHRASE_A
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
@pytest.mark.skip_t1
|
||||
@pytest.mark.setup_client(passphrase=True)
|
||||
def test_passphrase_always_on_device(client):
|
||||
# Let's start the communication by calling Initialize.
|
||||
session_id = _init_session(client)
|
||||
|
||||
# Force passphrase entry on Trezor.
|
||||
response = client.call(messages.ApplySettings(passphrase_always_on_device=True))
|
||||
assert isinstance(response, messages.Success)
|
||||
|
||||
# Since we enabled the always_on_device setting, Trezor will send ButtonRequests and ask for it on the device.
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.ButtonRequest)
|
||||
client.debug.input("") # Input empty passphrase.
|
||||
response = client.call_raw(messages.ButtonAck())
|
||||
assert isinstance(response, messages.PublicKey)
|
||||
assert response.xpub == XPUB_PASSPHRASE_NONE
|
||||
|
||||
# Passphrase will not be prompted. The session id stays the same and the passphrase is cached.
|
||||
_init_session(client, session_id=session_id)
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PublicKey)
|
||||
assert response.xpub == XPUB_PASSPHRASE_NONE
|
||||
|
||||
# In case we want to add a new passphrase we need to send session_id = None.
|
||||
_init_session(client)
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.ButtonRequest)
|
||||
client.debug.input("A") # Input empty passphrase.
|
||||
response = client.call_raw(messages.ButtonAck())
|
||||
assert isinstance(response, messages.PublicKey)
|
||||
assert response.xpub == XPUB_PASSPHRASE_A
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
@pytest.mark.skip_t2
|
||||
@pytest.mark.setup_client(passphrase=True)
|
||||
def test_passphrase_on_device_not_possible_on_t1(client):
|
||||
# This setting makes no sense on T1.
|
||||
response = client.call_raw(messages.ApplySettings(passphrase_always_on_device=True))
|
||||
assert isinstance(response, messages.Failure)
|
||||
assert response.code == FailureType.DataError
|
||||
|
||||
# T1 should not accept on_device request
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PassphraseRequest)
|
||||
response = client.call_raw(messages.PassphraseAck(on_device=True))
|
||||
assert isinstance(response, messages.Failure)
|
||||
assert response.code == FailureType.DataError
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
@pytest.mark.setup_client(passphrase=True)
|
||||
def test_passphrase_ack_mismatch(client):
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PassphraseRequest)
|
||||
response = client.call_raw(messages.PassphraseAck(passphrase="A", on_device=True))
|
||||
assert isinstance(response, messages.Failure)
|
||||
assert response.code == FailureType.DataError
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
@pytest.mark.setup_client(passphrase=True)
|
||||
def test_passphrase_missing(client):
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PassphraseRequest)
|
||||
response = client.call_raw(messages.PassphraseAck(passphrase=None))
|
||||
assert isinstance(response, messages.Failure)
|
||||
assert response.code == FailureType.DataError
|
||||
|
||||
response = client.call_raw(XPUB_REQUEST)
|
||||
assert isinstance(response, messages.PassphraseRequest)
|
||||
response = client.call_raw(messages.PassphraseAck(passphrase=None, on_device=False))
|
||||
assert isinstance(response, messages.Failure)
|
||||
assert response.code == FailureType.DataError
|
||||
|
||||
|
||||
def _get_xpub_cardano(client, passphrase):
|
||||
msg = messages.CardanoGetPublicKey(address_n=parse_path("44'/1815'/0'/0/0"))
|
||||
response = client.call_raw(msg)
|
||||
if passphrase is not None:
|
||||
assert isinstance(response, messages.PassphraseRequest)
|
||||
response = client.call_raw(messages.PassphraseAck(passphrase=passphrase))
|
||||
assert isinstance(response, messages.CardanoPublicKey)
|
||||
return response.xpub
|
||||
|
||||
|
||||
@pytest.mark.skip_ui
|
||||
@pytest.mark.skip_t1
|
||||
@pytest.mark.altcoin
|
||||
@pytest.mark.setup_client(passphrase=True)
|
||||
def test_cardano_passphrase(client):
|
||||
# Cardano uses a variation of BIP-39 so we need to ask for the passphrase again.
|
||||
|
||||
session_id = _init_session(client)
|
||||
|
||||
# GetPublicKey requires passphrase and since it is not cached,
|
||||
# Trezor will prompt for it.
|
||||
assert _get_xpub(client, passphrase="A") == XPUB_PASSPHRASE_A
|
||||
|
||||
# The passphrase is now cached for non-Cardano coins.
|
||||
assert _get_xpub(client, passphrase=None) == XPUB_PASSPHRASE_A
|
||||
|
||||
# Cardano will prompt for it again.
|
||||
assert _get_xpub_cardano(client, passphrase="B") == XPUB_CARDANO_PASSPHRASE_B
|
||||
|
||||
# But now also Cardano has it cached.
|
||||
assert _get_xpub_cardano(client, passphrase=None) == XPUB_CARDANO_PASSPHRASE_B
|
||||
|
||||
# And others behaviour did not change.
|
||||
assert _get_xpub(client, passphrase=None) == XPUB_PASSPHRASE_A
|
||||
|
||||
# Initialize with the session id does not destroy the state
|
||||
_init_session(client, session_id=session_id)
|
||||
assert _get_xpub(client, passphrase=None) == XPUB_PASSPHRASE_A
|
||||
assert _get_xpub_cardano(client, passphrase=None) == XPUB_CARDANO_PASSPHRASE_B
|
||||
|
||||
# New session will destroy the state
|
||||
_init_session(client)
|
||||
|
||||
# GetPublicKey must ask for passphrase again
|
||||
assert _get_xpub(client, passphrase="A") == XPUB_PASSPHRASE_A
|
||||
|
||||
# Cardano must also ask for passphrase again
|
||||
assert _get_xpub_cardano(client, passphrase="B") == XPUB_CARDANO_PASSPHRASE_B
|
||||
Loading…
Reference in new issue