Module substrateutils.helper

Helper functions - all functions in this file are pure with no side effects

View Source
"""

Helper functions - all functions in this file are pure with no side effects

"""

from hashlib import blake2b

from typing import Union

import scalecodec

import sr25519

import xxhash

from scalecodec.base import RuntimeConfigurationObject

from scalecodec.base import ScaleDecoder

from scalecodec.metadata import MetadataDecoder

from scalecodec.utils.ss58 import ss58_decode

def xx128(word: str) -> str:

    """

    Returns a xxh128 hash on provided word

    """

    a = bytearray(xxhash.xxh64(word, seed=0).digest())

    b = bytearray(xxhash.xxh64(word, seed=1).digest())

    a.reverse()

    b.reverse()

    return f"{a.hex()}{b.hex()}"

def get_prefix(escrow_address: str, address_type: int = 2) -> str:

    """

    Returns prefix containing the account ID of the address provided

    """

    module_prefix = xx128("Multisig") + xx128("Multisigs")

    account_id = ss58_decode(escrow_address, address_type)

    storage_key = bytearray(xxhash.xxh64(bytes.fromhex(account_id), seed=0).digest())

    storage_key.reverse()

    return f"{module_prefix}{storage_key.hex()}{account_id}"

def hash_call(call: "scalecodec.types.Call") -> str:

    """

    Returns a hashed call

    """

    call = bytes.fromhex(str(call.data)[2:])

    return f"0x{blake2b(call, digest_size=32).digest().hex()}"

