soroban_bls_signature.py

# pyright: reportMissingImports=false
"""Invoke the Soroban BLS signature custom account example.

This script follows the BLS example contract from:
https://github.com/stellar/soroban-examples/tree/main/bls_signature

Install the extra dependency first:

    pip install py-ecc

The deployed contract must be initialized with the aggregate G1 public key
derived from the three BLS secret scalars in ``SKS`` below. If the contract's
``agg_pk`` was created from different secret scalars, ``__check_auth`` will
reject the signatures generated by this script.
"""

import time
from hashlib import sha256

from py_ecc.bls.hash_to_curve import hash_to_G2  # type: ignore[import-not-found]
from py_ecc.optimized_bls12_381.optimized_curve import (  # type: ignore[import-not-found,import-untyped]
    curve_order,
    multiply,
    normalize,
)

from stellar_sdk import (
    InvokeHostFunction,
    Keypair,
    Network,
    SorobanServer,
    TransactionBuilder,
    scval,
)
from stellar_sdk import xdr as stellar_xdr
from stellar_sdk.auth import authorization_payload_hash, authorize_entry
from stellar_sdk.exceptions import PrepareTransactionException
from stellar_sdk.soroban_rpc import GetTransactionStatus, SendTransactionStatus

# The upstream example contract is both the custom account and the counter.
BLS_CONTRACT = "CBO4WJR2UG7N7JK2CK2YJCLXEK5T6FV4DM3MXWQS5YITZ5RTOTUFWWBO"
SOURCE_SK = "SAAPYAPTTRZMCUZFPG3G66V4ZMHTK4TWA6NS7U4F7Z3IMUD52EK4DDEV"

RPC_SERVER_URL = "https://soroban-testnet.stellar.org"
NETWORK_PASSPHRASE = Network.TESTNET_NETWORK_PASSPHRASE

DST = b"BLSSIG-V01-CS01-with-BLS12381G2_XMD:SHA-256_SSWU_RO_"

# These are the three BLS secret scalars used to initialize the example
# contract's aggregate public key. Keep them in sync with the deployed
# contract's constructor input.
SKS = [
    int.from_bytes(bytes.fromhex(secret), "big")
    for secret in [
        "7197d39d834199eb903efad5fa7cec3085d13e3af73e2ae978b294fb1e26fead",
        "628347027e5246340041b81c88ddb1aa854be4ffffc13378897961ac91de1589",
        "1c9043bdd86eb1157fc9fecb0d599c5ac88d94d510508fb0b1b7f2e64d1afa20",
    ]
]
SK_SUM = sum(SKS) % curve_order


def fp_to_bytes(value):
    return int(value).to_bytes(48, "big")


def fp2_to_bytes(value):
    c0, c1 = value.coeffs
    # Soroban encodes BLS12-381 Fp2 values as c1 || c0.
    return fp_to_bytes(c1) + fp_to_bytes(c0)


def g2_uncompressed(point):
    x, y = normalize(point)
    return fp2_to_bytes(x) + fp2_to_bytes(y)


def bls_sign(payload32: bytes) -> bytes:
    message_point = hash_to_G2(payload32, DST, sha256)  # type: ignore
    signature = g2_uncompressed(multiply(message_point, SK_SUM))
    if len(signature) != 192:
        raise RuntimeError(f"Expected a 192-byte BLS signature, got {len(signature)}.")
    return signature


def bls_signer(preimage: stellar_xdr.HashIDPreimage) -> stellar_xdr.SCVal:
    payload = authorization_payload_hash(preimage)
    return scval.to_bytes(bls_sign(payload))


source = Keypair.from_secret(SOURCE_SK)
server = SorobanServer(RPC_SERVER_URL)
account = server.load_account(source.public_key)

tx = (
    TransactionBuilder(account, NETWORK_PASSPHRASE, base_fee=1000)
    .set_timeout(300)
    .append_invoke_contract_function_op(
        contract_id=BLS_CONTRACT,
        function_name="increment",
        parameters=[],
    )
    .build()
)

# First simulation records the auth entry skeleton required by the contract.
sim1 = server.simulate_transaction(tx)
if sim1.error:
    raise RuntimeError(f"Initial simulation failed: {sim1.error}")
if not sim1.results or not sim1.results[0].auth:
    raise RuntimeError(f"Initial simulation did not return auth entries: {sim1}")

# Fill the auth entry with the exact SCVal shape expected by the BLS contract.
valid_until_ledger = sim1.latest_ledger + 20
op = tx.transaction.operations[0]
if not isinstance(op, InvokeHostFunction):
    raise TypeError("Expected an InvokeHostFunction operation.")
op.auth = [
    authorize_entry(
        sim1.results[0].auth[0],
        bls_signer,
        valid_until_ledger,
        NETWORK_PASSPHRASE,
    )
]

# Simulate again with auth present. This lets __check_auth run successfully and
# adds auth-dependent storage, such as Owners, to the final footprint.
sim2 = server.simulate_transaction(tx)
if sim2.error:
    raise RuntimeError(f"Authorized simulation failed: {sim2.error}")

# prepare_transaction keeps op.auth because it is already populated.
try:
    tx = server.prepare_transaction(tx, sim2)
except PrepareTransactionException as exc:
    raise RuntimeError(
        f"prepare_transaction failed: {exc.simulate_transaction_response}"
    ) from exc

tx.sign(source)
print(f"signed XDR:\n{tx.to_xdr()}")

send_resp = server.send_transaction(tx)
print(f"sent: {send_resp}")
if send_resp.status != SendTransactionStatus.PENDING:
    raise RuntimeError(f"send_transaction failed: {send_resp}")

while True:
    print("waiting...")
    get_resp = server.get_transaction(send_resp.hash)
    if get_resp.status != GetTransactionStatus.NOT_FOUND:
        break
    time.sleep(3)

if get_resp.status != GetTransactionStatus.SUCCESS:
    raise RuntimeError(f"Transaction failed: {get_resp}")
if get_resp.result_meta_xdr is None:
    raise RuntimeError("Successful transaction did not include result meta XDR.")

meta = stellar_xdr.TransactionMeta.from_xdr(get_resp.result_meta_xdr)
body = meta.v4 if meta.v4 is not None else meta.v3
if body is None or body.soroban_meta is None:
    raise RuntimeError("Transaction meta did not include Soroban metadata.")
if body.soroban_meta.return_value is None:
    raise RuntimeError("Transaction meta did not include a Soroban return value.")

counter = scval.from_uint32(body.soroban_meta.return_value)
print(f"counter = {counter}")