zeth.cli.utils Namespace Reference

Classes
class	ClientConfig
class	MixerDescription
class	NetworkConfig

Functions
NetworkConfig	get_eth_network (Optional[str] eth_network)
Any	open_web3_from_network (NetworkConfig eth_net)
str	load_contract_address (str contract_addr)
Any	open_web3_from_ctx (ClientConfig ctx)
Dict[str, Any]	get_erc20_abi ()
InstanceDescription	get_erc20_instance_description (str token_address)
None	write_mixer_description (str mixer_desc_file, MixerDescription mixer_desc)
MixerDescription	load_mixer_description (str mixer_desc_file)
MixerDescription	load_mixer_description_from_ctx (ClientConfig ctx)
str	get_zeth_address_file (ClientConfig ctx)
ZethAddressPub	load_zeth_address_public (ClientConfig ctx)
None	write_zeth_address_public (ZethAddressPub pub_addr, str pub_addr_file)
ZethAddressPriv	load_zeth_address_secret (ClientConfig ctx)
None	write_zeth_address_secret (ZethAddressPriv secret_addr, str addr_file)
ZethAddress	load_zeth_address (ClientConfig ctx)
Wallet	open_wallet (Any mixer_instance, ZethAddressPriv js_secret, ClientConfig ctx)
int	do_sync (Any web3, Wallet wallet, PairingParameters pp, Optional[str] wait_tx, Optional[Callable[[ZethNoteDescription], None]] callback=None, Optional[int] batch_size=None)
str	pub_address_file (str addr_file)
str	find_pub_address_file (str base_file)
ProverClient	create_prover_client (ClientConfig ctx)
MixerClient	create_mixer_client (ClientConfig ctx, Optional[ProverClient] prover_client=None)
Tuple[MixerClient, MixerDescription]	create_mixer_client_and_mixer_desc (ClientConfig ctx, Optional[ProverClient] prover_client=None)
str	zeth_note_short (ZethNoteDescription note_desc)
None	zeth_note_short_print (ZethNoteDescription note_desc)
Tuple[ZethAddressPub, EtherValue]	parse_output (str output_str)
str	load_eth_address (Optional[str] eth_addr)
None	write_eth_address (str eth_addr, str eth_addr_file)
Optional[bytes]	load_eth_private_key (Optional[str] private_key_file)
None	write_eth_private_key (bytes private_key, str private_key_file)

Function Documentation

create_mixer_client()

MixerClient zeth.cli.utils.create_mixer_client	(	ClientConfig	ctx,
		Optional[ProverClient]	prover_client = None
	)

Create a MixerClient for an existing deployment.

Definition at line 381 of file utils.py.

def create_mixer_client(
        ctx: ClientConfig,
        prover_client: Optional[ProverClient] = None) -> MixerClient:
    """
    Create a MixerClient for an existing deployment.
    """
    mixer_client, _ = create_mixer_client_and_mixer_desc(ctx, prover_client)
    return mixer_client
 
 

Here is the call graph for this function:

create_mixer_client_and_mixer_desc()

Tuple[MixerClient, MixerDescription] zeth.cli.utils.create_mixer_client_and_mixer_desc	(	ClientConfig	ctx,
		Optional[ProverClient]	prover_client = None
	)

Create a MixerClient and MixerDescription object, for an existing deployment.

Definition at line 391 of file utils.py.

def create_mixer_client_and_mixer_desc(
        ctx: ClientConfig,
        prover_client: Optional[ProverClient] = None
) -> Tuple[MixerClient, MixerDescription]:
    """
    Create a MixerClient and MixerDescription object, for an existing deployment.
    """
    web3 = open_web3_from_ctx(ctx)
    mixer_desc = load_mixer_description_from_ctx(ctx)
    mixer_instance = mixer_desc.mixer.instantiate(web3)
    if prover_client is None:
        prover_client = create_prover_client(ctx)
    prover_config = prover_client.get_configuration()
    mixer_client = MixerClient(web3, prover_config, mixer_instance)
    return (mixer_client, mixer_desc)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

create_prover_client()

ProverClient zeth.cli.utils.create_prover_client

(

ClientConfig

ctx

)

Create a prover client using the settings from the commands context.

Definition at line 373 of file utils.py.

def create_prover_client(ctx: ClientConfig) -> ProverClient:
    """
    Create a prover client using the settings from the commands context.
    """
    return ProverClient(
        ctx.prover_server_endpoint, ctx.prover_config_file)
 
 