def transfer_signature_payload(

    metadata: "MetadataDecoder",

    address: str,

    value: int,

    nonce: int,

    genesis_hash: str,

    spec_version: int,

    transaction_version: int = 3,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into unsigned transfer string

    """

    call = ScaleDecoder.get_decoder_class(

        "Call", metadata=metadata, runtime_config=runtime_config

    )

    call.encode(

        {

            "call_module": "Balances",

            "call_function": "transfer",

            "call_args": {"dest": address, "value": value},

        }

    )

    signature_payload = ScaleDecoder.get_decoder_class(

        "ExtrinsicPayloadValue", runtime_config=runtime_config

    )

    signature_payload.encode(

        {

            "call": str(call.data),

            "era": "00",

            "nonce": nonce,

            "tip": 0,

            "specVersion": spec_version,

            "transactionVersion": transaction_version,

            "genesisHash": genesis_hash,

            "blockHash": genesis_hash,

        }

    )

    return str(signature_payload.data)

def as_multi_signature_payload(

    metadata: "MetadataDecoder",

    spec_version: int,

    genesis_hash: str,

    nonce: int,

    to_address: str,

    amount: int,

    other_signatories: list,

    timepoint: Union[tuple, bool],

    threshold: int = 2,

    tip: int = 0,

    transaction_version: int = 3,

    max_weight: int = 0,

    store_call: bool = False,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into unsigned as_multi string

    """

    as_multi = ScaleDecoder.get_decoder_class(

        "Call", metadata=metadata, runtime_config=runtime_config

    )

    transfer = ScaleDecoder.get_decoder_class(

        "OpaqueCall", metadata=metadata, runtime_config=runtime_config

    )

    transfer.encode(

        {

            "call_module": "Balances",

            "call_function": "transfer",

            "call_args": {"dest": to_address, "value": amount},

        }

    )

    maybe_timepoint = (

        {"height": timepoint[0], "index": timepoint[1]} if timepoint else None

    )

    as_multi.encode(

        {

            "call_module": "Multisig",

            "call_function": "as_multi",

            "call_args": {

                "call": transfer.value,

                "maybe_timepoint": maybe_timepoint,

                "other_signatories": sorted(other_signatories),

                "threshold": threshold,

                "store_call": store_call,

                "max_weight": max_weight,

            },

        }

    )

    signature_payload = ScaleDecoder.get_decoder_class(

        "ExtrinsicPayloadValue", runtime_config=runtime_config

    )

    signature_payload.encode(

        {

            "call": str(as_multi.data),

            "era": "00",

            "nonce": nonce,

            "tip": tip,

            "specVersion": spec_version,

            "transactionVersion": transaction_version,

            "genesisHash": genesis_hash,

            "blockHash": genesis_hash,

        }

    )

    return str(signature_payload.data)

def _extrinsic_construction(

    metadata: "MetadataDecoder",

    account_id: str,

    signature: str,

    call_function: str,

    call_module: str,

    call_arguments: dict,

    nonce: int,

    tip: int = 0,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into extrinsic object

    """

    extrinsic = ScaleDecoder.get_decoder_class(

        "Extrinsic", metadata=metadata, runtime_config=runtime_config

    )

    extrinsic.encode(

        {

            "account_id": account_id,

            "signature_version": 1,

            "signature": signature,

            "call_function": call_function,

            "call_module": call_module,

            "call_args": call_arguments,

            "nonce": nonce,

            "era": "00",

            "tip": tip,

        }

    )

    print("Extrinsic", str(extrinsic.data))

    return str(extrinsic.data)

def unsigned_transfer_construction(

    metadata: "MetadataDecoder",

    account_id: str,

    signature: str,

    nonce: int,

    to_address: str,

    amount: int,

    tip: int = 0,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into a transfer extrinsic object

    """

    call_function = "transfer"

    call_module = "Balances"

    call_arguments = {"dest": to_address, "value": amount}

    return _extrinsic_construction(

        metadata,

        account_id,

        signature,

        call_function,

        call_module,

        call_arguments,

        nonce,

        tip,

        runtime_config,

    )

def unsigned_as_multi_construction(

    metadata: "MetadataDecoder",

    account_id: str,

    signature: str,

    nonce: int,

    to_address: str,

    amount: int,

    timepoint: Union[tuple, bool],

    other_signatories: list,

    max_weight: int = 0,

    store_call: bool = False,

    threshold: int = 2,

    tip: int = 0,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into an as_multi extrinsic object

    """

    call_function = "as_multi"

    call_module = "Multisig"

    transfer = ScaleDecoder.get_decoder_class(

        "OpaqueCall", metadata=metadata, runtime_config=runtime_config

    )

    transfer.encode(

        {

            "call_module": "Balances",

            "call_function": "transfer",

            "call_args": {"dest": to_address, "value": amount},

        }

    )

    maybe_timepoint = (

        {"height": timepoint[0], "index": timepoint[1]} if timepoint else None

    )

    call_arguments = {

        "call": transfer.value,

        "maybe_timepoint": maybe_timepoint,

        "other_signatories": sorted(other_signatories),

        "threshold": threshold,

        "store_call": store_call,

        "max_weight": max_weight,

    }

    return _extrinsic_construction(

        metadata,

        account_id,

        signature,

        call_function,

        call_module,

        call_arguments,

        nonce,

        tip,

        runtime_config,

    )

def sign_payload(keypair: tuple, payload: str) -> str:

    """

    Sign payload with keypair and return a signed hex string

    """

    if payload[0:2] == "0x":

        payload = payload[2:]

    signature = sr25519.sign(keypair, bytes.fromhex(payload))

    return signature.hex()

def hex_to_bytes(hex) -> bytes:

    """

    Generic hex to bytes conversion

    """

    if hex[0:2] == "0x":

        hex = hex[2:]

    return bytes.fromhex(hex)

Functions

as_multi_signature_payload

def as_multi_signature_payload(
    metadata: 'MetadataDecoder',
    spec_version: int,
    genesis_hash: str,
    nonce: int,
    to_address: str,
    amount: int,
    other_signatories: list,
    timepoint: Union[tuple, bool],
    threshold: int = 2,
    tip: int = 0,
    transaction_version: int = 3,
    max_weight: int = 0,
    store_call: bool = False,
    runtime_config: 'RuntimeConfigurationObject' = None
) -> str

Turn parameters gathered through side effects into unsigned as_multi string

View Source
def as_multi_signature_payload(

    metadata: "MetadataDecoder",

    spec_version: int,

    genesis_hash: str,

    nonce: int,

    to_address: str,

    amount: int,

    other_signatories: list,

    timepoint: Union[tuple, bool],

    threshold: int = 2,

    tip: int = 0,

    transaction_version: int = 3,

    max_weight: int = 0,

    store_call: bool = False,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into unsigned as_multi string

    """

    as_multi = ScaleDecoder.get_decoder_class(

        "Call", metadata=metadata, runtime_config=runtime_config

    )

    transfer = ScaleDecoder.get_decoder_class(

        "OpaqueCall", metadata=metadata, runtime_config=runtime_config

    )

    transfer.encode(

        {

            "call_module": "Balances",

            "call_function": "transfer",

            "call_args": {"dest": to_address, "value": amount},

        }

    )

    maybe_timepoint = (

        {"height": timepoint[0], "index": timepoint[1]} if timepoint else None

    )

    as_multi.encode(

        {

            "call_module": "Multisig",

            "call_function": "as_multi",

            "call_args": {

                "call": transfer.value,

                "maybe_timepoint": maybe_timepoint,

                "other_signatories": sorted(other_signatories),

                "threshold": threshold,

                "store_call": store_call,

                "max_weight": max_weight,

            },

        }

    )

    signature_payload = ScaleDecoder.get_decoder_class(

        "ExtrinsicPayloadValue", runtime_config=runtime_config

    )

    signature_payload.encode(

        {

            "call": str(as_multi.data),

            "era": "00",

            "nonce": nonce,

            "tip": tip,

            "specVersion": spec_version,

            "transactionVersion": transaction_version,

            "genesisHash": genesis_hash,

            "blockHash": genesis_hash,

        }

    )

    return str(signature_payload.data)

get_prefix

def get_prefix(
    escrow_address: str,
    address_type: int = 2
) -> str

Returns prefix containing the account ID of the address provided

View Source
def get_prefix(escrow_address: str, address_type: int = 2) -> str:

    """

    Returns prefix containing the account ID of the address provided

    """

    module_prefix = xx128("Multisig") + xx128("Multisigs")

    account_id = ss58_decode(escrow_address, address_type)

    storage_key = bytearray(xxhash.xxh64(bytes.fromhex(account_id), seed=0).digest())

    storage_key.reverse()

    return f"{module_prefix}{storage_key.hex()}{account_id}"

hash_call

def hash_call(
    call: 'scalecodec.types.Call'
) -> str

Returns a hashed call

View Source
def hash_call(call: "scalecodec.types.Call") -> str:

    """

    Returns a hashed call

    """

    call = bytes.fromhex(str(call.data)[2:])

    return f"0x{blake2b(call, digest_size=32).digest().hex()}"

hex_to_bytes

def hex_to_bytes(
    hex
) -> bytes

Generic hex to bytes conversion

View Source
def hex_to_bytes(hex) -> bytes:

    """

    Generic hex to bytes conversion

    """

    if hex[0:2] == "0x":

        hex = hex[2:]

    return bytes.fromhex(hex)

sign_payload

def sign_payload(
    keypair: tuple,
    payload: str
) -> str

Sign payload with keypair and return a signed hex string

View Source
def sign_payload(keypair: tuple, payload: str) -> str:

    """

    Sign payload with keypair and return a signed hex string

    """

    if payload[0:2] == "0x":

        payload = payload[2:]

    signature = sr25519.sign(keypair, bytes.fromhex(payload))

    return signature.hex()

transfer_signature_payload

def transfer_signature_payload(
    metadata: 'MetadataDecoder',
    address: str,
    value: int,
    nonce: int,
    genesis_hash: str,
    spec_version: int,
    transaction_version: int = 3,
    runtime_config: 'RuntimeConfigurationObject' = None
) -> str

Turn parameters gathered through side effects into unsigned transfer string

View Source
def transfer_signature_payload(

    metadata: "MetadataDecoder",

    address: str,

    value: int,

    nonce: int,

    genesis_hash: str,

    spec_version: int,

    transaction_version: int = 3,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into unsigned transfer string

    """

    call = ScaleDecoder.get_decoder_class(

        "Call", metadata=metadata, runtime_config=runtime_config

    )

    call.encode(

        {

            "call_module": "Balances",

            "call_function": "transfer",

            "call_args": {"dest": address, "value": value},

        }

    )

    signature_payload = ScaleDecoder.get_decoder_class(

        "ExtrinsicPayloadValue", runtime_config=runtime_config

    )

    signature_payload.encode(

        {

            "call": str(call.data),

            "era": "00",

            "nonce": nonce,

            "tip": 0,

            "specVersion": spec_version,

            "transactionVersion": transaction_version,

            "genesisHash": genesis_hash,

            "blockHash": genesis_hash,

        }

    )

    return str(signature_payload.data)

unsigned_as_multi_construction

def unsigned_as_multi_construction(
    metadata: 'MetadataDecoder',
    account_id: str,
    signature: str,
    nonce: int,
    to_address: str,
    amount: int,
    timepoint: Union[tuple, bool],
    other_signatories: list,
    max_weight: int = 0,
    store_call: bool = False,
    threshold: int = 2,
    tip: int = 0,
    runtime_config: 'RuntimeConfigurationObject' = None
) -> str

Turn parameters gathered through side effects into an as_multi extrinsic object

View Source
def unsigned_as_multi_construction(

    metadata: "MetadataDecoder",

    account_id: str,

    signature: str,

    nonce: int,

    to_address: str,

    amount: int,

    timepoint: Union[tuple, bool],

    other_signatories: list,

    max_weight: int = 0,

    store_call: bool = False,

    threshold: int = 2,

    tip: int = 0,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into an as_multi extrinsic object

    """

    call_function = "as_multi"

    call_module = "Multisig"

    transfer = ScaleDecoder.get_decoder_class(

        "OpaqueCall", metadata=metadata, runtime_config=runtime_config

    )

    transfer.encode(

        {

            "call_module": "Balances",

            "call_function": "transfer",

            "call_args": {"dest": to_address, "value": amount},

        }

    )

    maybe_timepoint = (

        {"height": timepoint[0], "index": timepoint[1]} if timepoint else None

    )

    call_arguments = {

        "call": transfer.value,

        "maybe_timepoint": maybe_timepoint,

        "other_signatories": sorted(other_signatories),

        "threshold": threshold,

        "store_call": store_call,

        "max_weight": max_weight,

    }

    return _extrinsic_construction(

        metadata,

        account_id,

        signature,

        call_function,

        call_module,

        call_arguments,

        nonce,

        tip,

        runtime_config,

    )

unsigned_transfer_construction

def unsigned_transfer_construction(
    metadata: 'MetadataDecoder',
    account_id: str,
    signature: str,
    nonce: int,
    to_address: str,
    amount: int,
    tip: int = 0,
    runtime_config: 'RuntimeConfigurationObject' = None
) -> str

Turn parameters gathered through side effects into a transfer extrinsic object

View Source
def unsigned_transfer_construction(

    metadata: "MetadataDecoder",

    account_id: str,

    signature: str,

    nonce: int,

    to_address: str,

    amount: int,

    tip: int = 0,

    runtime_config: "RuntimeConfigurationObject" = None,

) -> str:

    """

    Turn parameters gathered through side effects into a transfer extrinsic object

    """

    call_function = "transfer"

    call_module = "Balances"

    call_arguments = {"dest": to_address, "value": amount}

    return _extrinsic_construction(

        metadata,

        account_id,

        signature,

        call_function,

        call_module,

        call_arguments,

        nonce,

        tip,

        runtime_config,

    )

xx128

def xx128(
    word: str
) -> str

Returns a xxh128 hash on provided word

View Source
def xx128(word: str) -> str:

    """

    Returns a xxh128 hash on provided word

    """

    a = bytearray(xxhash.xxh64(word, seed=0).digest())

    b = bytearray(xxhash.xxh64(word, seed=1).digest())

    a.reverse()

    b.reverse()

    return f"{a.hex()}{b.hex()}"