151 lines
4.4 KiB
Python
151 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import secrets
|
|
from functools import wraps
|
|
|
|
from flask import Flask, jsonify, request, send_from_directory
|
|
|
|
try:
|
|
from bitcoinrpc.authproxy import AuthServiceProxy
|
|
except ImportError: # pragma: no cover
|
|
raise SystemExit("Missing python-bitcoinrpc. Install with: pip install python-bitcoinrpc")
|
|
|
|
|
|
RIN_RPC_HOST = os.environ.get("RIN_RPC_HOST", "127.0.0.1")
|
|
RIN_RPC_PORT = int(os.environ.get("RIN_RPC_PORT", "9556"))
|
|
RIN_RPC_USER = os.environ.get("RIN_RPC_USER", "rinrpc")
|
|
RIN_RPC_PASSWORD = os.environ.get("RIN_RPC_PASSWORD", "745ce784d5d537fc06105a1b935b7657903cfc71a5fb3b90")
|
|
RIN_WALLET_NAME = os.environ.get("RIN_WALLET_NAME", "main")
|
|
|
|
API_TOKEN = os.environ.get("RIN_WEB_WALLET_TOKEN")
|
|
if not API_TOKEN:
|
|
API_TOKEN = secrets.token_urlsafe(32)
|
|
print("[web-wallet] No RIN_WEB_WALLET_TOKEN provided. Generated one for this session.")
|
|
print(f"[web-wallet] API token: {API_TOKEN}")
|
|
|
|
|
|
def create_rpc_client() -> AuthServiceProxy:
|
|
url = f"http://{RIN_RPC_USER}:{RIN_RPC_PASSWORD}@{RIN_RPC_HOST}:{RIN_RPC_PORT}/wallet/{RIN_WALLET_NAME}"
|
|
return AuthServiceProxy(url, timeout=15)
|
|
|
|
|
|
def require_token(view_func):
|
|
@wraps(view_func)
|
|
def wrapper(*args, **kwargs):
|
|
header = request.headers.get("Authorization", "")
|
|
if not header.startswith("Bearer "):
|
|
return jsonify({"error": "missing_token"}), 401
|
|
token = header.split(" ", 1)[1]
|
|
if token != API_TOKEN:
|
|
return jsonify({"error": "invalid_token"}), 403
|
|
return view_func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
app = Flask(__name__, static_folder="static", static_url_path="")
|
|
|
|
|
|
def rpc_call(method: str, *params):
|
|
try:
|
|
rpc = create_rpc_client()
|
|
return rpc.__getattr__(method)(*params)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise RuntimeError(str(exc)) from exc
|
|
|
|
|
|
@app.route("/api/session", methods=["GET"])
|
|
def session_info():
|
|
return jsonify({
|
|
"wallet": RIN_WALLET_NAME,
|
|
"host": RIN_RPC_HOST,
|
|
})
|
|
|
|
|
|
@app.route("/api/address", methods=["POST"])
|
|
@require_token
|
|
def create_address():
|
|
data = request.get_json(silent=True) or {}
|
|
label = data.get("label", "")
|
|
try:
|
|
address = rpc_call("getnewaddress", label)
|
|
return jsonify({"address": address})
|
|
except RuntimeError as exc:
|
|
return jsonify({"error": str(exc)}), 500
|
|
|
|
|
|
@app.route("/api/balance", methods=["GET"])
|
|
@require_token
|
|
def get_balance():
|
|
try:
|
|
info = rpc_call("getwalletinfo")
|
|
confirmed = info.get("balance", 0)
|
|
unconfirmed = info.get("unconfirmed_balance", 0)
|
|
immature = info.get("immature_balance", 0)
|
|
total = confirmed + unconfirmed + immature
|
|
return jsonify({
|
|
"confirmed": confirmed,
|
|
"unconfirmed": unconfirmed,
|
|
"immature": immature,
|
|
"total": total,
|
|
})
|
|
except RuntimeError as exc:
|
|
return jsonify({"error": str(exc)}), 500
|
|
|
|
|
|
@app.route("/api/send", methods=["POST"])
|
|
@require_token
|
|
def send():
|
|
payload = request.get_json(force=True)
|
|
address = payload.get("address")
|
|
amount = payload.get("amount")
|
|
subtract_fee = bool(payload.get("subtractFee", True))
|
|
|
|
if not address or not isinstance(amount, (int, float)):
|
|
return jsonify({"error": "invalid_request"}), 400
|
|
|
|
if amount <= 0:
|
|
return jsonify({"error": "amount_must_be_positive"}), 400
|
|
|
|
try:
|
|
txid = rpc_call("sendtoaddress", address, float(amount), "", "", subtract_fee)
|
|
return jsonify({"txid": txid})
|
|
except RuntimeError as exc:
|
|
status = 400 if "Invalid RinCoin address" in str(exc) else 500
|
|
return jsonify({"error": str(exc)}), status
|
|
|
|
|
|
@app.route("/api/transactions", methods=["GET"])
|
|
@require_token
|
|
def list_transactions():
|
|
count = int(request.args.get("count", 10))
|
|
try:
|
|
txs = rpc_call("listtransactions", "*", count, 0, True)
|
|
return jsonify({"transactions": txs})
|
|
except RuntimeError as exc:
|
|
return jsonify({"error": str(exc)}), 500
|
|
|
|
|
|
@app.route("/")
|
|
def root():
|
|
return send_from_directory(app.static_folder, "index.html")
|
|
|
|
|
|
@app.route("/<path:path>")
|
|
def static_proxy(path):
|
|
return send_from_directory(app.static_folder, path)
|
|
|
|
|
|
def main():
|
|
port = int(os.environ.get("RIN_WEB_WALLET_PORT", "8787"))
|
|
app.run(host="127.0.0.1", port=port, debug=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|