From c606c31a3c6e126a64338a6dfd356951bf616daa Mon Sep 17 00:00:00 2001 From: Alexandru Popenta Date: Tue, 9 Dec 2025 15:56:17 +0200 Subject: [PATCH] update contract verify/unverify to work with new service --- multiversx_sdk_cli/cli_contracts.py | 52 +++++++++++- multiversx_sdk_cli/contract_verification.py | 90 +++++++++++---------- 2 files changed, 98 insertions(+), 44 deletions(-) diff --git a/multiversx_sdk_cli/cli_contracts.py b/multiversx_sdk_cli/cli_contracts.py index 539c9238..417998c4 100644 --- a/multiversx_sdk_cli/cli_contracts.py +++ b/multiversx_sdk_cli/cli_contracts.py @@ -28,7 +28,10 @@ from multiversx_sdk_cli.config import get_config_for_network_providers from multiversx_sdk_cli.config_env import MxpyEnv from multiversx_sdk_cli.constants import NUMBER_OF_SHARDS -from multiversx_sdk_cli.contract_verification import trigger_contract_verification +from multiversx_sdk_cli.contract_verification import ( + trigger_contract_verification, + trigger_contract_verification_from_existing, +) from multiversx_sdk_cli.docker import is_docker_installed, run_docker from multiversx_sdk_cli.errors import BadUsage, DockerMissingError, QueryContractError from multiversx_sdk_cli.ux import show_warning @@ -160,6 +163,34 @@ def setup_parser(args: list[str], subparsers: Any) -> Any: ) sub.set_defaults(func=verify) + sub = cli_shared.add_command_subparser( + subparsers, + "contract", + "verify-from-existing", + "Verify the authenticity of the code of a deployed Smart Contract from an already verified Smart Contract", + ) + + _add_contract_arg(sub) + sub.add_argument( + "--verified-contract", + required=True, + help="the bech32 address of the already verified contract", + ) + sub.add_argument( + "--verifier-url", + required=True, + help="the url of the service that validates the contract", + ) + sub.add_argument( + "--skip-confirmation", + "-y", + dest="skip_confirmation", + action="store_true", + default=False, + help="can be used to skip the confirmation prompt", + ) + sub.set_defaults(func=verify_from_existing) + sub = cli_shared.add_command_subparser( subparsers, "contract", @@ -568,6 +599,23 @@ def verify(args: Any) -> None: logger.info("Contract verification request completed!") +def verify_from_existing(args: Any) -> None: + if not args.skip_confirmation: + response = input( + "Are you sure you want to verify the contract? This will publish the contract's source code, which will be displayed on the MultiversX Explorer (y/n): " + ) + if response.lower() != "y": + logger.info("Contract verification cancelled.") + return + + contract = Address.new_from_bech32(args.contract) + verified_contract = Address.new_from_bech32(args.verified_contract) + verifier_url = args.verifier_url + + trigger_contract_verification_from_existing(contract, verified_contract, verifier_url) + logger.info("Contract verification request completed!") + + def unverify(args: Any) -> None: account = cli_shared.prepare_account(args) contract: str = args.contract @@ -593,7 +641,7 @@ def unverify(args: Any) -> None: headers = {"Content-type": "application/json"} response = requests.delete(verifier_url, json=request_payload, headers=headers) logger.info(f"Your request to unverify contract {contract} was submitted.") - print(response.json().get("message")) + utils.dump_out_json(response.json()) def do_reproducible_build(args: Any): diff --git a/multiversx_sdk_cli/contract_verification.py b/multiversx_sdk_cli/contract_verification.py index ab2c5c11..abb8e3a6 100644 --- a/multiversx_sdk_cli/contract_verification.py +++ b/multiversx_sdk_cli/contract_verification.py @@ -11,9 +11,6 @@ from multiversx_sdk_cli.errors import KnownError from multiversx_sdk_cli.utils import dump_out_json, read_json_file -HTTP_REQUEST_TIMEOUT = 408 -HTTP_SUCCESS = 200 - logger = logging.getLogger("cli.contracts.verifier") @@ -94,26 +91,31 @@ def trigger_contract_verification( request_dictionary = contract_verification.to_dictionary() url = f"{verifier_url}/verifier" - status_code, message, data = _do_post(url, request_dictionary) - - if status_code == HTTP_REQUEST_TIMEOUT: - task_id = data.get("taskId", "") - - if task_id: - query_status_with_task_id(verifier_url, task_id) - else: - dump_out_json(data) - elif status_code != HTTP_SUCCESS: - dump_out_json(data) - raise KnownError(f"Cannot verify contract: {message}") - else: - status = data.get("status", "") - if status: - logger.info(f"Task status: {status}") - dump_out_json(data) - else: - task_id = data.get("taskId", "") - query_status_with_task_id(verifier_url, task_id) + response = _do_post(url, request_dictionary) + + task_id: str = response.get("taskId", "") + if not task_id: + raise KnownError("No task ID received from the verifier.") + + logger.info(f"Contract verification triggered successfully. Task ID: {task_id}") + query_status_with_task_id(verifier_url, task_id) + + +def trigger_contract_verification_from_existing(contract: Address, verified_contract: Address, verifier_url: str): + payload = { + "contract": contract.to_bech32(), + "existingVerifiedContract": verified_contract.to_bech32(), + } + + url = f"{verifier_url}/verifier/from-existing" + response = _do_post(url, payload) + + task_id: str = response.get("taskId", "") + if not task_id: + raise KnownError("No task ID received from the verifier.") + + logger.info(f"Contract verification triggered successfully. Task ID: {task_id}") + query_status_with_task_id(verifier_url, task_id) def _create_request_signature(account: IAccount, contract_address: Address, request_payload: bytes) -> bytes: @@ -128,42 +130,46 @@ def query_status_with_task_id(url: str, task_id: str, interval: int = 10): old_status = "" while True: - _, _, response = _do_get(f"{url}/tasks/{task_id}") + response = _do_get(f"{url}/tasks/{task_id}") + try: + response.raise_for_status() + except requests.HTTPError as error: + data = response.json() + message = data.get("message", str(error)) + raise KnownError(f"Cannot verify contract: {message}", error) + + response = response.json() status = response.get("status", "") - if status == "finished": + if status == "error": + logger.error("Verification failed!") + dump_out_json(response) + break + elif status == "finished": logger.info("Verification finished!") dump_out_json(response) break elif status != old_status: logger.info(f"Task status: {status}") - dump_out_json(response) old_status = status time.sleep(interval) -def _do_post(url: str, payload: Any) -> tuple[int, str, dict[str, Any]]: +def _do_post(url: str, payload: Any) -> dict[str, str]: logger.debug(f"_do_post() to {url}") response = requests.post(url, json=payload) - try: + response.raise_for_status() + except requests.HTTPError as error: data = response.json() - message = data.get("message", "") - return response.status_code, message, data - except Exception as error: - logger.error(f"Erroneous response from {url}: {response.text}") - raise KnownError(f"Cannot parse response from {url}", error) + message = data.get("message", str(error)) + raise KnownError(f"Cannot verify contract: {message}", error) + return response.json() -def _do_get(url: str) -> tuple[int, str, dict[str, Any]]: + +def _do_get(url: str) -> requests.Response: logger.debug(f"_do_get() from {url}") response = requests.get(url) - - try: - data = response.json() - message = data.get("message", "") - return response.status_code, message, data - except Exception as error: - logger.error(f"Erroneous response from {url}: {response.text}") - raise KnownError(f"Cannot parse response from {url}", error) + return response