feat: Blockscout contract verification (#308)

implement Blockscout contract verification

main entry points are `boa.verify` and `boa.set_verifier(Blockscout(...))`

---------

Co-authored-by: Charles Cooper <[email protected]>

boa/__init__.py

@@ -21,6 +21,7 @@
 from boa.precompile import precompile
 from boa.test.strategies import fuzz
 from boa.util.open_ctx import Open
+from boa.verifiers import get_verifier, set_verifier, verify
 from boa.vm.py_evm import enable_pyevm_verbose_logging, patch_opcode
 
 # turn off tracebacks if we are in repl

boa/contracts/vyper/vyper_contract.py

@@ -26,7 +26,7 @@
 from vyper.codegen.module import generate_ir_for_module
 from vyper.compiler import CompilerData
 from vyper.compiler import output as compiler_output
-from vyper.compiler.output import build_abi_output
+from vyper.compiler.output import build_abi_output, build_solc_json
 from vyper.compiler.settings import OptimizationLevel, anchor_settings
 from vyper.exceptions import VyperException
 from vyper.ir.optimizer import optimize
@@ -123,9 +123,15 @@ def at(self, address: Any) -> "VyperContract":
         ret._set_bytecode(bytecode)
 
         ret.env.register_contract(address, ret)
-
         return ret
 
+    @cached_property
+    def standard_json(self):
+        """
+        Generates a standard JSON representation of the Vyper contract.
+        """
+        return build_solc_json(self.compiler_data)
+
     @cached_property
     def _constants(self):
         # Make constants available at compile time. Useful for testing. See #196
@@ -155,6 +161,10 @@ def __init__(
                 msg += f"{capabilities.describe_capabilities()}"
                 raise Exception(msg)
 
+    @cached_property
+    def deployer(self):
+        return VyperDeployer(self.compiler_data, filename=self.filename)
+
     @cached_property
     def abi(self):
         return build_abi_output(self.compiler_data)
@@ -204,10 +214,6 @@ def __init__(
 
         self.env.register_blueprint(compiler_data.bytecode, self)
 
-    @cached_property
-    def deployer(self):
-        return VyperDeployer(self.compiler_data, filename=self.filename)
-
 
 class FrameDetail(dict):
     def __init__(self, fn_name, *args, **kwargs):
@@ -631,11 +637,6 @@ def __repr__(self):
     def _immutables(self):
         return ImmutablesModel(self)
 
-    @cached_property
-    def deployer(self):
-        # TODO add test
-        return VyperDeployer(self.compiler_data, filename=self.filename)
-
     # is this actually useful?
     def at(self, address):
         return self.deployer.at(address)

boa/interpret.py

@@ -131,12 +131,13 @@ def compiler_data(
 ) -> CompilerData:
     global _disk_cache, _search_path
 
+    path = Path(contract_name)
+    resolved_path = Path(filename).resolve(strict=False)
+
     file_input = FileInput(
-        contents=source_code,
-        source_id=-1,
-        path=Path(contract_name),
-        resolved_path=Path(filename),
+        contents=source_code, source_id=-1, path=path, resolved_path=resolved_path
     )
+
     search_paths = get_search_paths(_search_path)
     input_bundle = FilesystemInputBundle(search_paths)
 
@@ -208,14 +209,16 @@ def loads_partial(
     dedent: bool = True,
     compiler_args: dict = None,
 ) -> VyperDeployer:
-    name = name or "VyperContract"  # TODO handle this upstream in CompilerData
+    name = name or "VyperContract"
     filename = filename or "<unknown>"
+
     if dedent:
         source_code = textwrap.dedent(source_code)
 
     version = _detect_version(source_code)
     if version is not None and version != vyper.__version__:
         filename = str(filename)  # help mypy
+        # TODO: pass name to loads_partial_vvm, not filename
         return _loads_partial_vvm(source_code, version, filename)
 
     compiler_args = compiler_args or {}

boa/verifiers.py

@@ -0,0 +1,162 @@
+import json
+import time
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from http import HTTPStatus
+from typing import Optional
+
+import requests
+
+from boa.util.abi import Address
+from boa.util.open_ctx import Open
+
+DEFAULT_BLOCKSCOUT_URI = "https://eth.blockscout.com"
+
+
+@dataclass
+class Blockscout:
+    """
+    Allows users to verify contracts on Blockscout.
+    This is independent of Vyper contracts, and can be used to verify any smart contract.
+    """
+
+    uri: str = DEFAULT_BLOCKSCOUT_URI
+    api_key: Optional[str] = None
+    timeout: timedelta = timedelta(minutes=2)
+    backoff: timedelta = timedelta(milliseconds=500)
+    backoff_factor: float = 1.1
+    retry_http_codes: tuple[int, ...] = (
+        HTTPStatus.NOT_FOUND,
+        HTTPStatus.INTERNAL_SERVER_ERROR,
+        HTTPStatus.SERVICE_UNAVAILABLE,
+        HTTPStatus.GATEWAY_TIMEOUT,
+    )
+
+    def verify(
+        self,
+        address: Address,
+        contract_name: str,
+        standard_json: dict,
+        license_type: str = None,
+        wait: bool = False,
+    ) -> Optional["VerificationResult"]:
+        """
+        Verify the Vyper contract on Blockscout.
+        :param address: The address of the contract.
+        :param contract_name: The name of the contract.
+        :param standard_json: The standard JSON output of the Vyper compiler.
+        :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
+        """
+        if license_type is None:
+            license_type = "none"
+
+        api_key = self.api_key or ""
+
+        url = f"{self.uri}/api/v2/smart-contracts/{address}/"
+        url += f"verification/via/vyper-standard-input?apikey={api_key}"
+        data = {
+            "compiler_version": standard_json["compiler_version"],
+            "license_type": license_type,
+        }
+        files = {
+            "files[0]": (
+                contract_name,
+                json.dumps(standard_json).encode("utf-8"),
+                "application/json",
+            )
+        }
+
+        response = requests.post(url, data=data, files=files)
+        response.raise_for_status()
+        print(response.json().get("message"))  # usually verification started
+
+        if not wait:
+            return VerificationResult(address, self)
+
+        self.wait_for_verification(address)
+        return None
+
+    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")
+
+    def is_verified(self, address: Address) -> bool:
+        api_key = self.api_key or ""
+        url = f"{self.uri}/api/v2/smart-contracts/{address}?apikey={api_key}"
+
+        response = requests.get(url)
+        if response.status_code in self.retry_http_codes:
+            return False
+        response.raise_for_status()
+        return True
+
+
+_verifier = Blockscout()
+
+
+@dataclass
+class VerificationResult:
+    address: Address
+    verifier: Blockscout
+
+    def wait_for_verification(self):
+        self.verifier.wait_for_verification(self.address)
+
+    def is_verified(self):
+        return self.verifier.is_verified(self.address)
+
+
+def _set_verifier(verifier):
+    global _verifier
+    _verifier = verifier
+
+
+def get_verifier():
+    global _verifier
+    return _verifier
+
+
+# TODO: maybe allow like `set_verifier("blockscout", *args, **kwargs)`
+def set_verifier(verifier):
+    return Open(get_verifier, _set_verifier, verifier)
+
+
+def verify(contract, verifier=None, license_type: str = None) -> VerificationResult:
+    """
+    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.
+    """
+    if verifier is None:
+        verifier = get_verifier()
+
+    if not hasattr(contract, "deployer") or not hasattr(
+        contract.deployer, "standard_json"
+    ):
+        raise ValueError(f"Not a contract! {contract}")
+
+    address = contract.address
+    return verifier.verify(
+        address=address,
+        standard_json=contract.deployer.standard_json,
+        contract_name=contract.contract_name,
+        license_type=license_type,
+    )

tests/integration/network/sepolia/test_sepolia_env.py

@@ -1,7 +1,10 @@
+import os
+
 import pytest
 
 import boa
 from boa.network import NetworkEnv
+from boa.verifiers import Blockscout
 
 # boa.env.anchor() does not work in prod environment
 pytestmark = pytest.mark.ignore_isolation
@@ -32,6 +35,15 @@ 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()
+
+
 def test_env_type():
     # sanity check
     assert isinstance(boa.env, NetworkEnv)