stack_wallet/lib/electrumx_rpc/rpc.dart

413 lines
11 KiB
Dart

/*
* This file is part of Stack Wallet.
*
* Copyright (c) 2023 Cypher Stack
* All Rights Reserved.
* The code is distributed under GPLv3 license, see LICENSE file for details.
* Generated by Cypher Stack on 2023-05-26
*
*/
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:flutter/foundation.dart';
import 'package:mutex/mutex.dart';
import 'package:stackwallet/exceptions/json_rpc/json_rpc_exception.dart';
import 'package:stackwallet/utilities/logger.dart';
import 'package:stackwallet/utilities/prefs.dart';
import 'package:tor_ffi_plugin/socks_socket.dart';
// Json RPC class to handle connecting to electrumx servers
class JsonRPC {
JsonRPC({
required this.host,
required this.port,
this.useSSL = false,
this.connectionTimeout = const Duration(seconds: 60),
required ({InternetAddress host, int port})? proxyInfo,
});
final bool useSSL;
final String host;
final int port;
final Duration connectionTimeout;
({InternetAddress host, int port})? proxyInfo;
final _requestMutex = Mutex();
final _JsonRPCRequestQueue _requestQueue = _JsonRPCRequestQueue();
Socket? _socket;
SOCKSSocket? _socksSocket;
StreamSubscription<List<int>>? _subscription;
void _dataHandler(List<int> data) {
_requestQueue.nextIncompleteReq.then((req) {
if (req != null) {
req.appendDataAndCheckIfComplete(data);
if (req.isComplete) {
_onReqCompleted(req);
}
} else {
Logging.instance.log(
"_dataHandler found a null req!",
level: LogLevel.Warning,
);
}
});
}
void _errorHandler(Object error, StackTrace trace) {
_requestQueue.nextIncompleteReq.then((req) {
if (req != null) {
req.completer.completeError(error, trace);
_onReqCompleted(req);
}
});
}
void _doneHandler() {
disconnect(reason: "JsonRPC _doneHandler() called");
}
void _onReqCompleted(_JsonRPCRequest req) {
_requestQueue.remove(req).then((_) {
// attempt to send next request
_sendNextAvailableRequest();
});
}
void _sendNextAvailableRequest() {
_requestQueue.nextIncompleteReq.then((req) {
if (req != null) {
if (!Prefs.instance.useTor) {
if (_socket == null) {
Logging.instance.log(
"JsonRPC _sendNextAvailableRequest attempted with"
" _socket=null on $host:$port",
level: LogLevel.Error,
);
}
// \r\n required by electrumx server
_socket!.write('${req.jsonRequest}\r\n');
} else {
if (_socksSocket == null) {
Logging.instance.log(
"JsonRPC _sendNextAvailableRequest attempted with"
" _socksSocket=null on $host:$port",
level: LogLevel.Error,
);
}
// \r\n required by electrumx server
_socksSocket?.write('${req.jsonRequest}\r\n');
}
// TODO different timeout length?
req.initiateTimeout(
onTimedOut: () {
_onReqCompleted(req);
},
);
}
});
}
Future<JsonRPCResponse> request(
String jsonRpcRequest,
Duration requestTimeout,
) async {
await _requestMutex.protect(() async {
if (!Prefs.instance.useTor) {
if (_socket == null) {
Logging.instance.log(
"JsonRPC request: opening socket $host:$port",
level: LogLevel.Info,
);
await _connect().timeout(requestTimeout, onTimeout: () {
throw Exception("Request timeout: $jsonRpcRequest");
});
}
} else {
if (_socksSocket == null) {
Logging.instance.log(
"JsonRPC request: opening SOCKS socket to $host:$port",
level: LogLevel.Info,
);
await _connect().timeout(requestTimeout, onTimeout: () {
throw Exception("Request timeout: $jsonRpcRequest");
});
}
}
});
final req = _JsonRPCRequest(
jsonRequest: jsonRpcRequest,
requestTimeout: requestTimeout,
completer: Completer<JsonRPCResponse>(),
);
final future = req.completer.future.onError(
(error, stackTrace) async {
await disconnect(
reason: "return req.completer.future.onError: $error\n$stackTrace",
);
return JsonRPCResponse(
exception: error is JsonRpcException
? error
: JsonRpcException(
"req.completer.future.onError: $error\n$stackTrace",
),
);
},
);
// if this is the only/first request then send it right away
await _requestQueue.add(
req,
onInitialRequestAdded: _sendNextAvailableRequest,
);
return future;
}
/// DO NOT set [ignoreMutex] to true unless fully aware of the consequences
Future<void> disconnect({
required String reason,
bool ignoreMutex = false,
}) async {
if (ignoreMutex) {
await _disconnectHelper(reason: reason);
} else {
await _requestMutex.protect(() async {
await _disconnectHelper(reason: reason);
});
}
}
Future<void> _disconnectHelper({required String reason}) async {
await _subscription?.cancel();
_subscription = null;
_socket?.destroy();
_socket = null;
await _socksSocket?.close();
_socksSocket = null;
// clean up remaining queue
await _requestQueue.completeRemainingWithError(
"JsonRPC disconnect() called with reason: \"$reason\"",
);
}
Future<void> _connect() async {
// ignore mutex is set to true here as _connect is already called within
// the mutex.protect block. Setting to false here leads to a deadlock
await disconnect(
reason: "New connection requested",
ignoreMutex: true,
);
if (!Prefs.instance.useTor) {
if (useSSL) {
_socket = await SecureSocket.connect(
host,
port,
timeout: connectionTimeout,
onBadCertificate: (_) => true,
); // TODO do not automatically trust bad certificates
} else {
_socket = await Socket.connect(
host,
port,
timeout: connectionTimeout,
);
}
_subscription = _socket!.listen(
_dataHandler,
onError: _errorHandler,
onDone: _doneHandler,
cancelOnError: true,
);
} else {
if (proxyInfo == null) {
throw JsonRpcException(
"JsonRPC.connect failed with useTor=${Prefs.instance.useTor} and proxyInfo is null");
}
// instantiate a socks socket at localhost and on the port selected by the tor service
_socksSocket = await SOCKSSocket.create(
proxyHost: proxyInfo!.host.address,
proxyPort: proxyInfo!.port,
sslEnabled: useSSL,
);
try {
Logging.instance.log(
"JsonRPC.connect(): connecting to SOCKS socket at $proxyInfo (SSL $useSSL)...",
level: LogLevel.Info);
await _socksSocket?.connect();
Logging.instance.log(
"JsonRPC.connect(): connected to SOCKS socket at $proxyInfo...",
level: LogLevel.Info);
} catch (e) {
Logging.instance.log(
"JsonRPC.connect(): failed to connect to SOCKS socket at $proxyInfo, $e",
level: LogLevel.Error);
throw JsonRpcException(
"JsonRPC.connect(): failed to connect to SOCKS socket at $proxyInfo, $e");
}
try {
Logging.instance.log(
"JsonRPC.connect(): connecting to $host:$port over SOCKS socket at $proxyInfo...",
level: LogLevel.Info);
await _socksSocket?.connectTo(host, port);
Logging.instance.log(
"JsonRPC.connect(): connected to $host:$port over SOCKS socket at $proxyInfo",
level: LogLevel.Info);
} catch (e) {
Logging.instance.log(
"JsonRPC.connect(): failed to connect to $host over tor proxy at $proxyInfo, $e",
level: LogLevel.Error);
throw JsonRpcException(
"JsonRPC.connect(): failed to connect to tor proxy, $e");
}
_subscription = _socksSocket!.listen(
_dataHandler,
onError: _errorHandler,
onDone: _doneHandler,
cancelOnError: true,
);
}
return;
}
}
class _JsonRPCRequestQueue {
final _lock = Mutex();
final List<_JsonRPCRequest> _rq = [];
Future<void> add(
_JsonRPCRequest req, {
VoidCallback? onInitialRequestAdded,
}) async {
return await _lock.protect(() async {
_rq.add(req);
if (_rq.length == 1) {
onInitialRequestAdded?.call();
}
});
}
Future<bool> remove(_JsonRPCRequest req) async {
return await _lock.protect(() async {
final result = _rq.remove(req);
return result;
});
}
Future<_JsonRPCRequest?> get nextIncompleteReq async {
return await _lock.protect(() async {
int removeCount = 0;
_JsonRPCRequest? returnValue;
for (final req in _rq) {
if (req.isComplete) {
removeCount++;
} else {
returnValue = req;
break;
}
}
_rq.removeRange(0, removeCount);
return returnValue;
});
}
Future<void> completeRemainingWithError(
String error, {
StackTrace? stackTrace,
}) async {
await _lock.protect(() async {
for (final req in _rq) {
if (!req.isComplete) {
req.completer.completeError(Exception(error), stackTrace);
}
}
_rq.clear();
});
}
Future<bool> get isEmpty async {
return await _lock.protect(() async {
return _rq.isEmpty;
});
}
}
class _JsonRPCRequest {
// 0x0A is newline
// https://electrumx-spesmilo.readthedocs.io/en/latest/protocol-basics.html
static const int separatorByte = 0x0A;
final String jsonRequest;
final Completer<JsonRPCResponse> completer;
final Duration requestTimeout;
final List<int> _responseData = [];
_JsonRPCRequest({
required this.jsonRequest,
required this.completer,
required this.requestTimeout,
});
void appendDataAndCheckIfComplete(List<int> data) {
_responseData.addAll(data);
if (data.last == separatorByte) {
try {
final response = json.decode(String.fromCharCodes(_responseData));
completer.complete(JsonRPCResponse(data: response));
} catch (e, s) {
Logging.instance.log(
"JsonRPC json.decode: $e\n$s",
level: LogLevel.Error,
);
completer.completeError(e, s);
}
}
}
void initiateTimeout({
required VoidCallback onTimedOut,
}) {
Future<void>.delayed(requestTimeout).then((_) {
if (!isComplete) {
completer.complete(
JsonRPCResponse(
data: null,
exception: JsonRpcException(
"_JsonRPCRequest timed out: $jsonRequest",
),
),
);
}
onTimedOut.call();
});
}
bool get isComplete => completer.isCompleted;
}
class JsonRPCResponse {
final dynamic data;
final JsonRpcException? exception;
JsonRPCResponse({this.data, this.exception});
}