Skip to content

aea.crypto.base

Abstract module wrapping the public and private key cryptography and ledger api.

Crypto Objects

class Crypto(Generic[EntityClass], ABC)

Base class for a crypto object.

__init__

def __init__(private_key_path: Optional[str] = None,
             password: Optional[str] = None,
             extra_entropy: Union[str, bytes, int] = "",
             entity: Optional[EntityClass] = None,
             **kwargs: Any) -> None

Initialize the crypto object.

The actual behaviour of this constructor is determined by the abstract methods 'generate_private_key()' and 'load_private_key_from_path(). Either way, the entity object will be accessible as a property.

Arguments:

  • private_key_path: the path to the private key. If None, the key will be generated by 'generate_private_key()'. If not None, the path will be processed by 'load_private_key_from_path()'.
  • password: the password to encrypt/decrypt the private key.
  • extra_entropy: add extra randomness to whatever randomness your OS can provide
  • entity: Custom entity object
  • kwargs: keyword arguments.

generate_private_key

@classmethod
@abstractmethod
def generate_private_key(cls,
                         extra_entropy: Union[str, bytes,
                                              int] = "") -> EntityClass

Generate a private key.

Arguments:

  • extra_entropy: add extra randomness to whatever randomness your OS can provide

Returns:

the entity object. Implementation dependent.

load_private_key_from_path

@classmethod
@abstractmethod
def load_private_key_from_path(cls,
                               file_name: str,
                               password: Optional[str] = None) -> EntityClass

Load a private key in hex format for raw private key and json format for encrypted private key from a file.

Arguments:

  • file_name: the path to the hex/json file.
  • password: the password to encrypt/decrypt the private key.

Returns:

the entity object.

entity

@property
def entity() -> EntityClass

Return an entity object.

Returns:

an entity object

private_key

@property
@abstractmethod
def private_key() -> str

Return a private key.

Returns:

a private key string

public_key

@property
@abstractmethod
def public_key() -> str

Return a public key.

Returns:

a public key string

address

@property
@abstractmethod
def address() -> str

Return the address.

Returns:

an address string

sign_message

@abstractmethod
def sign_message(message: bytes, is_deprecated_mode: bool = False) -> str

Sign a message in bytes string form.

Arguments:

  • message: the message to be signed
  • is_deprecated_mode: if the deprecated signing is used

Returns:

signature of the message in string form

sign_transaction

@abstractmethod
def sign_transaction(transaction: JSONLike) -> JSONLike

Sign a transaction in dict form.

Arguments:

  • transaction: the transaction to be signed

Returns:

signed transaction

load

@classmethod
def load(cls, private_key_file: str, password: Optional[str] = None) -> str

Load private key from file.

Arguments:

  • private_key_file: the file where the key is stored.
  • password: the password to encrypt/decrypt the private key.

Returns:

private_key in hex string format

dump

def dump(private_key_file: str, password: Optional[str] = None) -> None

Dump private key to file.

Arguments:

  • private_key_file: the file where the key is stored.
  • password: the password to encrypt/decrypt the private key.

encrypt

@abstractmethod
def encrypt(password: str) -> str

Encrypt the private key and return in json.

Arguments:

  • password: the password to decrypt.

Returns:

json string containing encrypted private key.

decrypt

@classmethod
@abstractmethod
def decrypt(cls, keyfile_json: str, password: str) -> str

Decrypt the private key and return in raw form.

Arguments:

  • keyfile_json: json string containing encrypted private key.
  • password: the password to decrypt.

Returns:

the raw private key.

Helper Objects

class Helper(ABC)

Interface for helper class usable as Mixin for LedgerApi or as standalone class.

is_transaction_settled

@staticmethod
@abstractmethod
def is_transaction_settled(tx_receipt: JSONLike) -> bool

Check whether a transaction is settled or not.

Arguments:

  • tx_receipt: the receipt associated to the transaction.

Returns:

True if the transaction has been settled, False o/w.

is_transaction_valid

@staticmethod
@abstractmethod
def is_transaction_valid(tx: JSONLike, seller: Address, client: Address,
                         tx_nonce: str, amount: int) -> bool

Check whether a transaction is valid or not.

Arguments:

  • tx: the transaction.
  • seller: the address of the seller.
  • client: the address of the client.
  • tx_nonce: the transaction nonce.
  • amount: the amount we expect to get from the transaction.

Returns:

True if the random_message is equals to tx['input']

get_contract_address

