Host-customized fork of https://github.com/tecnovert/basicswap/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

156 lines
4.8 KiB

# -*- coding: utf-8 -*-
# Copyright (c) 2020-2024 tecnovert
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.
import os
import json
import shlex
import urllib
import traceback
import subprocess
from xmlrpc.client import (
Fault,
Transport,
SafeTransport,
)
from .util import jsonDecimal
class Jsonrpc():
# __getattr__ complicates extending ServerProxy
def __init__(self, uri, transport=None, encoding=None, verbose=False,
allow_none=False, use_datetime=False, use_builtin_types=False,
*, context=None):
# establish a "logical" server connection
# get the url
parsed = urllib.parse.urlparse(uri)
if parsed.scheme not in ('http', 'https'):
raise OSError('unsupported XML-RPC protocol')
self.__host = parsed.netloc
self.__handler = parsed.path
if not self.__handler:
self.__handler = '/RPC2'
if transport is None:
handler = SafeTransport if parsed.scheme == 'https' else Transport
extra_kwargs = {}
transport = handler(use_datetime=use_datetime,
use_builtin_types=use_builtin_types,
**extra_kwargs)
self.__transport = transport
self.__encoding = encoding or 'utf-8'
self.__verbose = verbose
self.__allow_none = allow_none
self.__request_id = 1
def close(self):
if self.__transport is not None:
self.__transport.close()
def json_request(self, method, params):
try:
connection = self.__transport.make_connection(self.__host)
headers = self.__transport._extra_headers[:]
request_body = {
'method': method,
'params': params,
'id': self.__request_id
}
connection.putrequest('POST', self.__handler)
headers.append(('Content-Type', 'application/json'))
headers.append(('User-Agent', 'jsonrpc'))
self.__transport.send_headers(connection, headers)
self.__transport.send_content(connection, json.dumps(request_body, default=jsonDecimal).encode('utf-8'))
self.__request_id += 1
resp = connection.getresponse()
return resp.read()
except Fault:
raise
except Exception:
# All unexpected errors leave connection in
# a strange state, so we clear it.
self.__transport.close()
raise
def callrpc(rpc_port, auth, method, params=[], wallet=None, host='127.0.0.1'):
try:
url = 'http://{}@{}:{}/'.format(auth, host, rpc_port)
if wallet is not None:
url += 'wallet/' + urllib.parse.quote(wallet)
x = Jsonrpc(url)
v = x.json_request(method, params)
x.close()
r = json.loads(v.decode('utf-8'))
except Exception as ex:
traceback.print_exc()
raise ValueError('RPC server error ' + str(ex) + ', method: ' + method)
if 'error' in r and r['error'] is not None:
raise ValueError('RPC error ' + str(r['error']))
return r['result']
def openrpc(rpc_port, auth, wallet=None, host='127.0.0.1'):
try:
url = 'http://{}@{}:{}/'.format(auth, host, rpc_port)
if wallet is not None:
url += 'wallet/' + urllib.parse.quote(wallet)
return Jsonrpc(url)
except Exception as ex:
traceback.print_exc()
raise ValueError('RPC error ' + str(ex))
def callrpc_cli(bindir, datadir, chain, cmd, cli_bin='particl-cli', wallet=None):
cli_bin = os.path.join(bindir, cli_bin)
args = [cli_bin, ]
if chain != 'mainnet':
args.append('-' + chain)
args.append('-datadir=' + datadir)
if wallet is not None:
args.append('-rpcwallet=' + wallet)
args += shlex.split(cmd)
p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out = p.communicate()
if len(out[1]) > 0:
raise ValueError('RPC error ' + str(out[1]))
r = out[0].decode('utf-8').strip()
try:
r = json.loads(r)
except Exception:
pass
return r
def make_rpc_func(port, auth, wallet=None, host='127.0.0.1'):
port = port
auth = auth
wallet = wallet
host = host
def rpc_func(method, params=None, wallet_override=None):
nonlocal port, auth, wallet, host
return callrpc(port, auth, method, params, wallet if wallet_override is None else wallet_override, host)
return rpc_func
def escape_rpcauth(auth_str: str) -> str:
username, password = auth_str.split(':', 1)
password = urllib.parse.quote(password, safe='')
return f'{username}:{password}'