Added CLI functionality for basic diamond operations

pull/1/head
Neeraj Kashyap 2021-12-12 13:33:56 -08:00
rodzic db1437bafb
commit 65fdaa166e
3 zmienionych plików z 254 dodań i 1 usunięć

78
dao/abi.py 100644
Wyświetl plik

@ -0,0 +1,78 @@
"""
ABI utilities, because web3 doesn't do selectors well.
"""
import glob
import json
import os
from typing import Any, Dict, List, Optional
from web3 import Web3
def abi_input_signature(input_abi: Dict[str, Any]) -> str:
"""
Stringifies a function ABI input object according to the ABI specification:
https://docs.soliditylang.org/en/v0.5.3/abi-spec.html
"""
input_type = input_abi["type"]
if input_type.startswith("tuple"):
component_types = [
abi_input_signature(component) for component in input_abi["components"]
]
input_type = f"({','.join(component_types)}){input_type[len('tuple'):]}"
return input_type
def abi_function_signature(function_abi: Dict[str, Any]) -> str:
"""
Stringifies a function ABI according to the ABI specification:
https://docs.soliditylang.org/en/v0.5.3/abi-spec.html
"""
function_name = function_abi["name"]
function_arg_types = [
abi_input_signature(input_item) for input_item in function_abi["inputs"]
]
function_signature = f"{function_name}({','.join(function_arg_types)})"
return function_signature
def encode_function_signature(function_abi: Dict[str, Any]) -> Optional[str]:
"""
Encodes the given function (from ABI) with arguments arg_1, ..., arg_n into its 4 byte signature
by calculating:
keccak256("<function_name>(<arg_1_type>,...,<arg_n_type>")
If function_abi is not actually a function ABI (detected by checking if function_abi["type"] == "function),
returns None.
"""
if function_abi["type"] != "function":
return None
function_signature = abi_function_signature(function_abi)
encoded_signature = Web3.keccak(text=function_signature)[:4]
return encoded_signature.hex()
def project_abis(project_dir: str) -> Dict[str, List[Dict[str, Any]]]:
"""
Load all ABIs for project contracts and return then in a dictionary keyed by contract name.
Inputs:
- project_dir
Path to brownie project
"""
build_dir = os.path.join(project_dir, "build", "contracts")
build_files = glob.glob(os.path.join(build_dir, "*.json"))
abis: Dict[str, List[Dict[str, Any]]] = {}
for filepath in build_files:
contract_name, _ = os.path.splitext(os.path.basename(filepath))
with open(filepath, "r") as ifp:
contract_artifact = json.load(ifp)
contract_abi = contract_artifact.get("abi", [])
abis[contract_name] = contract_abi
return abis

Wyświetl plik

@ -1,5 +1,21 @@
import argparse
from . import diamond
def main():
print("Hello")
parser = argparse.ArgumentParser(
description="dao: The command line interface to Moonstream DAO"
)
parser.set_defaults(func=lambda _: parser.print_help())
dao_subparsers = parser.add_subparsers()
diamond_parser = diamond.generate_cli()
dao_subparsers.add_parser("diamond", parents=[diamond_parser], add_help=False)
args = parser.parse_args()
args.func(args)
if __name__ == "__main__":
main()

159
dao/diamond.py 100644
Wyświetl plik

