Skip to content

moonworm

moonworm is a code generation tool which makes it easy for anyone to interact with smart contracts on an EVM-based blockchain.

It is inspired by abigen and generates Python bindings to smart contracts when provided with their ABIs as inputs.

More importantly, moonworm also generates fully-featured command-line interfaces to those smart contracts, which is extremely useful when running live operations against deployed contracts. It is also helpful in creating bots that interact with deployed contracts.

moonworm is capable of generating Python bindings and CLIs which use any of the following libraries to interact with EVM-based blockchains:

  1. web3.py

  2. brownie

Finally, moonworm also makes it easy to gather information about smart contracts and how people are using them:

  1. watch - generates a crawler for a given contract at runtime

  2. find_deployment - uses a binary search to pin down the exact block at which a smart contract was deployed

Installation

Install moonworm using:

pip install moonworm

Uses

Generating brownie-compatible smart contract interfaces

generate_brownie_interface 
generate_brownie_interface(
    abi: List[Dict[str, Any]],
    contract_build: Dict[str, Any],
    contract_name: str,
    relative_path: str,
    cli: bool = True,
    format: bool = True,
    prod: bool = False,
) -> str
Source code in moonworm/generators/brownie.py 
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
def generate_brownie_interface(
    abi: List[Dict[str, Any]],
    contract_build: Dict[str, Any],
    contract_name: str,
    relative_path: str,
    cli: bool = True,
    format: bool = True,
    prod: bool = False,
) -> str:
    contract_class = generate_brownie_contract_class(abi, contract_name)
    module_body = [contract_class]

    if cli:
        contract_cli_functions = generate_brownie_cli(abi, contract_name)
        module_body.extend(contract_cli_functions)

    contract_body = cst.Module(body=module_body).code
    if prod:
        content = BROWNIE_INTERFACE_PROD_TEMPLATE.format(
            contract_abi=abi,
            contract_build={
                "bytecode": contract_build["bytecode"],
                "abi": contract_build["abi"],
                "contractName": contract_build["contractName"],
            },
            contract_body=contract_body,
            moonworm_version=MOONWORM_VERSION,
        )
    else:
        content = BROWNIE_INTERFACE_TEMPLATE.format(
            contract_body=contract_body,
            moonworm_version=MOONWORM_VERSION,
            relative_path=relative_path,
        )

    if format:
        content = format_code(content)

    return content

CLI: moonworm generate-brownie

To access this functionality from the moonworm command-line interface, use the moonworm generate-brownie command:

moonworm generate-brownie --help

Generating web3.py-compatible smart contract interfaces

generate_contract_interface_content 

generate_contract_interface_content(
    abi: List[Dict[str, Any]],
    abi_file_name: str,
    format: bool = True,
) -> str
Source code in moonworm/generators/basic.py 
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
def generate_contract_interface_content(
    abi: List[Dict[str, Any]], abi_file_name: str, format: bool = True
) -> str:
    contract_body = cst.Module(body=[generate_contract_class(abi)]).code

    content = INTERFACE_FILE_TEMPLATE.format(
        contract_body=contract_body,
        moonworm_version=MOONWORM_VERSION,
        abi_file_name=abi_file_name,
    )

    if format:
        content = format_code(content)

    return content

generate_contract_cli_content 

generate_contract_cli_content(
    abi: List[Dict[str, Any]],
    abi_file_name: str,
    format: bool = True,
) -> str
Source code in moonworm/generators/basic.py 
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
def generate_contract_cli_content(
    abi: List[Dict[str, Any]], abi_file_name: str, format: bool = True
) -> str:
    cli_body = cst.Module(body=[generate_argument_parser_function(abi)]).code

    content = CLI_FILE_TEMPLATE.format(
        cli_content=cli_body,
        moonworm_version=MOONWORM_VERSION,
        abi_file_name=abi_file_name,
    )

    if format:
        content = format_code(content)

    return content

CLI: moonworm generate

To access this functionality from the moonworm command-line interface, use the moonworm generate command:

moonworm generate --help

Crawling events and method calls to smart contracts

watch_contract 
watch_contract(
    web3: Web3,
    state_provider: EthereumStateProvider,
    contract_address: ChecksumAddress,
    contract_abi: List[Dict[str, Any]],
    num_confirmations: int = 10,
    sleep_time: float = 1,
    start_block: Optional[int] = None,
    end_block: Optional[int] = None,
    min_blocks_batch: int = 100,
    max_blocks_batch: int = 5000,
    batch_size_update_threshold: int = 100,
    only_events: bool = False,
    outfile: Optional[str] = None,
) -> None

