Merge pull request #21 from DanielSchiavini/feat/contract-verification

feat: implement zksync contract verification

.github/workflows/test.yaml

@@ -40,3 +40,5 @@ jobs:
             rm era_test_node.tar.gz
 
       - run: make coverage lint
+        env:
+          SEPOLIA_PKEY: ${{ secrets.SEPOLIA_PKEY }}

boa_zksync/__init__.py

@@ -1,10 +1,15 @@
 import boa
+from boa import get_verifier
+from boa.verifiers import VerificationResult
 
+from boa_zksync.contract import ZksyncContract
 from boa_zksync.environment import ZksyncEnv
 from boa_zksync.node import EraTestNode
+from boa_zksync.verifiers import ZksyncExplorer
 
 
-def set_zksync_env(url, nickname=None):
+def set_zksync_env(url, explorer_url=None, nickname=None):
+    boa.set_verifier(ZksyncExplorer(explorer_url))
     return boa.set_env(ZksyncEnv.from_url(url, nickname=nickname))
 
 
@@ -31,3 +36,14 @@ def set_zksync_browser_env(*args, **kwargs):
 boa.set_zksync_test_env = set_zksync_test_env
 boa.set_zksync_fork = set_zksync_fork
 boa.set_zksync_browser_env = set_zksync_browser_env
+
+
+def verify(contract: ZksyncContract, verifier=None, **kwargs) -> VerificationResult:
+    verifier = verifier or get_verifier()
+    return verifier.verify(
+        address=contract.address,
+        solc_json=contract.deployer.solc_json,
+        contract_name=contract.contract_name,
+        constructor_calldata=contract.constructor_calldata,
+        **kwargs,
+    )

boa_zksync/compile.py

@@ -1,4 +1,5 @@
 import json
+import re
 import subprocess
 from os import path
 from pathlib import Path
@@ -14,40 +15,46 @@
 def compile_zksync(
     contract_name: str, filename: str, compiler_args=None, source_code=None
 ) -> ZksyncCompilerData:
-    vyper_path = which("vyper")
+    vyper_path = which("vyper")  # make sure zkvyper uses the same vyper as boa
     assert vyper_path, "Vyper executable not found"
     compiler_args = compiler_args or []
