"""# Partisia Blockchain Client Unofficial library for reading contract states from Partisia Blockchain. This library is not officially associated with Partisia Blockchain nor Partisia Group ApS. """ import base64 import dataclasses import datetime import email.utils import hashlib import json import logging import time from collections.abc import Mapping from decimal import Decimal from typing import Any import ecdsa import pbcabi import requests from frozendict import frozendict from ._version import __version__ # noqa: F401 from .pbc_types import ( Address, HashSha256, Signature, SignedTransaction, SignedTransactionInnerPart, Transaction, ) logger = logging.getLogger(__name__) HOSTNAME_MAINNET = 'reader.partisiablockchain.com' HOSTNAME_TESTNET = 'node1.testnet.partisiablockchain.com' URL_ACCOUNT_PLUGIN = 'https://{hostname}/{shard}blockchain/accountPlugin/local' URL_ACCOUNT_PLUGIN_GLOBAL = 'https://{hostname}/{shard}blockchain/accountPlugin/global' URL_TRANSACTION = 'https://{hostname}/{shard}blockchain/transaction' URL_CONTRACT_STATE = 'https://{hostname}/{shard}blockchain/contracts/{address}?requireContractState=false' URL_CHAIN_ID = 'https://{hostname}/blockchain/chainId' URL_NONCE = 'https://{hostname}/{shard}blockchain/account/{address}' MPC_DECIMALS = 10000 def shard_id_for_address(address: str | Address) -> str: address = str(address) # Very rough implementation if address is None: msg = 'Address must not be None' raise TypeError(msg) if address.endswith('a'): return 'shards/Shard0/' if address.endswith('2'): return 'shards/Shard1/' return 'shards/Shard2/' @dataclasses.dataclass(frozen=True) class Balances: update_time: datetime.datetime mpc: Decimal byoc: Mapping[str, Decimal] @dataclasses.dataclass(frozen=True) class SenderAuthentication: secret_key: str def sender_address(self) -> Address: verifying_key_repr = ( self._signing_key().get_verifying_key().to_string('uncompressed') ) hashed = HashSha256.of_bytes(verifying_key_repr) return Address(b'\0' + hashed._bytes[-20:]) def sign_hash(self, _hash: HashSha256) -> Signature: return Signature(self._signing_key().sign(_hash._bytes)) def _signing_key(self): secret_exponent = int(self.secret_key, 16) return ecdsa.SigningKey.from_secret_exponent( secret_exponent, curve=ecdsa.SECP256k1, ) TRANSACTION_VALIDITY_DURATION = 60 def sign_transaction( sender_authentication: SenderAuthentication, nonce: int, gas_cost: int, chain_id: str, contract_address: Address | str, transaction_rpc: bytes, ) -> SignedTransaction: sender = sender_authentication.sender_address valid_to_time: int = int(time.time() + TRANSACTION_VALIDITY_DURATION) * 1000 inner: SignedTransactionInnerPart = SignedTransactionInnerPart( nonce, valid_to_time, gas_cost, Transaction(Address.from_string(contract_address), transaction_rpc), ) transaction_hash_bytes = inner.rpc_serialize() + chain_id.encode('utf8') transaction_hash: HashSha256 = HashSha256.of_bytes(transaction_hash_bytes) signature: Signature = sender_authentication.sign_hash(transaction_hash) return SignedTransaction(inner, signature, transaction_hash) @dataclasses.dataclass(frozen=True) class PbcClient: session: requests.Session sender_authentication: SenderAuthentication | None = None hostname: str = HOSTNAME_MAINNET def __post_init__(self): assert isinstance(self.session, requests.Session) assert isinstance(self.hostname, str) if self.sender_authentication is not None: assert isinstance(self.sender_authentication, SenderAuthentication) def send_transaction(self, contract_address: str, rpc: bytes, gas_cost: int): if self.sender_authentication is None: msg = 'PbcClient.sender_authentication required for send_transaction' raise Exception(msg) signed_transaction = sign_transaction( self.sender_authentication, self.get_sender_authentication_nonce(), gas_cost, self.get_chain_id(), contract_address, rpc, ) return self.send_signed_transaction(signed_transaction) def send_signed_transaction(self, signed_transaction: SignedTransaction): url = URL_TRANSACTION.format( hostname=self.hostname, shard=shard_id_for_address( signed_transaction.inner.transaction.contract_address, ), ) transaction_payload: str = base64.b64encode( signed_transaction.rpc_serialize(), ).decode('utf8') self._get_json( url, data={'transactionPayload': transaction_payload}, method='PUT', ) def _get_json( self, url: str, data: Mapping[str, Any] = frozendict(), method='POST', ) -> tuple[Any, datetime.datetime]: headers = { 'Content-Type': 'application/json', 'Accept': 'application/json', } response = self.session.request( method, url, headers=headers, data=json.dumps(data), ) response.raise_for_status() date_text = response.headers.get('last-modified') or response.headers.get( 'date', ) date = email.utils.parsedate_to_datetime(date_text) json_data = response.json() if json_data is None: msg = 'No result data for ' + url raise Exception(msg) return (json_data, date) def determine_coins(self) -> list[dict[str, Any]]: data: dict = {'path': []} url: str = URL_ACCOUNT_PLUGIN_GLOBAL.format( hostname=self.hostname, shard='', ) json_data, date = self._get_json(url, data=data) return json_data['coins']['coins'] def get_account_balances(self, address: str) -> Balances: coins = self.determine_coins() url = URL_ACCOUNT_PLUGIN.format( hostname=self.hostname, shard=shard_id_for_address(address), ) data: dict = { 'path': [ {'type': 'field', 'name': 'accounts'}, {'type': 'avl', 'keyType': 'BLOCKCHAIN_ADDRESS', 'key': address}, ], } account_data, date = self._get_json(url, data=data) byoc: dict[str, Decimal] = {} mpc = Decimal(account_data['mpcTokens']) / MPC_DECIMALS for coin_idx, amount_data in enumerate(account_data['accountCoins']): coin_data = coins[coin_idx] byoc_balance = Decimal(amount_data['balance']) denominator = Decimal(coin_data['conversionRate']['denominator']) native_balance = byoc_balance / denominator byoc[coin_data['symbol']] = native_balance del coin_idx, coin_data return Balances(date, mpc, byoc) def get_chain_id(self) -> str: url = URL_CHAIN_ID.format( hostname=self.hostname, ) return self._get_json(url, method='GET')[0]['chainId'] def get_sender_authentication_nonce(self) -> int: return self.get_nonce_for_account(self.sender_authentication.sender_address()) def get_nonce_for_account(self, address: str) -> int: url = URL_NONCE.format( hostname=self.hostname, shard=shard_id_for_address(address), address=address, ) return self._get_json(url, method='GET')[0]['nonce'] def get_contract_state(self, address: str) -> tuple[dict, datetime.datetime]: # TODO: Rename to get_contract_state_json url = URL_CONTRACT_STATE.format( hostname=self.hostname, shard=shard_id_for_address(address), address=address, ) data: dict = {'path': []} return self._get_json(url, data=data) def get_typed_contract_state(self, address: str) -> tuple[dict, datetime.datetime]: """ Only suitable for non-governance WASM contracts. """ file_abi = self.get_contract_abi(address) state, server_time = self.get_contract_state(address) state_bytes = base64.b64decode(state['state']['data']) state_deserialized = file_abi.contract.read_state(state_bytes) return state_deserialized, server_time def get_typed_contract_avl_tree( self, address: str, avl_tree_id: pbcabi.model.AvlTreeId, ) -> tuple[dict, datetime.datetime]: file_abi = self.get_contract_abi(address) state, server_time = self.get_contract_state(address) for avl_tree in state['avlTrees']: if avl_tree['key'] == avl_tree_id.avl_tree_id: break data = {} for key_and_value in avl_tree['value']['avlTree']: key_bytes = base64.b64decode(key_and_value['key']['data']['data']) value_bytes = base64.b64decode(key_and_value['value']['data']) key = file_abi.contract.read_state( key_bytes, avl_tree_id.type_spec.type_key, ) value = file_abi.contract.read_state( value_bytes, avl_tree_id.type_spec.type_value, ) data[key] = value # TODO: Type return data, server_time def get_contract_abi(self, address: str) -> pbcabi.model.FileAbi: url = URL_CONTRACT_STATE.format( hostname=self.hostname, shard=shard_id_for_address(address), address=address, ) meta, _ = self._get_json(url, method='GET') abi_bytes = base64.b64decode(meta['abi']) return pbcabi.model.FileAbi.read_from(abi_bytes)