update contract verify/unverify to work with new service

Author: popenta

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):

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