Here is the caller graph for this function:

do_sync()

int zeth.cli.utils.do_sync	(	Any	web3,
		Wallet	wallet,
		PairingParameters	pp,
		Optional[str]	wait_tx,
		Optional[Callable[[ZethNoteDescription], None]]	callback = None,
		Optional[int]	batch_size = None
	)

Implementation of sync, reused by several commands.  Returns the
block_number synced to.  Also updates and saves the MerkleTree.

Definition at line 289 of file utils.py.

def do_sync(
        web3: Any,
        wallet: Wallet,
        pp: PairingParameters,
        wait_tx: Optional[str],
        callback: Optional[Callable[[ZethNoteDescription], None]] = None,
        batch_size: Optional[int] = None) -> int:
    """
    Implementation of sync, reused by several commands.  Returns the
    block_number synced to.  Also updates and saves the MerkleTree.
    """
    def _do_sync() -> int:
        wallet_next_block = wallet.get_next_block()
        chain_block_number = get_block_number(web3)
 
        if chain_block_number >= wallet_next_block:
            new_merkle_root: Optional[bytes] = None
 
            print(f"SYNCHING blocks ({wallet_next_block} - {chain_block_number})")
            mixer_instance = wallet.mixer_instance
            for mix_result in get_mix_results(
                    web3,
                    mixer_instance,
                    wallet_next_block,
                    chain_block_number,
                    batch_size):
                new_merkle_root = mix_result.new_merkle_root
                for note_desc in wallet.receive_notes(
                        mix_result.output_events, pp):
                    if callback:
                        callback(note_desc)
 
                spent_commits = wallet.mark_nullifiers_used(mix_result.nullifiers)
                for commit in spent_commits:
                    print(f"    SPENT: {commit}")
 
            wallet.update_and_save_state(next_block=chain_block_number + 1)
 
            # Check merkle root and save the updated tree
            if new_merkle_root:
                our_merkle_root = wallet.merkle_tree.get_root()
                assert new_merkle_root == our_merkle_root
 
        return chain_block_number
 
    # Do a sync upfront (it would be a waste of time to wait for a tx before
    # syncing, as it can take time to traverse all blocks).  Then wait for a tx
    # if requested, and sync again.
 
    if wait_tx:
        _do_sync()
        tx_receipt = web3.eth.waitForTransactionReceipt(wait_tx, 10000)
        tx = web3.eth.getTransaction(wait_tx)
        gas_used = tx_receipt.gasUsed
        status = tx_receipt.status
        size_bytes = len(tx.input)
        print(
            f"{wait_tx[0:8]}: gasUsed={gas_used}, status={status}, "
            f"input_size={size_bytes}")
 
    return _do_sync()
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

find_pub_address_file()

str zeth.cli.utils.find_pub_address_file

(

str

base_file

)

Given a file name, which could point to a private or public key file, guess
at the name of the public key file.

Definition at line 359 of file utils.py.

def find_pub_address_file(base_file: str) -> str:
    """
    Given a file name, which could point to a private or public key file, guess
    at the name of the public key file.
    """
    pub_addr_file = pub_address_file(base_file)
    if exists(pub_addr_file):
        return pub_addr_file
    if exists(base_file):
        return base_file
 
    raise ClickException(f"No public key file {pub_addr_file} or {base_file}")
 
 

Here is the call graph for this function:

get_erc20_abi()

Dict[str, Any] zeth.cli.utils.get_erc20_abi

(

)

Definition at line 187 of file utils.py.

def get_erc20_abi() -> Dict[str, Any]:
    zeth_dir = get_zeth_dir()
    openzeppelin_dir = join(
        zeth_dir, "zeth_contracts", "node_modules", "openzeppelin-solidity")
    ierc20_path = join(
        openzeppelin_dir, "contracts", "token", "ERC20", "IERC20.sol")
    compiled_sol = compile_files([ierc20_path])
    erc20_interface = compiled_sol[ierc20_path + ":IERC20"]
    return erc20_interface["abi"]
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

get_erc20_instance_description()

InstanceDescription zeth.cli.utils.get_erc20_instance_description

(

str

token_address

)

Definition at line 198 of file utils.py.

def get_erc20_instance_description(token_address: str) -> InstanceDescription:
    return InstanceDescription(token_address, get_erc20_abi())
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