Watches a contract for events and calls.

Source code in moonworm/watch.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def watch_contract(
    web3: Web3,
    state_provider: EthereumStateProvider,
    contract_address: ChecksumAddress,
    contract_abi: List[Dict[str, Any]],
    num_confirmations: int = 10,
    sleep_time: float = 1,
    start_block: Optional[int] = None,
    end_block: Optional[int] = None,
    min_blocks_batch: int = 100,
    max_blocks_batch: int = 5000,
    batch_size_update_threshold: int = 100,
    only_events: bool = False,
    outfile: Optional[str] = None,
) -> None:
    """
    Watches a contract for events and calls.
    """

    def _crawl_events(
        event_abi, from_block: int, to_block: int, batch_size: int
    ) -> Tuple[List[Dict[str, Any]], int]:
        """
        Crawls events from the given block range.
        reduces the batch_size if response is failing.
        increases the batch_size if response is successful.
        """
        events = []
        current_from_block = from_block

        while current_from_block <= to_block:
            current_to_block = min(current_from_block + batch_size, to_block)
            try:
                events_chunk = _fetch_events_chunk(
                    web3,
                    event_abi,
                    current_from_block,
                    current_to_block,
                    [contract_address],
                )
                events.extend(events_chunk)
                current_from_block = current_to_block + 1
                if len(events) <= batch_size_update_threshold:
                    batch_size = min(batch_size * 2, max_blocks_batch)
            except Exception as e:
                if batch_size <= min_blocks_batch:
                    raise e
                time.sleep(0.1)
                batch_size = max(batch_size // 2, min_blocks_batch)
        return events, batch_size

    current_batch_size = min_blocks_batch
    state = MockState()
    crawler = FunctionCallCrawler(
        state,
        state_provider,
        contract_abi,
        [web3.toChecksumAddress(contract_address)],
    )

    event_abis = [item for item in contract_abi if item["type"] == "event"]

    if start_block is None:
        current_block = web3.eth.blockNumber - num_confirmations * 2
    else:
        current_block = start_block

    progress_bar = tqdm(unit=" blocks")
    progress_bar.set_description(f"Current block {current_block}")
    ofp = None
    if outfile is not None:
        ofp = open(outfile, "a")

    try:
        while end_block is None or current_block <= end_block:
            time.sleep(sleep_time)
            until_block = min(
                web3.eth.blockNumber - num_confirmations,
                current_block + current_batch_size,
            )
            if end_block is not None:
                until_block = min(until_block, end_block)
            if until_block < current_block:
                sleep_time *= 2
                continue

            sleep_time /= 2
            if not only_events:
                crawler.crawl(current_block, until_block)
                if state.state:
                    print("Got transaction calls:")
                    for call in state.state:
                        pp.pprint(call, width=200, indent=4)
                        if ofp is not None:
                            print(json.dumps(asdict(call)), file=ofp)
                            ofp.flush()
                    state.flush()

            for event_abi in event_abis:
                all_events, new_batch_size = _crawl_events(
                    event_abi, current_block, until_block, current_batch_size
                )

                if only_events:
                    # Updating batch size only in `--only-events` mode
                    # otherwise it will start taking too much if we also crawl transactions
                    current_batch_size = new_batch_size
                for event in all_events:
                    print("Got event:")
                    pp.pprint(event, width=200, indent=4)
                    if ofp is not None:
                        print(json.dumps(event), file=ofp)
                        ofp.flush()

            progress_bar.set_description(
                f"Current block {until_block}, Already watching for"
            )
            progress_bar.update(until_block - current_block + 1)
            current_block = until_block + 1
    finally:
        if ofp is not None:
            ofp.close()

State providers: Accessing blockchain state

watch_contract uses EthereumStateProvider objects to crawl blockchain state. Any object inheriting from the EthereumStateProvider base class will suffice.

EthereumStateProvider

Abstract class for Ethereum state provider. If you want to use a different state provider, you can implement this class.

Source code in moonworm/crawler/ethereum_state_provider.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class EthereumStateProvider(ABC):
    """
    Abstract class for Ethereum state provider.
    If you want to use a different state provider, you can implement this class.
    """

    @abstractmethod
    def get_last_block_number(self) -> int:
        """
        Returns the last block number.
        """
        pass

    @abstractmethod
    def get_block_timestamp(self, block_number: int) -> int:
        """
        Returns the timestamp of the block with the given block number.
        """
        pass

    @abstractmethod
    def get_transactions_to_address(
        self, address, block_number: int
    ) -> List[Dict[str, Any]]:
        """
        Returns all transactions to the given address in the given block number.
        """
        pass

For example:

Web3StateProvider

Bases: EthereumStateProvider

Implementation of EthereumStateProvider with web3.

Source code in moonworm/crawler/ethereum_state_provider.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class Web3StateProvider(EthereumStateProvider):
    """
    Implementation of EthereumStateProvider with web3.
    """

    def __init__(self, w3: Web3):
        self.w3 = w3

        self.blocks_cache = {}

    def get_transaction_reciept(self, transaction_hash: str) -> Dict[str, Any]:
        return self.w3.eth.get_transaction_receipt(transaction_hash)

    def get_last_block_number(self) -> int:
        return self.w3.eth.block_number

    def _get_block(self, block_number: int) -> Dict[str, Any]:
        if block_number in self.blocks_cache:
            return self.blocks_cache[block_number]
        block = self.w3.eth.getBlock(block_number, full_transactions=True)

        # clear cache if it grows too large
        if len(self.blocks_cache) > 50:
            self.blocks_cache = {}

        self.blocks_cache[block_number] = block
        return block

    def get_block_timestamp(self, block_number: int) -> int:
        block = self._get_block(block_number)
        return block["timestamp"]

    def get_transactions_to_address(
        self, address: ChecksumAddress, block_number: int
    ) -> List[Dict[str, Any]]:
        block = self._get_block(block_number)

        all_transactions = block["transactions"]
        return [tx for tx in all_transactions if tx.get("to") == address]

CLI: moonworm watch

To access this functionality from the moonworm command-line interface, use the moonworm watch command:

moonworm watch --help

Discovering the block at which a smart contract was deployed

find_deployment_block 
find_deployment_block(
    web3_client: Web3,
    contract_address: ChecksumAddress,
    web3_interval: float,
) -> Optional[int]

Note: We will assume no selfdestruct for now.

This means that, if the address does not currently contain code, we will assume it never contained code and is therefore not a smart contract address.

Source code in moonworm/deployment.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def find_deployment_block(
    web3_client: Web3,
    contract_address: ChecksumAddress,
    web3_interval: float,
) -> Optional[int]:
    """
    Note: We will assume no selfdestruct for now.

    This means that, if the address does not currently contain code, we will assume it never contained
    code and is therefore not a smart contract address.
    """
    log_prefix = f"find_deployment_block(web3_client, contract_address={contract_address}, web3_interval={web3_interval}) -- "

    logger.info(f"{log_prefix}Function invoked")
    config = {CONFIG_KEY_WEB3_INTERVAL: web3_interval}

    max_block = int(web3_client.eth.block_number)
    min_block = 0
    middle_block = int((min_block + max_block) / 2)

    was_deployed_at_max_block = was_deployed_at_block(
        web3_client, contract_address, max_block, config=config
    )
    if not was_deployed_at_max_block:
        logger.warn(f"{log_prefix}Address is not a smart contract")
        return None

    was_deployed: Dict[int, bool] = {
        max_block: was_deployed_at_max_block,
        min_block: was_deployed_at_block(
            web3_client, contract_address, min_block, config=config
        ),
        middle_block: was_deployed_at_block(
            web3_client, contract_address, middle_block, config=config
        ),
    }

    while max_block - min_block >= 2:
        logger.info(
            f"{log_prefix}Binary search -- max_block={max_block}, min_block={min_block}, middle_block={middle_block}"
        )
        if not was_deployed[min_block] and not was_deployed[middle_block]:
            min_block = middle_block
        else:
            max_block = middle_block

        middle_block = int((min_block + max_block) / 2)

        was_deployed[middle_block] = was_deployed_at_block(
            web3_client, contract_address, middle_block, config=config
        )

    if was_deployed[min_block]:
        return min_block
    return max_block

CLI: moonworm find-deployment

To access this functionality from the moonworm command-line interface, use the moonworm find-deployment command:

moonworm find-deployment --help
Last update: March 5, 2024
Created: March 5, 2024