@staticmethod
@abstractmethod
def get_contract_address(tx_receipt: JSONLike) -> Optional[str]

Get the contract address from a transaction receipt.

Arguments:

  • tx_receipt: the transaction digest

Returns:

the contract address if successful

generate_tx_nonce

@staticmethod
@abstractmethod
def generate_tx_nonce(seller: Address, client: Address) -> str

Generate a unique hash to distinguish transactions with the same terms.

Arguments:

  • seller: the address of the seller.
  • client: the address of the client.

Returns:

return the hash in hex.

get_address_from_public_key

@classmethod
@abstractmethod
def get_address_from_public_key(cls, public_key: str) -> str

Get the address from the public key.

Arguments:

  • public_key: the public key

Returns:

str

recover_message

@classmethod
@abstractmethod
def recover_message(cls,
                    message: bytes,
                    signature: str,
                    is_deprecated_mode: bool = False) -> Tuple[Address, ...]

Recover the addresses from the hash.

Arguments:

  • message: the message we expect
  • signature: the transaction signature
  • is_deprecated_mode: if the deprecated signing was used

Returns:

the recovered addresses

recover_public_keys_from_message

@classmethod
@abstractmethod
def recover_public_keys_from_message(
        cls,
        message: bytes,
        signature: str,
        is_deprecated_mode: bool = False) -> Tuple[str, ...]

Get the public key used to produce the signature of the message

Arguments:

  • message: raw bytes used to produce signature
  • signature: signature of the message
  • is_deprecated_mode: if the deprecated signing was used

Returns:

the recovered public keys

get_hash

@staticmethod
@abstractmethod
def get_hash(message: bytes) -> str

Get the hash of a message.

Arguments:

  • message: the message to be hashed.

Returns:

the hash of the message.

is_valid_address

@classmethod
@abstractmethod
def is_valid_address(cls, address: Address) -> bool

Check if the address is valid.

Arguments:

  • address: the address to validate

load_contract_interface

@classmethod
@abstractmethod
def load_contract_interface(cls, file_path: Path) -> Dict[str, str]

Load contract interface.

Arguments:

  • file_path: the file path to the interface

Returns:

the interface

LedgerApi Objects

class LedgerApi(Helper, ABC)

Interface for ledger APIs.

identifier

type: str

api

@property
@abstractmethod
def api() -> Any

Get the underlying API object.

This can be used for low-level operations with the concrete ledger APIs. If there is no such object, return None.

get_balance

@abstractmethod
def get_balance(address: Address, raise_on_try: bool = False) -> Optional[int]

Get the balance of a given account.

This usually takes the form of a web request to be waited synchronously.

Arguments:

  • address: the address.
  • raise_on_try: whether the method will raise or log on error

Returns:

the balance.

get_state

@abstractmethod
def get_state(callable_name: str,
              *args: Any,
              raise_on_try: bool = False,
              **kwargs: Any) -> Optional[JSONLike]

Call a specified function on the underlying ledger API.

This usually takes the form of a web request to be waited synchronously.

Arguments:

  • callable_name: the name of the API function to be called.
  • args: the positional arguments for the API function.
  • raise_on_try: whether the method will raise or log on error
  • kwargs: the keyword arguments for the API function.

Returns:

the ledger API response.

get_transfer_transaction

@abstractmethod
def get_transfer_transaction(sender_address: Address,
                             destination_address: Address, amount: int,
                             tx_fee: int, tx_nonce: str,
                             **kwargs: Any) -> Optional[JSONLike]

Submit a transfer transaction to the ledger.

Arguments:

  • sender_address: the sender address of the payer.
  • destination_address: the destination address of the payee.
  • amount: the amount of wealth to be transferred.
  • tx_fee: the transaction fee.
  • tx_nonce: verifies the authenticity of the tx
  • kwargs: the keyword arguments.

Returns:

the transfer transaction

send_signed_transaction

@abstractmethod
def send_signed_transaction(tx_signed: JSONLike,
                            raise_on_try: bool = False) -> Optional[str]

Send a signed transaction and wait for confirmation.

Use keyword arguments for the specifying the signed transaction payload.

Arguments:

  • tx_signed: the signed transaction
  • raise_on_try: whether the method will raise or log on error

Returns:

tx_digest, if present

get_transaction_receipt

@abstractmethod
def get_transaction_receipt(tx_digest: str,
                            raise_on_try: bool = False) -> Optional[JSONLike]

Get the transaction receipt for a transaction digest.

