feat: etherscan contract verification

boa/contracts/vyper/vyper_contract.py

@@ -570,11 +570,11 @@ def __init__(
         self.env.register_contract(self._address, self)
 
     def _run_init(self, *args, value=0, override_address=None, gas=None):
-        encoded_args = b""
+        self.constructor_calldata = b""
         if self._ctor:
-            encoded_args = self._ctor.prepare_calldata(*args)
+            self.constructor_calldata = self._ctor.prepare_calldata(*args)
 
-        initcode = self.compiler_data.bytecode + encoded_args
+        initcode = self.compiler_data.bytecode + self.constructor_calldata
         with self._anchor_source_map(self._deployment_source_map):
             address, computation = self.env.deploy(
                 bytecode=initcode,

boa/explorer.py

@@ -1,15 +1,22 @@
 import time
 from dataclasses import dataclass
+from datetime import timedelta
 from typing import Optional
 
+import boa
 from boa.rpc import json
+from boa.util.abi import Address
+from boa.verifiers import ContractVerifier, VerificationResult, _wait_until
 
 try:
     from requests_cache import CachedSession
 
+    def filter_fn(response):
+        return response.ok and _is_success_response(response.json())
+
     SESSION = CachedSession(
         "~/.cache/titanoboa/explorer_cache",
-        filter_fn=lambda response: _is_success_response(response.json()),
+        filter_fn=filter_fn,
         allowable_codes=[200],
         cache_control=True,
         expire_after=3600 * 6,
@@ -25,12 +32,108 @@
 
 
 @dataclass
-class Etherscan:
+class Etherscan(ContractVerifier[str]):
     uri: Optional[str] = DEFAULT_ETHERSCAN_URI
     api_key: Optional[str] = None
     num_retries: int = 10
     backoff_ms: int | float = 400.0
     backoff_factor: float = 1.1  # 1.1**10 ~= 2.59
+    timeout = timedelta(minutes=2)
+
+    def verify(
+        self,
+        address: Address,
+        contract_name: str,
+        solc_json: dict,
+        constructor_calldata: bytes,
+        license_type: str = "1",
+        wait: bool = False,
+    ) -> Optional["VerificationResult[str]"]:
+        """
+        Verify the Vyper contract on Etherscan.
+        :param address: The address of the contract.
+        :param contract_name: The name of the contract.
+        :param solc_json: The solc_json output of the Vyper compiler.
+        :param constructor_calldata: The calldata for the contract constructor.
+        :param license_type: The license to use for the contract. Defaults to "none".
+        :param wait: Whether to return a VerificationResult immediately
+                     or wait for verification to complete. Defaults to False
+        """
+        api_key = self.api_key or ""
+        data = {
+            "module": "contract",
+            "action": "verifysourcecode",
+            "apikey": api_key,
+            "chainId": boa.env.get_chain_id(),
+            "codeformat": "vyper-json",
+            "sourceCode": json.dumps(solc_json),
+            "constructorArguments": constructor_calldata.hex(),
+            "contractaddress": address,
+            "contractname": contract_name,
+            "compilerversion": "v0.4.0",
+            # todo: "compilerversion": solc_json["compiler_version"],
+            "licenseType": license_type,
+            "optimizationUsed": "1",
+        }
+
+        def verification_created():
+            # we need to retry until the contract is found by Etherscan
+            response = SESSION.post(self.uri, data=data)
+            response.raise_for_status()
+            response_json = response.json()
+            if response_json.get("status") == "1":
+                return response_json["result"]
+            if (
+                response_json.get("message") == "NOTOK"
+                and "Unable to locate ContractCode" not in response_json["result"]
+            ):
+                raise ValueError(f"Failed to verify: {response_json['result']}")
+            print(
+                f'Verification could not be created yet: {response_json["result"]}. Retrying...'
+            )
+            return None
+
+        identifier = _wait_until(
+            verification_created, timedelta(seconds=30), timedelta(seconds=5), 1.1
+        )
+        print(f"Verification started with identifier {identifier}")
+        if not wait:
+            return VerificationResult(identifier, self)
+
+        self.wait_for_verification(identifier)
+        return None
+
+    def wait_for_verification(self, identifier: str) -> None:
+        """
+        Waits for the contract to be verified on Etherscan.
+        :param identifier: The identifier of the contract.
+        """
+        _wait_until(
+            lambda: self.is_verified(identifier),
+            self.timeout,
+            self.backoff,
+            self.backoff_factor,
+        )
+        print("Contract verified!")
+
+    @property
+    def backoff(self):
+        return timedelta(milliseconds=self.backoff_ms)
+
+    def is_verified(self, identifier: str) -> bool:
+        api_key = self.api_key or ""
+        url = f"{self.uri}?module=contract&action=checkverifystatus"
+        url += f"&guid={identifier}&apikey={api_key}"
+
+        response = SESSION.get(url)
+        response.raise_for_status()
+        response_json = response.json()
+        if (
+            response_json.get("message") == "NOTOK"
+            and "Pending in queue" not in response_json["result"]
+        ):
+            raise ValueError(f"Failed to verify: {response_json['result']}")
+        return response_json.get("status") == "1"
 
     def __post_init__(self):
         if self.uri is None:

boa/verifiers.py

@@ -3,18 +3,38 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from http import HTTPStatus
-from typing import Optional
+from typing import Callable, Generic, Optional, TypeVar
 
 import requests
 
 from boa.util.abi import Address
 from boa.util.open_ctx import Open
 
 DEFAULT_BLOCKSCOUT_URI = "https://eth.blockscout.com"
+T = TypeVar("T")
+
+
+class ContractVerifier(Generic[T]):
+    def verify(
+        self,
+        address: Address,
+        contract_name: str,
+        solc_json: dict,
+        constructor_calldata: bytes,
+        license_type: str = "1",
+        wait: bool = False,
+    ) -> Optional["VerificationResult[T]"]:
+        raise NotImplementedError
+
+    def wait_for_verification(self, identifier: T) -> None:
+        raise NotImplementedError
+
+    def is_verified(self, identifier: T) -> bool:
+        raise NotImplementedError
 
 
 @dataclass
-class Blockscout:
+class Blockscout(ContractVerifier[Address]):
     """
     Allows users to verify contracts on Blockscout.
     This is independent of Vyper contracts, and can be used to verify any smart contract.
@@ -37,14 +57,16 @@ def verify(
         address: Address,
         contract_name: str,
         solc_json: dict,
-        license_type: str = None,
+        constructor_calldata: bytes,
+        license_type: str = "1",
         wait: bool = False,
-    ) -> Optional["VerificationResult"]:
+    ) -> Optional["VerificationResult[Address]"]:
         """
         Verify the Vyper contract on Blockscout.
         :param address: The address of the contract.
         :param contract_name: The name of the contract.
         :param solc_json: The solc_json output of the Vyper compiler.
+        :param constructor_calldata: The calldata for the constructor.
         :param license_type: The license to use for the contract. Defaults to "none".
         :param wait: Whether to return a VerificationResult immediately
                      or wait for verification to complete. Defaults to False
@@ -83,18 +105,15 @@ def wait_for_verification(self, address: Address) -> None:
         Waits for the contract to be verified on Blockscout.
         :param address: The address of the contract.
         """
-        timeout = datetime.now() + self.timeout
-        wait_time = self.backoff
-        while datetime.now() < timeout:
-            if self.is_verified(address):
-                msg = "Contract verified!"
-                msg += f" {self.uri}/address/{address}?tab=contract_code"
-                print(msg)
-                return
-            time.sleep(wait_time.total_seconds())
-            wait_time *= self.backoff_factor
-
-        raise TimeoutError("Timeout waiting for verification to complete")
+        _wait_until(
+            lambda: self.is_verified(address),
+            self.timeout,
+            self.backoff,
+            self.backoff_factor,
+        )
+        msg = "Contract verified!"
+        msg += f" {self.uri}/address/{address}?tab=contract_code"
+        print(msg)
 
     def is_verified(self, address: Address) -> bool:
         api_key = self.api_key or ""
@@ -107,19 +126,19 @@ def is_verified(self, address: Address) -> bool:
         return response.json().get("is_verified", False)
 
 
-_verifier = Blockscout()
+_verifier: ContractVerifier = Blockscout()
 
 
 @dataclass
-class VerificationResult:
-    address: Address
-    verifier: Blockscout
+class VerificationResult(Generic[T]):
+    identifier: T
+    verifier: ContractVerifier
 
     def wait_for_verification(self):
-        self.verifier.wait_for_verification(self.address)
+        self.verifier.wait_for_verification(self.identifier)
 
     def is_verified(self):
-        return self.verifier.is_verified(self.address)
+        return self.verifier.is_verified(self.identifier)
 
 
 def _set_verifier(verifier):
@@ -133,7 +152,7 @@ def get_verifier():
 
 
 # TODO: maybe allow like `set_verifier("blockscout", *args, **kwargs)`
-def set_verifier(verifier):
+def set_verifier(verifier: ContractVerifier):
     return Open(get_verifier, _set_verifier, verifier)
 
 
@@ -147,14 +166,14 @@ def get_verification_bundle(contract_like):
 
 # should we also add a `verify_deployment` function?
 def verify(
-    contract, verifier=None, license_type: str = None, wait=False
-) -> VerificationResult:
+    contract, verifier: ContractVerifier = None, wait=False, **kwargs
+) -> VerificationResult | None:
     """
     Verifies the contract on a block explorer.
     :param contract: The contract to verify.
     :param verifier: The block explorer verifier to use.
                      Defaults to get_verifier().
-    :param license_type: Optional license to use for the contract.
+    :param wait: Whether to wait for verification to complete.
     """
     if verifier is None:
         verifier = get_verifier()
@@ -166,6 +185,24 @@ def verify(
         address=contract.address,
         solc_json=bundle,
         contract_name=contract.contract_name,
-        license_type=license_type,
+        constructor_calldata=contract.constructor_calldata,
         wait=wait,
+        **kwargs,
     )
+
+
+def _wait_until(
+    predicate: Callable[[], T],
+    wait_for: timedelta,
+    backoff: timedelta,
+    backoff_factor: float,
+) -> T:
+    timeout = datetime.now() + wait_for
+    wait_time = backoff
+    while datetime.now() < timeout:
+        if result := predicate():
+            return result
+        time.sleep(wait_time.total_seconds())
+        wait_time *= backoff_factor
+
+    raise TimeoutError("Timeout waiting for verification to complete")

tests/integration/network/sepolia/test_sepolia_env.py

@@ -3,6 +3,7 @@
 import pytest
 
 import boa
+from boa import Etherscan
 from boa.deployments import DeploymentsDB, set_deployments_db
 from boa.network import NetworkEnv
 from boa.rpc import to_bytes
@@ -38,13 +39,24 @@ def simple_contract():
     return boa.loads(code, STARTING_SUPPLY)
 
 
-def test_verify(simple_contract):
-    api_key = os.getenv("BLOCKSCOUT_API_KEY")
-    blockscout = Blockscout("https://eth-sepolia.blockscout.com", api_key)
-    with boa.set_verifier(blockscout):
-        result = boa.verify(simple_contract, blockscout)
-        result.wait_for_verification()
-        assert result.is_verified()
+@pytest.fixture(scope="module", params=[Etherscan, Blockscout])
+def verifier(request):
+    if request.param == Blockscout:
+        api_key = os.getenv("BLOCKSCOUT_API_KEY")
+        verifier = Blockscout("https://eth-sepolia.blockscout.com", api_key)
+    elif request.param == Etherscan:
+        api_key = os.environ["ETHERSCAN_API_KEY"]
+        verifier = Etherscan("https://api-sepolia.etherscan.io/api", api_key)
+    else:
+        raise ValueError(f"Unknown verifier: {request.param}")
+    with boa.set_verifier(verifier):
+        yield verifier
+
+
+def test_verify(simple_contract, verifier):
+    result = boa.verify(simple_contract)
+    result.wait_for_verification()
+    assert result.is_verified()
 
 
 def test_env_type():