@ -0,0 +1,159 @@
"""
Generic diamond functionality for Moonstream contracts.
"""
import argparse
import os
from typing import Any, Dict, List, Set
from brownie import network
from . import (
abi,
Diamond,
DiamondCutFacet,
DiamondLoupeFacet,
OwnershipFacet,
)
FACETS: Dict[str, Any] = {
"DiamondCutFacet": DiamondCutFacet,
"DiamondLoupeFacet": DiamondLoupeFacet,
"OwnershipFacet": OwnershipFacet,
}
FACET_PRECEDENCE: List[str] = [
"DiamondCutFacet",
"OwnershipFacet",
"DiamondLoupeFacet",
]
FACET_ACTIONS: Dict[str, int] = {"add": 0, "replace": 1, "remove": 2}
ZERO_ADDRESS = "0x0000000000000000000000000000000000000000"
def facet_cut(
diamond_address: str,
facet_name: str,
facet_address: str,
action: str,
transaction_config: Dict[str, Any],
) -> Any:
"""
Cuts the given facet onto the given Diamond contract.
Resolves selectors in the precedence order defined by FACET_PRECEDENCE (highest precedence first).
"""
assert (
facet_name in FACETS
), f"Invalid facet: {facet_name}. Choices: {','.join(FACETS)}."
assert (
action in FACET_ACTIONS
), f"Invalid cut action: {action}. Choices: {','.join(FACET_ACTIONS)}."
project_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
abis = abi.project_abis(project_dir)
reserved_selectors: Set[str] = set()
for facet in FACET_PRECEDENCE:
if facet == facet_name:
break
facet_abi = abis.get(facet, [])
for item in facet_abi:
if item["type"] == "function":
reserved_selectors.add(abi.encode_function_signature(item))
facet_function_selectors: List[str] = []
facet_abi = abis.get(facet_name, [])
for item in facet_abi:
if item["type"] == "function":
function_selector = abi.encode_function_signature(item)
if function_selector not in reserved_selectors:
facet_function_selectors.append(function_selector)
target_address = facet_address
if FACET_ACTIONS[action] == 2:
target_address = ZERO_ADDRESS
diamond_cut_action = [
target_address,
FACET_ACTIONS[action],
facet_function_selectors,
]
diamond = DiamondCutFacet.DiamondCutFacet(diamond_address)
transaction = diamond.diamond_cut(
[diamond_cut_action], ZERO_ADDRESS, b"", transaction_config
)
return transaction
def handle_facet_cut(args: argparse.Namespace) -> None:
network.connect(args.network)
diamond_address = args.address
action = args.action
facet_name = args.facet_name
facet_address = args.facet_address
transaction_config = Diamond.get_transaction_config(args)
facet_cut(diamond_address, facet_name, facet_address, action, transaction_config)
def generate_cli() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="CLI to manage Moonstream DAO diamond contracts",
)
parser.set_defaults(func=lambda _: parser.print_help())
subcommands = parser.add_subparsers()
Diamond_parser = Diamond.generate_cli()
subcommands.add_parser("diamond", parents=[Diamond_parser], add_help=False)
facet_cut_parser = subcommands.add_parser("facet-cut")
Diamond.add_default_arguments(facet_cut_parser, transact=True)
facet_cut_parser.add_argument(
"--facet-name",
required=True,
choices=FACETS,
help="Name of facet to cut into or out of diamond",
)
facet_cut_parser.add_argument(
"--facet-address",
required=False,
default=ZERO_ADDRESS,
help=f"Address of deployed facet (default: {ZERO_ADDRESS})",
)
facet_cut_parser.add_argument(
"--action",
required=True,
choices=FACET_ACTIONS,
help="Diamond cut action to take on entire facet",
)
facet_cut_parser.set_defaults(func=handle_facet_cut)
DiamondCutFacet_parser = DiamondCutFacet.generate_cli()
subcommands.add_parser(
"diamond-cut", parents=[DiamondCutFacet_parser], add_help=False
)
DiamondLoupeFacet_parser = DiamondLoupeFacet.generate_cli()
subcommands.add_parser(
"diamond-loupe", parents=[DiamondLoupeFacet_parser], add_help=False
)
OwnershipFacet_parser = OwnershipFacet.generate_cli()
subcommands.add_parser("ownership", parents=[OwnershipFacet_parser], add_help=False)
return parser
def main() -> None:
parser = generate_cli()
args = parser.parse_args()
args.func(args)
if __name__ == "__main__":
main()