get_eth_network()

NetworkConfig zeth.cli.utils.get_eth_network

(

Optional[str]

eth_network

)

Parse the `eth_network` parameter to extract a URL. If `eth_network` does
not contain a URL, try interpreting it as a network name, otherwise
interpret it as a file to load the network config from. Fall back to a
default network config filename, and finally the default network name.

Definition at line 84 of file utils.py.

def get_eth_network(eth_network: Optional[str]) -> NetworkConfig:
    """
    Parse the `eth_network` parameter to extract a URL. If `eth_network` does
    not contain a URL, try interpreting it as a network name, otherwise
    interpret it as a file to load the network config from. Fall back to a
    default network config filename, and finally the default network name.
    """
    if eth_network is None:
        if exists(ETH_NETWORK_FILE_DEFAULT):
            eth_network = ETH_NETWORK_FILE_DEFAULT
        else:
            eth_network = ETH_NETWORK_DEFAULT
 
    if eth_network.startswith("http"):
        # When given only a url, assume the default network name
        return NetworkConfig(ETH_NETWORK_DEFAULT, eth_network)
 
    # Try loading from a file
    if exists(eth_network):
        with open(eth_network) as network_f:
            return NetworkConfig.from_json(network_f.read())
 
    # Assume a network name
    try:
        endpoint = ETH_RPC_ENDPOINT_DEFAULTS[eth_network]
        return NetworkConfig(eth_network, endpoint)
    except KeyError as ex:
        raise ClickException(f"invalid network name / url: {eth_network}") from ex
 
 

Here is the caller graph for this function:

get_zeth_address_file()

str zeth.cli.utils.get_zeth_address_file

(

ClientConfig

ctx

)

Definition at line 224 of file utils.py.

def get_zeth_address_file(ctx: ClientConfig) -> str:
    return ctx.address_file
 
 

Here is the caller graph for this function:

load_contract_address()

str zeth.cli.utils.load_contract_address

(

str

contract_addr

)

Parse a string as either an eth address, or a contract instance file.

Definition at line 121 of file utils.py.

def load_contract_address(contract_addr: str) -> str:
    """
    Parse a string as either an eth address, or a contract instance file.
    """
    if contract_addr.startswith("0x"):
        return Web3.toChecksumAddress(contract_addr)
    if exists(contract_addr):
        with open(contract_addr, "r") as instance_f:
            instance = InstanceDescription.from_json_dict(json.load(instance_f))
        return Web3.toChecksumAddress(instance.address)
    raise ClickException(
        f"failed to parse as address or instance file: {contract_addr}")
 
 

Here is the caller graph for this function:

load_eth_address()

str zeth.cli.utils.load_eth_address

(

Optional[str]

eth_addr

)

Given an --eth-addr command line param, either parse the address, load from
the file, or use a default file name.

Definition at line 444 of file utils.py.

def load_eth_address(eth_addr: Optional[str]) -> str:
    """
    Given an --eth-addr command line param, either parse the address, load from
    the file, or use a default file name.
    """
    eth_addr = eth_addr or ETH_ADDRESS_DEFAULT
    if eth_addr.startswith("0x"):
        return Web3.toChecksumAddress(eth_addr)
    if exists(eth_addr):
        with open(eth_addr, "r") as eth_addr_f:
            return Web3.toChecksumAddress(eth_addr_f.read().rstrip())
    raise ClickException(f"could find file or parse eth address: {eth_addr}")
 
 

Here is the caller graph for this function:

load_eth_private_key()

Optional[bytes] zeth.cli.utils.load_eth_private_key

(

Optional[str]

private_key_file

)

Definition at line 465 of file utils.py.

def load_eth_private_key(private_key_file: Optional[str]) -> Optional[bytes]:
    private_key_file = private_key_file or ETH_PRIVATE_KEY_FILE_DEFAULT
    if exists(private_key_file):
        with open(private_key_file, "rb") as private_key_f:
            return private_key_f.read(32)
    return None
 
 

Here is the caller graph for this function:

load_mixer_description()

MixerDescription zeth.cli.utils.load_mixer_description

(

str

mixer_desc_file

)

Return mixer and token (if present) contract instances

Definition at line 212 of file utils.py.

def load_mixer_description(mixer_desc_file: str) -> MixerDescription:
    """
    Return mixer and token (if present) contract instances
    """
    with open(mixer_desc_file, "r") as desc_f:
        return MixerDescription.from_json_dict(json.load(desc_f))
 
 

