proxmox/scripts/deployment/solana-transfer-native.py

#!/usr/bin/env python3
"""
Transfer native SOL (lamports) on Solana mainnet-beta (or any RPC you pass).

Loads ``SOLANA_RPC_URL`` and ``SOLANA_KEYPAIR_PATH`` from the environment when
set (after ``source scripts/lib/load-project-env.sh``). Submits via
``solana_jsonrpc.send_transaction_wire`` (``scripts/lib/solana_jsonrpc.py``) so
RPCs that return only a signature string for ``sendTransaction`` do not hit
``solana-py``'s ``SendTransactionResp`` parser (which can panic on ``missing field 'data'``).

Install (venv recommended)::

    pip install -r scripts/lib/requirements-solana-ops.txt
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import sys
from pathlib import Path

# ``scripts/lib`` is not a Python package; load ``solana_jsonrpc`` by path.
_LIB = Path(__file__).resolve().parents[1] / "lib"
if str(_LIB) not in sys.path:
    sys.path.insert(0, str(_LIB))

import solana_jsonrpc  # noqa: E402


def _load_keypair(path: Path):
    with path.open() as f:
        raw = json.load(f)
    if not isinstance(raw, list) or len(raw) != 64:
        raise SystemExit("keypair JSON must be a 64-element byte array (Solana CLI format)")
    from solders.keypair import Keypair

    return Keypair.from_bytes(bytes(raw))


def main() -> None:
    p = argparse.ArgumentParser(description="Send native SOL via JSON-RPC (robust sendTransaction parsing).")
    p.add_argument("--to", required=True, help="Destination base58 pubkey")
    p.add_argument(
        "--lamports",
        type=int,
        help="Amount to send (excludes fee; payer pays fee separately). Omit with --sweep-all",
    )
    p.add_argument(
        "--sweep-all",
        action="store_true",
        help="Send entire balance minus 5000 lamports legacy fee reserve",
    )
    p.add_argument("--fee-lamports", type=int, default=5000, help="Reserved for fee when using --sweep-all")
    p.add_argument("--rpc-url", default=os.environ.get("SOLANA_RPC_URL", "").strip())
    p.add_argument(
        "--keypair",
        type=Path,
        default=Path(os.environ.get("SOLANA_KEYPAIR_PATH", "").strip() or "."),
        help="Solana CLI JSON keypair path (default: SOLANA_KEYPAIR_PATH)",
    )
    p.add_argument("--skip-preflight", action="store_true")
    p.add_argument(
        "--dry-run",
        action="store_true",
        help="Print base64 wire and exit without sending",
    )
    p.add_argument(
        "--no-wait",
        action="store_true",
        help="Do not poll getSignatureStatuses after send (default: wait up to 90s)",
    )
    args = p.parse_args()

    try:
        from solders.hash import Hash
        from solders.pubkey import Pubkey
        from solders.system_program import TransferParams, transfer
        from solders.transaction import Transaction
    except ImportError:
        print(
            "Missing dependency: install with\n"
            "  pip install -r scripts/lib/requirements-solana-ops.txt",
            file=sys.stderr,
        )
        raise SystemExit(2) from None

    if not args.rpc_url:
        print("Set SOLANA_RPC_URL or pass --rpc-url", file=sys.stderr)
        raise SystemExit(2)
    if not args.keypair.is_file():
        print(f"Keypair not found: {args.keypair}", file=sys.stderr)
        raise SystemExit(2)

    kp = _load_keypair(args.keypair)
    dest = Pubkey.from_string(args.to)
    src = kp.pubkey()

    if args.dry_run:
        if args.sweep_all:
            raise SystemExit("--dry-run requires explicit --lamports (no balance query)")
        if args.lamports is None:
            raise SystemExit("Pass --lamports N with --dry-run")
        send_lamports = args.lamports
    else:
        bal = solana_jsonrpc.get_balance_lamports(args.rpc_url, str(src))
        if args.sweep_all:
            send_lamports = bal - args.fee_lamports
            if send_lamports <= 0:
                raise SystemExit("Nothing to sweep after fee reserve")
        elif args.lamports is not None:
            send_lamports = args.lamports
            if send_lamports <= 0:
                raise SystemExit("--lamports must be positive")
            if bal < send_lamports + args.fee_lamports:
                raise SystemExit(
                    f"Insufficient balance: have {bal} lamports, need {send_lamports + args.fee_lamports}"
                )
        else:
            raise SystemExit("Pass --lamports N or --sweep-all")

    bh_str = solana_jsonrpc.get_latest_blockhash(args.rpc_url)
    bh = Hash.from_string(bh_str)
    ix = transfer(TransferParams(from_pubkey=src, to_pubkey=dest, lamports=send_lamports))
    tx = Transaction.new_signed_with_payer([ix], src, [kp], bh)
    wire = bytes(tx)

    if args.dry_run:
        print("blockhash", bh_str)
        print("wire_b64", base64.b64encode(wire).decode("ascii"))
        return

    sig = solana_jsonrpc.send_transaction_wire(
        args.rpc_url,
        wire,
        skip_preflight=args.skip_preflight,
        preflight_commitment="confirmed",
    )
    print(sig)
    if not args.no_wait:
        st = solana_jsonrpc.wait_until_signature_confirmed(args.rpc_url, sig)
        print("confirmationStatus", st.get("confirmationStatus"), "slot", st.get("slot"), file=sys.stderr)


if __name__ == "__main__":
    main()