-    compile_result = subprocess.run(
-        [
-            "zkvyper",
-            # make sure zkvyper uses the same vyper as boa
-            "--vyper",
-            vyper_path,
-            # request JSON output
-            "-f",
-            "combined_json",
-            # pass any extra compiler args
-            *compiler_args,
-            # pass the file name
-            "--",
-            filename,
-        ],
-        capture_output=True,
+
+    result = _run_zkvyper(
+        "--vyper", vyper_path, "-f", "combined_json", *compiler_args, "--", filename
     )
+    output = json.loads(result)
 
-    assert compile_result.returncode == 0, compile_result.stderr.decode()
-    output = json.loads(compile_result.stdout.decode())
     if source_code is None:
         with open(filename) as file:
             source_code = file.read()
 
     compile_output = get_compiler_output(output)
     bytecode = to_bytes(compile_output.pop("bytecode"))
     return ZksyncCompilerData(
-        contract_name, source_code, compiler_args, bytecode, **compile_output
+        contract_name,
+        source_code,
+        _get_zkvyper_version(),
+        compiler_args,
+        bytecode,
+        **compile_output,
     )
 
 
+def _get_zkvyper_version():
+    output = _run_zkvyper("--version")
+    match = re.search(r"\b(v\d+\.\d+\.\d+\S*)", output)
+    if not match:
+        raise ValueError(f"Could not parse zkvyper version from {output}")
+    return match.group(0)
+
+
+def _run_zkvyper(*args):
+    compile_result = subprocess.run(["zkvyper", *args], capture_output=True)
+    assert compile_result.returncode == 0, compile_result.stderr.decode()
+    output_str = compile_result.stdout.decode()
+    return output_str
+
+
 def compile_zksync_source(
     source_code: str, name: str, compiler_args=None
 ) -> ZksyncCompilerData:

boa_zksync/contract.py

@@ -22,6 +22,7 @@
 
 if TYPE_CHECKING:
     from boa_zksync import ZksyncEnv
+    from boa_zksync.deployer import ZksyncDeployer
 
 
 class ZksyncContract(ABIContract):
@@ -78,14 +79,16 @@ def __init__(
         self.env.register_contract(address, self)
 
     def _run_init(self, *args, value=0, override_address=None, gas=None):
-        constructor_calldata = self._ctor.prepare_calldata(*args) if self._ctor else b""
+        self.constructor_calldata = (
+            self._ctor.prepare_calldata(*args) if self._ctor else b""
+        )
         address, bytecode = self.env.deploy_code(
             override_address=override_address,
             gas=gas,
             contract=self,
             bytecode=self.compiler_data.bytecode,
             value=value,
-            constructor_calldata=constructor_calldata,
+            constructor_calldata=self.constructor_calldata,
         )
         self.bytecode = bytecode
         return address
@@ -110,7 +113,7 @@ def override_vyper_namespace(self):
             yield
 
     @cached_property
-    def deployer(self):
+    def deployer(self) -> "ZksyncDeployer":
         from boa_zksync.deployer import ZksyncDeployer
 
         return ZksyncDeployer(

boa_zksync/deployer.py

@@ -87,9 +87,12 @@ def env(self) -> "ZksyncEnv":
         return env
 
     @cached_property
-    def solc_json(self):
+    def solc_json(self) -> dict:
         """
         A ZKsync compatible solc-json. Generates a solc "standard json" representation
         of the Vyper contract.
         """
-        return build_solc_json(self.zkvyper_data.vyper)
+        return {
+            "zkvyper_version": self.zkvyper_data.zkvyper_version,
+            **build_solc_json(self.zkvyper_data.vyper),
+        }

boa_zksync/types.py

@@ -15,6 +15,7 @@
 from eth_account import Account
 from eth_account.datastructures import SignedMessage
 from eth_account.messages import encode_typed_data
+from packaging.version import Version
 from rlp.sedes import BigEndianInt, Binary, List
 from vyper.compiler import CompilerData
 from vyper.compiler.settings import OptimizationLevel
@@ -233,6 +234,7 @@ class ZksyncCompilerData:
 
     contract_name: str
     source_code: str
+    zkvyper_version: Version
     compiler_args: list[str]
     bytecode: bytes
     method_identifiers: dict

boa_zksync/verifiers.py

@@ -0,0 +1,116 @@
+import re
+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.verifiers import VerificationResult
+
+DEFAULT_ZKSYNC_EXPLORER_URI = "https://zksync2-mainnet-explorer.zksync.io"
+
+
+@dataclass
+class ZksyncExplorer:
+    """
+    Allows users to verify contracts on the zksync explorer at https://explorer.zksync.io/
+    This is independent of Vyper contracts, and can be used to verify any smart contract.
+    """
+
+    uri: str = DEFAULT_ZKSYNC_EXPLORER_URI
+    api_key: Optional[str] = None  # todo: use or remove
+    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,
+        solc_json: dict,
+        constructor_calldata: bytes = b"",
+        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 solc_json: The solc_json output of the Vyper compiler.
+        :param constructor_calldata: The calldata for the constructor.
+        :param wait: Whether to return a VerificationResult immediately
+                     or wait for verification to complete. Defaults to False
+        """
+        url = f"{self.uri}/contract_verification"
+
+        body = {
+            "contractAddress": address,
+            "sourceCode": {
+                contract_name if name == "<unknown>" else name: asset["content"]
+                for name, asset in solc_json["sources"].items()
+            },
+            "codeFormat": "vyper-multi-file",
+            "contractName": contract_name,
+            "compilerVyperVersion": self._extract_version(
+                solc_json["compiler_version"]
+            ),
+            "compilerZkvyperVersion": solc_json["zkvyper_version"],
+            "constructorArguments": f"0x{constructor_calldata.hex()}",
+            "optimizationUsed": True,
+            # hardcoded in hardhat for some reason: https://github.com/matter-labs/hardhat-zksync/blob/187722e/packages/hardhat-zksync-verify-vyper/src/task-actions.ts#L110  # noqa: E501
+        }
+
+        response = requests.post(url, json=body)
+        response.raise_for_status()
+        verification_id = response.text
+        int(verification_id)  # raises ValueError if not an int
+
+        if not wait:
+            return VerificationResult(verification_id, self)  # type: ignore
+
+        self.wait_for_verification(verification_id)
+        return None
+
+    @staticmethod
+    def _extract_version(version: str):
+        # we only pass the first three digits of the version, as that's what the explorer expects
+        match = re.search(r"(\d+\.\d+\.\d+)", version)
+        assert match is not None, f"Could not extract version from {version}"
+        return match.group(0)
+
+    def wait_for_verification(self, verification_id: str) -> None:
+        """
+        Waits for the contract to be verified on Zksync Explorer.
+        :param verification_id: The ID of the contract verification.
+        """
+        timeout = datetime.now() + self.timeout
+        wait_time = self.backoff
+        while datetime.now() < timeout:
+            if self.is_verified(verification_id):
+                print("Contract verified!")
+                return
+            time.sleep(wait_time.total_seconds())
+            wait_time *= self.backoff_factor
+
+        raise TimeoutError("Timeout waiting for verification to complete")
+
+    def is_verified(self, verification_id: str) -> bool:
+        url = f"{self.uri}/contract_verification/{verification_id}"
+
+        response = requests.get(url)
+        if response.status_code in self.retry_http_codes:
+            return False
+        response.raise_for_status()
+
+        # known statuses: successful, failed, queued, in_progress
+        json = response.json()
+        if json["status"] == "failed":
+            raise ValueError(f"Verification failed: {json['error']}")
+        return json["status"] == "successful"

tests/conftest.py

@@ -10,6 +10,12 @@
 from boa_zksync.deployer import ZksyncDeployer
 
 STARTING_SUPPLY = 100
+ZKSYNC_SEPOLIA_RPC_URL = os.getenv(
+    "ZKSYNC_SEPOLIA_RPC_URL", "https://sepolia.era.zksync.dev"
+)
+ZKSYNC_SEPOLIA_EXPLORER_URL = os.getenv(
+    "ZKSYNC_SEPOLIA_EXPLORER_URL", "https://explorer.sepolia.era.zksync.dev"
+)
 
 
 @pytest.fixture(scope="module")
@@ -24,9 +30,8 @@ def zksync_env(account):
 @pytest.fixture(scope="module")
 def zksync_sepolia_fork(account):
     old_env = boa.env
-    fork_url = os.getenv("FORK_URL", "https://sepolia.era.zksync.dev")
     boa_zksync.set_zksync_fork(
-        fork_url,
+        ZKSYNC_SEPOLIA_RPC_URL,
         block_identifier=3000000,
         node_args=("--show-calls", "all", "--show-outputs", "true"),
     )
@@ -35,6 +40,21 @@ def zksync_sepolia_fork(account):
     boa.set_env(old_env)
 
 
+@pytest.fixture(scope="module")
+def zksync_sepolia_env():
+    key = os.getenv("SEPOLIA_PKEY")
+    if not key:
+        return pytest.skip("SEPOLIA_PKEY is not set, skipping test")
+
+    old_env = boa.env
+    boa_zksync.set_zksync_env(ZKSYNC_SEPOLIA_RPC_URL, ZKSYNC_SEPOLIA_EXPLORER_URL)
+    try:
+        boa.env.add_account(Account.from_key(key))
+        yield
+    finally:
+        boa.set_env(old_env)
+
+
 @pytest.fixture(scope="module")
 def account():
     # default rich account from era_test_node

tests/test_sepolia.py

@@ -2,6 +2,7 @@
 import pytest
 from boa.rpc import EthereumRPC
 
+import boa_zksync
 from boa_zksync import EraTestNode
 from boa_zksync.environment import ZERO_ADDRESS
 
@@ -38,3 +39,13 @@ def set_implementation(_implementation: address):
 def test_fork_rpc(zksync_sepolia_fork):
     assert isinstance(boa.env._rpc, EraTestNode)
     assert isinstance(boa.env._rpc.inner_rpc, EthereumRPC)
+
+
+@pytest.mark.ignore_isolation
+def test_real_deploy_and_verify(zksync_sepolia_env):
+    from tests.data import Counter
+
+    contract = Counter.deploy()
+    verify = boa_zksync.verify(contract)
+    verify.wait_for_verification()
+    assert verify.is_verified()