crytic_compile.platform.etherscan

  1"""
  2Etherscan platform.
  3"""
  4
  5import json
  6import logging
  7import os
  8import re
  9import urllib.request
 10from json.decoder import JSONDecodeError
 11from pathlib import Path, PurePosixPath
 12from typing import TYPE_CHECKING, Dict, List, Union, Tuple, Optional
 13
 14from crytic_compile.compilation_unit import CompilationUnit
 15from crytic_compile.compiler.compiler import CompilerVersion
 16from crytic_compile.platform import solc_standard_json
 17from crytic_compile.platform.abstract_platform import AbstractPlatform
 18from crytic_compile.platform.exceptions import InvalidCompilation
 19from crytic_compile.platform.types import Type
 20from crytic_compile.utils.naming import Filename
 21
 22# Cycle dependency
 23
 24if TYPE_CHECKING:
 25    from crytic_compile import CryticCompile
 26
 27LOGGER = logging.getLogger("CryticCompile")
 28
 29
 30# Etherscan v1 API style (per-scanner URL)
 31ETHERSCAN_BASE_V1 = "https://api%s/api?module=contract&action=getsourcecode&address=%s"
 32
 33# Etherscan v2 API style (unified)
 34ETHERSCAN_BASE_V2 = (
 35    "https://api.etherscan.io/v2/api?chainid=%s&module=contract&action=getsourcecode&address=%s"
 36)
 37
 38# Bytecode URL style (for scraping)
 39ETHERSCAN_BASE_BYTECODE = "https://%s/address/%s#code"
 40
 41# v1 style scanners
 42SUPPORTED_NETWORK_V1: Dict[str, Tuple[str, str]] = {
 43    # None at this time. External tracer instances not operated by Etherscan would be here
 44}
 45
 46# v2 style scanners
 47SUPPORTED_NETWORK_V2: Dict[str, Tuple[str, str]] = {
 48    # Key, (chainid, perfix_bytecode)
 49    "mainnet": ("1", "etherscan.io"),
 50    "sepolia": ("11155111", "sepolia.etherscan.io"),
 51    "holesky": ("17000", "holesky.etherscan.io"),
 52    "bsc": ("56", "bscscan.com"),
 53    "testnet.bsc": ("97", "testnet.bscscan.com"),
 54    "poly": ("137", "polygonscan.com"),
 55    "amoy.poly": ("80002", "amoy.polygonscan.com"),
 56    "polyzk": ("1101", "zkevm.polygonscan.com"),
 57    "cardona.polyzk": ("2442", "cardona-zkevm.polygonscan.com"),
 58    "base": ("8453", "basescan.org"),
 59    "sepolia.base": ("84532", "sepolia.basescan.org"),
 60    "arbi": ("42161", "arbiscan.io"),
 61    "nova.arbi": ("42170", "nova.arbiscan.io"),
 62    "sepolia.arbi": ("421614", "sepolia.arbiscan.io"),
 63    "linea": ("59144", "lineascan.build"),
 64    "sepolia.linea": ("59141", "sepolia.lineascan.build"),
 65    "ftm": ("250", "ftmscan.com"),
 66    "testnet.ftm": ("4002", "testnet.ftmscan.com"),
 67    "blast": ("81457", "blastscan.io"),
 68    "sepolia.blast": ("168587773", "sepolia.blastscan.io"),
 69    "optim": ("10", "optimistic.etherscan.io"),
 70    "sepolia.optim": ("11155420", "sepolia-optimism.etherscan.io"),
 71    "avax": ("43114", "snowscan.xyz"),
 72    "testnet.avax": ("43113", "testnet.snowscan.xyz"),
 73    "bttc": ("199", "bttcscan.com"),
 74    "testnet.bttc": ("1028", "testnet.bttcscan.com"),
 75    "celo": ("42220", "celoscan.io"),
 76    "alfajores.celo": ("44787", "alfajores.celoscan.io"),
 77    "cronos": ("25", "cronoscan.com"),
 78    "frax": ("252", "fraxscan.com"),
 79    "holesky.frax": ("2522", "holesky.fraxscan.com"),
 80    "gno": ("100", "gnosisscan.io"),
 81    "kroma": ("255", "kromascan.com"),
 82    "sepolia.kroma": ("2358", "sepolia.kromascan.com"),
 83    "mantle": ("5000", "mantlescan.xyz"),
 84    "sepolia.mantle": ("5003", "sepolia.mantlescan.xyz"),
 85    "moonbeam": ("1284", "moonbeam.moonscan.io"),
 86    "moonriver": ("1285", "moonriver.moonscan.io"),
 87    "moonbase": ("1287", "moonbase.moonscan.io"),
 88    "opbnb": ("204", "opbnb.bscscan.com"),
 89    "testnet.opbnb": ("5611", "opbnb-testnet.bscscan.com"),
 90    "scroll": ("534352", "scrollscan.com"),
 91    "sepolia.scroll": ("534351", "sepolia.scrollscan.com"),
 92    "taiko": ("167000", "taikoscan.io"),
 93    "hekla.taiko": ("167009", "hekla.taikoscan.io"),
 94    "wemix": ("1111", "wemixscan.com"),
 95    "testnet.wemix": ("1112", "testnet.wemixscan.com"),
 96    "era.zksync": ("324", "era.zksync.network"),
 97    "sepoliaera.zksync": ("300", "sepolia-era.zksync.network"),
 98    "xai": ("660279", "xaiscan.io"),
 99    "sepolia.xai": ("37714555429", "sepolia.xaiscan.io"),
100    "xdc": ("50", "xdcscan.com"),
101    "testnet.xdc": ("51", "testnet.xdcscan.com"),
102    "apechain": ("33139", "apescan.io"),
103    "curtis.apechain": ("33111", "curtis.apescan.io"),
104    "world": ("480", "worldscan.org"),
105    "sepolia.world": ("4801", "sepolia.worldscan.org"),
106    "sophon": ("50104", "sophscan.xyz"),
107    "testnet.sophon": ("531050104", "testnet.sophscan.xyz"),
108    "sonic": ("146", "sonicscan.org"),
109    "testnet.sonic": ("57054", "testnet.sonicscan.org"),
110    "unichain": ("130", "uniscan.xyz"),
111    "sepolia.unichain": ("1301", "sepolia.uniscan.xyz"),
112    "abstract": ("2741", "abscan.org"),
113    "sepolia.abstract": ("11124", "sepolia.abscan.org"),
114    "berachain": ("80094", "berascan.com"),
115}
116
117SUPPORTED_NETWORK = {**SUPPORTED_NETWORK_V1, **SUPPORTED_NETWORK_V2}
118
119
120def generate_supported_network_v2_list() -> None:
121    """Manual function to generate a dictionary for updating the SUPPORTED_NETWORK_V2 array"""
122
123    with urllib.request.urlopen("https://api.etherscan.io/v2/chainlist") as response:
124        items = response.read()
125    networks = json.loads(items)
126
127    id2name = {}
128    for name, (chainid, _) in SUPPORTED_NETWORK_V2.items():
129        id2name[chainid] = name
130
131    results = {}
132    for network in networks["result"]:
133        name = id2name.get(network["chainid"], f"{network['chainid']}")
134        results[name] = (
135            network["chainid"],
136            network["blockexplorer"].replace("https://", "").strip("/"),
137        )
138
139    print(results)
140
141
142def _handle_bytecode(crytic_compile: "CryticCompile", target: str, result_b: bytes) -> None:
143    """Parse the bytecode and populate CryticCompile info
144
145    Args:
146        crytic_compile (CryticCompile): Associate CryticCompile object
147        target (str): path to the target
148        result_b (bytes): text containing the bytecode
149    """
150
151    # There is no direct API to get the bytecode from etherscan
152    # The page changes from time to time, we use for now a simple parsing, it will not be robust
153    begin = """Search Algorithm">\nSimilar Contracts</button>\n"""
154    begin += """<div id="dividcode">\n<pre class=\'wordwrap\' style=\'height: 15pc;\'>0x"""
155    result = result_b.decode("utf8")
156    # Removing everything before the begin string
157    result = result[result.find(begin) + len(begin) :]
158    bytecode = result[: result.find("<")]
159
160    contract_name = f"Contract_{target}"
161
162    contract_filename = Filename(absolute="", relative="", short="", used="")
163
164    compilation_unit = CompilationUnit(crytic_compile, str(target))
165
166    source_unit = compilation_unit.create_source_unit(contract_filename)
167
168    source_unit.add_contract_name(contract_name)
169    compilation_unit.filename_to_contracts[contract_filename].add(contract_name)
170    source_unit.abis[contract_name] = {}
171    source_unit.bytecodes_init[contract_name] = bytecode
172    source_unit.bytecodes_runtime[contract_name] = ""
173    source_unit.srcmaps_init[contract_name] = []
174    source_unit.srcmaps_runtime[contract_name] = []
175
176    compilation_unit.compiler_version = CompilerVersion(
177        compiler="unknown", version="", optimized=False
178    )
179
180    crytic_compile.bytecode_only = True
181
182
183def _handle_single_file(
184    source_code: str, addr: str, prefix: Optional[str], contract_name: str, export_dir: str
185) -> str:
186    """Handle a result with a single file
187
188    Args:
189        source_code (str): source code
190        addr (str): contract address
191        prefix (Optional[str]): used to separate different chains
192        contract_name (str): contract name
193        export_dir (str): directory where the code will be saved
194
195    Returns:
196        str: filename containing the source code
197    """
198    if prefix:
199        filename = os.path.join(export_dir, f"{addr}{prefix}-{contract_name}.sol")
200    else:
201        filename = os.path.join(export_dir, f"{addr}-{contract_name}.sol")
202
203    with open(filename, "w", encoding="utf8") as file_desc:
204        file_desc.write(source_code)
205
206    return filename
207
208
209def _handle_multiple_files(
210    dict_source_code: Dict, addr: str, prefix: Optional[str], contract_name: str, export_dir: str
211) -> Tuple[List[str], str, Optional[List[str]]]:
212    """Handle a result with a multiple files. Generate multiple Solidity files
213
214    Args:
215        dict_source_code (Dict): dict result from etherscan
216        addr (str): contract address
217        prefix (Optional[str]): used to separate different chains
218        contract_name (str): contract name
219        export_dir (str): directory where the code will be saved
220
221    Returns:
222        Tuple[List[str], str]: filesnames, directory, where target_filename is the main file
223
224    Raises:
225        IOError: if the path is outside of the allowed directory
226    """
227    if prefix:
228        directory = os.path.join(export_dir, f"{addr}{prefix}-{contract_name}")
229    else:
230        directory = os.path.join(export_dir, f"{addr}-{contract_name}")
231
232    if "sources" in dict_source_code:
233        # etherscan might return an object with a sources prop, which contains an object with contract names as keys
234        source_codes = dict_source_code["sources"]
235    else:
236        # or etherscan might return an object with contract names as keys
237        source_codes = dict_source_code
238
239    filtered_paths: List[str] = []
240    for filename, source_code in source_codes.items():
241        path_filename = PurePosixPath(filename)
242        # Only keep solidity files
243        if path_filename.suffix not in [".sol", ".vy"]:
244            continue
245
246        # https://etherscan.io/address/0x19bb64b80cbf61e61965b0e5c2560cc7364c6546#code has an import of erc721a/contracts/ERC721A.sol
247        # if the full path is lost then won't compile
248        if "contracts" == path_filename.parts[0] and not filename.startswith("@"):
249            path_filename = PurePosixPath(
250                *path_filename.parts[path_filename.parts.index("contracts") :]
251            )
252
253        # Convert "absolute" paths such as "/interfaces/IFoo.sol" into relative ones.
254        # This is needed due to the following behavior from pathlib.Path:
255        # > When several absolute paths are given, the last is taken as an anchor
256        # We need to make sure this is relative, so that Path(directory, ...) remains anchored to directory
257        if path_filename.is_absolute():
258            path_filename = PurePosixPath(*path_filename.parts[1:])
259
260        filtered_paths.append(path_filename.as_posix())
261        path_filename_disk = Path(directory, path_filename)
262
263        allowed_path = os.path.abspath(directory)
264        if os.path.commonpath((allowed_path, os.path.abspath(path_filename_disk))) != allowed_path:
265            raise IOError(
266                f"Path '{path_filename_disk}' is outside of the allowed directory: {allowed_path}"
267            )
268        if not os.path.exists(path_filename_disk.parent):
269            os.makedirs(path_filename_disk.parent)
270        with open(path_filename_disk, "w", encoding="utf8") as file_desc:
271            file_desc.write(source_code["content"])
272
273    remappings = dict_source_code.get("settings", {}).get("remappings", None)
274
275    return list(filtered_paths), directory, _sanitize_remappings(remappings, directory)
276
277
278class Etherscan(AbstractPlatform):
279    """
280    Etherscan platform
281    """
282
283    NAME = "Etherscan"
284    PROJECT_URL = "https://etherscan.io/"
285    TYPE = Type.ETHERSCAN
286
287    # pylint: disable=too-many-locals,too-many-branches,too-many-statements
288    def compile(self, crytic_compile: "CryticCompile", **kwargs: str) -> None:
289        """Run the compilation
290
291        Args:
292            crytic_compile (CryticCompile): Associated CryticCompile object
293            **kwargs: optional arguments. Used "solc", "etherscan_only_source_code", "etherscan_only_bytecode",
294                "etherscan_api_key", "export_dir"
295
296        Raises:
297            InvalidCompilation: if etherscan returned an error, or its results were not correctly parsed
298        """
299
300        target = self._target
301
302        api_key_required = None
303
304        if target.startswith(tuple(SUPPORTED_NETWORK_V2)):
305            api_key_required = 2
306            prefix, addr = target.split(":", 2)
307            chainid, prefix_bytecode = SUPPORTED_NETWORK_V2[prefix]
308            etherscan_url = ETHERSCAN_BASE_V2 % (chainid, addr)
309            etherscan_bytecode_url = ETHERSCAN_BASE_BYTECODE % (prefix_bytecode, addr)
310        elif target.startswith(tuple(SUPPORTED_NETWORK_V1)):
311            api_key_required = 1
312            prefix = SUPPORTED_NETWORK_V1[target[: target.find(":") + 1]][0]
313            prefix_bytecode = SUPPORTED_NETWORK_V1[target[: target.find(":") + 1]][1]
314            addr = target[target.find(":") + 1 :]
315            etherscan_url = ETHERSCAN_BASE_V1 % (prefix, addr)
316            etherscan_bytecode_url = ETHERSCAN_BASE_BYTECODE % (prefix_bytecode, addr)
317        else:
318            api_key_required = 2
319            etherscan_url = ETHERSCAN_BASE_V2 % ("1", target)
320            etherscan_bytecode_url = ETHERSCAN_BASE_BYTECODE % ("etherscan.io", target)
321            addr = target
322            prefix = None
323
324        only_source = kwargs.get("etherscan_only_source_code", False)
325        only_bytecode = kwargs.get("etherscan_only_bytecode", False)
326
327        etherscan_api_key = kwargs.get("etherscan_api_key", None)
328        if etherscan_api_key is None:
329            etherscan_api_key = os.getenv("ETHERSCAN_API_KEY")
330
331        export_dir = kwargs.get("export_dir", "crytic-export")
332        export_dir = os.path.join(
333            export_dir, kwargs.get("etherscan_export_dir", "etherscan-contracts")
334        )
335
336        if api_key_required == 2 and etherscan_api_key:
337            etherscan_url += f"&apikey={etherscan_api_key}"
338            etherscan_bytecode_url += f"&apikey={etherscan_api_key}"
339        # API key handling for external tracers would be here e.g.
340        # elif api_key_required == 1 and avax_api_key and "snowtrace" in etherscan_url:
341        #    etherscan_url += f"&apikey={avax_api_key}"
342        #    etherscan_bytecode_url += f"&apikey={avax_api_key}"
343
344        source_code: str = ""
345        result: Dict[str, Union[bool, str, int]] = {}
346        contract_name: str = ""
347
348        if not only_bytecode:
349            # build object with headers, then send request
350            new_etherscan_url = urllib.request.Request(
351                etherscan_url,
352                headers={
353                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.80 Safari/537.36 crytic-compile/0"
354                },
355            )
356            with urllib.request.urlopen(new_etherscan_url) as response:
357                html = response.read()
358
359            info = json.loads(html)
360
361            if (
362                "result" in info
363                and "rate limit reached" in info["result"]
364                and "message" in info
365                and info["message"] == "NOTOK"
366            ):
367                LOGGER.error("Etherscan API rate limit exceeded")
368                raise InvalidCompilation("Etherscan API rate limit exceeded")
369
370            if "message" not in info:
371                LOGGER.error("Incorrect etherscan request")
372                raise InvalidCompilation("Incorrect etherscan request " + etherscan_url)
373
374            if not info["message"].startswith("OK") and "Invalid API Key" in info["result"]:
375                LOGGER.error("Invalid etherscan API Key")
376                raise InvalidCompilation("Invalid etherscan API Key: " + etherscan_url)
377
378            if not info["message"].startswith("OK"):
379                LOGGER.error("Contract has no public source code")
380                raise InvalidCompilation("Contract has no public source code: " + etherscan_url)
381
382            if "result" not in info:
383                LOGGER.error("Contract has no public source code")
384                raise InvalidCompilation("Contract has no public source code: " + etherscan_url)
385
386            result = info["result"][0]
387            # Assert to help mypy
388            assert isinstance(result["SourceCode"], str)
389            assert isinstance(result["ContractName"], str)
390            source_code = result["SourceCode"]
391            contract_name = result["ContractName"]
392
393        if source_code == "" and not only_source:
394            LOGGER.info("Source code not available, try to fetch the bytecode only")
395
396            req = urllib.request.Request(
397                etherscan_bytecode_url, headers={"User-Agent": "Mozilla/5.0"}
398            )
399            with urllib.request.urlopen(req) as response:
400                html = response.read()
401
402            _handle_bytecode(crytic_compile, target, html)
403            return
404
405        if source_code == "":
406            LOGGER.error("Contract has no public source code")
407            raise InvalidCompilation("Contract has no public source code: " + etherscan_url)
408
409        if not os.path.exists(export_dir):
410            os.makedirs(export_dir)
411
412        # Assert to help mypy
413        assert isinstance(result["CompilerVersion"], str)
414
415        compiler_version = re.findall(
416            r"\d+\.\d+\.\d+", _convert_version(result["CompilerVersion"])
417        )[0]
418
419        # etherscan can report "default" which is not a valid EVM version
420        evm_version: Optional[str] = None
421        if "EVMVersion" in result:
422            assert isinstance(result["EVMVersion"], str)
423            evm_version = result["EVMVersion"] if result["EVMVersion"] != "Default" else None
424
425        optimization_used: bool = result["OptimizationUsed"] == "1"
426
427        optimize_runs = None
428        if optimization_used:
429            optimize_runs = int(result["Runs"])
430
431        working_dir: Optional[str] = None
432        remappings: Optional[List[str]] = None
433
434        dict_source_code: Optional[Dict] = None
435        try:
436            # etherscan might return an object with two curly braces, {{ content }}
437            dict_source_code = json.loads(source_code[1:-1])
438            assert isinstance(dict_source_code, dict)
439            filenames, working_dir, remappings = _handle_multiple_files(
440                dict_source_code, addr, prefix, contract_name, export_dir
441            )
442        except JSONDecodeError:
443            try:
444                # or etherscan might return an object with single curly braces, { content }
445                dict_source_code = json.loads(source_code)
446                assert isinstance(dict_source_code, dict)
447                filenames, working_dir, remappings = _handle_multiple_files(
448                    dict_source_code, addr, prefix, contract_name, export_dir
449                )
450            except JSONDecodeError:
451                filenames = [
452                    _handle_single_file(source_code, addr, prefix, contract_name, export_dir)
453                ]
454
455        # viaIR is not exposed on the top level JSON offered by etherscan, so we need to inspect the settings
456        via_ir_enabled: Optional[bool] = None
457        if isinstance(dict_source_code, dict):
458            via_ir_enabled = dict_source_code.get("settings", {}).get("viaIR", None)
459
460        compilation_unit = CompilationUnit(crytic_compile, contract_name)
461
462        compilation_unit.compiler_version = CompilerVersion(
463            compiler=kwargs.get("solc", "solc"),
464            version=compiler_version,
465            optimized=optimization_used,
466            optimize_runs=optimize_runs,
467        )
468        compilation_unit.compiler_version.look_for_installed_version()
469
470        if "Proxy" in result and result["Proxy"] == "1":
471            assert "Implementation" in result
472            implementation = str(result["Implementation"])
473            if target.startswith(tuple(SUPPORTED_NETWORK)):
474                implementation = f"{target[:target.find(':')]}:{implementation}"
475            compilation_unit.implementation_address = implementation
476
477        solc_standard_json.standalone_compile(
478            filenames,
479            compilation_unit,
480            working_dir=working_dir,
481            remappings=remappings,
482            evm_version=evm_version,
483            via_ir=via_ir_enabled,
484        )
485
486        metadata_config = {
487            "solc_remaps": remappings if remappings else {},
488            "solc_solcs_select": compiler_version,
489            "solc_args": " ".join(
490                filter(
491                    None,
492                    [
493                        "--via-ir" if via_ir_enabled else "",
494                        "--optimize --optimize-runs " + str(optimize_runs) if optimize_runs else "",
495                        "--evm-version " + evm_version if evm_version else "",
496                    ],
497                )
498            ),
499        }
500
501        with open(
502            os.path.join(working_dir if working_dir else export_dir, "crytic_compile.config.json"),
503            "w",
504            encoding="utf-8",
505        ) as f:
506            json.dump(metadata_config, f)
507
508    def clean(self, **_kwargs: str) -> None:
509        pass
510
511    @staticmethod
512    def is_supported(target: str, **kwargs: str) -> bool:
513        """Check if the target is a etherscan project
514
515        Args:
516            target (str): path to the target
517            **kwargs: optional arguments. Used "etherscan_ignore"
518
519        Returns:
520            bool: True if the target is a etherscan project
521        """
522        etherscan_ignore = kwargs.get("etherscan_ignore", False)
523        if etherscan_ignore:
524            return False
525        if target.startswith(tuple(SUPPORTED_NETWORK)):
526            target = target[target.find(":") + 1 :]
527        return bool(re.match(r"^\s*0x[a-zA-Z0-9]{40}\s*$", target))
528
529    def is_dependency(self, _path: str) -> bool:
530        """Check if the path is a dependency
531
532        Args:
533            _path (str): path to the target
534
535        Returns:
536            bool: True if the target is a dependency
537        """
538        return False
539
540    def _guessed_tests(self) -> List[str]:
541        """Guess the potential unit tests commands
542
543        Returns:
544            List[str]: The guessed unit tests commands
545        """
546        return []
547
548
549def _convert_version(version: str) -> str:
550    """Convert the compiler version
551
552    Args:
553        version (str): original version
554
555    Returns:
556        str: converted version
557    """
558    if "+" in version:
559        return version[1 : version.find("+")]
560    return version[1:]
561
562
563def _sanitize_remappings(
564    remappings: Optional[List[str]], allowed_directory: str
565) -> Optional[List[str]]:
566    """Sanitize a list of remappings
567
568    Args:
569        remappings: (Optional[List[str]]): a list of remappings
570        allowed_directory: the allowed base directory for remaps
571
572    Returns:
573        Optional[List[str]]: a list of sanitized remappings
574    """
575
576    if remappings is None:
577        return remappings
578
579    allowed_path = os.path.abspath(allowed_directory)
580
581    remappings_clean: List[str] = []
582    for r in remappings:
583        split = r.split("=", 2)
584        if len(split) != 2:
585            LOGGER.warning("Invalid remapping %s", r)
586            continue
587
588        origin, dest = split[0], PurePosixPath(split[1])
589
590        # if path is absolute, relativize it
591        if dest.is_absolute():
592            dest = PurePosixPath(*dest.parts[1:])
593
594        dest_disk = Path(allowed_directory, dest)
595
596        if os.path.commonpath((allowed_path, os.path.abspath(dest_disk))) != allowed_path:
597            LOGGER.warning("Remapping %s=%s is potentially unsafe, skipping", origin, dest)
598            continue
599
600        # always use a trailing slash for the destination
601        remappings_clean.append(f"{origin}={str(dest / '_')[:-1]}")
602
603    return remappings_clean