diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 956cc1f0..15034023 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,7 +28,13 @@ jobs: uses: actions/checkout@v7 with: fetch-depth: 0 - submodules: "recursive" + + - name: Checkout test registries + timeout-minutes: 10 + run: >- + git submodule update --init --checkout + tests/registries/clear-signing-erc7730-registry + tests/registries/ledger-asset-dapps - name: Setup mise timeout-minutes: 10 diff --git a/.gitmodules b/.gitmodules index 448e07af..d6798df9 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,6 +1,8 @@ [submodule "clear-signing-erc7730-registry"] path = tests/registries/clear-signing-erc7730-registry url = https://github.com/LedgerHQ/clear-signing-erc7730-registry + update = none [submodule "ledger-asset-dapps"] path = tests/registries/ledger-asset-dapps url = https://github.com/LedgerHQ/ledger-asset-dapps + update = none diff --git a/docs/pages/developer.md b/docs/pages/developer.md index 573fc60c..0679a0a3 100644 --- a/docs/pages/developer.md +++ b/docs/pages/developer.md @@ -19,10 +19,14 @@ brew install mise 1. **Clone the repository:** ```bash -git clone --recursive git@github.com:LedgerHQ/python-erc7730.git +git clone git@github.com:LedgerHQ/python-erc7730.git cd python-erc7730 +git submodule update --init --checkout tests/registries/clear-signing-erc7730-registry tests/registries/ledger-asset-dapps ``` +The test registries are git submodules that are skipped by a recursive submodule update (so that `pip` can install the +library from git without an SSH key for a nested submodule), hence the explicit `--checkout`. + 2. **Install tools via mise:** ```bash diff --git a/docs/pages/lint.md b/docs/pages/lint.md index 39708fe2..fa28f7d3 100644 --- a/docs/pages/lint.md +++ b/docs/pages/lint.md @@ -1,14 +1,31 @@ # Linter checks list ## ABI checks +The four "not verified" / "not supported" / "could not fetch" checks below are reported as errors instead when `--require-verified` is passed, so that a strict run never passes without having checked every deployment. + +### Contract not verified +- **Level**: ⚠️ Warning +- **Message**: `contract
on chain is not verified on Sourcify, descriptor ABIs will not be validated` +- **Description**: The contract is not verified on Sourcify, so there is no reference ABI. Subsequent checks are skipped for the current deployment. + +### Proxy implementation not verified +- **Level**: ⚠️ Warning +- **Message**: `contract
on chain is a proxy, and its implementation
is not verified on Sourcify, descriptor ABIs will not be validated` +- **Description**: Sourcify resolved the contract as a proxy, but one of its implementations is not verified, so the reference ABI would be incomplete. Subsequent checks are skipped for the current deployment. + +### Chain not supported +- **Level**: ℹ️ Info +- **Message**: `chain is not supported by Sourcify, descriptor ABIs will not be validated` +- **Description**: Sourcify does not support the chain of the deployment, so no reference ABI can be fetched. Subsequent checks are skipped for the current deployment. + ### Could not fetch ABI - **Level**: ⚠️ Warning - **Message**: `Fetching reference ABI for chain id failed, descriptor ABIs will not be validated: ` -- **Description**: ABI fetch from external sources (Sourcify, then Etherscan as a fallback) has failed. Subsequents checks are skipped for the current deployment. +- **Description**: ABI fetch from Sourcify has failed for another reason, such as a rate limit, a network error or a proxy resolution error. Subsequent checks are skipped for the current deployment. -### Proxy Contract +### Deployment ABIs differ - **Level**: ⚠️ Warning -- **Message**: `Contract is likely to be a proxy, validation of descriptor ABIs skipped` -- **Description**: Contract detected as a potential proxy contract based on a simple heuristic. Subsequents checks are skipped. +- **Message**: `The reference ABIs of the deployments do not all expose the same functions, display fields are validated against each distinct reference ABI: :
, ...; :
, ...` +- **Description**: The deployments of the descriptor do not have the same reference ABI. Display fields are validated once per distinct ABI, so findings may apply to some chains only (the contract URL in each finding tells which). ### Extra function - **Level**: ⚠️ Warning diff --git a/docs/pages/usage_cli.md b/docs/pages/usage_cli.md index adadf384..86f29dcb 100644 --- a/docs/pages/usage_cli.md +++ b/docs/pages/usage_cli.md @@ -67,9 +67,9 @@ excluded, please add it to `excluded` list to avoid this warning. ➡️ checking tether/calldata-usdt.json… 🟠 warning: Function mismatch: Function approve(address,uint256) (selector: 0x095ea7b3) defined in descriptor ABIs does not match -reference ABI (see https://etherscan.io/address/0xdac17f958d2ee523a2206206994597c13d831ec7#code) +reference ABI (see https://repo.sourcify.dev/1/0xdac17f958d2ee523a2206206994597c13d831ec7) 🟠 warning: Function mismatch: Function transfer(address,uint256) (selector: 0xa9059cbb) defined in descriptor ABIs does not match -reference ABI (see https://etherscan.io/address/0xdac17f958d2ee523a2206206994597c13d831ec7#code) +reference ABI (see https://repo.sourcify.dev/1/0xdac17f958d2ee523a2206206994597c13d831ec7) 🔴 error: Invalid data path: "0xdAC17F958D2ee523a2206206994597C13D831ec7" is invalid, it must contain a data path to the address in the transaction data. It seems you are trying to use a constant address value instead, please note this feature is not supported (yet). @@ -78,13 +78,15 @@ checked 61 descriptor files, some errors found ❌ It can be called with single files or directories, in which case all descriptors will be checked. -Use `--skip-abi-validation` to disable external ABI comparisons against Sourcify/Etherscan (useful for offline runs or faster local checks). +Use `--skip-abi-validation` to disable external ABI comparisons against Sourcify (useful for offline runs or faster local checks). + +Use `--require-verified` to report contracts (or proxy implementations) that are not verified on Sourcify as errors instead of warnings. With the flag, a reference ABI that could not be fetched (rate limit, network error, proxy resolution failure) is an error too. ### `erc7730 generate` The `generate` command bootstraps a new descriptor file from ABIs or message schemas: ```shell -# fetch ABIs from sourcify/etherscan and generate a new calldata descriptor +# fetch ABIs from sourcify and generate a new calldata descriptor erc7730 generate --chain-id=1 --address=0x68b3465833fb72A70ecDF485E0e4C7bD8665Fc45 # generate a new calldata descriptor using given ABI file @@ -94,12 +96,11 @@ erc7730 generate --chain-id=1 --address=0x00000000000000000000000000000000000000 erc7730 generate --chain-id=1 --address=0x0000000000000000000000000000000000000000 --schema schemas.json ``` -ABIs are fetched from [Sourcify](https://sourcify.dev) first, which requires no API key. If the contract is not -verified on Sourcify, Etherscan is used as a fallback, which requires -[setting up an Etherscan API key](https://docs.etherscan.io/getting-started/viewing-api-usage-statistics): -```shell -export ETHERSCAN_API_KEY=XXXXXX -``` +ABIs are fetched from [Sourcify](https://sourcify.dev), which requires no API key. The contract (and its +implementations, if it is a proxy) must be verified on Sourcify. Sourcify responses are cached for one hour in +`~/.cache/erc7730` (or `$XDG_CACHE_HOME/erc7730`), set `ERC7730_NO_CACHE=1` to disable the cache. + +If you have a Sourcify API token, set `SOURCIFY_TOKEN` and it is sent with every Sourcify request. Please note that while the generator does its best to guess the right format based on fields name/type, the generated descriptor should be considered a starting point to refine. diff --git a/src/erc7730/common/abi.py b/src/erc7730/common/abi.py index 1c496725..b53a4e33 100644 --- a/src/erc7730/common/abi.py +++ b/src/erc7730/common/abi.py @@ -128,7 +128,6 @@ def function_to_selector(abi: Function) -> str: @dataclass(kw_only=True) class Functions: functions: dict[str, Function] - proxy: bool _READ_ONLY_MUTABILITIES = frozenset({StateMutability.pure, StateMutability.view}) @@ -141,14 +140,12 @@ def get_functions(abis: list[ABI], *, include_read_only: bool = False) -> Functi :param include_read_only: if False (default), filter out pure/view functions that cannot produce transactions :return: Functions dataclass with selector->Function mapping """ - functions = Functions(functions={}, proxy=False) + functions = Functions(functions={}) for abi in abis: if abi.type == "function": if not include_read_only and abi.stateMutability in _READ_ONLY_MUTABILITIES: continue functions.functions[function_to_selector(abi)] = abi - if abi.name in ("proxyType", "getImplementation", "implementation", "proxy__getImplementation"): - functions.proxy = True return functions diff --git a/src/erc7730/common/client.py b/src/erc7730/common/client.py index b99ca062..75a55281 100644 --- a/src/erc7730/common/client.py +++ b/src/erc7730/common/client.py @@ -10,7 +10,7 @@ from httpx_file import FileTransport from httpx_retries import RetryTransport from limiter import Limiter -from pydantic import ConfigDict, TypeAdapter, ValidationError +from pydantic import ConfigDict, Field, TypeAdapter, ValidationError from pydantic_string_url import FileUrl, HttpUrl from xdg_base_dirs import xdg_cache_home @@ -19,110 +19,169 @@ from erc7730.model.types import Address # ruff: noqa: UP047 +# ruff: noqa: N815 - camel case field names are tolerated to match schema ETHERSCAN = "api.etherscan.io" SOURCIFY = "sourcify.dev" +ERC7730_NO_CACHE = "ERC7730_NO_CACHE" +CACHE_TTL = 3600 + _T = TypeVar("_T") -class EtherscanChain(Model): - """Etherscan supported chain info.""" +class SourcifyChain(Model): + """Sourcify chain info, restricted to the fields used by this library.""" + + model_config = ConfigDict(strict=False, frozen=True, extra="ignore") + name: str + chainId: int + supported: bool = False + + +class SourcifyImplementation(Model): + """Sourcify proxy implementation, restricted to the fields used by this library.""" + + model_config = ConfigDict(strict=False, frozen=True, extra="ignore") + address: Address + + +class SourcifyError(Model): + """Sourcify error, restricted to the fields used by this library.""" + + model_config = ConfigDict(strict=False, frozen=True, extra="ignore") + customCode: str + message: str + + +class SourcifyProxyResolution(Model): + """Sourcify proxy resolution, restricted to the fields used by this library.""" model_config = ConfigDict(strict=False, frozen=True, extra="ignore") - chainname: str - chainid: int - blockexplorer: HttpUrl + isProxy: bool = False + implementations: list[SourcifyImplementation] = Field(default_factory=list) + proxyResolutionError: SourcifyError | None = None class SourcifyContract(Model): """Sourcify verified contract, restricted to the fields used by this library.""" model_config = ConfigDict(strict=False, frozen=True, extra="ignore") - abi: list[ABI] | None = None + abi: list[ABI] + proxyResolution: SourcifyProxyResolution | None = None + + +class ContractNotVerifiedError(Exception): + """Contract is not verified on Sourcify.""" + + +class ProxyImplementationNotVerifiedError(ContractNotVerifiedError): + """Contract is a proxy, and one of its implementations is not verified on Sourcify.""" + + +class ChainNotSupportedError(Exception): + """Chain is not supported by Sourcify.""" @cache -def get_supported_chains() -> list[EtherscanChain]: +def get_supported_chains() -> list[SourcifyChain]: """ - Get supported chains from Etherscan. + Get supported chains from Sourcify. - :return: Etherscan supported chains, with name/chain id/block explorer URL + :return: Sourcify supported chains, with name/chain id """ - return get(url=HttpUrl(f"https://{ETHERSCAN}/v2/chainlist"), model=list[EtherscanChain]) + chains = get(url=HttpUrl(f"https://{SOURCIFY}/server/chains"), model=list[SourcifyChain], force_cache=True) + return [chain for chain in chains if chain.supported] +@cache def get_contract_abis(chain_id: int, contract_address: Address) -> list[ABI]: """ - Get contract ABIs from Sourcify, falling back to Etherscan if the contract is not available on Sourcify. + Get contract ABIs from Sourcify. :param chain_id: EIP-155 chain ID :param contract_address: EVM contract address :return: deserialized list of ABIs - :raises Exception: if contract source is not available, API key not setup, or unexpected response + :raises ContractNotVerifiedError: if contract, or one of its proxy implementations, is not verified on Sourcify + :raises ChainNotSupportedError: if chain is not supported by Sourcify + :raises Exception: if unexpected response """ - try: - if (abis := get_contract_abis_from_sourcify(chain_id, contract_address)) is not None: - return abis - sourcify_error = "contract source is not available on Sourcify" - except Exception as e: - sourcify_error = str(e) - - try: - return get_contract_abis_from_etherscan(chain_id, contract_address) - except Exception as e: - raise Exception(f"{sourcify_error}, and fetching from Etherscan failed: {e}") from e + return get_contract_abis_from_sourcify(chain_id, contract_address) -def get_contract_abis_from_sourcify(chain_id: int, contract_address: Address) -> list[ABI] | None: +def get_contract_abis_from_sourcify(chain_id: int, contract_address: Address) -> list[ABI]: """ Get contract ABIs from Sourcify. + If Sourcify resolves the contract as a proxy, the ABIs of its implementations are appended to the ABIs of the + proxy, following nested proxies. + :param chain_id: EIP-155 chain ID :param contract_address: EVM contract address - :return: deserialized list of ABIs, or None if chain is not supported or contract source is not available + :return: deserialized list of ABIs + :raises ContractNotVerifiedError: if contract is not verified on Sourcify + :raises ProxyImplementationNotVerifiedError: if contract is a proxy and an implementation is not verified + :raises ChainNotSupportedError: if chain is not supported by Sourcify :raises Exception: if unexpected response """ + contract = _get_sourcify_contract(chain_id, contract_address) + abis = list(contract.abi) + visited = {contract_address.lower()} + pending = _get_implementation_addresses(contract) + while pending: + if (address := pending.pop(0)).lower() in visited: + continue + visited.add(address.lower()) + try: + implementation = _get_sourcify_contract(chain_id, address) + except ContractNotVerifiedError as e: + raise ProxyImplementationNotVerifiedError( + f"contract {contract_address} on chain {chain_id} is a proxy, and its implementation {address} is not " + f"verified on Sourcify" + ) from e + abis.extend(implementation.abi) + pending.extend(_get_implementation_addresses(implementation)) + return abis + + +def _get_implementation_addresses(contract: SourcifyContract) -> list[Address]: + if (resolution := contract.proxyResolution) is None or not resolution.isProxy: + return [] + return [implementation.address for implementation in resolution.implementations] + + +def _get_sourcify_contract(chain_id: int, contract_address: Address) -> SourcifyContract: try: - return get( + contract = get( url=HttpUrl(f"https://{SOURCIFY}/server/v2/contract/{chain_id}/{contract_address}"), - fields="abi", + fields="abi,proxyResolution", model=SourcifyContract, - ).abi + force_cache=True, + ) except HTTPStatusError as e: if e.response.status_code == codes.NOT_FOUND: - return None # contract source is not available on Sourcify + raise ContractNotVerifiedError( + f"contract {contract_address} on chain {chain_id} is not verified on Sourcify" + ) from e if e.response.status_code == codes.BAD_REQUEST: - return None # chain id is not supported by Sourcify + error = _parse_sourcify_error(e.response) + if error is not None and error.customCode == "unsupported_chain": + raise ChainNotSupportedError(f"chain {chain_id} is not supported by Sourcify") from e + raise Exception(f"Sourcify rejected the request: {error.message if error else e}") from e if e.response.status_code == codes.TOO_MANY_REQUESTS: raise Exception("Sourcify rate limit exceeded, please retry") from e raise e + # proxy resolution is computed at request time, a failure must not be mistaken for a regular contract + if (resolution := contract.proxyResolution) is not None and (error := resolution.proxyResolutionError) is not None: + raise Exception(f"Sourcify could not resolve whether {contract_address} is a proxy: {error.message}") + return contract -def get_contract_abis_from_etherscan(chain_id: int, contract_address: Address) -> list[ABI]: - """ - Get contract ABIs from Etherscan. - - :param chain_id: EIP-155 chain ID - :param contract_address: EVM contract address - :return: deserialized list of ABIs - :raises Exception: if chain id not supported, API key not setup, or unexpected response - """ +def _parse_sourcify_error(response: Response) -> SourcifyError | None: try: - return get( - url=HttpUrl(f"https://{ETHERSCAN}/v2/api"), - chainid=chain_id, - module="contract", - action="getabi", - address=contract_address, - model=list[ABI], - ) - except Exception as e: - if "Contract source code not verified" in str(e): - raise Exception("contract source is not available on Etherscan") from e - if "Max calls per sec rate limit reached" in str(e): - raise Exception("Etherscan rate limit exceeded, please retry") from e - raise e + return SourcifyError.model_validate_json(response.read()) + except ValidationError: + return None def get_contract_explorer_url(chain_id: int, contract_address: Address) -> HttpUrl: @@ -131,18 +190,12 @@ def get_contract_explorer_url(chain_id: int, contract_address: Address) -> HttpU :param chain_id: EIP-155 chain ID :param contract_address: EVM contract address - :return: URL to the contract explorer site - :raises NotImplementedError: if chain id not supported + :return: URL to the contract on the Sourcify repository """ - for chain in get_supported_chains(): - if chain.chainid == chain_id: - return HttpUrl(f"{chain.blockexplorer}/address/{contract_address}#code") - raise NotImplementedError( - f"Chain ID {chain_id} is not supported, please report this to authors of python-erc7730 library" - ) + return HttpUrl(f"https://repo.{SOURCIFY}/{chain_id}/{contract_address}") -def get(model: type[_T], url: HttpUrl | FileUrl, **params: Any) -> _T: +def get(model: type[_T], url: HttpUrl | FileUrl, *, force_cache: bool = False, **params: Any) -> _T: """ Fetch data from a file or an HTTP URL and deserialize it. @@ -150,13 +203,18 @@ def get(model: type[_T], url: HttpUrl | FileUrl, **params: Any) -> _T: - GitHub: adaptation to "raw.githubusercontent.com" - Etherscan: rate limiting, API key parameter injection, "result" field unwrapping + Responses are cached on disk according to their caching headers, unless the ERC7730_NO_CACHE environment variable + is set. + :param url: URL to get data from :param model: Pydantic model to deserialize the data + :param force_cache: cache the response even if it has no caching headers (for CACHE_TTL seconds) + :param params: query parameters :return: deserialized response :raises Exception: if URL type is not supported, API key not setup, or unexpected response """ with _client() as client: - response = client.get(url, params=params).raise_for_status().content + response = client.get(url, params=params, extensions={"force_cache": force_cache}).raise_for_status().content try: return TypeAdapter(model).validate_json(response) except ValidationError as e: @@ -168,12 +226,14 @@ def _client() -> Client: Create a new HTTP client with GitHub and Etherscan specific transports. :return: """ - cache_storage = FileStorage(base_path=xdg_cache_home() / "erc7730", ttl=7 * 24 * 3600, check_ttl_every=24 * 3600) - http_transport = HTTPTransport() + http_transport: BaseTransport = HTTPTransport() http_transport = GithubTransport(http_transport) http_transport = EtherscanTransport(http_transport) + http_transport = SourcifyTransport(http_transport) http_transport = RetryTransport(transport=http_transport) - http_transport = CacheTransport(transport=http_transport, storage=cache_storage) + if os.environ.get(ERC7730_NO_CACHE) is None: + cache_storage = FileStorage(base_path=xdg_cache_home() / "erc7730", ttl=CACHE_TTL, check_ttl_every=CACHE_TTL) + http_transport = CacheTransport(transport=http_transport, storage=cache_storage) file_transport = FileTransport() # TODO file storage: authorize relative paths only transports = {"https://": http_transport, "file://": file_transport} @@ -213,6 +273,23 @@ def handle_request(self, request: Request) -> Response: return super().handle_request(request) +@final +class SourcifyTransport(DelegateTransport): + """Sourcify specific transport for handling token header injection.""" + + SOURCIFY_TOKEN = "SOURCIFY_TOKEN" # nosec B105 - environment variable name, not a secret + + @override + def handle_request(self, request: Request) -> Response: + if request.url.host != SOURCIFY: + return super().handle_request(request) + + # add token if provided, it exempts the caller from rate limiting + if (token := os.environ.get(self.SOURCIFY_TOKEN)) is not None: + request.headers.update({"X-Sourcify-Token": token}) + return super().handle_request(request) + + @final class EtherscanTransport(DelegateTransport): """Etherscan specific transport for handling rate limiting, API key parameter injection, response unwrapping.""" diff --git a/src/erc7730/convert/resolved/v2/address.py b/src/erc7730/convert/resolved/v2/address.py new file mode 100644 index 00000000..3ecec772 --- /dev/null +++ b/src/erc7730/convert/resolved/v2/address.py @@ -0,0 +1,43 @@ +from collections.abc import Sequence +from typing import Any + +from pydantic import TypeAdapter, ValidationError + +from erc7730.common.output import OutputAdder +from erc7730.model.types import Address, MixedCaseAddress + +_ADDRESS = TypeAdapter(MixedCaseAddress) + + +def resolved_address(value: Any, out: OutputAdder) -> Address | None: + """ + Validate a value that a parameter resolved to as an address. + + An address written in the parameter itself is validated by the input model. A value that comes from a constant + (`$.metadata.constants...`) is not, so it is validated here with the same rules: the shape, and the EIP-55 + checksum if the address is mixed-case. + + :param value: resolved value + :param out: error handler + :return: the address, or None if an error was reported + """ + try: + return Address(_ADDRESS.validate_python(value)) + except ValidationError as e: + return out.error(title="Invalid address", message=e.errors()[0]["msg"]) + + +def resolved_addresses(values: Sequence[Any], out: OutputAdder) -> list[Address] | None: + """ + Validate a list of values that a parameter resolved to as addresses. + + :param values: resolved values + :param out: error handler + :return: the addresses, or None after the first error was reported + """ + addresses: list[Address] = [] + for value in values: + if (address := resolved_address(value, out)) is None: + return None + addresses.append(address) + return addresses diff --git a/src/erc7730/convert/resolved/v2/constants.py b/src/erc7730/convert/resolved/v2/constants.py index 7fdcbd39..26386db0 100644 --- a/src/erc7730/convert/resolved/v2/constants.py +++ b/src/erc7730/convert/resolved/v2/constants.py @@ -2,7 +2,7 @@ from collections.abc import Sequence from typing import Any, assert_never, override -from pydantic import TypeAdapter, ValidationError +from pydantic import TypeAdapter from typing_extensions import TypeVar from erc7730.common.output import OutputAdder @@ -12,7 +12,7 @@ from erc7730.model.input.v2.display import InputMapReference from erc7730.model.paths import ROOT_DESCRIPTOR_PATH, ArrayElement, ContainerPath, DataPath, DescriptorPath, Field from erc7730.model.paths.path_ops import descriptor_path_append, to_absolute -from erc7730.model.types import MixedCaseAddress +from erc7730.model.types import ADDRESS_PATTERN _T = TypeVar("_T", covariant=True) @@ -83,8 +83,8 @@ def assert_not_address(path: DataPath | ContainerPath) -> bool: case DataPath(): if path.absolute: return True - try: - TypeAdapter(MixedCaseAddress).validate_strings(str(path)) + # the shape only: an address with a wrong checksum is still an address, not a path + if ADDRESS_PATTERN.fullmatch(str(path)): out.error( title="Invalid data path", message=f""""{path}" is invalid, it must contain a data path to the address in the """ @@ -92,8 +92,7 @@ def assert_not_address(path: DataPath | ContainerPath) -> bool: "use the adequate parameter to provide a constant value.", ) return False - except ValidationError: - return True + return True case _: assert_never(path) diff --git a/src/erc7730/convert/resolved/v2/parameters.py b/src/erc7730/convert/resolved/v2/parameters.py index 2252d70e..0c9c7b0c 100644 --- a/src/erc7730/convert/resolved/v2/parameters.py +++ b/src/erc7730/convert/resolved/v2/parameters.py @@ -2,6 +2,7 @@ from erc7730.common.abi import ABIDataType from erc7730.common.output import OutputAdder +from erc7730.convert.resolved.v2.address import resolved_addresses from erc7730.convert.resolved.v2.constants import ConstantProvider from erc7730.convert.resolved.v2.enums import get_enum, get_enum_id from erc7730.convert.resolved.v2.values import resolve_path_or_constant_value @@ -89,10 +90,10 @@ def resolve_address_name_parameters( resolved_sender = constants.resolve_or_none(sender_addr_input, out) if resolved_sender is None: sender_address = None - elif isinstance(resolved_sender, str): - sender_address = [Address(resolved_sender)] - elif isinstance(resolved_sender, list): - sender_address = [Address(addr) for addr in resolved_sender] + elif isinstance(resolved_sender, str | list): + senders = [resolved_sender] if isinstance(resolved_sender, str) else resolved_sender + if (sender_address := resolved_addresses(senders, out)) is None: + return None else: raise Exception("Invalid senderAddress type") @@ -119,10 +120,10 @@ def resolve_interoperable_address_name_parameters( resolved_sender = constants.resolve_or_none(sender_addr_input, out) if resolved_sender is None: sender_address = None - elif isinstance(resolved_sender, str): - sender_address = [Address(resolved_sender)] - elif isinstance(resolved_sender, list): - sender_address = [Address(addr) for addr in resolved_sender] + elif isinstance(resolved_sender, str | list): + senders = [resolved_sender] if isinstance(resolved_sender, str) else resolved_sender + if (sender_address := resolved_addresses(senders, out)) is None: + return None else: raise Exception("Invalid senderAddress type") @@ -241,17 +242,15 @@ def resolve_token_amount_parameters( list[DescriptorPathStr | MixedCaseAddress] | MixedCaseAddress | None, constants.resolve_or_none(params.nativeCurrencyAddress, out), ) - resolved_addresses: list[Address] | None + native_addresses: list[Address] | None if input_addresses is None: - resolved_addresses = None - elif isinstance(input_addresses, list): - resolved_addresses = [] - for input_address in input_addresses: - if (resolved_address := constants.resolve(input_address, out)) is None: - return None - resolved_addresses.append(Address(resolved_address)) - elif isinstance(input_addresses, str): - resolved_addresses = [Address(input_addresses)] + native_addresses = None + elif isinstance(input_addresses, str | list): + inputs = [input_addresses] if isinstance(input_addresses, str) else input_addresses + # an element of the list can be a path to a constant + resolved_inputs = [constants.resolve(i, out) for i in inputs] + if None in resolved_inputs or (native_addresses := resolved_addresses(resolved_inputs, out)) is None: + return None else: raise Exception("Invalid nativeCurrencyAddress type") @@ -279,7 +278,7 @@ def resolve_token_amount_parameters( return ResolvedTokenAmountParameters( token=token_resolved, - nativeCurrencyAddress=resolved_addresses, + nativeCurrencyAddress=native_addresses, threshold=resolved_threshold, message=constants.resolve_or_none(params.message, out), chainId=resolved_chain_id, diff --git a/src/erc7730/convert/resolved/v2/values.py b/src/erc7730/convert/resolved/v2/values.py index 55ef0739..bc8351aa 100644 --- a/src/erc7730/convert/resolved/v2/values.py +++ b/src/erc7730/convert/resolved/v2/values.py @@ -4,6 +4,7 @@ from erc7730.common.abi import ABIDataType from erc7730.common.output import OutputAdder +from erc7730.convert.resolved.v2.address import resolved_address from erc7730.convert.resolved.v2.constants import ConstantProvider from erc7730.model.input.v2.display import InputFieldBase from erc7730.model.input.v2.format import FieldFormat @@ -62,6 +63,9 @@ def resolve_field_value( abi_type=abi_type, constants=constants, out=out, + # an ERC-7930 interoperable address has the address type family, but is a longer binary value + evm_address=abi_type == ABIDataType.ADDRESS + and input_field_format != FieldFormat.INTEROPERABLE_ADDRESS_NAME, ) ) is None: return out.error(title="Invalid field", message="Field must have either a path or a value.") @@ -75,6 +79,7 @@ def resolve_path_or_constant_value( abi_type: ABIDataType, constants: ConstantProvider, out: OutputAdder, + evm_address: bool | None = None, ) -> ResolvedValue | None: """ Resolve value, as a data path or constant value. @@ -85,6 +90,8 @@ def resolve_path_or_constant_value( :param abi_type: expected encoded value data type :param constants: descriptor paths constants resolver :param out: error handler + :param evm_address: whether a constant value must be a 20 bytes address (shape and EIP-55 checksum); by default, + whether the data type is an address :return: resolved value or None if error or value resolves to None """ if input_path is not None: @@ -103,6 +110,11 @@ def resolve_path_or_constant_value( if (value := constants.resolve(input_value, out)) is None: return None + if evm_address is None: + evm_address = abi_type == ABIDataType.ADDRESS + if evm_address and resolved_address(value, out) is None: + return None + if not isinstance(value, str | bool | int | float): return out.error( title="Invalid constant value", diff --git a/src/erc7730/generate/generate.py b/src/erc7730/generate/generate.py index 6b1c0149..cad03638 100644 --- a/src/erc7730/generate/generate.py +++ b/src/erc7730/generate/generate.py @@ -56,7 +56,7 @@ def generate_descriptor( Generate an ERC-7730 descriptor. If an EIP-712 schema is provided, an EIP-712 descriptor is generated for this schema, otherwise a calldata - descriptor. If no ABI is supplied, the ABIs are fetched from Sourcify or Etherscan using the chain id / contract + descriptor. If no ABI is supplied, the ABIs are fetched from Sourcify using the chain id / contract address. :param chain_id: contract chain id @@ -109,8 +109,8 @@ def _generate_context_calldata( if abi is not None: abis = TypeAdapter(list[ABI]).validate_json(abi) - elif (abis := get_contract_abis(chain_id, contract_address)) is None: - raise Exception("Failed to fetch contract ABIs") + else: + abis = get_contract_abis(chain_id, contract_address) functions = list(get_functions(abis, include_read_only=True).functions.values()) diff --git a/src/erc7730/lint/lint.py b/src/erc7730/lint/lint.py index 5376f46b..313e68c6 100644 --- a/src/erc7730/lint/lint.py +++ b/src/erc7730/lint/lint.py @@ -31,10 +31,12 @@ from erc7730.model.input.descriptor import InputERC7730Descriptor -def lint_all_and_print_errors(paths: list[Path], gha: bool = False, skip_abi_validation: bool = False) -> bool: +def lint_all_and_print_errors( + paths: list[Path], gha: bool = False, skip_abi_validation: bool = False, require_verified: bool = False +) -> bool: out = GithubAnnotationsAdder() if gha else DropFileOutputAdder(delegate=ConsoleOutputAdder()) - count = lint_all(paths, out, skip_abi_validation=skip_abi_validation) + count = lint_all(paths, out, skip_abi_validation=skip_abi_validation, require_verified=require_verified) if out.has_errors: print(f"[bold][red]checked {count} descriptor files, some errors found ❌[/red][/bold]") @@ -48,7 +50,9 @@ def lint_all_and_print_errors(paths: list[Path], gha: bool = False, skip_abi_val return True -def lint_all(paths: list[Path], out: OutputAdder, skip_abi_validation: bool = False) -> int: +def lint_all( + paths: list[Path], out: OutputAdder, skip_abi_validation: bool = False, require_verified: bool = False +) -> int: """ Lint all ERC-7730 descriptor files at given paths. @@ -56,6 +60,9 @@ def lint_all(paths: list[Path], out: OutputAdder, skip_abi_validation: bool = Fa :param paths: paths to apply linter on :param out: output adder + :param skip_abi_validation: skip ABI comparison with Sourcify reference data + :param require_verified: report contracts that are not verified on Sourcify, and reference ABIs that could not be + fetched, as errors instead of warnings :return: number of files checked """ linters = [ @@ -65,7 +72,7 @@ def lint_all(paths: list[Path], out: OutputAdder, skip_abi_validation: bool = Fa ValidateMaxLengthLinter(), ] if not skip_abi_validation: - linters.insert(0, ValidateABILinter()) + linters.insert(0, ValidateABILinter(require_verified=require_verified)) linter = MultiLinter(linters) files = list(get_erc7730_files(*paths, out=out)) diff --git a/src/erc7730/lint/lint_validate_abi.py b/src/erc7730/lint/lint_validate_abi.py index 68b6c0aa..da37a33d 100644 --- a/src/erc7730/lint/lint_validate_abi.py +++ b/src/erc7730/lint/lint_validate_abi.py @@ -16,6 +16,13 @@ class ValidateABILinter(ERC7730Linter): - => compares the two ABIs """ + def __init__(self, require_verified: bool = False) -> None: + """ + :param require_verified: report a contract that is not verified on Sourcify as an error instead of a warning + (as well as a reference ABI that could not be fetched, for instance because of a rate limit) + """ + self.require_verified = require_verified + @override def lint(self, descriptor: ResolvedERC7730Descriptor, out: OutputAdder) -> None: if isinstance(descriptor.context, ResolvedEIP712Context): @@ -28,37 +35,38 @@ def lint(self, descriptor: ResolvedERC7730Descriptor, out: OutputAdder) -> None: def _validate_eip712_schemas(cls, context: ResolvedEIP712Context, out: OutputAdder) -> None: pass # not implemented - @classmethod - def _validate_contract_abis(cls, context: ResolvedContractContext, out: OutputAdder) -> None: + def _validate_contract_abis(self, context: ResolvedContractContext, out: OutputAdder) -> None: if not isinstance(context.contract.abi, list): raise ValueError("Contract ABIs should have been resolved") if (deployments := context.contract.deployments) is None: return for deployment in deployments: + skipped = "descriptor ABIs will not be validated" + unverified = out.error if self.require_verified else out.warning + unsupported = out.error if self.require_verified else out.info + failed = out.error if self.require_verified else out.warning try: - if (abis := client.get_contract_abis(deployment.chainId, deployment.address)) is None: - continue + abis = client.get_contract_abis(deployment.chainId, deployment.address) + except client.ProxyImplementationNotVerifiedError as e: + unverified(title="Proxy implementation not verified", message=f"{e}, {skipped}") + continue + except client.ContractNotVerifiedError as e: + unverified(title="Contract not verified", message=f"{e}, {skipped}") + continue + except client.ChainNotSupportedError as e: + unsupported(title="Chain not supported", message=f"{e}, {skipped}") + continue except Exception as e: - out.warning( + failed( title="Could not fetch ABI", - message=f"Fetching reference ABI for chain id {deployment.chainId} failed, descriptor ABIs will " - f"not be validated: {e}", + message=f"Fetching reference ABI for chain id {deployment.chainId} failed, {skipped}: {e}", ) continue reference_abis = get_functions(abis) descriptor_abis = get_functions(context.contract.abi) - try: - url = client.get_contract_explorer_url(deployment.chainId, deployment.address) - except NotImplementedError: - url = f"" - - if reference_abis.proxy: - return out.info( - title="Proxy contract", - message=f"Contract {url} is likely to be a proxy, validation of descriptor ABIs skipped", - ) + url = client.get_contract_explorer_url(deployment.chainId, deployment.address) for selector, abi in descriptor_abis.functions.items(): if selector not in reference_abis.functions: diff --git a/src/erc7730/lint/v2/lint.py b/src/erc7730/lint/v2/lint.py index 515f9e4c..c47018f8 100644 --- a/src/erc7730/lint/v2/lint.py +++ b/src/erc7730/lint/v2/lint.py @@ -23,11 +23,11 @@ from erc7730.model.input.v2.descriptor import InputERC7730Descriptor -def lint_all_and_print_errors(paths: list[Path], gha: bool = False) -> bool: +def lint_all_and_print_errors(paths: list[Path], gha: bool = False, require_verified: bool = False) -> bool: """Lint all ERC-7730 v2 descriptor files at given paths and print results.""" out = GithubAnnotationsAdder() if gha else DropFileOutputAdder(delegate=ConsoleOutputAdder()) - count = lint_all(paths, out) + count = lint_all(paths, out, require_verified=require_verified) if out.has_errors: print(f"[bold][red]checked {count} v2 descriptor files, some errors found ❌[/red][/bold]") @@ -41,7 +41,7 @@ def lint_all_and_print_errors(paths: list[Path], gha: bool = False) -> bool: return True -def lint_all(paths: list[Path], out: OutputAdder) -> int: +def lint_all(paths: list[Path], out: OutputAdder, require_verified: bool = False) -> int: """ Lint all ERC-7730 v2 descriptor files at given paths. @@ -49,11 +49,13 @@ def lint_all(paths: list[Path], out: OutputAdder) -> int: :param paths: paths to apply linter on :param out: output adder + :param require_verified: report contracts that are not verified on Sourcify, and reference ABIs that could not be + fetched, as errors instead of warnings :return: number of files checked """ linter = MultiLinter( [ - ValidateDisplayFieldsLinter(), + ValidateDisplayFieldsLinter(require_verified=require_verified), ValidateEIP712KeysLinter(), ClassifyTransactionTypeLinter(), ValidateMaxLengthLinter(), diff --git a/src/erc7730/lint/v2/lint_transaction_type_classifier.py b/src/erc7730/lint/v2/lint_transaction_type_classifier.py index 37b59779..8009cb23 100644 --- a/src/erc7730/lint/v2/lint_transaction_type_classifier.py +++ b/src/erc7730/lint/v2/lint_transaction_type_classifier.py @@ -32,12 +32,12 @@ class ClassifyTransactionTypeLinter(ERC7730Linter): def lint( self, input_descriptor: InputERC7730Descriptor, descriptor: ResolvedERC7730Descriptor, out: OutputAdder ) -> None: - if (tx_class := self._determine_tx_class(descriptor)) is None: + if (tx_class := self._determine_tx_class(descriptor, out)) is None: return None DisplayFormatChecker(tx_class, descriptor.display).check(out) @classmethod - def _determine_tx_class(cls, descriptor: ResolvedERC7730Descriptor) -> TxClass | None: + def _determine_tx_class(cls, descriptor: ResolvedERC7730Descriptor, out: OutputAdder) -> TxClass | None: match descriptor.context: case ResolvedEIP712Context(): # In v2, no schemas — classify from format keys (primaryType) @@ -47,18 +47,24 @@ def _determine_tx_class(cls, descriptor: ResolvedERC7730Descriptor) -> TxClass | return None case ResolvedContractContext(): # Try to classify from fetched ABI - return cls._classify_from_fetched_abi(descriptor.context) + return cls._classify_from_fetched_abi(descriptor.context, out) @classmethod - def _classify_from_fetched_abi(cls, context: ResolvedContractContext) -> TxClass | None: + def _classify_from_fetched_abi(cls, context: ResolvedContractContext, out: OutputAdder) -> TxClass | None: if (deployments := context.contract.deployments) is None: return None for deployment in deployments: try: - if (abis := client.get_contract_abis(deployment.chainId, deployment.address)) is not None: - return ABIClassifier().classify(list(abis)) - except Exception: # nosec B112 - intentional: try next deployment on failure + abis = client.get_contract_abis(deployment.chainId, deployment.address) + except Exception as e: + # the display fields linter already reports the failure, only trace which deployment was skipped + out.debug( + title="Transaction type not classified", + message=f"Fetching reference ABI for chain id {deployment.chainId} failed, trying next deployment: " + f"{e}", + ) continue + return ABIClassifier().classify(list(abis)) return None diff --git a/src/erc7730/lint/v2/lint_validate_display_fields.py b/src/erc7730/lint/v2/lint_validate_display_fields.py index 86d6e82e..d11f4aad 100644 --- a/src/erc7730/lint/v2/lint_validate_display_fields.py +++ b/src/erc7730/lint/v2/lint_validate_display_fields.py @@ -1,36 +1,39 @@ """ -V2 linter that validates display fields against reference ABIs fetched from Sourcify or Etherscan. +V2 linter that validates display fields against reference ABIs fetched from Sourcify. In v2, ABI and EIP-712 schemas are NOT embedded in the descriptor. Instead: - - For contract context: fetch ABI from Sourcify or Etherscan, validate display field paths match ABI params, + - For contract context: fetch ABI from Sourcify, validate display field paths match ABI params, and check selector exhaustiveness. - For EIP-712 context: no schema to validate against (no-op). """ -from typing import final, override +import json +from typing import Any, final, override + +from pydantic_string_url import HttpUrl from erc7730.common import client -from erc7730.common.abi import compute_signature, get_functions, parse_signature, signature_to_selector +from erc7730.common.abi import Functions, compute_signature, get_functions, parse_signature, signature_to_selector from erc7730.common.output import OutputAdder from erc7730.lint.v2 import ERC7730Linter from erc7730.lint.v2.path_schemas import compute_format_schema_paths -from erc7730.model.abi import Function +from erc7730.model.abi import Component, Function, InputOutput from erc7730.model.input.v2.descriptor import InputERC7730Descriptor from erc7730.model.paths import DataPath, Field from erc7730.model.paths.path_ops import data_path_starts_with from erc7730.model.paths.path_schemas import compute_abi_schema_paths -from erc7730.model.resolved.v2.context import ResolvedContractContext, ResolvedEIP712Context +from erc7730.model.resolved.v2.context import ResolvedContractContext, ResolvedDeployment, ResolvedEIP712Context from erc7730.model.resolved.v2.descriptor import ResolvedERC7730Descriptor @final class ValidateDisplayFieldsLinter(ERC7730Linter): """ - Validates display fields against reference ABIs fetched from Sourcify or Etherscan. + Validates display fields against reference ABIs fetched from Sourcify. For contract context: - - Fetches ABI from Sourcify or Etherscan for each deployment - - Validates that display field paths exist in the ABI + - Fetches ABI from Sourcify for each deployment, and reports deployments not exposing the same functions + - Validates that display field paths exist in the ABI (once per distinct reference ABI) - Validates that all ABI function params have display fields - Checks that all selectors in the ABI have corresponding display formats @@ -38,6 +41,13 @@ class ValidateDisplayFieldsLinter(ERC7730Linter): - No schema available in v2 resolved model, so no validation is performed """ + def __init__(self, require_verified: bool = False) -> None: + """ + :param require_verified: report a contract that is not verified on Sourcify as an error instead of a warning + (as well as a reference ABI that could not be fetched, for instance because of a rate limit) + """ + self.require_verified = require_verified + @override def lint( self, input_descriptor: InputERC7730Descriptor, descriptor: ResolvedERC7730Descriptor, out: OutputAdder @@ -48,9 +58,8 @@ def lint( case ResolvedContractContext(): self._validate_contract_display_fields(input_descriptor, descriptor, out) - @classmethod def _validate_contract_display_fields( - cls, input_descriptor: InputERC7730Descriptor, descriptor: ResolvedERC7730Descriptor, out: OutputAdder + self, input_descriptor: InputERC7730Descriptor, descriptor: ResolvedERC7730Descriptor, out: OutputAdder ) -> None: context = descriptor.context if not isinstance(context, ResolvedContractContext): @@ -59,44 +68,93 @@ def _validate_contract_display_fields( if (deployments := context.contract.deployments) is None: return - # Try to fetch ABI from Sourcify or Etherscan for the first deployment that succeeds - reference_abis = None - explorer_url = None + # Fetch the reference ABI of every deployment, and group deployments exposing the same functions, so that + # each distinct ABI is validated once and deployments diverging from the others are reported + groups: dict[str, tuple[Functions, list[ResolvedDeployment]]] = {} for deployment in deployments: + skipped = "display fields will not be validated against ABI" + unverified = out.error if self.require_verified else out.warning + unsupported = out.error if self.require_verified else out.info + failed = out.error if self.require_verified else out.warning try: - if (abis := client.get_contract_abis(deployment.chainId, deployment.address)) is None: - continue + abis = client.get_contract_abis(deployment.chainId, deployment.address) + except client.ProxyImplementationNotVerifiedError as e: + unverified(title="Proxy implementation not verified", message=f"{e}, {skipped}") + continue + except client.ContractNotVerifiedError as e: + unverified(title="Contract not verified", message=f"{e}, {skipped}") + continue + except client.ChainNotSupportedError as e: + unsupported(title="Chain not supported", message=f"{e}, {skipped}") + continue except Exception as e: - out.warning( + failed( title="Could not fetch ABI", - message=f"Fetching reference ABI for chain id {deployment.chainId} failed, display fields will " - f"not be validated against ABI: {e}", + message=f"Fetching reference ABI for chain id {deployment.chainId} failed, {skipped}: {e}", ) continue reference_abis = get_functions(abis) - try: - explorer_url = client.get_contract_explorer_url(deployment.chainId, deployment.address) - except NotImplementedError: - explorer_url = f"" - break - - if reference_abis is None: - return + groups.setdefault(self._external_abi_key(reference_abis), (reference_abis, []))[1].append(deployment) - if reference_abis.proxy: - return out.info( - title="Proxy contract", - message=f"Contract {explorer_url} is likely to be a proxy, validation of display fields skipped", + if len(groups) > 1: + out.warning( + title="Deployment ABIs differ", + message="The reference ABIs of the deployments do not all expose the same functions, display fields " + "are validated against each distinct reference ABI: " + + "; ".join(", ".join(f"{d.chainId}:{d.address}" for d in ds) for _, ds in groups.values()), ) + for reference_abis, group_deployments in groups.values(): + explorer_url = client.get_contract_explorer_url(group_deployments[0].chainId, group_deployments[0].address) + self._validate_display_fields(input_descriptor, descriptor, reference_abis, explorer_url, out) + + @classmethod + def _external_abi_key(cls, reference_abis: Functions) -> str: + """ + Compute a key identifying the functions as seen by a caller of the contract. + + Compiler details such as internal types, or legacy fields, are left out, so that deployments compiled from + slightly different sources but exposing the same functions are grouped together. + + :param reference_abis: functions of a reference ABI + :return: key equal for two ABIs exposing the same functions + """ + return json.dumps( + { + selector: { + "name": abi.name, + "inputs": [cls._external_parameter(param) for param in abi.inputs or []], + "outputs": [cls._external_parameter(param) for param in abi.outputs or []], + "stateMutability": abi.stateMutability, + } + for selector, abi in sorted(reference_abis.functions.items()) + } + ) + + @classmethod + def _external_parameter(cls, param: InputOutput | Component) -> dict[str, Any]: + return { + "name": param.name, + "type": param.type, + "components": [cls._external_parameter(component) for component in param.components or []], + } + + def _validate_display_fields( + self, + input_descriptor: InputERC7730Descriptor, + descriptor: ResolvedERC7730Descriptor, + reference_abis: Functions, + explorer_url: HttpUrl, + out: OutputAdder, + ) -> None: # Build ABI paths by selector abi_paths_by_selector: dict[str, set[DataPath]] = {} for selector, abi in reference_abis.functions.items(): abi_paths_by_selector[selector] = compute_abi_schema_paths(abi) # Parse the input format keys, which carry the parameter names resolution reduced to selectors - declared_abis_by_selector = cls._parse_declared_abis(input_descriptor) + declared_abis_by_selector = self._parse_declared_abis(input_descriptor) # Validate display field paths against ABI paths for selector, fmt in descriptor.display.formats.items(): @@ -110,7 +168,7 @@ def _validate_contract_display_fields( format_paths = compute_format_schema_paths(fmt) abi_paths = abi_paths_by_selector[selector] - unnamed_parameter_names = cls._unnamed_parameter_names( + unnamed_parameter_names = self._unnamed_parameter_names( reference_abis.functions[selector], declared_abis_by_selector.get(selector) ) @@ -119,7 +177,7 @@ def _validate_contract_display_fields( # (e.g. defining a field for an array root covers all nested elements). for path in format_paths.data_paths - abi_paths: if not any(data_path_starts_with(abi_path, path) for abi_path in abi_paths): - if cls._root_name(path) in unnamed_parameter_names: + if self._root_name(path) in unnamed_parameter_names: continue out.error( title="Invalid display field", diff --git a/src/erc7730/main.py b/src/erc7730/main.py index 13c04979..72c22299 100644 --- a/src/erc7730/main.py +++ b/src/erc7730/main.py @@ -103,17 +103,27 @@ def command_lint( paths: Annotated[list[Path], Argument(help="The files or directory paths to lint")], gha: Annotated[bool, Option(help="Enable Github annotations output")] = False, skip_abi_validation: Annotated[ - bool, Option("--skip-abi-validation", help="Skip ABI comparison with Sourcify/Etherscan reference data") + bool, Option("--skip-abi-validation", help="Skip ABI comparison with Sourcify reference data") + ] = False, + require_verified: Annotated[ + bool, + Option( + "--require-verified", + help="Report contracts that are not verified on Sourcify, and reference ABIs that could not be fetched, " + "as errors", + ), ] = False, v2: Annotated[ bool, Option("--v2", help="Use v2 model for validation (auto-detected from $schema if not set)") ] = False, ) -> None: if v2 or _any_v2_descriptor(paths): - if not lint_all_and_print_errors_v2(paths, gha): + if not lint_all_and_print_errors_v2(paths, gha, require_verified=require_verified): raise Exit(1) else: - if not lint_all_and_print_errors_v1(paths, gha, skip_abi_validation=skip_abi_validation): + if not lint_all_and_print_errors_v1( + paths, gha, skip_abi_validation=skip_abi_validation, require_verified=require_verified + ): raise Exit(1) diff --git a/src/erc7730/model/types.py b/src/erc7730/model/types.py index 9b249163..6c74a2d3 100644 --- a/src/erc7730/model/types.py +++ b/src/erc7730/model/types.py @@ -5,12 +5,40 @@ JSON schema: https://github.com/LedgerHQ/clear-signing-erc7730-registry/blob/master/specs/erc7730-v1.schema.json """ +import re from typing import Annotated -from pydantic import BeforeValidator, Field +from eth_utils.address import to_checksum_address +from pydantic import AfterValidator, BeforeValidator, Field +from pydantic_core import PydanticCustomError from erc7730.common.pydantic import ErrorTypeLabel +ADDRESS_PATTERN = re.compile(r"^0x[a-fA-F0-9]{40}$") +"""The shape of an address, without the checksum.""" + + +def validate_address_checksum(value: str) -> str: + """ + Reject an address that is neither in lowercase nor in its EIP-55 checksum form. + + An address written in lowercase only carries no checksum and is accepted as is. An address with an uppercase + letter claims a checksum, so a mismatch is most likely a typo or a corrupted copy. This includes an address + written in uppercase only: EIP-55 does not define it as a form without checksum. + + The chain-specific checksum of EIP-1191 (used by Rootstock) is not supported: the type does not know the chain. + """ + if value == value.lower(): + return value + if value != (expected := to_checksum_address(value)): + raise PydanticCustomError( + "address_checksum", + 'invalid EIP-55 checksum for address "{value}", expected "{expected}" (or the address in lowercase)', + {"value": value, "expected": expected}, + ) + return value + + Id = Annotated[ str, Field( @@ -30,12 +58,14 @@ description="An Ethereum contract address, can be lowercase or EIP-55.", min_length=42, max_length=42, - pattern=r"^0x[a-fA-F0-9]+$", + pattern=ADDRESS_PATTERN.pattern, ), ErrorTypeLabel( '20 bytes, hexadecimal Ethereum address prefixed with "0x" (EIP-55 or lowercase), such as ' + '"0xdac17f958d2ee523a2206206994597c13d831ec7".' ), + # after the label wrapper, so that a checksum error keeps its own message + AfterValidator(validate_address_checksum), ] Address = Annotated[ diff --git a/tests/common/test_client.py b/tests/common/test_client.py index 24cec886..a3bcd549 100644 --- a/tests/common/test_client.py +++ b/tests/common/test_client.py @@ -1,3 +1,7 @@ +from typing import Any + +import pytest +from httpx import BaseTransport, HTTPStatusError, Request, Response, codes from pydantic_string_url import HttpUrl from erc7730.common import client @@ -8,9 +12,9 @@ def test_get_supported_chains() -> None: result = client.get_supported_chains() assert result is not None assert len(result) >= 50 - names = {chain.chainname for chain in result} + names = {chain.name for chain in result} assert "Ethereum Mainnet" in names - assert "Sepolia Testnet" in names + assert "Ethereum Sepolia Testnet" in names assert "BNB Smart Chain Mainnet" in names assert "BNB Smart Chain Testnet" in names assert "Polygon Mainnet" in names @@ -40,6 +44,11 @@ def test_get_supported_chains() -> None: assert "Taiko Mainnet" in names +def test_get_contract_explorer_url() -> None: + result = client.get_contract_explorer_url(chain_id=1, contract_address="0x06012c8cf97bead5deae237070f9587f8e7a266d") + assert result == "https://repo.sourcify.dev/1/0x06012c8cf97bead5deae237070f9587f8e7a266d" + + def test_get_contract_abis() -> None: result = client.get_contract_abis(chain_id=1, contract_address="0x06012c8cf97bead5deae237070f9587f8e7a266d") assert result is not None @@ -55,17 +64,101 @@ def test_get_contract_abis_from_sourcify() -> None: def test_get_contract_abis_from_sourcify_unverified_contract() -> None: - result = client.get_contract_abis_from_sourcify( - chain_id=1, contract_address="0x0000000000000000000000000000000000000001" - ) - assert result is None + with pytest.raises(client.ContractNotVerifiedError): + client.get_contract_abis_from_sourcify( + chain_id=1, contract_address="0x0000000000000000000000000000000000000001" + ) def test_get_contract_abis_from_sourcify_unsupported_chain() -> None: + with pytest.raises(client.ChainNotSupportedError): + client.get_contract_abis_from_sourcify( + chain_id=99999999, contract_address="0x06012c8cf97bead5deae237070f9587f8e7a266d" + ) + + +def test_get_contract_abis_from_sourcify_proxy() -> None: + # USDC is a proxy, transfer() is only defined in the ABI of its implementation result = client.get_contract_abis_from_sourcify( - chain_id=99999999, contract_address="0x06012c8cf97bead5deae237070f9587f8e7a266d" + chain_id=1, contract_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" + ) + assert result is not None + names = {abi.name for abi in result if abi.type == "function"} + assert "upgradeTo" in names + assert "transfer" in names + + +def test_get_contract_abis_unverified_proxy_implementation(monkeypatch: pytest.MonkeyPatch) -> None: + proxy = client.SourcifyContract.model_validate( + { + "abi": [], + "proxyResolution": { + "isProxy": True, + "implementations": [{"address": "0x0000000000000000000000000000000000000001"}], + }, + } ) - assert result is None + + def get(model: Any, url: str, **params: Any) -> Any: + if url.endswith("eb48"): + return proxy + response = Response(status_code=codes.NOT_FOUND, request=Request("GET", url)) + raise HTTPStatusError("not verified", request=response.request, response=response) + + client.get_contract_abis.cache_clear() + monkeypatch.setattr(client, "get", get) + with pytest.raises(client.ProxyImplementationNotVerifiedError, match="0x0000000000000000000000000000000000000001"): + client.get_contract_abis(chain_id=1, contract_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48") + + +def test_get_contract_abis_proxy_resolution_error(monkeypatch: pytest.MonkeyPatch) -> None: + contract = client.SourcifyContract.model_validate( + { + "abi": [], + "proxyResolution": { + "proxyResolutionError": { + "customCode": "proxy_resolution_error", + "message": "Error while running proxy detection and implementation resolution", + } + }, + } + ) + client.get_contract_abis.cache_clear() + monkeypatch.setattr(client, "get", lambda model, url, **params: contract) + with pytest.raises(Exception, match="could not resolve whether"): + client.get_contract_abis(chain_id=1, contract_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48") + + +class _RecordingTransport(BaseTransport): + """Transport that records the requests it receives and answers with an empty JSON list.""" + + def __init__(self) -> None: + self.requests: list[Request] = [] + + def handle_request(self, request: Request) -> Response: + self.requests.append(request) + return Response(status_code=codes.OK, json=[]) + + +@pytest.mark.parametrize( + ("token", "url", "expected"), + [ + ("secret", "https://sourcify.dev/server/chains", "secret"), + (None, "https://sourcify.dev/server/chains", None), + ("secret", "https://api.etherscan.io/v2/chainlist", None), + ], +) +def test_sourcify_transport_token_header( + monkeypatch: pytest.MonkeyPatch, token: str | None, url: str, expected: str | None +) -> None: + if token is None: + monkeypatch.delenv(client.SourcifyTransport.SOURCIFY_TOKEN, raising=False) + else: + monkeypatch.setenv(client.SourcifyTransport.SOURCIFY_TOKEN, token) + delegate = _RecordingTransport() + client.SourcifyTransport(delegate).handle_request(Request("GET", url)) + assert len(delegate.requests) == 1 + assert delegate.requests[0].headers.get("X-Sourcify-Token") == expected def test_get_from_github() -> None: diff --git a/tests/convert/resolved/data/literal_values_input.json b/tests/convert/resolved/data/literal_values_input.json index 8ca0f8b5..975819e8 100644 --- a/tests/convert/resolved/data/literal_values_input.json +++ b/tests/convert/resolved/data/literal_values_input.json @@ -5,7 +5,7 @@ "deployments": [ { "chainId": 1, - "address": "0x0000000000000000000000000000000000000aAa" + "address": "0x0000000000000000000000000000000000000aaa" } ], "abi": [ @@ -30,7 +30,7 @@ }, { "label": "Test - addressName", - "value": "0x0000000000000000000000000000000000000Bbb", + "value": "0x0000000000000000000000000000000000000bbb", "format": "addressName", "params": { "types": [ "eoa" ], "sources": [ "ens" ]} }, @@ -41,7 +41,7 @@ }, { "label": "Test - calldata", - "value": "0x0000000000000000000000000000000000000Bbb", + "value": "0x0000000000000000000000000000000000000bbb", "format": "calldata", "params": { "callee": "0x0000000000000000000000000000000000000001" } }, diff --git a/tests/convert/resolved/data/literal_values_resolved.json b/tests/convert/resolved/data/literal_values_resolved.json index 06b8905e..ae8b09e2 100644 --- a/tests/convert/resolved/data/literal_values_resolved.json +++ b/tests/convert/resolved/data/literal_values_resolved.json @@ -27,7 +27,7 @@ "type": "constant", "type_family": "address", "type_size": 20, - "value": "0x0000000000000000000000000000000000000Bbb", + "value": "0x0000000000000000000000000000000000000bbb", "raw": "0x0000000000000000000000000000000000000bbb" }, "label": "Test - addressName", @@ -44,7 +44,7 @@ "type": "constant", "type_family": "bytes", "type_size": 20, - "value": "0x0000000000000000000000000000000000000Bbb", + "value": "0x0000000000000000000000000000000000000bbb", "raw": "0x0000000000000000000000000000000000000bbb" }, "label": "Test - calldata", diff --git a/tests/convert/resolved/data/minimal_contract_input.json b/tests/convert/resolved/data/minimal_contract_input.json index a927cb85..c18f1af1 100644 --- a/tests/convert/resolved/data/minimal_contract_input.json +++ b/tests/convert/resolved/data/minimal_contract_input.json @@ -5,7 +5,7 @@ "deployments": [ { "chainId": 1, - "address": "0x0000000000000000000000000000000000000aAa" + "address": "0x0000000000000000000000000000000000000aaa" } ], "abi": [ diff --git a/tests/lint/test_lint_validate_abi.py b/tests/lint/test_lint_validate_abi.py new file mode 100644 index 00000000..478b3573 --- /dev/null +++ b/tests/lint/test_lint_validate_abi.py @@ -0,0 +1,43 @@ +import pytest + +from erc7730.common import client +from erc7730.common.output import ListOutputAdder, Output +from erc7730.lint.lint_validate_abi import ValidateABILinter +from erc7730.model.resolved.context import ResolvedContractContext + +CONTEXT = ResolvedContractContext.model_validate( + { + "contract": { + "abi": [], + "deployments": [{"chainId": 1, "address": "0x0000000000000000000000000000000000000001"}], + } + } +) + +NOT_VERIFIED = client.ContractNotVerifiedError("contract 0x1 on chain 1 is not verified on Sourcify") +IMPLEMENTATION_NOT_VERIFIED = client.ProxyImplementationNotVerifiedError("contract 0x1 on chain 1 is a proxy") +CHAIN_NOT_SUPPORTED = client.ChainNotSupportedError("chain 1 is not supported by Sourcify") +RATE_LIMITED = Exception("Sourcify rate limit exceeded, please retry") + + +@pytest.mark.parametrize( + ("error", "title", "default_level"), + [ + (NOT_VERIFIED, "Contract not verified", Output.Level.WARNING), + (IMPLEMENTATION_NOT_VERIFIED, "Proxy implementation not verified", Output.Level.WARNING), + (CHAIN_NOT_SUPPORTED, "Chain not supported", Output.Level.INFO), + (RATE_LIMITED, "Could not fetch ABI", Output.Level.WARNING), + ], +) +def test_fetch_failure_is_an_error_only_when_verified_is_required( + monkeypatch: pytest.MonkeyPatch, error: Exception, title: str, default_level: Output.Level +) -> None: + def get_contract_abis(chain_id: int, contract_address: str) -> None: + raise error + + monkeypatch.setattr(client, "get_contract_abis", get_contract_abis) + + for require_verified, level in ((False, default_level), (True, Output.Level.ERROR)): + out = ListOutputAdder() + ValidateABILinter(require_verified=require_verified)._validate_contract_abis(CONTEXT, out) + assert [(output.title, output.level) for output in out.outputs] == [(title, level)] diff --git a/tests/model/test_types.py b/tests/model/test_types.py new file mode 100644 index 00000000..83782adc --- /dev/null +++ b/tests/model/test_types.py @@ -0,0 +1,52 @@ +import pytest +from pydantic import TypeAdapter, ValidationError + +from erc7730.model.input.v2.context import InputDeployment +from erc7730.model.types import MixedCaseAddress + +CHECKSUMMED = "0xb426b5AE61c23fF1B901a8AD1f1A3921D1E9D2F1" +WRONG_CHECKSUM = "0xb426B5aE61c23Ff1b901A8ad1F1A3921D1E9D2f1" +UPPERCASE = "0x" + CHECKSUMMED[2:].upper() + + +@pytest.mark.parametrize( + "address", + [ + CHECKSUMMED, + CHECKSUMMED.lower(), + "0x0000000000000000000000000000000000000000", + "0x1234567890123456789012345678901234567890", + ], +) +def test_mixed_case_address_accepted(address: str) -> None: + assert TypeAdapter(MixedCaseAddress).validate_python(address) == address + + +@pytest.mark.parametrize("address", [WRONG_CHECKSUM, UPPERCASE]) +def test_mixed_case_address_wrong_checksum(address: str) -> None: + with pytest.raises(ValidationError) as e: + TypeAdapter(MixedCaseAddress).validate_python(address) + message = e.value.errors()[0]["msg"] + assert "invalid EIP-55 checksum" in message + assert address in message + assert CHECKSUMMED in message + + +@pytest.mark.parametrize( + "address", + [ + "0xb426", + "b426b5AE61c23fF1B901a8AD1f1A3921D1E9D2F1", + "0xZZ26b5AE61c23fF1B901a8AD1f1A3921D1E9D2F1", + CHECKSUMMED.lower() + "\n", + ], +) +def test_mixed_case_address_wrong_shape(address: str) -> None: + with pytest.raises(ValidationError) as e: + TypeAdapter(MixedCaseAddress).validate_python(address) + assert "expected a 20 bytes, hexadecimal Ethereum address" in e.value.errors()[0]["msg"] + + +def test_deployment_wrong_checksum() -> None: + with pytest.raises(ValidationError, match="invalid EIP-55 checksum"): + InputDeployment(chainId=1, address=WRONG_CHECKSUM) diff --git a/tests/resources/lint/calldata-two-deployments.json b/tests/resources/lint/calldata-two-deployments.json new file mode 100644 index 00000000..b80e34da --- /dev/null +++ b/tests/resources/lint/calldata-two-deployments.json @@ -0,0 +1,36 @@ +{ + "$schema": "https://eips.ethereum.org/assets/eip-7730/erc7730-v2.schema.json", + "context": { + "$id": "Test Token", + "contract": { + "deployments": [ + { "chainId": 1, "address": "0x0000000000000000000000000000000000000001" }, + { "chainId": 137, "address": "0x0000000000000000000000000000000000000002" } + ] + } + }, + "metadata": { + "owner": "Test", + "info": { "url": "https://example.com", "deploymentDate": "2024-01-01T00:00:00Z" }, + "token": { "ticker": "TST", "name": "Test Token", "decimals": 18 }, + "contractName": "Test Token" + }, + "display": { + "formats": { + "transfer(address _to, uint256 _value)": { + "intent": "Send", + "fields": [ + { "path": "#._value", "label": "Amount", "format": "tokenAmount", "params": { "tokenPath": "@.to" } }, + { "path": "#._to", "label": "To", "format": "addressName", "params": { "types": ["eoa"] } } + ] + }, + "approve(address _spender, uint256 _value)": { + "intent": "Approve", + "fields": [ + { "path": "#._spender", "label": "Spender", "format": "addressName", "params": { "types": ["contract"] } }, + { "path": "#._value", "label": "Amount", "format": "tokenAmount", "params": { "tokenPath": "@.to" } } + ] + } + } + } +} diff --git a/tests/v2/convert/resolved/test_address.py b/tests/v2/convert/resolved/test_address.py new file mode 100644 index 00000000..864102a1 --- /dev/null +++ b/tests/v2/convert/resolved/test_address.py @@ -0,0 +1,154 @@ +""" +Address checksum checks on the way from the input to the resolved descriptor. + +An address written in a parameter is validated by the input model. An address that comes from a constant, or that is +the literal value of a field, is only seen by the resolver, so these tests go through the whole conversion. +""" + +from typing import Any + +import pytest + +from erc7730.common.output import ListOutputAdder +from erc7730.convert.resolved.v2.address import resolved_address, resolved_addresses +from erc7730.convert.resolved.v2.convert_erc7730_input_to_resolved import ERC7730InputToResolved +from erc7730.model.input.v2.descriptor import InputERC7730Descriptor + +CHECKSUMMED = "0xb426b5AE61c23fF1B901a8AD1f1A3921D1E9D2F1" +WRONG_CHECKSUM = "0xb426B5aE61c23Ff1b901A8ad1F1A3921D1E9D2f1" +OTHER_WRONG_CHECKSUM = "0xDAC17F958D2ee523a2206206994597C13D831ec7" + +# an ERC-7930 interoperable address: longer than 20 bytes, so not an address for the checksum check +INTEROPERABLE_ADDRESS = "0x00010000010114dac17f958d2ee523a2206206994597c13d831ec7" + +# fields with an address parameter, written with a path to the constant `addr` +CONSTANT = "$.metadata.constants.addr" +FIELDS_WITH_ADDRESS_PARAMETER = { + "token": {"path": "amount", "format": "tokenAmount", "params": {"token": CONSTANT}}, + "nativeCurrencyAddress": {"path": "amount", "format": "tokenAmount", "params": {"nativeCurrencyAddress": CONSTANT}}, + "senderAddress": {"path": "to", "format": "addressName", "params": {"senderAddress": CONSTANT}}, + "callee": {"path": "data", "format": "calldata", "params": {"callee": CONSTANT}}, + "spender": {"path": "data", "format": "calldata", "params": {"callee": CHECKSUMMED, "spender": CONSTANT}}, + "collection": {"path": "amount", "format": "nftName", "params": {"collection": CONSTANT}}, +} + +# formats whose field value is an address +ADDRESS_FORMATS: dict[str, dict[str, Any]] = { + "addressName": {"format": "addressName", "params": {"types": ["eoa"]}}, + "tokenTicker": {"format": "tokenTicker"}, +} + + +def _descriptor(field: dict[str, Any], constants: dict[str, Any] | None = None) -> dict[str, Any]: + """A contract descriptor with one field, which may refer to the constants.""" + return { + "context": { + "$id": "test", + "contract": {"deployments": [{"chainId": 1, "address": "0x0000000000000000000000000000000000000001"}]}, + }, + "metadata": {"owner": "Test", "constants": constants or {}}, + "display": { + "formats": { + "transfer(address to,uint256 amount,bytes data)": { + "intent": "Transfer", + "fields": [{"label": "Field", **field}], + } + } + }, + } + + +def _convert(descriptor: dict[str, Any]) -> ListOutputAdder: + out = ListOutputAdder() + ERC7730InputToResolved().convert(InputERC7730Descriptor.model_validate(descriptor, strict=False), out) + return out + + +@pytest.mark.parametrize("address", [CHECKSUMMED, CHECKSUMMED.lower()]) +def test_resolved_address_accepted(address: str) -> None: + out = ListOutputAdder() + assert resolved_address(address, out) == address + assert out.outputs == [] + + +@pytest.mark.parametrize( + "value,expected_error", + [ + (WRONG_CHECKSUM, "invalid EIP-55 checksum"), + ("0xb426", "expected a 20 bytes, hexadecimal Ethereum address"), + (42, "expected a 20 bytes, hexadecimal Ethereum address"), + ], +) +def test_resolved_address_rejected(value: object, expected_error: str) -> None: + out = ListOutputAdder() + assert resolved_address(value, out) is None + assert [o.title for o in out.outputs] == ["Invalid address"] + assert expected_error in out.outputs[0].message + + +def test_resolved_addresses_accepted() -> None: + out = ListOutputAdder() + assert resolved_addresses([CHECKSUMMED, CHECKSUMMED.lower()], out) == [CHECKSUMMED, CHECKSUMMED.lower()] + assert out.outputs == [] + + +def test_resolved_addresses_stops_at_first_error() -> None: + out = ListOutputAdder() + assert resolved_addresses([CHECKSUMMED, WRONG_CHECKSUM, OTHER_WRONG_CHECKSUM], out) is None + assert len(out.outputs) == 1 + assert WRONG_CHECKSUM in out.outputs[0].message + + +@pytest.mark.parametrize("param", FIELDS_WITH_ADDRESS_PARAMETER) +def test_constant_address_accepted(param: str) -> None: + out = _convert(_descriptor(FIELDS_WITH_ADDRESS_PARAMETER[param], {"addr": CHECKSUMMED})) + assert out.outputs == [] + + +@pytest.mark.parametrize("param", FIELDS_WITH_ADDRESS_PARAMETER) +def test_constant_address_wrong_checksum(param: str) -> None: + out = _convert(_descriptor(FIELDS_WITH_ADDRESS_PARAMETER[param], {"addr": WRONG_CHECKSUM})) + # the converter may add a generic error at the parameter after the specific one + assert out.outputs[0].title == "Invalid address" + assert "invalid EIP-55 checksum" in out.outputs[0].message + assert CHECKSUMMED in out.outputs[0].message + + +def test_constant_address_in_list_wrong_checksum() -> None: + field = { + "path": "amount", + "format": "tokenAmount", + "params": {"nativeCurrencyAddress": ["0x0000000000000000000000000000000000000002", CONSTANT]}, + } + out = _convert(_descriptor(field, {"addr": WRONG_CHECKSUM})) + assert out.outputs[0].title == "Invalid address" + assert WRONG_CHECKSUM in out.outputs[0].message + + +def test_literal_parameter_wrong_checksum_rejected_by_model() -> None: + field = {"path": "amount", "format": "tokenAmount", "params": {"token": WRONG_CHECKSUM}} + with pytest.raises(ValueError, match="invalid EIP-55 checksum"): + InputERC7730Descriptor.model_validate(_descriptor(field), strict=False) + + +@pytest.mark.parametrize("fmt", ADDRESS_FORMATS) +@pytest.mark.parametrize("value", [WRONG_CHECKSUM, CONSTANT]) +def test_field_value_wrong_checksum(fmt: str, value: str) -> None: + """The value of a field is a scalar for the input model, so only the resolver can check it.""" + out = _convert(_descriptor({"value": value, **ADDRESS_FORMATS[fmt]}, {"addr": WRONG_CHECKSUM})) + assert out.outputs[0].title == "Invalid address" + assert "invalid EIP-55 checksum" in out.outputs[0].message + + +@pytest.mark.parametrize("fmt", ADDRESS_FORMATS) +@pytest.mark.parametrize("value", [CHECKSUMMED, CONSTANT]) +def test_field_value_accepted(fmt: str, value: str) -> None: + out = _convert(_descriptor({"value": value, **ADDRESS_FORMATS[fmt]}, {"addr": CHECKSUMMED})) + assert out.outputs == [] + + +@pytest.mark.parametrize("value", [INTEROPERABLE_ADDRESS, CONSTANT]) +def test_interoperable_address_value_is_not_checked(value: str) -> None: + field = {"value": value, "format": "interoperableAddressName", "params": {"types": ["eoa"]}} + out = _convert(_descriptor(field, {"addr": INTEROPERABLE_ADDRESS})) + assert out.outputs == [] diff --git a/tests/v2/lint/conftest.py b/tests/v2/lint/conftest.py new file mode 100644 index 00000000..84c98188 --- /dev/null +++ b/tests/v2/lint/conftest.py @@ -0,0 +1,27 @@ +from collections.abc import Callable + +import pytest + +from erc7730.common import client +from erc7730.common.output import ListOutputAdder +from erc7730.lint.v2.lint import lint_all +from erc7730.model.abi import ABI +from tests.files import TEST_RESOURCES + +# a descriptor with deployments on chains 1 and 137, and formats for transfer and approve with named parameters +TWO_DEPLOYMENTS = TEST_RESOURCES / "lint" / "calldata-two-deployments.json" + +LintDescriptor = Callable[[Callable[[int], list[ABI]], bool], ListOutputAdder] + + +@pytest.fixture +def lint_descriptor(monkeypatch: pytest.MonkeyPatch) -> LintDescriptor: + """Lint the two deployments descriptor, with reference ABIs (or failures) provided per chain id.""" + + def lint(abis_of: Callable[[int], list[ABI]], require_verified: bool = False) -> ListOutputAdder: + monkeypatch.setattr(client, "get_contract_abis", lambda chain_id, contract_address: abis_of(chain_id)) + out = ListOutputAdder() + lint_all([TWO_DEPLOYMENTS], out, require_verified=require_verified) + return out + + return lint diff --git a/tests/v2/lint/test_lint_require_verified.py b/tests/v2/lint/test_lint_require_verified.py new file mode 100644 index 00000000..3d6d9d44 --- /dev/null +++ b/tests/v2/lint/test_lint_require_verified.py @@ -0,0 +1,54 @@ +from collections.abc import Callable + +import pytest + +from erc7730.common import client +from erc7730.common.output import ListOutputAdder, Output +from erc7730.model.abi import ABI +from tests.v2.lint.conftest import LintDescriptor + +NOT_VERIFIED = client.ContractNotVerifiedError("contract 0x1 on chain 1 is not verified on Sourcify") +IMPLEMENTATION_NOT_VERIFIED = client.ProxyImplementationNotVerifiedError("contract 0x1 on chain 1 is a proxy") +CHAIN_NOT_SUPPORTED = client.ChainNotSupportedError("chain 1 is not supported by Sourcify") +RATE_LIMITED = Exception("Sourcify rate limit exceeded, please retry") + + +def raising(error: Exception) -> Callable[[int], list[ABI]]: + def abis_of(chain_id: int) -> list[ABI]: + raise error + + return abis_of + + +def level_of(out: ListOutputAdder, title: str) -> Output.Level: + return next(output.level for output in out.outputs if output.title == title) + + +@pytest.mark.parametrize( + ("error", "title", "default_level"), + [ + (NOT_VERIFIED, "Contract not verified", Output.Level.WARNING), + (IMPLEMENTATION_NOT_VERIFIED, "Proxy implementation not verified", Output.Level.WARNING), + (CHAIN_NOT_SUPPORTED, "Chain not supported", Output.Level.INFO), + ], +) +def test_unverified_contract_is_an_error_only_when_required( + lint_descriptor: LintDescriptor, error: Exception, title: str, default_level: Output.Level +) -> None: + out = lint_descriptor(raising(error), False) + assert level_of(out, title) == default_level + assert not out.has_errors + + out = lint_descriptor(raising(error), True) + assert level_of(out, title) == Output.Level.ERROR + assert out.has_errors + + +def test_fetch_failure_is_an_error_only_when_verified_is_required(lint_descriptor: LintDescriptor) -> None: + out = lint_descriptor(raising(RATE_LIMITED), False) + assert level_of(out, "Could not fetch ABI") == Output.Level.WARNING + assert not out.has_errors + + out = lint_descriptor(raising(RATE_LIMITED), True) + assert level_of(out, "Could not fetch ABI") == Output.Level.ERROR + assert out.has_errors diff --git a/tests/v2/lint/test_lint_validate_all_deployments.py b/tests/v2/lint/test_lint_validate_all_deployments.py new file mode 100644 index 00000000..3e6c064f --- /dev/null +++ b/tests/v2/lint/test_lint_validate_all_deployments.py @@ -0,0 +1,44 @@ +from erc7730.common import client +from erc7730.common.abi import parse_signature +from erc7730.common.output import ListOutputAdder +from erc7730.model.abi import ABI +from tests.v2.lint.conftest import LintDescriptor + +# the functions the descriptor declares formats for +TRANSFER = parse_signature("transfer(address _to, uint256 _value)") +APPROVE = parse_signature("approve(address _spender, uint256 _value)") +# same functions with different parameter names, so display field paths do not match +TRANSFER_RENAMED = parse_signature("transfer(address to, uint256 value)") +APPROVE_RENAMED = parse_signature("approve(address spender, uint256 value)") + + +def titles(out: ListOutputAdder, title: str) -> int: + return sum(1 for output in out.outputs if output.title == title) + + +def test_deployments_with_the_same_abi_are_validated_once(lint_descriptor: LintDescriptor) -> None: + out = lint_descriptor(lambda chain_id: [TRANSFER, APPROVE], False) + assert titles(out, "Deployment ABIs differ") == 0 + assert titles(out, "Invalid display field") == 0 + + +def test_deployments_with_different_abis_are_reported_and_each_validated(lint_descriptor: LintDescriptor) -> None: + def abis_of(chain_id: int) -> list[ABI]: + return [TRANSFER, APPROVE] if chain_id == 1 else [TRANSFER_RENAMED, APPROVE_RENAMED] + + out = lint_descriptor(abis_of, False) + assert titles(out, "Deployment ABIs differ") == 1 + # the chain 137 ABI names the parameters differently, so the 4 display fields are invalid against it only + assert titles(out, "Invalid display field") == 4 + + +def test_a_deployment_that_cannot_be_fetched_does_not_stop_the_others(lint_descriptor: LintDescriptor) -> None: + def abis_of(chain_id: int) -> list[ABI]: + if chain_id == 1: + raise client.ContractNotVerifiedError("contract 0x1 on chain 1 is not verified on Sourcify") + return [TRANSFER_RENAMED, APPROVE_RENAMED] + + out = lint_descriptor(abis_of, False) + assert titles(out, "Contract not verified") == 1 + assert titles(out, "Deployment ABIs differ") == 0 + assert titles(out, "Invalid display field") == 4