From bdd25da3873ffbc0dfa26c72266ef71ba06a9841 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 26 Oct 2021 23:03:40 -0400 Subject: [PATCH 1/8] solution: Websocket Ethereum proxy fix: #110 --- .../io/emeraldpay/dshackle/ProxyStarter.kt | 4 +- .../emeraldpay/dshackle/config/ProxyConfig.kt | 1 + .../emeraldpay/dshackle/proxy/BaseHandler.kt | 86 ++++++++++ .../emeraldpay/dshackle/proxy/HttpHandler.kt | 86 ++++++++++ .../emeraldpay/dshackle/proxy/ProxyServer.kt | 104 ++---------- .../emeraldpay/dshackle/proxy/ReadRpcJson.kt | 70 ++++---- .../dshackle/proxy/WebsocketHandler.kt | 155 ++++++++++++++++++ .../dshackle/rpc/NativeSubscribe.kt | 4 +- .../dshackle/proxy/BaseHandlerSpec.groovy | 105 ++++++++++++ ...rverSpec.groovy => HttpHandlerSpec.groovy} | 145 ++++++++-------- .../dshackle/proxy/ReadRpcJsonSpec.groovy | 8 + .../proxy/WebsocketHandlerSpec.groovy | 117 +++++++++++++ 12 files changed, 681 insertions(+), 204 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy rename src/test/groovy/io/emeraldpay/dshackle/proxy/{ProxyServerSpec.groovy => HttpHandlerSpec.groovy} (67%) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt b/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt index f1f248c8..8d05044f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt @@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.proxy.ProxyServer import io.emeraldpay.dshackle.proxy.ReadRpcJson import io.emeraldpay.dshackle.proxy.WriteRpcJson import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.rpc.NativeSubscribe import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -37,6 +38,7 @@ class ProxyStarter( @Autowired private val readRpcJson: ReadRpcJson, @Autowired private val writeRpcJson: WriteRpcJson, @Autowired private val nativeCall: NativeCall, + @Autowired private val nativeSubscribe: NativeSubscribe, @Autowired private val tlsSetup: TlsSetup, @Autowired private val accessHandlerHttp: AccessHandlerHttp, // depend on Monitoring, declared here just to ensure it's properly initialized before the Proxy @@ -54,7 +56,7 @@ class ProxyStarter( log.debug("Proxy server is not configured") return } - val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall, tlsSetup, accessHandlerHttp.factory) + val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, tlsSetup, accessHandlerHttp.factory) server.start() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt index 2c53e5cb..fc01f00d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt @@ -28,6 +28,7 @@ open class ProxyConfig { } var enabled: Boolean = true + var websocketEnabled: Boolean = true /** * Host to bind server. Default: 127.0.0.1 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt new file mode 100644 index 00000000..8ca2fdbd --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt @@ -0,0 +1,86 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.proxy + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.grpc.Chain +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import java.util.concurrent.TimeUnit + +abstract class BaseHandler( + private val writeRpcJson: WriteRpcJson, + private val nativeCall: NativeCall, + private val requestMetrics: ProxyServer.RequestMetricsFactory, +) { + + companion object { + private val log = LoggerFactory.getLogger(BaseHandler::class.java) + } + + fun execute(chain: Chain, call: ProxyCall, handler: AccessHandlerHttp.RequestHandler): Publisher { + // return empty response for empty request + if (call.items.isEmpty()) { + return if (call.type == ProxyCall.RpcType.BATCH) { + Mono.just("[]") + } else { + Mono.just("") + } + } + val jsons = execute(chain, call.items, handler) + .transform(writeRpcJson.toJsons(call)) + return if (call.type == ProxyCall.RpcType.SINGLE) { + jsons.next() + } else { + jsons.transform(writeRpcJson.asArray()) + } + } + + fun execute(chain: Chain, items: List, handler: AccessHandlerHttp.RequestHandler): Flux { + val startTime = System.currentTimeMillis() + // during the execution we know only ID of the call, so we use it to find the origin call and associated metrics + val metricById = { id: Int -> + items.find { it.id == id }?.let { item -> + requestMetrics.get(chain, item.method) + } + } + val request = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChain(Common.ChainRef.forNumber(chain.id)) + .addAllItems(items) + .build() + handler.onRequest(request) + return nativeCall + .nativeCallResult(Mono.just(request)) + .doOnNext { + metricById(it.id)?.let { metrics -> + metrics.requestMetric.increment() + metrics.callMetric.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) + } + handler.onResponse(it) + } + .doOnError { + // when error happened the whole flux is stopped and no result is produced, so we should mark all the requests as failed + items.forEach { item -> + requestMetrics.get(chain, item.method).errorMetric.increment() + } + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt new file mode 100644 index 00000000..6ebe5177 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt @@ -0,0 +1,86 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.proxy + +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.etherjar.rpc.RpcException +import io.emeraldpay.grpc.Chain +import io.netty.buffer.ByteBuf +import io.netty.buffer.Unpooled +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import org.springframework.http.HttpHeaders +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.netty.http.server.HttpServerRequest +import reactor.netty.http.server.HttpServerResponse +import java.util.function.BiFunction + +/** + * Responds to HTTP requests made to the Ethereum Proxy Server + */ +class HttpHandler( + private val readRpcJson: ReadRpcJson, + writeRpcJson: WriteRpcJson, + nativeCall: NativeCall, + private val accessHandler: AccessHandlerHttp.HandlerFactory, + requestMetrics: ProxyServer.RequestMetricsFactory, +) : BaseHandler(writeRpcJson, nativeCall, requestMetrics) { + + companion object { + private val log = LoggerFactory.getLogger(HttpHandler::class.java) + } + + fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { + return BiFunction { req, resp -> + // handle access events + val eventHandler = accessHandler.create(req, routeConfig.blockchain) + val request = req.receive() + .aggregate() + .asByteArray() + val results = processRequest(routeConfig.blockchain, request, eventHandler) + // make sure that the access log handler is closed at the end, so it can render the logs + .doFinally { eventHandler.close() } + resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json") + .send(results) + } + } + + fun processRequest( + chain: Chain, + request: Mono, + handler: AccessHandlerHttp.RequestHandler + ): Flux { + return request + .map(readRpcJson) + .flatMapMany { call -> + execute(chain, call, handler) + } + .onErrorResume(RpcException::class.java) { err -> + val id = err.details?.let { + if (it is JsonRpcResponse.Id) it else JsonRpcResponse.NumberId(-1) + } ?: JsonRpcResponse.NumberId(-1) + + val json = JsonRpcResponse.error(err.code, err.rpcMessage, id) + Mono.just(Global.objectMapper.writeValueAsString(json)) + } + .map { Unpooled.wrappedBuffer(it.toByteArray()) } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index 6e5ab4f9..a16c40a6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -16,36 +16,23 @@ */ package io.emeraldpay.dshackle.proxy -import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.api.proto.Common import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.TlsSetup import io.emeraldpay.dshackle.config.ProxyConfig import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall -import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse -import io.emeraldpay.etherjar.rpc.RpcException +import io.emeraldpay.dshackle.rpc.NativeSubscribe import io.emeraldpay.grpc.Chain import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.Timer -import io.netty.buffer.ByteBuf -import io.netty.buffer.Unpooled import io.netty.channel.ChannelHandler import io.netty.channel.ChannelHandlerContext -import org.reactivestreams.Publisher import org.slf4j.LoggerFactory -import org.springframework.http.HttpHeaders -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono import reactor.netty.http.server.HttpServer -import reactor.netty.http.server.HttpServerRequest -import reactor.netty.http.server.HttpServerResponse import reactor.netty.http.server.HttpServerRoutes import java.util.EnumMap -import java.util.concurrent.TimeUnit import java.util.concurrent.locks.ReentrantReadWriteLock -import java.util.function.BiFunction import kotlin.concurrent.read import kotlin.concurrent.write @@ -57,6 +44,7 @@ class ProxyServer( private val readRpcJson: ReadRpcJson, private val writeRpcJson: WriteRpcJson, private val nativeCall: NativeCall, + private val nativeSubscribe: NativeSubscribe, private val tlsSetup: TlsSetup, private val accessHandler: AccessHandlerHttp.HandlerFactory ) { @@ -92,6 +80,11 @@ class ProxyServer( StandardRequestMetrics() } + private val httpHandler = HttpHandler(readRpcJson, writeRpcJson, nativeCall, accessHandler, requestMetrics) + private val wsHandler: WebsocketHandler? = if (config.websocketEnabled) { + WebsocketHandler(readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, requestMetrics) + } else null + fun start() { if (!config.enabled) { log.debug("Proxy server is not enabled") @@ -116,88 +109,11 @@ class ProxyServer( fun setupRoutes(routes: HttpServerRoutes) { config.routes.forEach { routeConfig -> - routes.post("/" + routeConfig.id, proxy(routeConfig)) - } - } - - fun execute(chain: Chain, call: ProxyCall, handler: AccessHandlerHttp.RequestHandler): Publisher { - // return empty response for empty request - if (call.items.isEmpty()) { - return if (call.type == ProxyCall.RpcType.BATCH) { - Mono.just("[]") - } else { - Mono.just("") + routes.post("/" + routeConfig.id, httpHandler.proxy(routeConfig)) + if (config.websocketEnabled && wsHandler != null) { + routes.ws("/" + routeConfig.id, wsHandler.proxy(routeConfig)) } } - val startTime = System.currentTimeMillis() - // during the execution we know only ID of the call, we use it to find the origin call and associated metrics - val metricById = { id: Int -> - call.items.find { it.id == id }?.let { item -> - requestMetrics.get(chain, item.method) - } - } - val request = BlockchainOuterClass.NativeCallRequest.newBuilder() - .setChain(Common.ChainRef.forNumber(chain.id)) - .addAllItems(call.items) - .build() - handler.onRequest(request) - val jsons = nativeCall - .nativeCallResult(Mono.just(request)) - .doOnNext { - metricById(it.id)?.requestMetric?.increment() - } - .doOnNext { - handler.onResponse(it) - metricById(it.id)?.callMetric?.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS) - } - .doOnError { - // when error happened the whole flux is stopped and no result is produced, so we should mark all the requests as failed - call.items.forEach { item -> - requestMetrics.get(chain, item.method).errorMetric.increment() - } - } - .transform(writeRpcJson.toJsons(call)) - return if (call.type == ProxyCall.RpcType.SINGLE) { - jsons.next() - } else { - jsons.transform(writeRpcJson.asArray()) - } - } - - fun processRequest( - chain: Chain, - request: Mono, - handler: AccessHandlerHttp.RequestHandler - ): Flux { - return request - .map(readRpcJson) - .flatMapMany { call -> - execute(chain, call, handler) - } - .onErrorResume(RpcException::class.java) { err -> - val id = err.details?.let { - if (it is JsonRpcResponse.Id) it else JsonRpcResponse.NumberId(-1) - } ?: JsonRpcResponse.NumberId(-1) - - val json = JsonRpcResponse.error(err.code, err.rpcMessage, id) - Mono.just(Global.objectMapper.writeValueAsString(json)) - } - .map { Unpooled.wrappedBuffer(it.toByteArray()) } - } - - fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { - return BiFunction { req, resp -> - // handle access events - val eventHandler = accessHandler.create(req, routeConfig.blockchain) - val request = req.receive() - .aggregate() - .asByteArray() - val results = processRequest(routeConfig.blockchain, request, eventHandler) - // make sure that the access log handler is closed at the end, so it can render the logs - .doFinally { eventHandler.close() } - resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json") - .send(results) - } } interface RequestMetricsFactory { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt index a5901f4c..feb785e9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt @@ -28,7 +28,6 @@ import org.slf4j.LoggerFactory import org.springframework.stereotype.Service import java.io.IOException import java.util.function.Function -import java.util.stream.Collectors /** * Reader for JSON RPC request @@ -41,11 +40,11 @@ open class ReadRpcJson : Function { private val spaces = " \n\t".toByteArray() } - private val jsonExtractor: Function, RequestJson> + val jsonExtractor: (Map<*, *>) -> RequestJson private val objectMapper: ObjectMapper = Global.objectMapper init { - jsonExtractor = Function { json -> + jsonExtractor = { json -> if (json["id"] == null) { throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID is not set") } @@ -126,33 +125,11 @@ open class ReadRpcJson : Function { * Convert payload to the proxy call details */ override fun apply(data: ByteArray): ProxyCall { - val list: MutableList> + val list: List> try { val type = getType(data) - if (ProxyCall.RpcType.BATCH == type) { - list = objectMapper.readerFor(MutableList::class.java).readValue(data) - } else { - list = ArrayList(1) - val json = objectMapper.readerFor(MutableMap::class.java).readValue>(data) - list.add(json) - } - val context = ProxyCall(type) - // our internal ids for calls - var seq = 0 - val batch = list.stream() - .map>(jsonExtractor) - .map { json -> - val id = seq++ - context.ids[id] = json.id - BlockchainOuterClass.NativeCallItem.newBuilder() - .setId(id) - .setMethod(json.method) - .setPayload(ByteString.copyFrom(objectMapper.writeValueAsBytes(json.params))) - .build() - } - .collect(Collectors.toList()) - context.items.addAll(batch) - return context + list = extract(type, data) + return convertMapToNativeCall(type, list) } catch (e: RpcException) { throw e } catch (e: Exception) { @@ -160,4 +137,41 @@ open class ReadRpcJson : Function { throw RpcException(RpcResponseError.CODE_INVALID_JSON, e.message) } } + + fun extract(type: ProxyCall.RpcType, data: ByteArray): List> { + return if (ProxyCall.RpcType.BATCH == type) { + objectMapper.readerFor(MutableList::class.java).readValue(data) + } else { + val list = ArrayList>(1) + val json = objectMapper.readerFor(MutableMap::class.java).readValue>(data) + list.add(json) + list + } + } + + fun convertMapToNativeCall(type: ProxyCall.RpcType, list: List>): ProxyCall { + return convertToNativeCall(type, list.map(jsonExtractor)) + } + + fun convertToNativeCall(type: ProxyCall.RpcType, list: List>): ProxyCall { + val context = ProxyCall(type) + val batch = convertToNativeCall(0, context, list) + context.items.addAll(batch) + return context + } + + fun convertToNativeCall(seqStart: Int, context: ProxyCall, items: List>): List { + // internal ids for calls + var seq = seqStart + return items + .map { json -> + val id = seq++ + context.ids[id] = json.id + BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(id) + .setMethod(json.method) + .setPayload(ByteString.copyFrom(objectMapper.writeValueAsBytes(json.params))) + .build() + } + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt new file mode 100644 index 00000000..e36007d2 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt @@ -0,0 +1,155 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.proxy + +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.rpc.NativeSubscribe +import io.emeraldpay.etherjar.rpc.json.RequestJson +import io.emeraldpay.etherjar.rpc.json.ResponseJson +import io.emeraldpay.grpc.Chain +import io.netty.buffer.ByteBufInputStream +import io.netty.buffer.Unpooled +import org.apache.commons.lang3.StringUtils +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.netty.http.websocket.WebsocketInbound +import reactor.netty.http.websocket.WebsocketOutbound +import java.util.concurrent.atomic.AtomicLong +import java.util.function.BiFunction + +/** + * Responds to Websocket requests made to the Ethereum Proxy Server + */ +class WebsocketHandler( + private val readRpcJson: ReadRpcJson, + writeRpcJson: WriteRpcJson, + nativeCall: NativeCall, + private val nativeSubscribe: NativeSubscribe, + requestMetrics: ProxyServer.RequestMetricsFactory, +) : BaseHandler(writeRpcJson, nativeCall, requestMetrics) { + + companion object { + private val log = LoggerFactory.getLogger(WebsocketHandler::class.java) + } + + private val subscriptionId = AtomicLong(0) + + fun nextSubscriptionId(): String { + val n = subscriptionId.incrementAndGet() + return StringUtils.leftPad(n.toString(16), 16, "0") + } + + fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { + return BiFunction { req, resp -> + val requests: Flux> = req.aggregateFrames() + .receiveFrames() + .map { ByteBufInputStream(it.content()).readAllBytes() } + .flatMap(this@WebsocketHandler::parseRequest) + + val eventHandler: AccessHandlerHttp.RequestHandler = AccessHandlerHttp.NoOpHandler() + val responses = respond(routeConfig.blockchain, requests, eventHandler) + .map { Unpooled.wrappedBuffer(it.toByteArray()) } + + resp.send(responses) + .then() + } + } + + fun parseRequest(data: ByteArray): Mono> { + // try to parse JSON call. If received an invalid value just silently ignore it, that's what other Ethereum servers do + try { + val type = readRpcJson.getType(data) + // WS is not supposed to have batches, so ignore them too + if (type != ProxyCall.RpcType.SINGLE) { + return Mono.empty() + } + val items = readRpcJson.extract(type, data) + if (items.isEmpty()) { + //empty should never happen for a SINGLE type of request, but anyway, just return nothing + return Mono.empty() + } + return Mono + .just(items.first()) + .map(readRpcJson.jsonExtractor) + .onErrorResume { Mono.empty() } + } catch (t: Throwable) { + return Mono.empty() + } + } + + fun respond(blockchain: Chain, requests: Flux>, eventHandler: AccessHandlerHttp.RequestHandler): Flux { + return requests.flatMap { call -> + val method = call.method + if (method == "eth_subscribe") { + val methodParams = splitMethodParams(call.params) + if (methodParams != null) { + val subscriptionId = nextSubscriptionId() + // first need to respond with ID of the subscription, and the following responses would have it in "subscription" param + val start = ResponseJson().also { + it.id = call.id + it.result = subscriptionId + } + // produce actual responses + val responses = nativeSubscribe + .subscribe(blockchain, methodParams.first, methodParams.second) + .map { event -> + WsSubscriptionResponse(params = WsSubscriptionData(event, subscriptionId)) + } + Flux.concat(Mono.just(start), responses) + .map { Global.objectMapper.writeValueAsString(it) } + } else { + Mono.empty() + } + } else { + val proxyCall = readRpcJson.convertToNativeCall(ProxyCall.RpcType.SINGLE, listOf(call)) + execute(blockchain, proxyCall, eventHandler) + } + } + } + + fun splitMethodParams(params: List): Pair? { + if (params.isEmpty()) { + return null + } + if (params.size == 1) { + return Pair(params.first().toString(), null) + } + if (params.size == 2) { + return Pair(params.first().toString(), params[1]) + } + return null + } + + // classes only to render WebSocket subscription response. + // the difference with standard JSON RPC responses that it + // (1) it doesn't have id on the top level, but rather as part of params, + // and (2) it has the `method` field + data class WsSubscriptionResponse( + val jsonrpc: String = "2.0", + val method: String = "eth_subscription", + val params: WsSubscriptionData, + ) + + data class WsSubscriptionData( + val result: Any?, + val subscription: String + ) +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index fd8ffbe2..63640e3a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt @@ -33,7 +33,7 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono @Service -class NativeSubscribe( +open class NativeSubscribe( @Autowired private val multistreamHolder: MultistreamHolder ) { @@ -81,7 +81,7 @@ class NativeSubscribe( } } - fun subscribe(chain: Chain, method: String, params: Any?): Flux { + open fun subscribe(chain: Chain, method: String, params: Any?): Flux { val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) return (up as EthereumMultistream) .getSubscribe() diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy new file mode 100644 index 00000000..607c1bab --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy @@ -0,0 +1,105 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.proxy + +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.grpc.Chain +import org.jetbrains.annotations.NotNull +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import spock.lang.Specification + +import java.time.Duration + +class BaseHandlerSpec extends Specification { + + def requestHandler = new AccessHandlerHttp.NoOpHandler() + + def "Return empty for empty single call"() { + setup: + def handler = new BaseHandlerImpl(new WriteRpcJson(), Stub(NativeCall), Stub(ProxyServer.RequestMetricsFactory)) + when: + def act = Mono.from(handler.execute(Chain.ETHEREUM, new ProxyCall(ProxyCall.RpcType.SINGLE), requestHandler)) + .block(Duration.ofSeconds(1)) + then: + act == "" + } + + def "Return empty array for empty batch call"() { + setup: + def handler = new BaseHandlerImpl(new WriteRpcJson(), Stub(NativeCall), Stub(ProxyServer.RequestMetricsFactory)) + when: + def act = Mono.from(handler.execute(Chain.ETHEREUM, new ProxyCall(ProxyCall.RpcType.BATCH), requestHandler)) + .block(Duration.ofSeconds(1)) + then: + act == "[]" + } + + def "Execute single call"() { + setup: + def nativeCall = Mock(NativeCall) + def handler = new BaseHandlerImpl(new WriteRpcJson(), nativeCall, Stub(ProxyServer.RequestMetricsFactory)) + + def request = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test") + .setId(0) + .build() + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.items.add(request) + call.ids[0] = 5 + def response = new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) + when: + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler)) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '{"jsonrpc":"2.0","id":5,"result":{"foo": 1}}' + 1 * nativeCall.nativeCallResult(_) >> Flux.fromIterable([response]) + } + + def "Execute batch call with one item"() { + setup: + def nativeCall = Mock(NativeCall) + def handler = new BaseHandlerImpl(new WriteRpcJson(), nativeCall, Stub(ProxyServer.RequestMetricsFactory)) + + def request = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test") + .setId(0) + .build() + def call = new ProxyCall(ProxyCall.RpcType.BATCH) + call.items.add(request) + call.ids[0] = 5 + def response = new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) + when: + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler)) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '[{"jsonrpc":"2.0","id":5,"result":{"foo": 1}}]' + 1 * nativeCall.nativeCallResult(_) >> Flux.fromIterable([response]) + } + + class BaseHandlerImpl extends BaseHandler { + + BaseHandlerImpl(@NotNull WriteRpcJson writeRpcJson, @NotNull NativeCall nativeCall, @NotNull ProxyServer.RequestMetricsFactory requestMetrics) { + super(writeRpcJson, nativeCall, requestMetrics) + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy similarity index 67% rename from src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy index 8bd5d75b..59b1d211 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy @@ -1,6 +1,5 @@ /** - * Copyright (c) 2020 ETCDEV GmbH - * Copyright (c) 2020 EmeraldPay, Inc + * Copyright (c) 2021 EmeraldPay, Inc * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,11 +18,8 @@ package io.emeraldpay.dshackle.proxy import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.TlsSetup -import io.emeraldpay.dshackle.config.ProxyConfig import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall -import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.grpc.Chain @@ -35,64 +31,7 @@ import spock.lang.Specification import java.time.Duration import java.util.function.Function -class ProxyServerSpec extends Specification { - - def "Uses NativeCall"() { - setup: - NativeCall nativeCall = Mock(NativeCall) - def predefined = { a -> Flux.just("hello") } as Function - - WriteRpcJson writeRpcJson = Mock { - 1 * toJsons(_) >> predefined - } - - ProxyServer server = new ProxyServer( - new ProxyConfig(), - new ReadRpcJson(), - writeRpcJson, - nativeCall, - new TlsSetup(TestingCommons.fileResolver()), - new AccessHandlerHttp.NoOpFactory() - ) - - def call = new ProxyCall(ProxyCall.RpcType.SINGLE) - call.ids[1] = 1 - call.items.add( - BlockchainOuterClass.NativeCallItem.newBuilder() - .setMethod("eth_hello") - .build() - ) - when: - def act = server.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler()) - - then: - 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, "".bytes, null)) - StepVerifier.create(act) - .expectNext("hello") - .expectComplete() - .verify(Duration.ofSeconds(1)) - } - - def "Return error on invalid request"() { - setup: - ReadRpcJson read = Mock(ReadRpcJson) { - 1 * apply(_) >> { throw new RpcException(-32123, "test", new JsonRpcResponse.NumberId(4)) } - } - def server = new ProxyServer( - Stub(ProxyConfig), - read, - Stub(WriteRpcJson), Stub(NativeCall), Stub(TlsSetup), - new AccessHandlerHttp.NoOpFactory() - ) - when: - def act = server.processRequest(Chain.ETHEREUM, Mono.just("".bytes), new AccessHandlerHttp.NoOpHandler()) - .map { new String(it.array()) } - then: - StepVerifier.create(act) - .expectNext('{"jsonrpc":"2.0","id":4,"error":{"code":-32123,"message":"test"}}') - .expectComplete() - .verify(Duration.ofSeconds(1)) - } +class HttpHandlerSpec extends Specification { def "Calls access log handler"() { setup: @@ -107,30 +46,78 @@ class ProxyServerSpec extends Specification { .addItems(reqItem) .build() - - ReadRpcJson read = Mock(ReadRpcJson) { - 1 * apply(_) >> new ProxyCall(ProxyCall.RpcType.SINGLE).tap { it.items.add(reqItem) } - } NativeCall nativeCall = Mock(NativeCall) { 1 * nativeCallResult(_) >> Flux.fromIterable([respItem]) } - def handler = Mock(AccessHandlerHttp.RequestHandler.class) - - def server = new ProxyServer( - Stub(ProxyConfig), - read, - new WriteRpcJson(), - nativeCall, - Stub(TlsSetup), - new AccessHandlerHttp.NoOpFactory() + def accessHandler = Mock(AccessHandlerHttp.RequestHandler) + def accessHandlerFactory = Mock(AccessHandlerHttp.HandlerFactory) { + _ * it.create(_,) >> accessHandler + } + def handler = new HttpHandler( + new ReadRpcJson(), new WriteRpcJson(), + nativeCall, accessHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) ) when: - server.processRequest(Chain.ETHEREUM, Mono.just("".bytes), handler) + handler.execute(Chain.ETHEREUM, [reqItem], accessHandler) .blockLast() then: - 1 * handler.onRequest(req) - 1 * handler.onResponse(respItem) + 1 * accessHandler.onRequest(req) + 1 * accessHandler.onResponse(respItem) + } + + def "Return error on invalid request"() { + setup: + ReadRpcJson read = Mock(ReadRpcJson) { + 1 * apply(_) >> { throw new RpcException(-32123, "test", new JsonRpcResponse.NumberId(4)) } + } + + def handler = new HttpHandler( + read, new WriteRpcJson(), + Stub(NativeCall), Stub(AccessHandlerHttp.HandlerFactory), Stub(ProxyServer.RequestMetricsFactory) + ) + when: + + def act = handler.processRequest(Chain.ETHEREUM, Mono.just("".bytes), new AccessHandlerHttp.NoOpHandler()) + .map { new String(it.array()) } + then: + StepVerifier.create(act) + .expectNext('{"jsonrpc":"2.0","id":4,"error":{"code":-32123,"message":"test"}}') + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Uses NativeCall"() { + setup: + NativeCall nativeCall = Mock(NativeCall) + def predefined = { a -> Flux.just("hello") } as Function + + WriteRpcJson writeRpcJson = Mock { + 1 * toJsons(_) >> predefined + } + + + def handler = new HttpHandler( + new ReadRpcJson(), writeRpcJson, + nativeCall, Stub(AccessHandlerHttp.HandlerFactory), Stub(ProxyServer.RequestMetricsFactory) + ) + + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.ids[1] = 1 + call.items.add( + BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_hello") + .build() + ) + when: + def act = handler.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler()) + + then: + 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, "".bytes, null)) + StepVerifier.create(act) + .expectNext("hello") + .expectComplete() + .verify(Duration.ofSeconds(1)) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy index 53b31983..3cab7db4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy @@ -224,4 +224,12 @@ class ReadRpcJsonSpec extends Specification { t.rpcMessage.toLowerCase() == "params must be an array" t.details == new JsonRpcResponse.NumberId(2) } + + def "Error if json is broken"() { + when: + reader.apply('{"id":2, "method":"net_peerCount", "params"'.bytes) + then: + def t = thrown(RpcException) + t.code == -32700 + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy new file mode 100644 index 00000000..35027283 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy @@ -0,0 +1,117 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.proxy + +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.rpc.NativeSubscribe +import io.emeraldpay.etherjar.rpc.json.RequestJson +import io.emeraldpay.grpc.Chain +import reactor.core.publisher.Flux +import spock.lang.Specification + +import java.time.Duration + +class WebsocketHandlerSpec extends Specification { + + def requestHandler = new AccessHandlerHttp.NoOpHandler() + + def "Parse standard RPC request"() { + setup: + def handler = new WebsocketHandler( + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + ) + when: + def act = handler.parseRequest('{"id": 5, "jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["0x100001", false]}'.bytes) + .block(Duration.ofSeconds(1)) + + then: + act.id == 5 + act.method == "eth_getBlockByNumber" + act.params == ["0x100001", false] + } + + def "Parse to empty an invalid request"() { + setup: + def handler = new WebsocketHandler( + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + ) + when: + def act = handler.parseRequest('hello world'.bytes) + .block(Duration.ofSeconds(1)) + + then: + act == null + } + + def "Parse to empty a batch request"() { + setup: + def req1 = '{"id": 5, "jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["0x100001", false]}' + def handler = new WebsocketHandler( + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + ) + when: + def act = handler.parseRequest("[$req1]".bytes) + .block(Duration.ofSeconds(1)) + + then: + act == null + } + + def "Respond to a single call"() { + setup: + def response = new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) + + def nativeCall = Mock(NativeCall) { + 1 * it.nativeCallResult(_) >> Flux.fromIterable([response]) + } + def handler = new WebsocketHandler( + new ReadRpcJson(), new WriteRpcJson(), nativeCall, Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + ) + + def request = new RequestJson("foo_test", [], 2) + when: + def act = handler.respond(Chain.ETHEREUM, Flux.just(request), requestHandler) + .single() + .block(Duration.ofSeconds(1)) + then: + act == '{"jsonrpc":"2.0","id":2,"result":{"foo": 1}}' + } + + def "Respond to a subscription call"() { + setup: + def response1 = [foo: 1] + def response2 = [foo: 2] + + def nativeSubscribe = Mock(NativeSubscribe) { + 1 * it.subscribe(Chain.ETHEREUM, "foo_test", null) >> Flux.fromIterable([response1, response2]) + } + def handler = new WebsocketHandler( + new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), nativeSubscribe, Stub(ProxyServer.RequestMetricsFactory) + ) + + def request = new RequestJson("eth_subscribe", ["foo_test"], 2) + when: + def act = handler.respond(Chain.ETHEREUM, Flux.just(request), requestHandler) + .collectList() + .block(Duration.ofSeconds(1)) + then: + act[0] == '{"jsonrpc":"2.0","id":2,"result":"0000000000000001"}' + act[1] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":1},"subscription":"0000000000000001"}}' + act[2] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":2},"subscription":"0000000000000001"}}' + act.size() == 3 + } +} From dc3c7b1249e03c2f7e820fa2db566ccda4372740 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 26 Oct 2021 23:18:24 -0400 Subject: [PATCH 2/8] problem: ktlint --- .../kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt index e36007d2..567944fe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt @@ -83,7 +83,7 @@ class WebsocketHandler( } val items = readRpcJson.extract(type, data) if (items.isEmpty()) { - //empty should never happen for a SINGLE type of request, but anyway, just return nothing + // empty should never happen for a SINGLE type of request, but anyway, just return nothing return Mono.empty() } return Mono From 133937822c9bb801fabcfebcf24ad978cd13aa16 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 27 Oct 2021 20:34:04 -0400 Subject: [PATCH 3/8] solution: configuration for Websocket proxy --- .../emeraldpay/dshackle/config/ProxyConfigReader.kt | 3 +++ .../dshackle/config/ProxyConfigReaderSpec.groovy | 12 ++++++++++++ src/test/resources/dshackle-proxy-no-ws.yaml | 6 ++++++ 3 files changed, 21 insertions(+) create mode 100644 src/test/resources/dshackle-proxy-no-ws.yaml diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt index e75f80c4..390453f0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt @@ -58,6 +58,9 @@ class ProxyConfigReader : YamlConfigReader(), ConfigReader { getValueAsBool(input, "enabled")?.let { config.enabled = it } + getValueAsBool(input, "websocket")?.let { + config.websocketEnabled = it + } val currentRoutes = HashSet() getList(input, "routes")?.let { routes -> config.routes = routes.value.map { route -> diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy index 3cdbe6dc..6ac07a94 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy @@ -30,6 +30,7 @@ class ProxyConfigReaderSpec extends Specification { then: act.enabled + act.websocketEnabled act.port == 8080 act.host == '127.0.0.1' act.routes.size() == 1 @@ -39,6 +40,17 @@ class ProxyConfigReaderSpec extends Specification { } } + def "Read proxy config with websocket disabled"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("dshackle-proxy-no-ws.yaml") + when: + def act = reader.read(config) + + then: + act.enabled + !act.websocketEnabled + } + def "Read proxy config with two elements"() { setup: def config = this.class.getClassLoader().getResourceAsStream("dshackle-proxy-two.yaml") diff --git a/src/test/resources/dshackle-proxy-no-ws.yaml b/src/test/resources/dshackle-proxy-no-ws.yaml new file mode 100644 index 00000000..faaf01b8 --- /dev/null +++ b/src/test/resources/dshackle-proxy-no-ws.yaml @@ -0,0 +1,6 @@ +proxy: + port: 8080 + websocket: false + routes: + - id: ethereum + blockchain: ethereum \ No newline at end of file From 89eec9958a12555f7f1d967f5be0348731e5e335 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 27 Oct 2021 21:18:58 -0400 Subject: [PATCH 4/8] problem: Proxy doesn't render internal exceptions to JSON RPC errors --- .../emeraldpay/dshackle/proxy/WriteRpcJson.kt | 3 --- .../dshackle/proxy/WriteRpcJsonSpec.groovy | 27 ++++++++++++++++--- 2 files changed, 24 insertions(+), 6 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt index e7c6a691..c0ed6ee4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt @@ -63,9 +63,6 @@ open class WriteRpcJson { Mono.empty() } } - .onErrorContinue { t, _ -> - log.warn("Failed to convert to JSON", t) - } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy index a4f1d13d..7ddf8f09 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy @@ -16,10 +16,8 @@ */ package io.emeraldpay.dshackle.proxy -import com.google.protobuf.ByteString -import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.rpc.NativeCall -import io.emeraldpay.dshackle.test.TestingCommons +import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux import spock.lang.Specification @@ -143,4 +141,27 @@ class WriteRpcJsonSpec extends Specification { act[1] == '{"jsonrpc":"2.0","id":11,"error":{"code":-32002,"message":"oops"}}' act[2] == '{"jsonrpc":"2.0","id":15,"result":{"hash": "0x2484f459dc"}}' } + + def "Write JSON RPC error on exception"() { + setup: + def writer = new WriteRpcJson() { + @Override + String toJson(@NotNull ProxyCall call, @NotNull NativeCall.CallResult response) { + throw new NativeCall.CallFailure(1, new IllegalStateException("TEST")) + } + } + + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.ids[1] = 10 + def data = [ + new NativeCall.CallResult(1, '"0x1"'.bytes, null), + ] + when: + def act = Flux.fromIterable(data) + .transform(writer.toJsons(call)) + .collectList() + .block(Duration.ofSeconds(1)) + then: + act[0] == '{"jsonrpc":"2.0","id":10,"error":{"code":-32003,"message":"TEST"}}' + } } From 92657cf6354dcbe09fd8f7eb0baba5bdd29684e4 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 27 Oct 2021 21:19:32 -0400 Subject: [PATCH 5/8] solution: test ProxyServer --- .../emeraldpay/dshackle/proxy/ProxyServer.kt | 20 ++-- .../dshackle/proxy/ProxyServerSpec.groovy | 92 +++++++++++++++++++ 2 files changed, 106 insertions(+), 6 deletions(-) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index a16c40a6..314a28ff 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -41,12 +41,12 @@ import kotlin.concurrent.write */ class ProxyServer( private var config: ProxyConfig, - private val readRpcJson: ReadRpcJson, - private val writeRpcJson: WriteRpcJson, - private val nativeCall: NativeCall, - private val nativeSubscribe: NativeSubscribe, + readRpcJson: ReadRpcJson, + writeRpcJson: WriteRpcJson, + nativeCall: NativeCall, + nativeSubscribe: NativeSubscribe, private val tlsSetup: TlsSetup, - private val accessHandler: AccessHandlerHttp.HandlerFactory + accessHandler: AccessHandlerHttp.HandlerFactory ) { companion object { @@ -90,7 +90,10 @@ class ProxyServer( log.debug("Proxy server is not enabled") return } - log.info("Listening Proxy on ${config.host}:${config.port}") + log.info("Start HTTP JSON RPC Proxy on ${connectAddress("http")}") + if (config.websocketEnabled) { + log.info("Start Websocket JSON RPC Proxy on ${connectAddress("ws")}") + } var serverBuilder = HttpServer.create() .doOnChannelInit { _, channel, _ -> channel.pipeline().addFirst(errorHandler) @@ -116,6 +119,11 @@ class ProxyServer( } } + fun connectAddress(baseSchema: String): String { + val schema = if (config.tls != null) baseSchema + "s" else baseSchema + return "$schema://${config.host}:${config.port}" + } + interface RequestMetricsFactory { fun get(chain: Chain, method: String): RequestMetrics } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy new file mode 100644 index 00000000..4e225e01 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy @@ -0,0 +1,92 @@ +package io.emeraldpay.dshackle.proxy + +import io.emeraldpay.dshackle.TlsSetup +import io.emeraldpay.dshackle.config.AuthConfig +import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.rpc.NativeSubscribe +import io.emeraldpay.grpc.Chain +import reactor.netty.http.server.HttpServerRoutes +import spock.lang.Specification + +class ProxyServerSpec extends Specification { + + def "Setup routes"() { + setup: + def config1 = new ProxyConfig() + config1.routes = [ + new ProxyConfig.Route("test", Chain.ETHEREUM) + ] + def proxyServer = new ProxyServer( + config1, + new ReadRpcJson(), new WriteRpcJson(), + Stub(NativeCall), Stub(NativeSubscribe), + Stub(TlsSetup), new AccessHandlerHttp.NoOpFactory() + ) + + def routes = Mock(HttpServerRoutes) + when: + proxyServer.setupRoutes(routes) + + then: + 1 * routes.post("/test", _) + 1 * routes.ws("/test", _) + } + + def "Setup routes when WS is disabled"() { + setup: + def config1 = new ProxyConfig() + config1.websocketEnabled = false + config1.routes = [ + new ProxyConfig.Route("test", Chain.ETHEREUM) + ] + def proxyServer = new ProxyServer( + config1, + new ReadRpcJson(), new WriteRpcJson(), + Stub(NativeCall), Stub(NativeSubscribe), + Stub(TlsSetup), new AccessHandlerHttp.NoOpFactory() + ) + + def routes = Mock(HttpServerRoutes) + when: + proxyServer.setupRoutes(routes) + + then: + 1 * routes.post("/test", _) + 0 * routes.ws(_, _) + } + + def "Generate Connect Address"() { + def config1 = new ProxyConfig() + config1.host = "192.168.0.1" + config1.port = 1000 + def proxyServer = new ProxyServer( + config1, + new ReadRpcJson(), new WriteRpcJson(), + Stub(NativeCall), Stub(NativeSubscribe), + Stub(TlsSetup), new AccessHandlerHttp.NoOpFactory() + ) + when: + def act = proxyServer.connectAddress("http") + then: + act == "http://192.168.0.1:1000" + } + + def "Generate Connect Address with TLS"() { + def config1 = new ProxyConfig() + config1.host = "192.168.0.1" + config1.port = 1000 + config1.tls = new AuthConfig.ServerTlsAuth() + def proxyServer = new ProxyServer( + config1, + new ReadRpcJson(), new WriteRpcJson(), + Stub(NativeCall), Stub(NativeSubscribe), + Stub(TlsSetup), new AccessHandlerHttp.NoOpFactory() + ) + when: + def act = proxyServer.connectAddress("ws") + then: + act == "wss://192.168.0.1:1000" + } +} From e001913acb03af42ac820eb308a3df085c26233c Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 28 Oct 2021 23:46:32 -0400 Subject: [PATCH 6/8] solution: AccessLog for Websocket requests --- .../monitoring/accesslog/AccessHandlerGrpc.kt | 2 +- .../monitoring/accesslog/AccessHandlerHttp.kt | 161 ++++++++++++++---- .../dshackle/monitoring/accesslog/Events.kt | 2 +- .../monitoring/accesslog/EventsBuilder.kt | 107 ++++++++++-- .../emeraldpay/dshackle/proxy/ProxyServer.kt | 2 +- .../dshackle/proxy/WebsocketHandler.kt | 25 ++- .../proxy/WebsocketHandlerSpec.groovy | 11 +- 7 files changed, 258 insertions(+), 52 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt index 7e3c849f..aac2d8c7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt @@ -130,7 +130,7 @@ class AccessHandlerGrpc( ): ServerCall.Listener { return process( call, headers, next, - EventsBuilder.NativeSubscribe() as EventsBuilder.RequestReply<*, ReqT, RespT> + EventsBuilder.NativeSubscribe(Events.Channel.GRPC) as EventsBuilder.RequestReply<*, ReqT, RespT> ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerHttp.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerHttp.kt index eb468ddf..503361dc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerHttp.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerHttp.kt @@ -8,6 +8,7 @@ import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import reactor.netty.http.server.HttpServerRequest +import reactor.netty.http.websocket.WebsocketInbound import java.time.Instant import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock @@ -25,6 +26,9 @@ class AccessHandlerHttp( companion object { private val log = LoggerFactory.getLogger(AccessHandlerHttp::class.java) + + private val NO_SUBSCRIBE = NoOnSubscriptionHandler() + private val NO_REQUEST = NoOpHandler() } /** @@ -38,11 +42,21 @@ class AccessHandlerHttp( interface HandlerFactory { fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler + fun start(req: WebsocketInbound, blockchain: Chain): WsHandlerFactory + } + + interface WsHandlerFactory { + fun call(): RequestHandler + fun subscribe(): SubscriptionHandler } class NoOpFactory : HandlerFactory { override fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler { - return NoOpHandler() + return NO_REQUEST + } + + override fun start(req: WebsocketInbound, blockchain: Chain): WsHandlerFactory { + return NO_REQUEST } } @@ -50,6 +64,10 @@ class AccessHandlerHttp( override fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler { return StandardHandler(accessLogWriter, req, blockchain) } + + override fun start(req: WebsocketInbound, blockchain: Chain): WsHandlerFactory { + return StandardWsHandlerFactory(accessLogWriter, req, blockchain) + } } interface RequestHandler { @@ -58,7 +76,12 @@ class AccessHandlerHttp( fun onResponse(callResult: NativeCall.CallResult) } - class NoOpHandler : RequestHandler { + interface SubscriptionHandler { + fun onRequest(request: Pair) + fun onResponse(msgSize: Long) + } + + class NoOpHandler : RequestHandler, WsHandlerFactory { override fun close() { } @@ -67,37 +90,47 @@ class AccessHandlerHttp( override fun onResponse(callResult: NativeCall.CallResult) { } + + override fun call(): RequestHandler { + return this + } + + override fun subscribe(): SubscriptionHandler { + return NO_SUBSCRIBE + } } - class StandardHandler( - private val accessLogWriter: AccessLogWriter, - private val httpRequest: HttpServerRequest, - private val blockchain: Chain - ) : RequestHandler { - - private var request: BlockchainOuterClass.NativeCallRequest? = null - private val responses = ArrayList() - private val updateLock = ReentrantLock() - - override fun close() { - if (request == null) { - return - } - val responseTime = Instant.now() - val builder = EventsBuilder.NativeCall() - builder.withChain(blockchain.id) - builder.start(httpRequest) - builder.onRequest(request!!) - responses - .map { - builder.onReply(it, Events.Channel.JSONRPC).also { item -> - // since for JSON RPC you get a single response then the timestamp of all items included in it must have the same timestamp - item.ts = responseTime - } - } - .let(accessLogWriter::submit) + class NoOnSubscriptionHandler : SubscriptionHandler { + override fun onRequest(request: Pair) { } + override fun onResponse(msgSize: Long) { + } + } + + class StandardWsHandlerFactory( + private val accessLogWriter: AccessLogWriter, + private val wsRequest: WebsocketInbound, + private val blockchain: Chain + ) : WsHandlerFactory { + + override fun call(): RequestHandler { + return WsRequestHandler(accessLogWriter, wsRequest, blockchain) + } + + override fun subscribe(): SubscriptionHandler { + return WsSubscriptionHandler(accessLogWriter, wsRequest, blockchain) + } + } + + abstract class AbstractRequestHandler( + private val accessLogWriter: AccessLogWriter, + private val channel: Events.Channel + ) : RequestHandler { + protected var request: BlockchainOuterClass.NativeCallRequest? = null + protected val responses = ArrayList() + protected val updateLock = ReentrantLock() + override fun onRequest(request: BlockchainOuterClass.NativeCallRequest) { this.request = request } @@ -107,5 +140,75 @@ class AccessHandlerHttp( responses.add(callResult) } } + + fun onClose(builder: EventsBuilder.NativeCall) { + val responseTime = Instant.now() + responses + .map { + builder.onReply(it, channel).also { item -> + // since for JSON RPC you get a single response then the timestamp of all items included in it must have the same timestamp + item.ts = responseTime + } + } + .let(accessLogWriter::submit) + } + } + + class StandardHandler( + accessLogWriter: AccessLogWriter, + private val httpRequest: HttpServerRequest, + private val blockchain: Chain + ) : RequestHandler, AbstractRequestHandler(accessLogWriter, Events.Channel.JSONRPC) { + + override fun close() { + if (request == null) { + return + } + val builder = EventsBuilder.NativeCall() + builder.withChain(blockchain.id) + builder.start(httpRequest) + builder.onRequest(request!!) + onClose(builder) + } + } + + class WsRequestHandler( + accessLogWriter: AccessLogWriter, + private val wsRequest: WebsocketInbound, + private val blockchain: Chain + ) : RequestHandler, AbstractRequestHandler(accessLogWriter, Events.Channel.WSJSONRPC) { + + override fun close() { + if (request == null) { + return + } + val builder = EventsBuilder.NativeCall() + builder.withChain(blockchain.id) + builder.start(wsRequest) + builder.onRequest(request!!) + onClose(builder) + } + } + + class WsSubscriptionHandler( + private val accessLogWriter: AccessLogWriter, + private val wsRequest: WebsocketInbound, + private val blockchain: Chain + ) : SubscriptionHandler { + + private var builder: EventsBuilder.NativeSubscribeHttp? = null + + override fun onRequest(request: Pair) { + val builder = EventsBuilder.NativeSubscribeHttp(Events.Channel.WSJSONRPC, blockchain) + builder.start(wsRequest) + builder.onRequest(request) + this.builder = builder + } + + override fun onResponse(msgSize: Long) { + builder + ?.onReply(msgSize) + ?.let(accessLogWriter::submit) + } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt index 793fb820..d3d95268 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt @@ -28,7 +28,7 @@ class Events { } enum class Channel { - GRPC, JSONRPC + GRPC, JSONRPC, WSJSONRPC } abstract class Base( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt index ed2a5de7..375d1386 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt @@ -21,9 +21,11 @@ import io.emeraldpay.grpc.Chain import io.grpc.Attributes import io.grpc.Grpc import io.grpc.Metadata +import io.netty.handler.codec.http.HttpHeaders import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import reactor.netty.http.server.HttpServerRequest +import reactor.netty.http.websocket.WebsocketInbound import java.net.InetAddress import java.net.InetSocketAddress import java.time.Instant @@ -44,12 +46,16 @@ class EventsBuilder { fun start(request: HttpServerRequest) } + interface StartingWsRequest { + fun start(request: WebsocketInbound) + } + interface RequestReply : StartingHttp2Request { fun onRequest(msg: Req) fun onReply(msg: Resp): E } - abstract class Base : StartingHttp2Request, StartingHttp1Request { + abstract class Base : StartingHttp2Request, StartingHttp1Request, StartingWsRequest { companion object { private val remoteIpHeaders = listOf( "x-real-ip", @@ -136,17 +142,9 @@ class EventsBuilder { override fun start(request: HttpServerRequest) { val headers = request.requestHeaders() - val userAgent = headers.get("user-agent") - ?.let(this@Base::clean) - ?: "" + val userAgent = getUserAgent(headers) val ips = ArrayList() - remoteIpHeaders.forEach { key -> - headers.get(key)?.let { - it.trim().ifEmpty { null } - ?.let(this@Base::toInetAddress) - ?.let(ips::add) - } - } + extractIps(headers, ips) request.remoteAddress()?.let { addr -> ips.add(addr.address) } @@ -161,6 +159,53 @@ class EventsBuilder { ) } + override fun start(request: WebsocketInbound) { + val headers = request.headers() + val userAgent = getUserAgent(headers) + val ips = ArrayList() + extractIps(headers, ips) + // class WebsocketServerOperations, which is an implementation for the Websocket server connection, has a remoteAddress method + // But the class, and it's parent HttpServerOperations, are both private and cannot be used directly, + // so we try to access the field via reflection when it's possible + val remoteAddress: InetSocketAddress? = request.javaClass.methods + .find { it.name == "remoteAddress" } + ?.let { + if (it.canAccess(request) || it.trySetAccessible()) { + it.invoke(request) as InetSocketAddress + } else { + null + } + } + remoteAddress?.let { addr -> + ips.add(addr.address) + } + val ip = findBestIp(ips)?.hostAddress ?: "" + this.requestDetails = this.requestDetails + .copy( + remote = Events.Remote( + ips = ips.map { it.hostAddress }, + ip = ip, + userAgent = userAgent + ) + ) + } + + fun getUserAgent(headers: HttpHeaders): String { + return headers.get("user-agent") + ?.let(this@Base::clean) + ?: "" + } + + fun extractIps(headers: HttpHeaders, ips: MutableList) { + remoteIpHeaders.forEach { key -> + headers.get(key)?.let { + it.trim().ifEmpty { null } + ?.let(this@Base::toInetAddress) + ?.let(ips::add) + } + } + } + fun withChain(chain: Int): T { this.chainId = chain this.chain = Chain.byId(chainId) @@ -301,7 +346,9 @@ class EventsBuilder { } } - class NativeSubscribe : + class NativeSubscribe( + val channel: Events.Channel + ) : Base(), RequestReply { var item: Events.NativeSubscribeItemDetails? = null @@ -331,6 +378,42 @@ class EventsBuilder { } } + class NativeSubscribeHttp( + val channel: Events.Channel, + chain: Chain, + ) : + Base(), + RequestReply, Long> { + var item: Events.NativeSubscribeItemDetails? = null + val replies = HashMap() + + init { + withChain(chain.id) + } + + override fun getT(): NativeSubscribeHttp { + return this + } + + override fun onRequest(msg: Pair) { + this.item = Events.NativeSubscribeItemDetails( + msg.first, + msg.second?.size?.toLong() ?: 0L + ) + } + + override fun onReply(msg: Long): Events.NativeSubscribe { + return Events.NativeSubscribe( + request = requestDetails, + blockchain = chain, + nativeSubscribe = item!!, + payloadSizeBytes = msg, + id = UUID.randomUUID(), + channel = channel + ) + } + } + class Describe : Base(), RequestReply { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index 314a28ff..6b10b06a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -82,7 +82,7 @@ class ProxyServer( private val httpHandler = HttpHandler(readRpcJson, writeRpcJson, nativeCall, accessHandler, requestMetrics) private val wsHandler: WebsocketHandler? = if (config.websocketEnabled) { - WebsocketHandler(readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, requestMetrics) + WebsocketHandler(readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, accessHandler, requestMetrics) } else null fun start() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt index 567944fe..34385553 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt @@ -43,6 +43,7 @@ class WebsocketHandler( writeRpcJson: WriteRpcJson, nativeCall: NativeCall, private val nativeSubscribe: NativeSubscribe, + private val accessHandler: AccessHandlerHttp.HandlerFactory, requestMetrics: ProxyServer.RequestMetricsFactory, ) : BaseHandler(writeRpcJson, nativeCall, requestMetrics) { @@ -64,7 +65,8 @@ class WebsocketHandler( .map { ByteBufInputStream(it.content()).readAllBytes() } .flatMap(this@WebsocketHandler::parseRequest) - val eventHandler: AccessHandlerHttp.RequestHandler = AccessHandlerHttp.NoOpHandler() + val eventHandler = accessHandler.start(req, routeConfig.blockchain) + val responses = respond(routeConfig.blockchain, requests, eventHandler) .map { Unpooled.wrappedBuffer(it.toByteArray()) } @@ -95,13 +97,22 @@ class WebsocketHandler( } } - fun respond(blockchain: Chain, requests: Flux>, eventHandler: AccessHandlerHttp.RequestHandler): Flux { + fun respond(blockchain: Chain, requests: Flux>, eventHandlerFactory: AccessHandlerHttp.WsHandlerFactory): Flux { return requests.flatMap { call -> val method = call.method + if (method == "eth_subscribe") { val methodParams = splitMethodParams(call.params) if (methodParams != null) { + val eventHandler: AccessHandlerHttp.SubscriptionHandler = eventHandlerFactory.subscribe() val subscriptionId = nextSubscriptionId() + eventHandler.onRequest( + methodParams.let { mp -> + // TODO ineffective to encode the params each time just to get size, ideally should get a reference to the original JSON bytes + // but it doesn't happen very ofter, only on initial subscribe only for logs with filter + Pair(mp.first, mp.second?.let { Global.objectMapper.writeValueAsBytes(it) }) + } + ) // first need to respond with ID of the subscription, and the following responses would have it in "subscription" param val start = ResponseJson().also { it.id = call.id @@ -115,12 +126,20 @@ class WebsocketHandler( } Flux.concat(Mono.just(start), responses) .map { Global.objectMapper.writeValueAsString(it) } + .doOnNext { + eventHandler.onResponse(it.length.toLong()) + } } else { + // TODO should it produce a 404 to the AccessLog? Mono.empty() } } else { + val eventHandler: AccessHandlerHttp.RequestHandler = eventHandlerFactory.call() val proxyCall = readRpcJson.convertToNativeCall(ProxyCall.RpcType.SINGLE, listOf(call)) - execute(blockchain, proxyCall, eventHandler) + Mono.from(execute(blockchain, proxyCall, eventHandler)) + // thought the event handler is used in execute + // it still needs to be closed at the end, so it can render the logs + .doFinally { eventHandler.close() } } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy index 35027283..1ca8415d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy @@ -27,12 +27,13 @@ import java.time.Duration class WebsocketHandlerSpec extends Specification { + def requestHandlerFactory = new AccessHandlerHttp.NoOpFactory() def requestHandler = new AccessHandlerHttp.NoOpHandler() def "Parse standard RPC request"() { setup: def handler = new WebsocketHandler( - new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) ) when: def act = handler.parseRequest('{"id": 5, "jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["0x100001", false]}'.bytes) @@ -47,7 +48,7 @@ class WebsocketHandlerSpec extends Specification { def "Parse to empty an invalid request"() { setup: def handler = new WebsocketHandler( - new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) ) when: def act = handler.parseRequest('hello world'.bytes) @@ -61,7 +62,7 @@ class WebsocketHandlerSpec extends Specification { setup: def req1 = '{"id": 5, "jsonrpc": "2.0", "method": "eth_getBlockByNumber", "params": ["0x100001", false]}' def handler = new WebsocketHandler( - new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + new ReadRpcJson(), Stub(WriteRpcJson), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) ) when: def act = handler.parseRequest("[$req1]".bytes) @@ -79,7 +80,7 @@ class WebsocketHandlerSpec extends Specification { 1 * it.nativeCallResult(_) >> Flux.fromIterable([response]) } def handler = new WebsocketHandler( - new ReadRpcJson(), new WriteRpcJson(), nativeCall, Stub(NativeSubscribe), Stub(ProxyServer.RequestMetricsFactory) + new ReadRpcJson(), new WriteRpcJson(), nativeCall, Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) ) def request = new RequestJson("foo_test", [], 2) @@ -100,7 +101,7 @@ class WebsocketHandlerSpec extends Specification { 1 * it.subscribe(Chain.ETHEREUM, "foo_test", null) >> Flux.fromIterable([response1, response2]) } def handler = new WebsocketHandler( - new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), nativeSubscribe, Stub(ProxyServer.RequestMetricsFactory) + new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), nativeSubscribe, requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) ) def request = new RequestJson("eth_subscribe", ["foo_test"], 2) From f1720ad3542be316849f59e7c4aa9da6f6cc8ad5 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 29 Oct 2021 16:12:54 -0400 Subject: [PATCH 7/8] solution: websocket unsubscribe --- .../emeraldpay/dshackle/proxy/ProxyServer.kt | 1 + .../dshackle/proxy/WebsocketHandler.kt | 49 +++++++++++++++++-- .../proxy/WebsocketHandlerSpec.groovy | 49 +++++++++++++++++-- 3 files changed, 90 insertions(+), 9 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index 6b10b06a..f0df599d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -112,6 +112,7 @@ class ProxyServer( fun setupRoutes(routes: HttpServerRoutes) { config.routes.forEach { routeConfig -> + // TODO implement a manual handling of the routes and WS upgrade to have a better control over the connection and improve the access logging routes.post("/" + routeConfig.id, httpHandler.proxy(routeConfig)) if (config.websocketEnabled && wsHandler != null) { routes.ws("/" + routeConfig.id, wsHandler.proxy(routeConfig)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt index 34385553..9b692f43 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt @@ -15,6 +15,8 @@ */ package io.emeraldpay.dshackle.proxy +import com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.ProxyConfig import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp @@ -25,11 +27,11 @@ import io.emeraldpay.etherjar.rpc.json.ResponseJson import io.emeraldpay.grpc.Chain import io.netty.buffer.ByteBufInputStream import io.netty.buffer.Unpooled -import org.apache.commons.lang3.StringUtils import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.publisher.Sinks import reactor.netty.http.websocket.WebsocketInbound import reactor.netty.http.websocket.WebsocketOutbound import java.util.concurrent.atomic.AtomicLong @@ -55,11 +57,14 @@ class WebsocketHandler( fun nextSubscriptionId(): String { val n = subscriptionId.incrementAndGet() - return StringUtils.leftPad(n.toString(16), 16, "0") + return n.toString(16) } fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { return BiFunction { req, resp -> + // each connection keeps a list of subscription controllers + val control = HashMap>() + val requests: Flux> = req.aggregateFrames() .receiveFrames() .map { ByteBufInputStream(it.content()).readAllBytes() } @@ -67,7 +72,7 @@ class WebsocketHandler( val eventHandler = accessHandler.start(req, routeConfig.blockchain) - val responses = respond(routeConfig.blockchain, requests, eventHandler) + val responses = respond(routeConfig.blockchain, control, requests, eventHandler) .map { Unpooled.wrappedBuffer(it.toByteArray()) } resp.send(responses) @@ -97,7 +102,12 @@ class WebsocketHandler( } } - fun respond(blockchain: Chain, requests: Flux>, eventHandlerFactory: AccessHandlerHttp.WsHandlerFactory): Flux { + fun respond( + blockchain: Chain, + control: MutableMap>, + requests: Flux>, + eventHandlerFactory: AccessHandlerHttp.WsHandlerFactory + ): Flux { return requests.flatMap { call -> val method = call.method @@ -113,6 +123,8 @@ class WebsocketHandler( Pair(mp.first, mp.second?.let { Global.objectMapper.writeValueAsBytes(it) }) } ) + val currentControl = Sinks.one() + control[subscriptionId] = currentControl // first need to respond with ID of the subscription, and the following responses would have it in "subscription" param val start = ResponseJson().also { it.id = call.id @@ -124,6 +136,7 @@ class WebsocketHandler( .map { event -> WsSubscriptionResponse(params = WsSubscriptionData(event, subscriptionId)) } + .takeUntilOther(currentControl.asMono()) Flux.concat(Mono.just(start), responses) .map { Global.objectMapper.writeValueAsString(it) } .doOnNext { @@ -133,6 +146,34 @@ class WebsocketHandler( // TODO should it produce a 404 to the AccessLog? Mono.empty() } + } else if (method == "eth_unsubscribe") { + val id = call.params?.getOrNull(0) ?: "" + + // put it to the Access Log with fake id=0 (it doesn't matter, except the later reference) + val eventHandler: AccessHandlerHttp.RequestHandler = eventHandlerFactory.call() + eventHandler.onRequest( + BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChainValue(blockchain.id) + .addItems( + BlockchainOuterClass.NativeCallItem.newBuilder() + .setId(0) + .setMethod("eth_unsubscribe") + .setPayload(ByteString.copyFromUtf8("[\"$id\"]")) + .build() + ) + .build() + ) + + val p = control.remove(id.toString()) + val success = p?.tryEmitValue(true)?.isSuccess ?: false + val response = ResponseJson().also { + it.id = call.id + it.result = success + } + Mono.just(response) + .map { Global.objectMapper.writeValueAsString(it) } + .doOnNext { eventHandler.onResponse(NativeCall.CallResult.ok(0, it.toByteArray())) } + .doFinally { eventHandler.close() } } else { val eventHandler: AccessHandlerHttp.RequestHandler = eventHandlerFactory.call() val proxyCall = readRpcJson.convertToNativeCall(ProxyCall.RpcType.SINGLE, listOf(call)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy index 1ca8415d..35b2b3fb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.rpc.NativeSubscribe import io.emeraldpay.etherjar.rpc.json.RequestJson import io.emeraldpay.grpc.Chain import reactor.core.publisher.Flux +import reactor.core.publisher.Sinks import spock.lang.Specification import java.time.Duration @@ -85,7 +86,7 @@ class WebsocketHandlerSpec extends Specification { def request = new RequestJson("foo_test", [], 2) when: - def act = handler.respond(Chain.ETHEREUM, Flux.just(request), requestHandler) + def act = handler.respond(Chain.ETHEREUM, new HashMap>(), Flux.just(request), requestHandler) .single() .block(Duration.ofSeconds(1)) then: @@ -106,13 +107,51 @@ class WebsocketHandlerSpec extends Specification { def request = new RequestJson("eth_subscribe", ["foo_test"], 2) when: - def act = handler.respond(Chain.ETHEREUM, Flux.just(request), requestHandler) + def act = handler.respond(Chain.ETHEREUM, new HashMap>(), Flux.just(request), requestHandler) .collectList() .block(Duration.ofSeconds(1)) then: - act[0] == '{"jsonrpc":"2.0","id":2,"result":"0000000000000001"}' - act[1] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":1},"subscription":"0000000000000001"}}' - act[2] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":2},"subscription":"0000000000000001"}}' + act[0] == '{"jsonrpc":"2.0","id":2,"result":"1"}' + act[1] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":1},"subscription":"1"}}' + act[2] == '{"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{"foo":2},"subscription":"1"}}' act.size() == 3 } + + def "Unsubscribe"() { + setup: + + def handler = new WebsocketHandler( + new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) + ) + + def control = new HashMap>() + Sinks.One sink = Sinks.one(); + control["5"] = sink + def request = new RequestJson("eth_unsubscribe", ["5"], 0) + when: + def act = handler.respond(Chain.ETHEREUM, control, Flux.just(request), requestHandler) + .single() + .block(Duration.ofSeconds(1)) + def sinkResponse = sink.asMono().block() + then: + act == '{"jsonrpc":"2.0","id":0,"result":true}' + sinkResponse != null + } + + def "Unsubscribe when no subscription"() { + setup: + + def handler = new WebsocketHandler( + new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), Stub(NativeSubscribe), requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) + ) + + def control = new HashMap>() + def request = new RequestJson("eth_unsubscribe", ["5"], 0) + when: + def act = handler.respond(Chain.ETHEREUM, control, Flux.just(request), requestHandler) + .single() + .block(Duration.ofSeconds(1)) + then: + act == '{"jsonrpc":"2.0","id":0,"result":false}' + } } From b05b69080473c82f828c0def433260015f29a69e Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 29 Oct 2021 17:55:45 -0400 Subject: [PATCH 8/8] solution: docs for WebSocket support --- README.adoc | 26 ++++++++++++++++++++++---- docs/03-server-config.adoc | 8 +++++++- docs/reference-configuration.adoc | 23 +++++++++++++++-------- 3 files changed, 44 insertions(+), 13 deletions(-) diff --git a/README.adoc b/README.adoc index 8ac0390f..e1d2271d 100644 --- a/README.adoc +++ b/README.adoc @@ -24,7 +24,7 @@ It automatically verifies their availability and the current status of the netwo Provides: -- Standard Bitcoin and Ethereum JSON RPC API +- Standard Bitcoin and Ethereum JSON RPC API over HTTP and WebSocket - Enhanced gRPC-based API, with upstream selection, async execution, etc - **Secure** TLS with optional client authentication - Blockchain-aware edge **caching**, in memory and Redis @@ -107,12 +107,12 @@ Which sets the following: - gRPC access through 0.0.0.0:2449 ** TLS security is disabled (_please don't use in production!_) -- JSON RPC access through 0.0.0.0:8545 +- JSON RPC access through 0.0.0.0:8545 (both HTTP and WebsScket) ** proxy requests to Ethereum and Kovan upstreams ** request path for Ethereum Mainnet is `/eth`, `/kovan` for Kovan Testnet, and `/btc` for bitcoin ** i.e. call Ethereum Mainnet by `POST http://127.0.0.0:8545/eth` with JSON RPC payload - two upstreams, one for Ethereum Mainnet and another for Kovan Testnet (both upstreams are configured to use Infura endpoint) -- for Ethereum Mainnet it connects using JSON RPC and Websockets connections, +- for Ethereum Mainnet it connects using JSON RPC and WebSocket connections, - for Bitcoin Mainet only JSON RPC is used - `${INFURA_USER}` will be provided through environment variable @@ -146,7 +146,7 @@ Tools such as https://github.com/fullstorydev/grpcurl[gRPCurl] can automatically Alternatively you can connect to port 8545 with traditional JSON RPC requests -==== Access using JSON RPC +==== Access using JSON RPC over HTTP Dshackle implements standard JSON RPC interface, providing additional caching layer, upstream readiness/liveness checks, retry and other features for building Fault Tolerant services. @@ -165,6 +165,24 @@ curl --request POST \ {"jsonrpc":"2.0","id":1,"result":"0x72fa5e0181"} ---- +==== Access using JSON RPC over WebSocket + +Or the same Proxy URL can be accessed through WebSocket + +[source,bash] +---- +websocat ws://localhost:8545/eth +---- + +Then make RPC calls or subscriptions: + +---- +> | {"jsonrpc":"2.0", "id": 1, "method": "eth_subscribe", "params": ["newHeads"]} + +< | {"jsonrpc":"2.0","id":1,"result":"1f8"} +< | {"jsonrpc":"2.0","method":"eth_subscription","params":{"result":{....},"subscription":"1f8"}} +---- + ==== Access using gRPC NOTE: It's not necessary to use gRPC, as Dshackle can provide standard JSON RPC proxy, but Dshackle gRPC interface improves performance and provides additional features. diff --git a/docs/03-server-config.adoc b/docs/03-server-config.adoc index b1e45fc9..2fb2895c 100644 --- a/docs/03-server-config.adoc +++ b/docs/03-server-config.adoc @@ -62,6 +62,10 @@ a| `upstreams` === Enabling JSON RPC proxy +In addition to the gRPC protocol, Dshackle provides access compatible with Bitcoin and Ethereum JSON RPC. +The same server can be accessible as an HTTP JSON RPC and WebSocket JSON RPC. +For Ethereum, besides the standard RPC calls, it provides RPC subscriptions with `eth_subscribe` method. + .Example proxy: [source,yaml] ---- @@ -82,7 +86,9 @@ cluster: With that configuration Dshackle starts a JSON RPC proxy: - JSON RPC server is listening on `0.0.0.0:8080` -- `http://0.0.0.0:8080/eth` provides access to Ethereum API routed to an available upstream +- `http://0.0.0.0:8080/eth` (and `ws://0.0.0.0:8080/eth`) provides access to Ethereum API routed to an available upstream + +NOTE: Same URL should be used to access both HTTP RPC and WebSocket RPC .Full configuration: [source,yaml] diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index 16f06420..00f26f23 100644 --- a/docs/reference-configuration.adoc +++ b/docs/reference-configuration.adoc @@ -50,6 +50,7 @@ cache: proxy: host: 0.0.0.0 port: 8080 + websocket: true tls: enabled: true server: @@ -373,20 +374,26 @@ proxy: | `host` | `127.0.0.0` -| Host to bind gRPC server +| Host to bind HTTP server | `port` -| `2449` -| Port to bind gRPC server +| `8080` +| Port to bind HTT server +| `websocket` +| `true` +| Enable WebSocket Proxy | `tls` | -| Setup TLS configuration for the Proxy server. See <> section +| Setup TLS configuration for the Proxy server. +See <> section | `routes` | -a| Routing paths for Proxy.The proxy will handle requests as `https://${HOST}:${PORT}/${ROUTE_ID}` (or `http://` if TLS is not enabled) +a| Routing paths for Proxy. +The proxy will handle requests as `https://${HOST}:${PORT}/${ROUTE_ID}` (or `http://` if TLS is not enabled). +For WebSocket it's `wss` / `ws`, accordingly. |=== .Route config @@ -663,14 +670,14 @@ rpc: ---- | `ws.url` -| Websocket URL to connect to. +| WebSocket URL to connect to. Optional, but optimizes performance if it's available. | `ws.origin` -| HTTP `Origin` if required by Websocket remote server. +| HTTP `Origin` if required by WebSocket remote server. | `ws.basic-auth` + ... -| Websocket Basic Auth configuration, if required by the remote server +| WebSocket Basic Auth configuration, if required by the remote server | `ws.frameSize` | WebSocket frame size limit.