Here is the caller graph for this function:

load_mixer_description_from_ctx()

MixerDescription zeth.cli.utils.load_mixer_description_from_ctx

(

ClientConfig

ctx

)

Definition at line 220 of file utils.py.

def load_mixer_description_from_ctx(ctx: ClientConfig) -> MixerDescription:
    return load_mixer_description(ctx.instance_file)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

load_zeth_address()

ZethAddress zeth.cli.utils.load_zeth_address

(

ClientConfig

ctx

)

Load a ZethAddress secret from a file, and the associated public address,
and return as a ZethAddress.

Definition at line 265 of file utils.py.

def load_zeth_address(ctx: ClientConfig) -> ZethAddress:
    """
    Load a ZethAddress secret from a file, and the associated public address,
    and return as a ZethAddress.
    """
    return ZethAddress.from_secret_public(
        load_zeth_address_secret(ctx),
        load_zeth_address_public(ctx))
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

load_zeth_address_public()

ZethAddressPub zeth.cli.utils.load_zeth_address_public

(

ClientConfig

ctx

)

Load a ZethAddressPub from a key file.

Definition at line 228 of file utils.py.

def load_zeth_address_public(ctx: ClientConfig) -> ZethAddressPub:
    """
    Load a ZethAddressPub from a key file.
    """
    secret_key_file = get_zeth_address_file(ctx)
    pub_addr_file = pub_address_file(secret_key_file)
    with open(pub_addr_file, "r") as pub_addr_f:
        return ZethAddressPub.parse(pub_addr_f.read())
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

load_zeth_address_secret()

ZethAddressPriv zeth.cli.utils.load_zeth_address_secret

(

ClientConfig

ctx

)

Read ZethAddressPriv

Definition at line 247 of file utils.py.

def load_zeth_address_secret(ctx: ClientConfig) -> ZethAddressPriv:
    """
    Read ZethAddressPriv
    """
    addr_file = get_zeth_address_file(ctx)
    with open(addr_file, "r") as addr_f:
        return ZethAddressPriv.from_json(addr_f.read())
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

open_wallet()

Wallet zeth.cli.utils.open_wallet	(	Any	mixer_instance,
		ZethAddressPriv	js_secret,
		ClientConfig	ctx
	)

Load a wallet using a secret key.

Definition at line 275 of file utils.py.

def open_wallet(
        mixer_instance: Any,
        js_secret: ZethAddressPriv,
        ctx: ClientConfig) -> Wallet:
    """
    Load a wallet using a secret key.
    """
    wallet_dir = ctx.wallet_dir
    prover_config = create_prover_client(ctx).get_configuration()
    tree_hash = get_tree_hash_for_pairing(prover_config.pairing_parameters.name)
    return Wallet(
        mixer_instance, WALLET_USERNAME, wallet_dir, js_secret, tree_hash)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

open_web3_from_ctx()

Any zeth.cli.utils.open_web3_from_ctx

(

ClientConfig

ctx

)

Definition at line 135 of file utils.py.

def open_web3_from_ctx(ctx: ClientConfig) -> Any:
    eth_net = get_eth_network(ctx.eth_network)
    return open_web3_from_network(eth_net)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

open_web3_from_network()

Any zeth.cli.utils.open_web3_from_network

(

NetworkConfig

eth_net

)

Definition at line 114 of file utils.py.

def open_web3_from_network(eth_net: NetworkConfig) -> Any:
    return open_web3(
        url=eth_net.endpoint,
        certificate=eth_net.certificate,
        insecure=eth_net.insecure)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

parse_output()

Tuple[ZethAddressPub, EtherValue] zeth.cli.utils.parse_output

(

str

output_str

)

Parse a string of the form "<receiver_pub_address>,<value>" to an output
specification. <receiver_pub_address> can be a file name containing the
address. "<value>" is interpreted as the <default-address-file>,<value>.

Definition at line 421 of file utils.py.

