contract_verification.py

import hashlib
import json
import logging
import time
from pathlib import Path
from typing import Any, Optional, Protocol

import requests
from multiversx_sdk import Address, Message

from multiversx_sdk_cli.errors import KnownError
from multiversx_sdk_cli.utils import dump_out_json, read_json_file

logger = logging.getLogger("cli.contracts.verifier")


# fmt: off
class IAccount(Protocol):
    def sign_message(self, message: Message) -> bytes:
        ...
# fmt: on


class ContractVerificationRequest:
    def __init__(
        self,
        contract: Address,
        source_code: dict[str, Any],
        signature: bytes,
        docker_image: str,
        contract_variant: Optional[str],
    ) -> None:
        self.contract = contract
        self.source_code = source_code
        self.signature = signature
        self.docker_image = docker_image
        self.contract_variant = contract_variant

    def to_dictionary(self) -> dict[str, Any]:
        return {
            "signature": self.signature.hex(),
            "payload": {
                "contract": self.contract.to_bech32(),
                "dockerImage": self.docker_image,
                "sourceCode": self.source_code,
                "contractVariant": self.contract_variant,
            },
        }


class ContractVerificationPayload:
    def __init__(
        self,
        contract: Address,
        source_code: dict[str, Any],
        docker_image: str,
        contract_variant: Optional[str],
    ) -> None:
        self.contract = contract
        self.source_code = source_code
        self.docker_image = docker_image
        self.contract_variant = contract_variant

    def serialize(self) -> str:
        payload = {
            "contract": self.contract.to_bech32(),
            "dockerImage": self.docker_image,
            "sourceCode": self.source_code,
            "contractVariant": self.contract_variant,
        }

        return json.dumps(payload, separators=(",", ":"))


def trigger_contract_verification(
    packaged_source: Path,
    owner: IAccount,
    contract: Address,
    verifier_url: str,
    docker_image: str,
    contract_variant: Optional[str],
):
    source_code = read_json_file(packaged_source)

    payload = ContractVerificationPayload(contract, source_code, docker_image, contract_variant).serialize()
    signature = _create_request_signature(owner, contract, payload.encode())
    contract_verification = ContractVerificationRequest(
        contract, source_code, signature, docker_image, contract_variant
    )

    request_dictionary = contract_verification.to_dictionary()

    url = f"{verifier_url}/verifier"
    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:
    hashed_payload: str = hashlib.sha256(request_payload).hexdigest()
    raw_data_to_sign = f"{contract_address.to_bech32()}{hashed_payload}"

    return account.sign_message(Message(raw_data_to_sign.encode()))


def query_status_with_task_id(url: str, task_id: str, interval: int = 10):
    logger.info("Please wait while we verify your contract. This may take a while.")
    old_status = ""

    while True:
        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 == "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}")
            old_status = status

        time.sleep(interval)


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", str(error))
        raise KnownError(f"Cannot verify contract: {message}", error)

    return response.json()


def _do_get(url: str) -> requests.Response:
    logger.debug(f"_do_get() from {url}")
    response = requests.get(url)
    return response