style(core/apps): use new syntax for typing

pull/1382/head
Pavol Rusnak 4 years ago committed by matejcik
parent 0036686588
commit 1e8673bf5f

@ -139,9 +139,9 @@ async def handle_Ping(ctx: wire.Context, msg: Ping) -> Success:
async def handle_DoPreauthorized(
ctx: wire.Context, msg: DoPreauthorized
) -> protobuf.MessageType:
authorization = storage.cache.get(
authorization: Authorization = storage.cache.get(
storage.cache.APP_BASE_AUTHORIZATION
) # type: Authorization
)
if not authorization:
raise wire.ProcessError("No preauthorized operation")
@ -157,9 +157,7 @@ async def handle_DoPreauthorized(
def set_authorization(authorization: Authorization) -> None:
previous = storage.cache.get(
storage.cache.APP_BASE_AUTHORIZATION
) # type: Authorization
previous: Authorization = storage.cache.get(storage.cache.APP_BASE_AUTHORIZATION)
if previous:
previous.__del__()
storage.cache.set(storage.cache.APP_BASE_AUTHORIZATION, authorization)
@ -168,9 +166,9 @@ def set_authorization(authorization: Authorization) -> None:
async def handle_CancelAuthorization(
ctx: wire.Context, msg: CancelAuthorization
) -> protobuf.MessageType:
authorization = storage.cache.get(
authorization: Authorization = storage.cache.get(
storage.cache.APP_BASE_AUTHORIZATION
) # type: Authorization
)
if not authorization:
raise wire.ProcessError("No preauthorized operation")

@ -35,12 +35,14 @@ MULTISIG_OUTPUT_SCRIPT_TYPES = (
OutputScriptType.PAYTOWITNESS,
)
CHANGE_OUTPUT_TO_INPUT_SCRIPT_TYPES = {
CHANGE_OUTPUT_TO_INPUT_SCRIPT_TYPES: Dict[
EnumTypeOutputScriptType, EnumTypeInputScriptType
] = {
OutputScriptType.PAYTOADDRESS: InputScriptType.SPENDADDRESS,
OutputScriptType.PAYTOMULTISIG: InputScriptType.SPENDMULTISIG,
OutputScriptType.PAYTOP2SHWITNESS: InputScriptType.SPENDP2SHWITNESS,
OutputScriptType.PAYTOWITNESS: InputScriptType.SPENDWITNESS,
} # type: Dict[EnumTypeOutputScriptType, EnumTypeInputScriptType]
}
INTERNAL_INPUT_SCRIPT_TYPES = tuple(CHANGE_OUTPUT_TO_INPUT_SCRIPT_TYPES.values())
CHANGE_OUTPUT_SCRIPT_TYPES = tuple(CHANGE_OUTPUT_TO_INPUT_SCRIPT_TYPES.keys())

@ -19,7 +19,7 @@ if False:
from .authorization import CoinJoinAuthorization
class MsgWithCoinName(Protocol):
coin_name = ... # type: str
coin_name: str
class MsgWithAddressScriptType(Protocol):
# XXX should be Bip32Path but that fails

@ -46,9 +46,9 @@ async def sign_tx(
authorization: Optional[CoinJoinAuthorization] = None,
) -> TxRequest:
if authorization:
approver = approvers.CoinJoinApprover(
approver: approvers.Approver = approvers.CoinJoinApprover(
msg, coin, authorization
) # type: approvers.Approver
)
else:
approver = approvers.BasicApprover(msg, coin)
@ -64,8 +64,8 @@ async def sign_tx(
signer = signer_class(msg, keychain, coin, approver).signer()
res = None # type: Union[TxAckType, bool, None]
field_cache = {} # type: FieldCache
res: Union[TxAckType, bool, None] = None
field_cache: FieldCache = {}
while True:
req = signer.send(res)
if isinstance(req, tuple):

@ -293,7 +293,7 @@ class CoinJoinApprover(Approver):
# Add the coordinator fee for the last group of outputs.
self._new_group(0)
decimal_divisor = pow(10, FEE_PER_ANONYMITY_DECIMALS + 2) # type: float
decimal_divisor: float = pow(10, FEE_PER_ANONYMITY_DECIMALS + 2)
return (
self.coordinator_fee_base
* self.authorization.fee_per_anonymity

@ -81,10 +81,10 @@ class Bitcoin:
self.approver = approver
# set of indices of inputs which are segwit
self.segwit = set() # type: Set[int]
self.segwit: Set[int] = set()
# set of indices of inputs which are external
self.external = set() # type: Set[int]
self.external: Set[int] = set()
# transaction and signature serialization
self.serialized_tx = writers.empty_bytearray(_MAX_SERIALIZED_CHUNK_SIZE)
@ -102,7 +102,7 @@ class Bitcoin:
# h_inputs is a digest of the inputs streamed for approval in Step 1, which
# is used to ensure that the inputs streamed for verification in Step 3 are
# the same as those in Step 1.
self.h_inputs = None # type: Optional[bytes]
self.h_inputs: Optional[bytes] = None
progress.init(tx.inputs_count, tx.outputs_count)

@ -43,7 +43,7 @@ class MatchChecker(Generic[T]):
UNDEFINED = object()
def __init__(self) -> None:
self.attribute = self.UNDEFINED # type: Union[object, T]
self.attribute: Union[object, T] = self.UNDEFINED
self.read_only = False # Failsafe to ensure that add_input() is not accidentally called after output_matches().
def attribute_from_tx(self, txio: Union[TxInput, TxOutput]) -> T:

@ -31,8 +31,8 @@ class SignatureVerifier:
coin: CoinInfo,
):
self.threshold = 1
self.public_keys = [] # type: List[bytes]
self.signatures = [] # type: List[Tuple[bytes, int]]
self.public_keys: List[bytes] = []
self.signatures: List[Tuple[bytes, int]] = []
if not script_sig:
if not witness:

@ -31,7 +31,7 @@ async def verify_message(ctx: wire.Context, msg: VerifyMessage) -> Success:
recid = signature[0]
if recid >= 27 and recid <= 34:
# p2pkh
script_type = SPENDADDRESS # type: EnumTypeInputScriptType
script_type: EnumTypeInputScriptType = SPENDADDRESS
elif recid >= 35 and recid <= 38:
# segwit-in-p2sh
script_type = SPENDP2SHWITNESS

@ -149,7 +149,7 @@ def _cbor_decode(cbor: bytes) -> Tuple[Value, bytes]:
return (data[0:ln].decode(), data[ln:])
elif fb_type == _CBOR_ARRAY:
if fb_aux == _CBOR_VAR_FOLLOWS:
res = [] # type: Value
res: Value = []
data = cbor[1:]
while True:
item, data = _cbor_decode(data)

@ -41,7 +41,7 @@ async def confirm(
cancel_style,
major_confirm,
)
dialog = content # type: ui.Layout
dialog: ui.Layout = content
else:
dialog = Confirm(
content, confirm, confirm_style, cancel, cancel_style, major_confirm
@ -97,7 +97,7 @@ async def hold_to_confirm(
content.pages[-1] = HoldToConfirm(
content.pages[-1], confirm, confirm_style, loader_style, cancel
)
dialog = content # type: ui.Layout
dialog: ui.Layout = content
else:
dialog = HoldToConfirm(content, confirm, confirm_style, loader_style, cancel)

@ -54,8 +54,8 @@ FORBIDDEN_KEY_PATH = wire.DataError("Forbidden key path")
class LRUCache:
def __init__(self, size: int) -> None:
self.size = size
self.cache_keys = [] # type: List[Any]
self.cache = {} # type: Dict[Any, Deletable]
self.cache_keys: List[Any] = []
self.cache: Dict[Any, Deletable] = {}
def insert(self, key: Any, value: Deletable) -> None:
if key in self.cache_keys:
@ -126,7 +126,7 @@ class Keychain:
new_root: Callable[[], NodeType],
) -> NodeType:
cached_prefix = tuple(path[:prefix_len])
cached_root = self._cache.get(cached_prefix) # type: Optional[NodeType]
cached_root: Optional[NodeType] = self._cache.get(cached_prefix)
if cached_root is None:
cached_root = new_root()
cached_root.derive_path(cached_prefix)

@ -70,7 +70,7 @@ async def show_pubkey(ctx: wire.Context, pubkey: bytes) -> None:
async def show_xpub(ctx: wire.Context, xpub: str, desc: str, cancel: str) -> bool:
pages = [] # type: List[ui.Component]
pages: List[ui.Component] = []
for lines in chunks(list(chunks(xpub, 16)), 5):
text = Text(desc, ui.ICON_RECEIVE, ui.GREEN)
text.mono(*lines)

@ -13,9 +13,9 @@ def read_setting() -> EnumTypeSafetyCheckLevel:
"""
Returns the effective safety check level.
"""
temporary_safety_check_level = storage.cache.get(
APP_COMMON_SAFETY_CHECKS_TEMPORARY
) # type: Optional[EnumTypeSafetyCheckLevel]
temporary_safety_check_level: Optional[
EnumTypeSafetyCheckLevel
] = storage.cache.get(APP_COMMON_SAFETY_CHECKS_TEMPORARY)
if temporary_safety_check_level is not None:
return temporary_safety_check_level
else:

@ -19,7 +19,7 @@ async def _wrong_card_dialog(ctx: wire.GenericContext) -> bool:
text.br_half()
if SD_CARD_HOT_SWAPPABLE:
text.normal("Please insert the", "correct SD card for", "this device.")
btn_confirm = "Retry" # type: Optional[str]
btn_confirm: Optional[str] = "Retry"
btn_cancel = "Abort"
else:
text.normal("Please unplug the", "device and insert the", "correct SD card.")
@ -35,7 +35,7 @@ async def insert_card_dialog(ctx: wire.GenericContext) -> bool:
text.br_half()
if SD_CARD_HOT_SWAPPABLE:
text.normal("Please insert your", "SD card.")
btn_confirm = "Retry" # type: Optional[str]
btn_confirm: Optional[str] = "Retry"
btn_cancel = "Abort"
else:
text.normal("Please unplug the", "device and insert your", "SD card.")

@ -23,7 +23,7 @@ if __debug__:
save_screen = False
save_screen_directory = "."
reset_internal_entropy = None # type: Optional[bytes]
reset_internal_entropy: Optional[bytes] = None
reset_current_words = loop.chan()
reset_word_index = loop.chan()
@ -37,7 +37,7 @@ if __debug__:
debuglink_decision_chan = loop.chan()
layout_change_chan = loop.chan()
current_content = [] # type: List[str]
current_content: List[str] = []
watch_layout_changes = False
def screenshot() -> bool:

@ -16,7 +16,7 @@ if False:
from apps.common.keychain import MsgOut, Handler, HandlerWithKeychain
class MsgWithAddressN(MessageType, Protocol):
address_n = ... # type: paths.Bip32Path
address_n: paths.Bip32Path
# We believe Ethereum should use 44'/60'/a' for everything, because it is

@ -166,7 +166,7 @@ async def _process_words(
share = None
if not is_slip39: # BIP-39
secret = recover.process_bip39(words) # type: Optional[bytes]
secret: Optional[bytes] = recover.process_bip39(words)
else:
secret, share = recover.process_slip39(words)

@ -115,7 +115,7 @@ class Bip39Keyboard(ui.Layout):
("abc", "def", "ghi", "jkl", "mno", "pqr", "stu", "vwx", "yz")
)
]
self.pending_button = None # type: Optional[Button]
self.pending_button: Optional[Button] = None
self.pending_index = 0
def dispatch(self, event: int, x: int, y: int) -> None:

@ -28,8 +28,8 @@ class InputButton(Button):
def __init__(self, area: ui.Area, keyboard: "Slip39Keyboard") -> None:
super().__init__(area, "")
self.word = ""
self.pending_button = None # type: Optional[Button]
self.pending_index = None # type: Optional[int]
self.pending_button: Optional[Button] = None
self.pending_index: Optional[int] = None
self.keyboard = keyboard
self.disable()
@ -118,7 +118,7 @@ class Slip39Keyboard(ui.Layout):
("ab", "cd", "ef", "ghij", "klm", "nopq", "rs", "tuv", "wxyz")
)
]
self.pending_button = None # type: Optional[Button]
self.pending_button: Optional[Button] = None
self.pending_index = 0
self.button_sequence = ""
self.mask = slip39.KEYBOARD_FULL_MASK