def parse_output(output_str: str) -> Tuple[ZethAddressPub, EtherValue]:
    """
    Parse a string of the form "<receiver_pub_address>,<value>" to an output
    specification. <receiver_pub_address> can be a file name containing the
    address. "<value>" is interpreted as the <default-address-file>,<value>.
    """
    parts = output_str.split(",")
    if len(parts) == 1:
        addr = ZETH_PUBLIC_ADDRESS_FILE_DEFAULT
        value = parts[0]
    elif len(parts) == 2:
        addr = parts[0]
        value = parts[1]
    else:
        raise ClickException(f"invalid output spec: {output_str}")
 
    if exists(addr):
        with open(addr, "r") as addr_f:
            addr = addr_f.read()
 
    return (ZethAddressPub.parse(addr), EtherValue(value))
 
 

Here is the caller graph for this function:

pub_address_file()

str zeth.cli.utils.pub_address_file

(

str

addr_file

)

The name of a public address file, given the secret address file.

Definition at line 352 of file utils.py.

def pub_address_file(addr_file: str) -> str:
    """
    The name of a public address file, given the secret address file.
    """
    return splitext(addr_file)[0] + ".pub"
 
 

Here is the caller graph for this function:

write_eth_address()

None zeth.cli.utils.write_eth_address	(	str	eth_addr,
		str	eth_addr_file
	)

Definition at line 458 of file utils.py.

def write_eth_address(eth_addr: str, eth_addr_file: str) -> None:
    if exists(eth_addr_file):
        raise ClickException(f"refusing to overwrite address \"{eth_addr_file}\"")
    with open(eth_addr_file, "w") as eth_addr_f:
        eth_addr_f.write(eth_addr)
 
 

Here is the caller graph for this function:

write_eth_private_key()

None zeth.cli.utils.write_eth_private_key	(	bytes	private_key,
		str	private_key_file
	)

Definition at line 473 of file utils.py.

def write_eth_private_key(private_key: bytes, private_key_file: str) -> None:
    if exists(private_key_file):
        raise ClickException(
            f"refusing to overwrite private key \"{private_key_file}\"")
    with open(private_key_file, "wb") as private_key_f:
        private_key_f.write(private_key)

Here is the caller graph for this function:

write_mixer_description()

None zeth.cli.utils.write_mixer_description	(	str	mixer_desc_file,
		MixerDescription	mixer_desc
	)

Write the mixer (and token) instance information

Definition at line 202 of file utils.py.

def write_mixer_description(
        mixer_desc_file: str,
        mixer_desc: MixerDescription) -> None:
    """
    Write the mixer (and token) instance information
    """
    with open(mixer_desc_file, "w") as instance_f:
        json.dump(mixer_desc.to_json_dict(), instance_f)
 
 

Here is the caller graph for this function:

write_zeth_address_public()

None zeth.cli.utils.write_zeth_address_public	(	ZethAddressPub	pub_addr,
		str	pub_addr_file
	)

Write a ZethAddressPub to a file

Definition at line 238 of file utils.py.

def write_zeth_address_public(
        pub_addr: ZethAddressPub, pub_addr_file: str) -> None:
    """
    Write a ZethAddressPub to a file
    """
    with open(pub_addr_file, "w") as pub_addr_f:
        pub_addr_f.write(str(pub_addr))
 
 

Here is the caller graph for this function:

write_zeth_address_secret()

None zeth.cli.utils.write_zeth_address_secret	(	ZethAddressPriv	secret_addr,
		str	addr_file
	)

Write ZethAddressPriv to file

Definition at line 256 of file utils.py.

def write_zeth_address_secret(
        secret_addr: ZethAddressPriv, addr_file: str) -> None:
    """
    Write ZethAddressPriv to file
    """
    with open(addr_file, "w") as addr_f:
        addr_f.write(secret_addr.to_json())
 
 

Here is the caller graph for this function:

zeth_note_short()

str zeth.cli.utils.zeth_note_short

(

ZethNoteDescription

note_desc

)

Generate a short human-readable description of a commitment.

Definition at line 408 of file utils.py.

def zeth_note_short(note_desc: ZethNoteDescription) -> str:
    """
    Generate a short human-readable description of a commitment.
    """
    value = from_zeth_units(int(note_desc.note.value, 16)).ether()
    cm = short_commitment(note_desc.commitment)
    return f"{cm}: value={value} ETH, addr={note_desc.address}"
 
 

Here is the call graph for this function:

zeth_note_short_print()

None zeth.cli.utils.zeth_note_short_print

(

ZethNoteDescription

note_desc

)

Definition at line 417 of file utils.py.

def zeth_note_short_print(note_desc: ZethNoteDescription) -> None:
    print(f" NEW NOTE: {zeth_note_short(note_desc)}")