gabrielkerekes
3 years ago
committed by
matejcik
parent
2c503b16f5
commit
2313293477
@ -0,0 +1,241 @@
|
||||
from trezor.crypto import hashlib
|
||||
from trezor.crypto.curve import ed25519
|
||||
from trezor.messages import CardanoAddressType
|
||||
|
||||
from apps.common import cbor
|
||||
|
||||
from ..common.seed import remove_ed25519_prefix
|
||||
from .address import derive_address_bytes, derive_human_readable_address
|
||||
from .helpers import INVALID_AUXILIARY_DATA, bech32
|
||||
from .helpers.bech32 import HRP_JORMUN_PUBLIC_KEY
|
||||
from .helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT
|
||||
from .layout import confirm_catalyst_registration, show_auxiliary_data_hash
|
||||
|
||||
if False:
|
||||
from typing import Union
|
||||
from trezor import wire
|
||||
|
||||
from trezor.messages.CardanoCatalystRegistrationParametersType import (
|
||||
CardanoCatalystRegistrationParametersType,
|
||||
)
|
||||
from trezor.messages.CardanoTxAuxiliaryDataType import CardanoTxAuxiliaryDataType
|
||||
|
||||
CatalystRegistrationPayload = dict[int, Union[bytes, int]]
|
||||
CatalystRegistrationSignature = dict[int, bytes]
|
||||
CatalystRegistration = dict[
|
||||
int, Union[CatalystRegistrationPayload, CatalystRegistrationSignature]
|
||||
]
|
||||
|
||||
from . import seed
|
||||
|
||||
AUXILIARY_DATA_HASH_SIZE = 32
|
||||
CATALYST_VOTING_PUBLIC_KEY_LENGTH = 32
|
||||
CATALYST_REGISTRATION_HASH_SIZE = 32
|
||||
|
||||
METADATA_KEY_CATALYST_REGISTRATION = 61284
|
||||
METADATA_KEY_CATALYST_REGISTRATION_SIGNATURE = 61285
|
||||
|
||||
|
||||
def validate_auxiliary_data(
|
||||
keychain: seed.Keychain,
|
||||
auxiliary_data: CardanoTxAuxiliaryDataType | None,
|
||||
protocol_magic: int,
|
||||
network_id: int,
|
||||
) -> None:
|
||||
if not auxiliary_data:
|
||||
return
|
||||
|
||||
fields_provided = 0
|
||||
if auxiliary_data.blob:
|
||||
fields_provided += 1
|
||||
_validate_auxiliary_data_blob(auxiliary_data.blob)
|
||||
if auxiliary_data.catalyst_registration_parameters:
|
||||
fields_provided += 1
|
||||
_validate_catalyst_registration_parameters(
|
||||
keychain,
|
||||
auxiliary_data.catalyst_registration_parameters,
|
||||
protocol_magic,
|
||||
network_id,
|
||||
)
|
||||
|
||||
if fields_provided != 1:
|
||||
raise INVALID_AUXILIARY_DATA
|
||||
|
||||
|
||||
def _validate_auxiliary_data_blob(auxiliary_data_blob: bytes) -> None:
|
||||
try:
|
||||
# validation to prevent CBOR injection and invalid CBOR
|
||||
# we don't validate data format, just that it's a valid CBOR
|
||||
cbor.decode(auxiliary_data_blob)
|
||||
except Exception:
|
||||
raise INVALID_AUXILIARY_DATA
|
||||
|
||||
|
||||
def _validate_catalyst_registration_parameters(
|
||||
keychain: seed.Keychain,
|
||||
catalyst_registration_parameters: CardanoCatalystRegistrationParametersType,
|
||||
protocol_magic: int,
|
||||
network_id: int,
|
||||
) -> None:
|
||||
if (
|
||||
len(catalyst_registration_parameters.voting_public_key)
|
||||
!= CATALYST_VOTING_PUBLIC_KEY_LENGTH
|
||||
):
|
||||
raise INVALID_AUXILIARY_DATA
|
||||
|
||||
if not SCHEMA_STAKING_ANY_ACCOUNT.match(
|
||||
catalyst_registration_parameters.staking_path
|
||||
):
|
||||
raise INVALID_AUXILIARY_DATA
|
||||
|
||||
address_parameters = catalyst_registration_parameters.reward_address_parameters
|
||||
if address_parameters.address_type == CardanoAddressType.BYRON:
|
||||
raise INVALID_AUXILIARY_DATA
|
||||
|
||||
# try to derive the address to validate it
|
||||
derive_address_bytes(keychain, address_parameters, protocol_magic, network_id)
|
||||
|
||||
|
||||
async def show_auxiliary_data(
|
||||
ctx: wire.Context,
|
||||
keychain: seed.Keychain,
|
||||
auxiliary_data: CardanoTxAuxiliaryDataType | None,
|
||||
protocol_magic: int,
|
||||
network_id: int,
|
||||
) -> None:
|
||||
if not auxiliary_data:
|
||||
return
|
||||
|
||||
if auxiliary_data.catalyst_registration_parameters:
|
||||
await _show_catalyst_registration(
|
||||
ctx,
|
||||
keychain,
|
||||
auxiliary_data.catalyst_registration_parameters,
|
||||
protocol_magic,
|
||||
network_id,
|
||||
)
|
||||
|
||||
auxiliary_data_bytes = get_auxiliary_data_cbor(
|
||||
keychain, auxiliary_data, protocol_magic, network_id
|
||||
)
|
||||
|
||||
auxiliary_data_hash = hash_auxiliary_data(bytes(auxiliary_data_bytes))
|
||||
await show_auxiliary_data_hash(ctx, auxiliary_data_hash)
|
||||
|
||||
|
||||
async def _show_catalyst_registration(
|
||||
ctx: wire.Context,
|
||||
keychain: seed.Keychain,
|
||||
catalyst_registration_parameters: CardanoCatalystRegistrationParametersType,
|
||||
protocol_magic: int,
|
||||
network_id: int,
|
||||
) -> None:
|
||||
public_key = catalyst_registration_parameters.voting_public_key
|
||||
encoded_public_key = bech32.encode(HRP_JORMUN_PUBLIC_KEY, public_key)
|
||||
staking_path = catalyst_registration_parameters.staking_path
|
||||
reward_address = derive_human_readable_address(
|
||||
keychain,
|
||||
catalyst_registration_parameters.reward_address_parameters,
|
||||
protocol_magic,
|
||||
network_id,
|
||||
)
|
||||
nonce = catalyst_registration_parameters.nonce
|
||||
|
||||
await confirm_catalyst_registration(
|
||||
ctx, encoded_public_key, staking_path, reward_address, nonce
|
||||
)
|
||||
|
||||
|
||||
def get_auxiliary_data_cbor(
|
||||
keychain: seed.Keychain,
|
||||
auxiliary_data: CardanoTxAuxiliaryDataType,
|
||||
protocol_magic: int,
|
||||
network_id: int,
|
||||
) -> bytes:
|
||||
if auxiliary_data.blob:
|
||||
return auxiliary_data.blob
|
||||
elif auxiliary_data.catalyst_registration_parameters:
|
||||
cborized_catalyst_registration = _cborize_catalyst_registration(
|
||||
keychain,
|
||||
auxiliary_data.catalyst_registration_parameters,
|
||||
protocol_magic,
|
||||
network_id,
|
||||
)
|
||||
return cbor.encode(_wrap_metadata(cborized_catalyst_registration))
|
||||
else:
|
||||
raise INVALID_AUXILIARY_DATA
|
||||
|
||||
|
||||
def _cborize_catalyst_registration(
|
||||
keychain: seed.Keychain,
|
||||
catalyst_registration_parameters: CardanoCatalystRegistrationParametersType,
|
||||
protocol_magic: int,
|
||||
network_id: int,
|
||||
) -> CatalystRegistration:
|
||||
staking_node = keychain.derive(catalyst_registration_parameters.staking_path)
|
||||
staking_key = remove_ed25519_prefix(staking_node.public_key())
|
||||
|
||||
catalyst_registration_payload: CatalystRegistrationPayload = {
|
||||
1: catalyst_registration_parameters.voting_public_key,
|
||||
2: staking_key,
|
||||
3: derive_address_bytes(
|
||||
keychain,
|
||||
catalyst_registration_parameters.reward_address_parameters,
|
||||
protocol_magic,
|
||||
network_id,
|
||||
),
|
||||
4: catalyst_registration_parameters.nonce,
|
||||
}
|
||||
|
||||
catalyst_registration_payload_signature = (
|
||||
_create_catalyst_registration_payload_signature(
|
||||
keychain,
|
||||
catalyst_registration_payload,
|
||||
catalyst_registration_parameters.staking_path,
|
||||
)
|
||||
)
|
||||
catalyst_registration_signature = {1: catalyst_registration_payload_signature}
|
||||
|
||||
return {
|
||||
METADATA_KEY_CATALYST_REGISTRATION: catalyst_registration_payload,
|
||||
METADATA_KEY_CATALYST_REGISTRATION_SIGNATURE: catalyst_registration_signature,
|
||||
}
|
||||
|
||||
|
||||
def _create_catalyst_registration_payload_signature(
|
||||
keychain: seed.Keychain,
|
||||
catalyst_registration_payload: CatalystRegistrationPayload,
|
||||
path: list[int],
|
||||
) -> bytes:
|
||||
node = keychain.derive(path)
|
||||
|
||||
encoded_catalyst_registration = cbor.encode(
|
||||
{METADATA_KEY_CATALYST_REGISTRATION: catalyst_registration_payload}
|
||||
)
|
||||
|
||||
catalyst_registration_hash = hashlib.blake2b(
|
||||
data=encoded_catalyst_registration,
|
||||
outlen=CATALYST_REGISTRATION_HASH_SIZE,
|
||||
).digest()
|
||||
|
||||
return ed25519.sign_ext(
|
||||
node.private_key(), node.private_key_ext(), catalyst_registration_hash
|
||||
)
|
||||
|
||||
|
||||
def _wrap_metadata(metadata: dict) -> tuple[dict, tuple]:
|
||||
"""
|
||||
A new structure of metadata is used after Cardano Mary era. The metadata
|
||||
is wrapped in a tuple and auxiliary_scripts may follow it. Cardano
|
||||
tooling uses this new format of "wrapped" metadata even if no
|
||||
auxiliary_scripts are included. So we do the same here.
|
||||
|
||||
https://github.com/input-output-hk/cardano-ledger-specs/blob/f7deb22be14d31b535f56edc3ca542c548244c67/shelley-ma/shelley-ma-test/cddl-files/shelley-ma.cddl#L212
|
||||
"""
|
||||
return metadata, ()
|
||||
|
||||
|
||||
def hash_auxiliary_data(auxiliary_data: bytes) -> bytes:
|
||||
return hashlib.blake2b(
|
||||
data=auxiliary_data, outlen=AUXILIARY_DATA_HASH_SIZE
|
||||
).digest()
|
||||
@ -0,0 +1,37 @@
|
||||
# Automatically generated by pb2py
|
||||
# fmt: off
|
||||
import protobuf as p
|
||||
|
||||
from .CardanoAddressParametersType import CardanoAddressParametersType
|
||||
|
||||
if __debug__:
|
||||
try:
|
||||
from typing import Dict, List, Optional # noqa: F401
|
||||
from typing_extensions import Literal # noqa: F401
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
class CardanoCatalystRegistrationParametersType(p.MessageType):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
voting_public_key: bytes,
|
||||
reward_address_parameters: CardanoAddressParametersType,
|
||||
nonce: int,
|
||||
staking_path: Optional[List[int]] = None,
|
||||
) -> None:
|
||||
self.staking_path = staking_path if staking_path is not None else []
|
||||
self.voting_public_key = voting_public_key
|
||||
self.reward_address_parameters = reward_address_parameters
|
||||
self.nonce = nonce
|
||||
|
||||
@classmethod
|
||||
def get_fields(cls) -> Dict:
|
||||
return {
|
||||
1: ('voting_public_key', p.BytesType, p.FLAG_REQUIRED),
|
||||
2: ('staking_path', p.UVarintType, p.FLAG_REPEATED),
|
||||
3: ('reward_address_parameters', CardanoAddressParametersType, p.FLAG_REQUIRED),
|
||||
4: ('nonce', p.UVarintType, p.FLAG_REQUIRED),
|
||||
}
|
||||
@ -0,0 +1,37 @@
|
||||
# Automatically generated by pb2py
|
||||
# fmt: off
|
||||
from .. import protobuf as p
|
||||
|
||||
from .CardanoAddressParametersType import CardanoAddressParametersType
|
||||
|
||||
if __debug__:
|
||||
try:
|
||||
from typing import Dict, List, Optional # noqa: F401
|
||||
from typing_extensions import Literal # noqa: F401
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
class CardanoCatalystRegistrationParametersType(p.MessageType):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
voting_public_key: bytes,
|
||||
reward_address_parameters: CardanoAddressParametersType,
|
||||
nonce: int,
|
||||
staking_path: Optional[List[int]] = None,
|
||||
) -> None:
|
||||
self.staking_path = staking_path if staking_path is not None else []
|
||||
self.voting_public_key = voting_public_key
|
||||
self.reward_address_parameters = reward_address_parameters
|
||||
self.nonce = nonce
|
||||
|
||||
@classmethod
|
||||
def get_fields(cls) -> Dict:
|
||||
return {
|
||||
1: ('voting_public_key', p.BytesType, p.FLAG_REQUIRED),
|
||||
2: ('staking_path', p.UVarintType, p.FLAG_REPEATED),
|
||||
3: ('reward_address_parameters', CardanoAddressParametersType, p.FLAG_REQUIRED),
|
||||
4: ('nonce', p.UVarintType, p.FLAG_REQUIRED),
|
||||
}
|
||||
Loading…
Reference in new issue