@ -53,12 +53,12 @@ async def request_mnemonic(
) -> Optional[str]:
await button_request(ctx, code=ButtonRequestType.MnemonicInput)
words = [] # type: List[str]
words: List[str] = []
for i in range(word_count):
if backup_types.is_slip39_word_count(word_count):
keyboard = Slip39Keyboard(
keyboard: Union[Slip39Keyboard, Bip39Keyboard] = Slip39Keyboard(
"Type word %s of %s:" % (i + 1, word_count)
) # type: Union[Slip39Keyboard, Bip39Keyboard]
)
else:
keyboard = Bip39Keyboard("Type word %s of %s:" % (i + 1, word_count))
@ -86,7 +86,7 @@ async def show_remaining_shares(
shares_remaining: List[int],
group_threshold: int,
) -> None:
pages = [] # type: List[ui.Component]
pages: List[ui.Component] = []
for remaining, group in groups:
if 0 < remaining < MAX_SHARE_COUNT:
text = Text("Remaining Shares")

@ -37,11 +37,11 @@ class State:
- spend private/public key
- and its corresponding address
"""
self.creds = None # type: Optional[AccountCreds]
self.creds: Optional[AccountCreds] = None
# HMAC/encryption keys used to protect offloaded data
self.key_hmac = None # type: Optional[bytes]
self.key_enc = None # type: Optional[bytes]
self.key_hmac: Optional[bytes] = None
self.key_enc: Optional[bytes] = None
"""
Transaction keys
@ -51,8 +51,8 @@ class State:
- for subaddresses the `r` is commonly denoted as `s`, however it is still just a random number
- the keys are used to derive the one time address and its keys (P = H(A*r)*G + B)
"""
self.tx_priv = None # type: Sc25519
self.tx_pub = None # type: Ge25519
self.tx_priv: Sc25519 = None
self.tx_pub: Ge25519 = None
"""
In some cases when subaddresses are used we need more tx_keys
@ -77,8 +77,8 @@ class State:
self.account_idx = 0
# contains additional tx keys if need_additional_tx_keys is True
self.additional_tx_private_keys = [] # type: List[Sc25519]
self.additional_tx_public_keys = [] # type: List[bytes]
self.additional_tx_private_keys: List[Sc25519] = []
self.additional_tx_public_keys: List[bytes] = []
# currently processed input/output index
self.current_input_index = -1
@ -92,36 +92,36 @@ class State:
self.summary_outs_money = 0
# output commitments
self.output_pk_commitments = [] # type: List[bytes]
self.output_pk_commitments: List[bytes] = []
self.output_amounts = [] # type: List[int]
self.output_amounts: List[int] = []
# output *range proof* masks. HP10+ makes them deterministic.
self.output_masks = [] # type: List[Sc25519]
self.output_masks: List[Sc25519] = []
# the range proofs are calculated in batches, this denotes the grouping
self.rsig_grouping = [] # type: List[int]
self.rsig_grouping: List[int] = []
# is range proof computing offloaded or not
self.rsig_offload = False
# sum of all inputs' pseudo out masks
self.sumpouts_alphas = crypto.sc_0() # type: Sc25519
self.sumpouts_alphas: Sc25519 = crypto.sc_0()
# sum of all output' pseudo out masks
self.sumout = crypto.sc_0() # type: Sc25519
self.sumout: Sc25519 = crypto.sc_0()
self.subaddresses = {} # type: Subaddresses
self.subaddresses: Subaddresses = {}
# TX_EXTRA_NONCE extra field for tx.extra, due to sort_tx_extra()
self.extra_nonce = None
# contains an array where each item denotes the input's position
# (inputs are sorted by key images)
self.source_permutation = [] # type: List[int]
self.source_permutation: List[int] = []
# Last key image seen. Used for input permutation correctness check
self.last_ki = None # type: Optional[bytes]
self.last_ki: Optional[bytes] = None
# Encryption key to release to host after protocol ends without error
self.opening_key = None # type: Optional[bytes]
self.opening_key: Optional[bytes] = None
# Step transition automaton
self.last_step = self.STEP_INIT
@ -132,7 +132,7 @@ class State:
See Monero-Trezor documentation section 3.3 for more details.
"""
self.tx_prefix_hasher = KeccakXmrArchive()
self.tx_prefix_hash = None # type: Optional[bytes]
self.tx_prefix_hash: Optional[bytes] = None
"""
Full message hasher/hash that is to be signed using MLSAG.
@ -140,7 +140,7 @@ class State:
See Monero-Trezor documentation section 3.3 for more details.
"""
self.full_message_hasher = PreMlsagHasher()
self.full_message = None # type: Optional[bytes]
self.full_message: Optional[bytes] = None
def mem_trace(self, x=None, collect=False):
if __debug__:

@ -10,7 +10,7 @@ DEFAULT_ICON = "apps/webauthn/res/icon_webauthn.toif"
class ConfirmInfo:
def __init__(self) -> None:
self.app_icon = None # type: Optional[bytes]
self.app_icon: Optional[bytes] = None
def get_header(self) -> str:
raise NotImplementedError

@ -55,11 +55,11 @@ _U2F_KEY_PATH = const(0x8055_3246)
class Credential:
def __init__(self) -> None:
self.index = None # type: Optional[int]
self.id = b"" # type: bytes
self.rp_id = "" # type: str
self.rp_id_hash = b"" # type: bytes
self.user_id = None # type: Optional[bytes]
self.index: Optional[int] = None
self.id: bytes = b""
self.rp_id: str = ""
self.rp_id_hash: bytes = b""
self.user_id: Optional[bytes] = None
def __lt__(self, other: "Credential") -> bool:
raise NotImplementedError
@ -107,14 +107,14 @@ class Credential:
class Fido2Credential(Credential):
def __init__(self) -> None:
super().__init__()
self.rp_name = None # type: Optional[str]
self.user_name = None # type: Optional[str]
self.user_display_name = None # type: Optional[str]
self.creation_time = 0 # type: int
self.hmac_secret = False # type: bool
self.use_sign_count = False # type: bool
self.algorithm = _DEFAULT_ALGORITHM # type: int
self.curve = _DEFAULT_CURVE # type: int
self.rp_name: Optional[str] = None
self.user_name: Optional[str] = None
self.user_display_name: Optional[str] = None
self.creation_time: int = 0
self.hmac_secret: bool = False
self.use_sign_count: bool = False
self.algorithm: int = _DEFAULT_ALGORITHM
self.curve: int = _DEFAULT_CURVE
def __lt__(self, other: Credential) -> bool:
# Sort FIDO2 credentials newest first amongst each other.
@ -359,7 +359,7 @@ class Fido2Credential(Credential):
class U2fCredential(Credential):
def __init__(self) -> None:
super().__init__()
self.node = None # type: Optional[bip32.HDNode]
self.node: Optional[bip32.HDNode] = None
def __lt__(self, other: "Credential") -> bool:
# Sort U2F credentials after FIDO2 credentials.

@ -563,7 +563,7 @@ async def handle_reports(iface: io.HID) -> None:
dialog_mgr.get_cid(),
_CID_BROADCAST,
):
resp = cmd_error(req.cid, _ERR_CHANNEL_BUSY) # type: Optional[Cmd]
resp: Optional[Cmd] = cmd_error(req.cid, _ERR_CHANNEL_BUSY)
else:
resp = dispatch_cmd(req, dialog_mgr)
if resp is not None:
@ -771,7 +771,7 @@ class Fido2Unlock(Fido2State):
super().__init__(req.cid, dialog_mgr.iface)
self.process_func = process_func
self.req = req
self.resp = None # type: Optional[Cmd]
self.resp: Optional[Cmd] = None
self.dialog_mgr = dialog_mgr
async def confirm_dialog(self) -> Union[bool, "State"]:
@ -992,11 +992,11 @@ class DialogManager:
self._clear()
def _clear(self) -> None:
self.state = None # type: Optional[State]
self.state: Optional[State] = None
self.deadline = 0
self.result = _RESULT_NONE
self.workflow = None # type: Optional[loop.spawn]
self.keepalive = None # type: Optional[Coroutine]
self.workflow: Optional[loop.spawn] = None
self.keepalive: Optional[Coroutine] = None
def _workflow_is_running(self) -> bool:
return self.workflow is not None and not self.workflow.finished
@ -1211,7 +1211,7 @@ def cmd_wink(req: Cmd) -> Cmd:
def msg_register(req: Msg, dialog_mgr: DialogManager) -> Cmd:
if not config.is_unlocked():
new_state = U2fUnlock(req.cid, dialog_mgr.iface) # type: State
new_state: State = U2fUnlock(req.cid, dialog_mgr.iface)
dialog_mgr.set_state(new_state)
return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)
@ -1292,7 +1292,7 @@ def msg_register_sign(challenge: bytes, cred: U2fCredential) -> bytes:
def msg_authenticate(req: Msg, dialog_mgr: DialogManager) -> Cmd:
if not config.is_unlocked():
new_state = U2fUnlock(req.cid, dialog_mgr.iface) # type: State
new_state: State = U2fUnlock(req.cid, dialog_mgr.iface)
dialog_mgr.set_state(new_state)
return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)
@ -1427,7 +1427,7 @@ def credentials_from_descriptor_list(
def distinguishable_cred_list(credentials: Iterable[Credential]) -> List[Credential]:
"""Reduces the input to a list of credentials which can be distinguished by
the user. It is assumed that all input credentials share the same RP ID."""
cred_list = [] # type: List[Credential]
cred_list: List[Credential] = []
for cred in credentials:
for i, prev_cred in enumerate(cred_list):
if prev_cred.account_name() == cred.account_name():

Loading…
Cancel
Save