Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _validate(msg) -> int:
if storage.is_initialized():
raise wire.UnexpectedMessage("Already initialized")
if not msg.mnemonics:
raise wire.ProcessError("No mnemonic provided")
word_count = len(msg.mnemonics[0].split(" "))
for m in msg.mnemonics[1:]:
if word_count != len(m.split(" ")):
raise wire.ProcessError(
"All shares are required to have the same number of words"
)
return word_count
newpin = await request_pin_confirm(ctx, cancellable=False)
# save into storage
if not msg.dry_run:
if msg.pin_protection:
config.change_pin(pin_to_int(""), pin_to_int(newpin))
storage.load_settings(label=msg.label, use_passphrase=msg.passphrase_protection)
storage.load_mnemonic(mnemonic=mnemonic, needs_backup=False, no_backup=False)
return Success(message="Device recovered")
else:
if storage.get_mnemonic() == mnemonic:
return Success(
message="The seed is valid and matches the one in the device"
)
else:
raise wire.ProcessError(
"The seed is valid but does not match the one in the device"
)
raise ProcessError('Public key provided but no payload to encrypt')
if transfer.payload:
if len(transfer.payload) > NEM_MAX_PLAIN_PAYLOAD_SIZE:
raise ProcessError('Payload too large')
if transfer.public_key and len(transfer.payload) > NEM_MAX_ENCRYPTED_PAYLOAD_SIZE:
raise ProcessError('Payload too large')
if not nem.validate_address(transfer.recipient, network):
raise ProcessError('Invalid recipient address')
for m in transfer.mosaics:
if m.namespace is None:
raise ProcessError('No mosaic namespace provided')
if m.mosaic is None:
raise ProcessError('No mosaic name provided')
if m.quantity is None:
raise ProcessError('No mosaic quantity provided')
def _validate_supply_change(supply_change: NEMMosaicSupplyChange):
if supply_change.namespace is None:
raise ProcessError('No namespace provided')
if supply_change.mosaic is None:
raise ProcessError('No mosaic provided')
if supply_change.type is None:
raise ProcessError('No type provided')
elif supply_change.type not in [NEMSupplyChangeType.SupplyChange_Decrease, NEMSupplyChangeType.SupplyChange_Increase]:
raise ProcessError('Invalid supply change type')
if supply_change.delta is None:
raise ProcessError('No delta provided')
raise ProcessError('No levy address provided')
if mosaic_creation.definition.levy_namespace is None:
raise ProcessError('No levy namespace provided')
if mosaic_creation.definition.levy_mosaic is None:
raise ProcessError('No levy mosaic name provided')
if mosaic_creation.definition.divisibility is None:
raise ProcessError('No divisibility provided')
if mosaic_creation.definition.supply is None:
raise ProcessError('No supply provided')
if mosaic_creation.definition.mutable_supply is None:
raise ProcessError('No supply mutability provided')
if mosaic_creation.definition.transferable is None:
raise ProcessError('No mosaic transferability provided')
if mosaic_creation.definition.description is None:
raise ProcessError('No description provided')
if mosaic_creation.definition.divisibility > NEM_MAX_DIVISIBILITY:
raise ProcessError('Invalid divisibility provided')
if mosaic_creation.definition.supply > NEM_MAX_SUPPLY:
raise ProcessError('Invalid supply provided')
if not nem.validate_address(mosaic_creation.definition.levy_address, network):
raise ProcessError('Invalid levy address')
raise ProcessError('Definition supply needs to be provided when divisibility is')
if mosaic_creation.definition.levy is not None:
if mosaic_creation.definition.fee is None:
raise ProcessError('No levy fee provided')
if mosaic_creation.definition.levy_address is None:
raise ProcessError('No levy address provided')
if mosaic_creation.definition.levy_namespace is None:
raise ProcessError('No levy namespace provided')
if mosaic_creation.definition.levy_mosaic is None:
raise ProcessError('No levy mosaic name provided')
if mosaic_creation.definition.divisibility is None:
raise ProcessError('No divisibility provided')
if mosaic_creation.definition.supply is None:
raise ProcessError('No supply provided')
if mosaic_creation.definition.mutable_supply is None:
raise ProcessError('No supply mutability provided')
if mosaic_creation.definition.transferable is None:
raise ProcessError('No mosaic transferability provided')
if mosaic_creation.definition.description is None:
raise ProcessError('No description provided')
if mosaic_creation.definition.divisibility > NEM_MAX_DIVISIBILITY:
raise ProcessError('Invalid divisibility provided')
if mosaic_creation.definition.supply > NEM_MAX_SUPPLY:
raise ProcessError('Invalid supply provided')
if not nem.validate_address(mosaic_creation.definition.levy_address, network):
raise ProcessError('Invalid levy address')
def _validate_aggregate_modification(
aggregate_modification: NEMAggregateModification,
creation: bool = False):
if creation and not aggregate_modification.modifications:
raise ProcessError('No modifications provided')
for m in aggregate_modification.modifications:
if not m.type:
raise ProcessError('No modification type provided')
if m.type not in (
NEMModificationType.CosignatoryModification_Add,
NEMModificationType.CosignatoryModification_Delete
):
raise ProcessError('Unknown aggregate modification')
if creation and m.type == NEMModificationType.CosignatoryModification_Delete:
raise ProcessError('Cannot remove cosignatory when converting account')
_validate_public_key(m.public_key, 'Invalid cosignatory public key provided')
async def get_keychain(ctx: wire.Context) -> Keychain:
if not storage.is_initialized():
raise wire.ProcessError("Device is not initialized")
# derive the root node from mnemonic and passphrase
passphrase = cache.get_passphrase()
if passphrase is None:
passphrase = await protect_by_passphrase(ctx)
cache.set_passphrase(passphrase)
root = bip32.from_mnemonic_cardano(mnemonic.restore(), passphrase)
# derive the namespaced root node
for i in SEED_NAMESPACE:
root.derive_cardano(i)
keychain = Keychain(SEED_NAMESPACE, root)
return keychain
def _validate_single_tx(msg: NEMSignTx):
# ensure exactly one transaction is provided
tx_count = \
bool(msg.transfer) + \
bool(msg.provision_namespace) + \
bool(msg.mosaic_creation) + \
bool(msg.supply_change) + \
bool(msg.aggregate_modification) + \
bool(msg.importance_transfer)
if tx_count == 0:
raise ProcessError('No transaction provided')
if tx_count > 1:
raise ProcessError('More than one transaction provided')
async def reset_device(ctx, msg):
# validate parameters and device state
if msg.strength not in (128, 192, 256):
raise wire.ProcessError("Invalid strength (has to be 128, 192 or 256 bits)")
if msg.display_random and (msg.skip_backup or msg.no_backup):
raise wire.ProcessError("Can't show internal entropy when backup is skipped")
if storage.is_initialized():
raise wire.UnexpectedMessage("Already initialized")
text = Text("Create a new wallet", ui.ICON_RESET, new_lines=False)
text.normal("Do you really want to")
text.br()
text.normal("create a new wallet?")
text.br()
text.br_half()
text.normal("By continuing you agree")
text.br()
text.normal("to")
text.bold("https://trezor.io/tos")