1# This file is part of the Trezor project. 2# 3# Copyright (C) 2012-2022 SatoshiLabs and contributors 4# 5# This library is free software: you can redistribute it and/or modify 6# it under the terms of the GNU Lesser General Public License version 3 7# as published by the Free Software Foundation. 8# 9# This library is distributed in the hope that it will be useful, 10# but WITHOUT ANY WARRANTY; without even the implied warranty of 11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 12# GNU Lesser General Public License for more details. 13# 14# You should have received a copy of the License along with this library. 15# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>. 16 17from ipaddress import ip_address 18from itertools import chain 19from typing import ( 20 TYPE_CHECKING, 21 Any, 22 Dict, 23 Iterable, 24 Iterator, 25 List, 26 Optional, 27 Sequence, 28 Tuple, 29 Union, 30) 31 32from . import exceptions, messages, tools 33from .tools import expect 34 35if TYPE_CHECKING: 36 from .client import TrezorClient 37 from .protobuf import MessageType 38 39SIGNING_MODE_IDS = { 40 "ORDINARY_TRANSACTION": messages.CardanoTxSigningMode.ORDINARY_TRANSACTION, 41 "POOL_REGISTRATION_AS_OWNER": messages.CardanoTxSigningMode.POOL_REGISTRATION_AS_OWNER, 42 "MULTISIG_TRANSACTION": messages.CardanoTxSigningMode.MULTISIG_TRANSACTION, 43} 44 45PROTOCOL_MAGICS = {"mainnet": 764824073, "testnet": 42} 46NETWORK_IDS = {"mainnet": 1, "testnet": 0} 47 48REQUIRED_FIELDS_TRANSACTION = ("inputs", "outputs") 49REQUIRED_FIELDS_INPUT = ("prev_hash", "prev_index") 50REQUIRED_FIELDS_CERTIFICATE = ("type",) 51REQUIRED_FIELDS_POOL_PARAMETERS = ( 52 "pool_id", 53 "vrf_key_hash", 54 "pledge", 55 "cost", 56 "margin", 57 "reward_account", 58 "owners", 59) 60REQUIRED_FIELDS_TOKEN_GROUP = ("policy_id", "tokens") 61REQUIRED_FIELDS_CATALYST_REGISTRATION = ( 62 "voting_public_key", 63 "staking_path", 64 "nonce", 65 "reward_address_parameters", 66) 67 68INCOMPLETE_OUTPUT_ERROR_MESSAGE = "The output is missing some fields" 69 70INVALID_OUTPUT_TOKEN_BUNDLE_ENTRY = "The output's token_bundle entry is invalid" 71INVALID_MINT_TOKEN_BUNDLE_ENTRY = "The mint token_bundle entry is invalid" 72 73InputWithPath = Tuple[messages.CardanoTxInput, List[int]] 74AssetGroupWithTokens = Tuple[messages.CardanoAssetGroup, List[messages.CardanoToken]] 75OutputWithAssetGroups = Tuple[messages.CardanoTxOutput, List[AssetGroupWithTokens]] 76OutputItem = Union[ 77 messages.CardanoTxOutput, messages.CardanoAssetGroup, messages.CardanoToken 78] 79CertificateItem = Union[ 80 messages.CardanoTxCertificate, 81 messages.CardanoPoolOwner, 82 messages.CardanoPoolRelayParameters, 83] 84MintItem = Union[ 85 messages.CardanoTxMint, messages.CardanoAssetGroup, messages.CardanoToken 86] 87PoolOwnersAndRelays = Tuple[ 88 List[messages.CardanoPoolOwner], List[messages.CardanoPoolRelayParameters] 89] 90CertificateWithPoolOwnersAndRelays = Tuple[ 91 messages.CardanoTxCertificate, Optional[PoolOwnersAndRelays] 92] 93Path = List[int] 94Witness = Tuple[Path, bytes] 95AuxiliaryDataSupplement = Dict[str, Union[int, bytes]] 96SignTxResponse = Dict[str, Union[bytes, List[Witness], AuxiliaryDataSupplement]] 97 98 99def parse_optional_bytes(value: Optional[str]) -> Optional[bytes]: 100 return bytes.fromhex(value) if value is not None else None 101 102 103def parse_optional_int(value: Optional[str]) -> Optional[int]: 104 return int(value) if value is not None else None 105 106 107def create_address_parameters( 108 address_type: messages.CardanoAddressType, 109 address_n: List[int], 110 address_n_staking: Optional[List[int]] = None, 111 staking_key_hash: Optional[bytes] = None, 112 block_index: Optional[int] = None, 113 tx_index: Optional[int] = None, 114 certificate_index: Optional[int] = None, 115 script_payment_hash: Optional[bytes] = None, 116 script_staking_hash: Optional[bytes] = None, 117) -> messages.CardanoAddressParametersType: 118 certificate_pointer = None 119 120 if address_type in ( 121 messages.CardanoAddressType.POINTER, 122 messages.CardanoAddressType.POINTER_SCRIPT, 123 ): 124 certificate_pointer = _create_certificate_pointer( 125 block_index, tx_index, certificate_index 126 ) 127 128 return messages.CardanoAddressParametersType( 129 address_type=address_type, 130 address_n=address_n, 131 address_n_staking=address_n_staking, 132 staking_key_hash=staking_key_hash, 133 certificate_pointer=certificate_pointer, 134 script_payment_hash=script_payment_hash, 135 script_staking_hash=script_staking_hash, 136 ) 137 138 139def _create_certificate_pointer( 140 block_index: Optional[int], 141 tx_index: Optional[int], 142 certificate_index: Optional[int], 143) -> messages.CardanoBlockchainPointerType: 144 if block_index is None or tx_index is None or certificate_index is None: 145 raise ValueError("Invalid pointer parameters") 146 147 return messages.CardanoBlockchainPointerType( 148 block_index=block_index, tx_index=tx_index, certificate_index=certificate_index 149 ) 150 151 152def parse_input(tx_input: dict) -> InputWithPath: 153 if not all(k in tx_input for k in REQUIRED_FIELDS_INPUT): 154 raise ValueError("The input is missing some fields") 155 156 path = tools.parse_path(tx_input.get("path", "")) 157 return ( 158 messages.CardanoTxInput( 159 prev_hash=bytes.fromhex(tx_input["prev_hash"]), 160 prev_index=tx_input["prev_index"], 161 ), 162 path, 163 ) 164 165 166def parse_output(output: dict) -> OutputWithAssetGroups: 167 contains_address = "address" in output 168 contains_address_type = "addressType" in output 169 170 if "amount" not in output: 171 raise ValueError(INCOMPLETE_OUTPUT_ERROR_MESSAGE) 172 if not (contains_address or contains_address_type): 173 raise ValueError(INCOMPLETE_OUTPUT_ERROR_MESSAGE) 174 175 address = None 176 address_parameters = None 177 token_bundle = [] 178 179 if contains_address: 180 address = output["address"] 181 182 if contains_address_type: 183 address_parameters = _parse_address_parameters( 184 output, INCOMPLETE_OUTPUT_ERROR_MESSAGE 185 ) 186 187 if "token_bundle" in output: 188 token_bundle = _parse_token_bundle(output["token_bundle"], is_mint=False) 189 190 return ( 191 messages.CardanoTxOutput( 192 address=address, 193 address_parameters=address_parameters, 194 amount=int(output["amount"]), 195 asset_groups_count=len(token_bundle), 196 ), 197 token_bundle, 198 ) 199 200 201def _parse_token_bundle( 202 token_bundle: Iterable[dict], is_mint: bool 203) -> List[AssetGroupWithTokens]: 204 error_message: str 205 if is_mint: 206 error_message = INVALID_MINT_TOKEN_BUNDLE_ENTRY 207 else: 208 error_message = INVALID_OUTPUT_TOKEN_BUNDLE_ENTRY 209 210 result = [] 211 for token_group in token_bundle: 212 if not all(k in token_group for k in REQUIRED_FIELDS_TOKEN_GROUP): 213 raise ValueError(error_message) 214 215 tokens = _parse_tokens(token_group["tokens"], is_mint) 216 217 result.append( 218 ( 219 messages.CardanoAssetGroup( 220 policy_id=bytes.fromhex(token_group["policy_id"]), 221 tokens_count=len(tokens), 222 ), 223 tokens, 224 ) 225 ) 226 227 return result 228 229 230def _parse_tokens(tokens: Iterable[dict], is_mint: bool) -> List[messages.CardanoToken]: 231 error_message: str 232 if is_mint: 233 error_message = INVALID_MINT_TOKEN_BUNDLE_ENTRY 234 else: 235 error_message = INVALID_OUTPUT_TOKEN_BUNDLE_ENTRY 236 237 result = [] 238 for token in tokens: 239 if "asset_name_bytes" not in token: 240 raise ValueError(error_message) 241 242 mint_amount = None 243 amount = None 244 if is_mint: 245 if "mint_amount" not in token: 246 raise ValueError(error_message) 247 mint_amount = int(token["mint_amount"]) 248 else: 249 if "amount" not in token: 250 raise ValueError(error_message) 251 amount = int(token["amount"]) 252 253 result.append( 254 messages.CardanoToken( 255 asset_name_bytes=bytes.fromhex(token["asset_name_bytes"]), 256 amount=amount, 257 mint_amount=mint_amount, 258 ) 259 ) 260 261 return result 262 263 264def _parse_address_parameters( 265 address_parameters: dict, error_message: str 266) -> messages.CardanoAddressParametersType: 267 if "addressType" not in address_parameters: 268 raise ValueError(error_message) 269 270 payment_path = tools.parse_path(address_parameters.get("path", "")) 271 staking_path = tools.parse_path(address_parameters.get("stakingPath", "")) 272 staking_key_hash_bytes = parse_optional_bytes( 273 address_parameters.get("stakingKeyHash") 274 ) 275 script_payment_hash = parse_optional_bytes( 276 address_parameters.get("scriptPaymentHash") 277 ) 278 script_staking_hash = parse_optional_bytes( 279 address_parameters.get("scriptStakingHash") 280 ) 281 282 return create_address_parameters( 283 messages.CardanoAddressType(address_parameters["addressType"]), 284 payment_path, 285 staking_path, 286 staking_key_hash_bytes, 287 address_parameters.get("blockIndex"), 288 address_parameters.get("txIndex"), 289 address_parameters.get("certificateIndex"), 290 script_payment_hash, 291 script_staking_hash, 292 ) 293 294 295def parse_native_script(native_script: dict) -> messages.CardanoNativeScript: 296 if "type" not in native_script: 297 raise ValueError("Script is missing some fields") 298 299 type = native_script["type"] 300 scripts = [ 301 parse_native_script(sub_script) 302 for sub_script in native_script.get("scripts", ()) 303 ] 304 305 key_hash = parse_optional_bytes(native_script.get("key_hash")) 306 key_path = tools.parse_path(native_script.get("key_path", "")) 307 required_signatures_count = parse_optional_int( 308 native_script.get("required_signatures_count") 309 ) 310 invalid_before = parse_optional_int(native_script.get("invalid_before")) 311 invalid_hereafter = parse_optional_int(native_script.get("invalid_hereafter")) 312 313 return messages.CardanoNativeScript( 314 type=type, 315 scripts=scripts, 316 key_hash=key_hash, 317 key_path=key_path, 318 required_signatures_count=required_signatures_count, 319 invalid_before=invalid_before, 320 invalid_hereafter=invalid_hereafter, 321 ) 322 323 324def parse_certificate(certificate: dict) -> CertificateWithPoolOwnersAndRelays: 325 CERTIFICATE_MISSING_FIELDS_ERROR = ValueError( 326 "The certificate is missing some fields" 327 ) 328 329 if not all(k in certificate for k in REQUIRED_FIELDS_CERTIFICATE): 330 raise CERTIFICATE_MISSING_FIELDS_ERROR 331 332 certificate_type = certificate["type"] 333 334 if certificate_type == messages.CardanoCertificateType.STAKE_DELEGATION: 335 if "pool" not in certificate: 336 raise CERTIFICATE_MISSING_FIELDS_ERROR 337 338 path, script_hash = _parse_path_or_script_hash( 339 certificate, CERTIFICATE_MISSING_FIELDS_ERROR 340 ) 341 342 return ( 343 messages.CardanoTxCertificate( 344 type=certificate_type, 345 path=path, 346 pool=bytes.fromhex(certificate["pool"]), 347 script_hash=script_hash, 348 ), 349 None, 350 ) 351 elif certificate_type in ( 352 messages.CardanoCertificateType.STAKE_REGISTRATION, 353 messages.CardanoCertificateType.STAKE_DEREGISTRATION, 354 ): 355 path, script_hash = _parse_path_or_script_hash( 356 certificate, CERTIFICATE_MISSING_FIELDS_ERROR 357 ) 358 359 return ( 360 messages.CardanoTxCertificate( 361 type=certificate_type, path=path, script_hash=script_hash 362 ), 363 None, 364 ) 365 elif certificate_type == messages.CardanoCertificateType.STAKE_POOL_REGISTRATION: 366 pool_parameters = certificate["pool_parameters"] 367 368 if any( 369 required_param not in pool_parameters 370 for required_param in REQUIRED_FIELDS_POOL_PARAMETERS 371 ): 372 raise CERTIFICATE_MISSING_FIELDS_ERROR 373 374 pool_metadata: Optional[messages.CardanoPoolMetadataType] 375 if pool_parameters.get("metadata") is not None: 376 pool_metadata = messages.CardanoPoolMetadataType( 377 url=pool_parameters["metadata"]["url"], 378 hash=bytes.fromhex(pool_parameters["metadata"]["hash"]), 379 ) 380 else: 381 pool_metadata = None 382 383 owners = [ 384 _parse_pool_owner(pool_owner) 385 for pool_owner in pool_parameters.get("owners", []) 386 ] 387 relays = [ 388 _parse_pool_relay(pool_relay) 389 for pool_relay in pool_parameters.get("relays", []) 390 ] 391 392 return ( 393 messages.CardanoTxCertificate( 394 type=certificate_type, 395 pool_parameters=messages.CardanoPoolParametersType( 396 pool_id=bytes.fromhex(pool_parameters["pool_id"]), 397 vrf_key_hash=bytes.fromhex(pool_parameters["vrf_key_hash"]), 398 pledge=int(pool_parameters["pledge"]), 399 cost=int(pool_parameters["cost"]), 400 margin_numerator=int(pool_parameters["margin"]["numerator"]), 401 margin_denominator=int(pool_parameters["margin"]["denominator"]), 402 reward_account=pool_parameters["reward_account"], 403 metadata=pool_metadata, 404 owners_count=len(owners), 405 relays_count=len(relays), 406 ), 407 ), 408 (owners, relays), 409 ) 410 else: 411 raise ValueError("Unknown certificate type") 412 413 414def _parse_path_or_script_hash( 415 obj: dict, error: ValueError 416) -> Tuple[List[int], Optional[bytes]]: 417 if "path" not in obj and "script_hash" not in obj: 418 raise error 419 420 path = tools.parse_path(obj.get("path", "")) 421 script_hash = parse_optional_bytes(obj.get("script_hash")) 422 423 return path, script_hash 424 425 426def _parse_pool_owner(pool_owner: dict) -> messages.CardanoPoolOwner: 427 if "staking_key_path" in pool_owner: 428 return messages.CardanoPoolOwner( 429 staking_key_path=tools.parse_path(pool_owner["staking_key_path"]) 430 ) 431 432 return messages.CardanoPoolOwner( 433 staking_key_hash=bytes.fromhex(pool_owner["staking_key_hash"]) 434 ) 435 436 437def _parse_pool_relay(pool_relay: dict) -> messages.CardanoPoolRelayParameters: 438 pool_relay_type = messages.CardanoPoolRelayType(pool_relay["type"]) 439 440 if pool_relay_type == messages.CardanoPoolRelayType.SINGLE_HOST_IP: 441 ipv4_address_packed = ( 442 ip_address(pool_relay["ipv4_address"]).packed 443 if "ipv4_address" in pool_relay 444 else None 445 ) 446 ipv6_address_packed = ( 447 ip_address(pool_relay["ipv6_address"]).packed 448 if "ipv6_address" in pool_relay 449 else None 450 ) 451 452 return messages.CardanoPoolRelayParameters( 453 type=pool_relay_type, 454 port=int(pool_relay["port"]), 455 ipv4_address=ipv4_address_packed, 456 ipv6_address=ipv6_address_packed, 457 ) 458 elif pool_relay_type == messages.CardanoPoolRelayType.SINGLE_HOST_NAME: 459 return messages.CardanoPoolRelayParameters( 460 type=pool_relay_type, 461 port=int(pool_relay["port"]), 462 host_name=pool_relay["host_name"], 463 ) 464 elif pool_relay_type == messages.CardanoPoolRelayType.MULTIPLE_HOST_NAME: 465 return messages.CardanoPoolRelayParameters( 466 type=pool_relay_type, 467 host_name=pool_relay["host_name"], 468 ) 469 470 raise ValueError("Unknown pool relay type") 471 472 473def parse_withdrawal(withdrawal: dict) -> messages.CardanoTxWithdrawal: 474 WITHDRAWAL_MISSING_FIELDS_ERROR = ValueError( 475 "The withdrawal is missing some fields" 476 ) 477 478 if "amount" not in withdrawal: 479 raise WITHDRAWAL_MISSING_FIELDS_ERROR 480 481 path, script_hash = _parse_path_or_script_hash( 482 withdrawal, WITHDRAWAL_MISSING_FIELDS_ERROR 483 ) 484 485 return messages.CardanoTxWithdrawal( 486 path=path, 487 amount=int(withdrawal["amount"]), 488 script_hash=script_hash, 489 ) 490 491 492def parse_auxiliary_data( 493 auxiliary_data: Optional[dict], 494) -> Optional[messages.CardanoTxAuxiliaryData]: 495 if auxiliary_data is None: 496 return None 497 498 AUXILIARY_DATA_MISSING_FIELDS_ERROR = ValueError( 499 "Auxiliary data is missing some fields" 500 ) 501 502 # include all provided fields so we can test validation in FW 503 hash = parse_optional_bytes(auxiliary_data.get("hash")) 504 505 catalyst_registration_parameters = None 506 if "catalyst_registration_parameters" in auxiliary_data: 507 catalyst_registration = auxiliary_data["catalyst_registration_parameters"] 508 if not all( 509 k in catalyst_registration for k in REQUIRED_FIELDS_CATALYST_REGISTRATION 510 ): 511 raise AUXILIARY_DATA_MISSING_FIELDS_ERROR 512 513 catalyst_registration_parameters = ( 514 messages.CardanoCatalystRegistrationParametersType( 515 voting_public_key=bytes.fromhex( 516 catalyst_registration["voting_public_key"] 517 ), 518 staking_path=tools.parse_path(catalyst_registration["staking_path"]), 519 nonce=catalyst_registration["nonce"], 520 reward_address_parameters=_parse_address_parameters( 521 catalyst_registration["reward_address_parameters"], 522 str(AUXILIARY_DATA_MISSING_FIELDS_ERROR), 523 ), 524 ) 525 ) 526 527 if hash is None and catalyst_registration_parameters is None: 528 raise AUXILIARY_DATA_MISSING_FIELDS_ERROR 529 530 return messages.CardanoTxAuxiliaryData( 531 hash=hash, 532 catalyst_registration_parameters=catalyst_registration_parameters, 533 ) 534 535 536def parse_mint(mint: Iterable[dict]) -> List[AssetGroupWithTokens]: 537 return _parse_token_bundle(mint, is_mint=True) 538 539 540def parse_additional_witness_request( 541 additional_witness_request: dict, 542) -> Path: 543 if "path" not in additional_witness_request: 544 raise ValueError("Invalid additional witness request") 545 546 return tools.parse_path(additional_witness_request["path"]) 547 548 549def _get_witness_requests( 550 inputs: Sequence[InputWithPath], 551 certificates: Sequence[CertificateWithPoolOwnersAndRelays], 552 withdrawals: Sequence[messages.CardanoTxWithdrawal], 553 additional_witness_requests: Sequence[Path], 554 signing_mode: messages.CardanoTxSigningMode, 555) -> List[messages.CardanoTxWitnessRequest]: 556 paths = set() 557 558 # don't gather paths from tx elements in MULTISIG_TRANSACTION signing mode 559 if signing_mode != messages.CardanoTxSigningMode.MULTISIG_TRANSACTION: 560 for _, path in inputs: 561 if path: 562 paths.add(tuple(path)) 563 for certificate, pool_owners_and_relays in certificates: 564 if ( 565 certificate.type 566 in ( 567 messages.CardanoCertificateType.STAKE_DEREGISTRATION, 568 messages.CardanoCertificateType.STAKE_DELEGATION, 569 ) 570 and certificate.path 571 ): 572 paths.add(tuple(certificate.path)) 573 elif ( 574 certificate.type 575 == messages.CardanoCertificateType.STAKE_POOL_REGISTRATION 576 and pool_owners_and_relays is not None 577 ): 578 owners, _ = pool_owners_and_relays 579 for pool_owner in owners: 580 if pool_owner.staking_key_path: 581 paths.add(tuple(pool_owner.staking_key_path)) 582 for withdrawal in withdrawals: 583 if withdrawal.path: 584 paths.add(tuple(withdrawal.path)) 585 586 # add additional_witness_requests in all cases 587 for additional_witness_request in additional_witness_requests: 588 paths.add(tuple(additional_witness_request)) 589 590 sorted_paths = sorted([list(path) for path in paths]) 591 return [messages.CardanoTxWitnessRequest(path=path) for path in sorted_paths] 592 593 594def _get_input_items(inputs: List[InputWithPath]) -> Iterator[messages.CardanoTxInput]: 595 for input, _ in inputs: 596 yield input 597 598 599def _get_output_items(outputs: List[OutputWithAssetGroups]) -> Iterator[OutputItem]: 600 for output, asset_groups in outputs: 601 yield output 602 for asset_group, tokens in asset_groups: 603 yield asset_group 604 yield from tokens 605 606 607def _get_certificate_items( 608 certificates: Sequence[CertificateWithPoolOwnersAndRelays], 609) -> Iterator[CertificateItem]: 610 for certificate, pool_owners_and_relays in certificates: 611 yield certificate 612 if pool_owners_and_relays is not None: 613 owners, relays = pool_owners_and_relays 614 yield from owners 615 yield from relays 616 617 618def _get_mint_items(mint: Sequence[AssetGroupWithTokens]) -> Iterator[MintItem]: 619 yield messages.CardanoTxMint(asset_groups_count=len(mint)) 620 for asset_group, tokens in mint: 621 yield asset_group 622 yield from tokens 623 624 625# ====== Client functions ====== # 626 627 628@expect(messages.CardanoAddress, field="address", ret_type=str) 629def get_address( 630 client: "TrezorClient", 631 address_parameters: messages.CardanoAddressParametersType, 632 protocol_magic: int = PROTOCOL_MAGICS["mainnet"], 633 network_id: int = NETWORK_IDS["mainnet"], 634 show_display: bool = False, 635 derivation_type: messages.CardanoDerivationType = messages.CardanoDerivationType.ICARUS, 636) -> "MessageType": 637 return client.call( 638 messages.CardanoGetAddress( 639 address_parameters=address_parameters, 640 protocol_magic=protocol_magic, 641 network_id=network_id, 642 show_display=show_display, 643 derivation_type=derivation_type, 644 ) 645 ) 646 647 648@expect(messages.CardanoPublicKey) 649def get_public_key( 650 client: "TrezorClient", 651 address_n: List[int], 652 derivation_type: messages.CardanoDerivationType = messages.CardanoDerivationType.ICARUS, 653) -> "MessageType": 654 return client.call( 655 messages.CardanoGetPublicKey( 656 address_n=address_n, derivation_type=derivation_type 657 ) 658 ) 659 660 661@expect(messages.CardanoNativeScriptHash) 662def get_native_script_hash( 663 client: "TrezorClient", 664 native_script: messages.CardanoNativeScript, 665 display_format: messages.CardanoNativeScriptHashDisplayFormat = messages.CardanoNativeScriptHashDisplayFormat.HIDE, 666 derivation_type: messages.CardanoDerivationType = messages.CardanoDerivationType.ICARUS, 667) -> "MessageType": 668 return client.call( 669 messages.CardanoGetNativeScriptHash( 670 script=native_script, 671 display_format=display_format, 672 derivation_type=derivation_type, 673 ) 674 ) 675 676 677def sign_tx( 678 client: "TrezorClient", 679 signing_mode: messages.CardanoTxSigningMode, 680 inputs: List[InputWithPath], 681 outputs: List[OutputWithAssetGroups], 682 fee: int, 683 ttl: Optional[int], 684 validity_interval_start: Optional[int], 685 certificates: Sequence[CertificateWithPoolOwnersAndRelays] = (), 686 withdrawals: Sequence[messages.CardanoTxWithdrawal] = (), 687 protocol_magic: int = PROTOCOL_MAGICS["mainnet"], 688 network_id: int = NETWORK_IDS["mainnet"], 689 auxiliary_data: Optional[messages.CardanoTxAuxiliaryData] = None, 690 mint: Sequence[AssetGroupWithTokens] = (), 691 additional_witness_requests: Sequence[Path] = (), 692 derivation_type: messages.CardanoDerivationType = messages.CardanoDerivationType.ICARUS, 693) -> Dict[str, Any]: 694 UNEXPECTED_RESPONSE_ERROR = exceptions.TrezorException("Unexpected response") 695 696 witness_requests = _get_witness_requests( 697 inputs, certificates, withdrawals, additional_witness_requests, signing_mode 698 ) 699 700 response = client.call( 701 messages.CardanoSignTxInit( 702 signing_mode=signing_mode, 703 inputs_count=len(inputs), 704 outputs_count=len(outputs), 705 fee=fee, 706 ttl=ttl, 707 validity_interval_start=validity_interval_start, 708 certificates_count=len(certificates), 709 withdrawals_count=len(withdrawals), 710 protocol_magic=protocol_magic, 711 network_id=network_id, 712 has_auxiliary_data=auxiliary_data is not None, 713 minting_asset_groups_count=len(mint), 714 witness_requests_count=len(witness_requests), 715 derivation_type=derivation_type, 716 ) 717 ) 718 if not isinstance(response, messages.CardanoTxItemAck): 719 raise UNEXPECTED_RESPONSE_ERROR 720 721 for tx_item in chain( 722 _get_input_items(inputs), 723 _get_output_items(outputs), 724 _get_certificate_items(certificates), 725 withdrawals, 726 ): 727 response = client.call(tx_item) 728 if not isinstance(response, messages.CardanoTxItemAck): 729 raise UNEXPECTED_RESPONSE_ERROR 730 731 sign_tx_response: Dict[str, Any] = {} 732 733 if auxiliary_data is not None: 734 auxiliary_data_supplement = client.call(auxiliary_data) 735 if not isinstance( 736 auxiliary_data_supplement, messages.CardanoTxAuxiliaryDataSupplement 737 ): 738 raise UNEXPECTED_RESPONSE_ERROR 739 if ( 740 auxiliary_data_supplement.type 741 != messages.CardanoTxAuxiliaryDataSupplementType.NONE 742 ): 743 sign_tx_response[ 744 "auxiliary_data_supplement" 745 ] = auxiliary_data_supplement.__dict__ 746 747 response = client.call(messages.CardanoTxHostAck()) 748 if not isinstance(response, messages.CardanoTxItemAck): 749 raise UNEXPECTED_RESPONSE_ERROR 750 751 if mint: 752 for mint_item in _get_mint_items(mint): 753 response = client.call(mint_item) 754 if not isinstance(response, messages.CardanoTxItemAck): 755 raise UNEXPECTED_RESPONSE_ERROR 756 757 sign_tx_response["witnesses"] = [] 758 for witness_request in witness_requests: 759 response = client.call(witness_request) 760 if not isinstance(response, messages.CardanoTxWitnessResponse): 761 raise UNEXPECTED_RESPONSE_ERROR 762 sign_tx_response["witnesses"].append( 763 { 764 "type": response.type, 765 "pub_key": response.pub_key, 766 "signature": response.signature, 767 "chain_code": response.chain_code, 768 } 769 ) 770 771 response = client.call(messages.CardanoTxHostAck()) 772 if not isinstance(response, messages.CardanoTxBodyHash): 773 raise UNEXPECTED_RESPONSE_ERROR 774 sign_tx_response["tx_hash"] = response.tx_hash 775 776 response = client.call(messages.CardanoTxHostAck()) 777 if not isinstance(response, messages.CardanoSignTxFinished): 778 raise UNEXPECTED_RESPONSE_ERROR 779 780 return sign_tx_response 781