Arguments:

  • tx_digest: the digest associated to the transaction.
  • raise_on_try: whether the method will raise or log on error

Returns:

the tx receipt, if present

get_transaction

@abstractmethod
def get_transaction(tx_digest: str,
                    raise_on_try: bool = False) -> Optional[JSONLike]

Get the transaction for a transaction digest.

Arguments:

  • tx_digest: the digest associated to the transaction.
  • raise_on_try: whether the method will raise or log on error

Returns:

the tx, if present

send_signed_transactions

@abstractmethod
def send_signed_transactions(signed_transactions: List[JSONLike],
                             raise_on_try: bool = False,
                             **kwargs: Any) -> Optional[List[str]]

Atomically send multiple of transactions.

Arguments:

  • signed_transactions: the signed transactions to bundle together and send.
  • raise_on_try: whether the method will raise or log on error
  • kwargs: the keyword arguments.

Returns:

the transaction digest if the transactions went through, None otherwise.

get_contract_instance

@abstractmethod
def get_contract_instance(contract_interface: Dict[str, str],
                          contract_address: Optional[str] = None) -> Any

Get the instance of a contract.

Arguments:

  • contract_interface: the contract interface.
  • contract_address: the contract address.

Returns:

the contract instance

get_deploy_transaction

@abstractmethod
def get_deploy_transaction(contract_interface: Dict[str, str],
                           deployer_address: Address,
                           raise_on_try: bool = False,
                           **kwargs: Any) -> Optional[JSONLike]

Get the transaction to deploy the smart contract.

Arguments:

  • contract_interface: the contract interface.
  • deployer_address: The address that will deploy the contract.
  • raise_on_try: whether the method will raise or log on error
  • kwargs: the keyword arguments.

Returns:

tx: the transaction dictionary.

update_with_gas_estimate

@abstractmethod
def update_with_gas_estimate(transaction: JSONLike) -> JSONLike

Attempts to update the transaction with a gas estimate

Arguments:

  • transaction: the transaction

Returns:

the updated transaction

contract_method_call

@abstractmethod
def contract_method_call(contract_instance: Any, method_name: str,
                         **method_args: Any) -> Optional[JSONLike]

Call a contract's method

Arguments:

  • contract_instance: the contract to use
  • method_name: the contract method to call
  • method_args: the contract call parameters

build_transaction

@abstractmethod
def build_transaction(contract_instance: Any,
                      method_name: str,
                      method_args: Optional[Dict],
                      tx_args: Optional[Dict],
                      raise_on_try: bool = False) -> Optional[JSONLike]

Prepare a transaction

Arguments:

  • contract_instance: the contract to use
  • method_name: the contract method to call
  • method_args: the contract parameters
  • tx_args: the transaction parameters
  • raise_on_try: whether the method will raise or log on error

get_transaction_transfer_logs

@abstractmethod
def get_transaction_transfer_logs(
        contract_instance: Any,
        tx_hash: str,
        target_address: Optional[str] = None) -> Optional[JSONLike]

Get all transfer events derived from a transaction.

Arguments:

  • contract_instance: the contract
  • tx_hash: the transaction hash
  • target_address: optional address to filter transfer events to just those that affect it

filter_event

@abstractmethod
def filter_event(event: Any, match_single: Dict[str, Any],
                 match_any: Dict[str, Any], to_block: int, from_block: int,
                 batch_size: int, max_retries: int, reduce_factor: float,
                 timeout: int) -> Optional[JSONLike]

Filter an event using batching to avoid RPC timeouts.

Arguments:

  • event: the event to filter for.
  • match_single: the filter parameters with value checking against the event abi. It allows for defining a single match value.
  • match_any: the filter parameters with value checking against the event abi. It allows for defining multiple match values.
  • to_block: the block to which to filter.
  • from_block: the block from which to start filtering.
  • batch_size: the blocks' batch size of the filtering.
  • max_retries: the maximum number of retries.
  • reduce_factor: the percentage by which the batch size is reduced in case of a timeout.
  • timeout: a timeout in seconds to interrupt the operation in case the RPC request hangs.

FaucetApi Objects

class FaucetApi(ABC)

Interface for testnet faucet APIs.

identifier

type: str

network_name

type: str

get_wealth

@abstractmethod
def get_wealth(address: Address, url: Optional[str] = None) -> None

Get wealth from the faucet for the provided address.

Arguments:

  • address: the address.
  • url: the url

Returns:

None