xrp.py

import json
import logging
import random
import requests
from utils import format_as_padded_hex


def pick_random_payment_tx(rpc_url, txs) -> dict:
    tx_type = ""
    while tx_type != "Payment":
        tx_hash = random.choice(txs)
        logging.info(f"Checking transaction {tx_hash}")
        payload = {
            "method": "tx",
            "params": [
                {
                    "transaction": tx_hash,
                    "binary": False,
                    "api_version": 2,
                }
            ],
        }
        tx = requests.post(rpc_url, json=payload).json()
        tx_type = tx.get("result").get("tx_json").get("TransactionType")

        if tx_type == "Payment":
            return tx_hash

    return


def verifier_request(verifier_url, api_key, tx_id, attestation_type="Payment"):
    url = f"{verifier_url}/verifier/xrp/{attestation_type}/prepareRequest"
    headers = {"x-api-key": api_key, "Content-Type": "application/json"}
    payload = {
        "attestationType": format_as_padded_hex(attestation_type),
        "sourceId": format_as_padded_hex("XRP"),
        "requestBody": {"transactionId": tx_id, "inUtxo": "0", "utxo": "0"},
    }

    response = requests.post(url, headers=headers, json=payload)

    return json.loads(response.text)


def get_abi_bytes(rpc, verifier):
    payload = {
        "method": "ledger",
        "params": [{"ledger_index": "validated", "transactions": True}],
    }
    txs = (
        requests.post(rpc["url"], json=payload)
        .json()
        .get("result", {})
        .get("ledger", {})
        .get("transactions", [])
    )
    tx_is_valid = False
    while not tx_is_valid:
        tx_hash = pick_random_payment_tx(rpc["url"], txs)
        verifier_res = verifier_request(verifier["url"], verifier["api_key"], tx_hash)
        if verifier_res["status"] != "VALID":
            logging.error("Verifier request is not valid")
        else:
            tx_is_valid = True

    return verifier_res["abiEncodedRequest"]