From 26f8dd294e94e4fe2b7e3874ac993ced5b60639d Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 19 Mar 2020 23:34:32 -0400 Subject: [PATCH] solution: basic implementation for Proxy endpoint --- README.adoc | 75 ++++++-- build.gradle | 5 +- docs/01-architecture-intro.adoc | 33 ++-- docs/02-quick-start.adoc | 35 +++- docs/03-server-config.adoc | 29 ++- docs/10-client-libraries.adoc | 2 +- gradle.properties | 2 +- .../dshackle/DshackleEnvironment.kt | 21 ++- .../io/emeraldpay/dshackle/GrpcServer.kt | 2 +- .../io/emeraldpay/dshackle/ProxyStarter.kt | 55 ++++++ .../dshackle/config/InvalidConfigException.kt | 28 +++ .../emeraldpay/dshackle/config/ProxyConfig.kt | 56 ++++++ .../dshackle/config/ProxyConfigReader.kt | 81 ++++++++ .../dshackle/config/UpstreamsConfigReader.kt | 90 +-------- .../dshackle/config/YamlConfigReader.kt | 123 ++++++++++++ .../io/emeraldpay/dshackle/proxy/ProxyCall.kt | 57 ++++++ .../emeraldpay/dshackle/proxy/ProxyServer.kt | 95 ++++++++++ .../emeraldpay/dshackle/proxy/ReadRpcJson.kt | 145 ++++++++++++++ .../emeraldpay/dshackle/proxy/WriteRpcJson.kt | 92 +++++++++ .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 6 +- .../config/ProxyConfigReaderSpec.groovy | 67 +++++++ .../dshackle/proxy/ProxyServerSpec.groovy | 65 +++++++ .../dshackle/proxy/ReadRpcJsonSpec.groovy | 81 ++++++++ .../dshackle/proxy/WriteRpcJsonSpec.groovy | 178 ++++++++++++++++++ src/test/resources/dshackle-proxy-basic.yaml | 5 + src/test/resources/dshackle-proxy-max.yaml | 9 + src/test/resources/dshackle-proxy-two.yaml | 7 + 27 files changed, 1309 insertions(+), 135 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/InvalidConfigException.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy create mode 100644 src/test/resources/dshackle-proxy-basic.yaml create mode 100644 src/test/resources/dshackle-proxy-max.yaml create mode 100644 src/test/resources/dshackle-proxy-two.yaml diff --git a/README.adoc b/README.adoc index f7bbda10..e3100f2f 100644 --- a/README.adoc +++ b/README.adoc @@ -24,18 +24,18 @@ The main goals of the project is to: The main features and advantages are: -- leveraging gRPC and HTTP2 protocols, with server push and asynchronous communications, to simplify and optimize standard - access patterns +- provides standard JSON RPC as smart load balancing proxy to active nodes +- in addition to JSON RPC it provides gRPC protocol, with server push and asynchronous communications, to simplify and optimize standard access patterns - targeting Kubernetes architecture -- automatically distributing access to API through multiple different target nodes, taking into account their current - availability and status -- allowing to build a mesh network of routers in different regions sharing a set of underlying nodes, with automatic - rebalancing and smart routing +- automatically distributing access to API through multiple different target nodes, taking into account their current availability and status +- allowing to build a mesh network of routers in different regions sharing a set of underlying nodes, with automatic rebalancing and smart routing - caching data on the edge -- providing monitoring (ex. Prometheus) and externalizable logging +- providing monitoring (ex. +Prometheus) and externalizable logging - configurable access authentication and authorization, including TLS certificates -Dshackle connects to several upstreams via JSON RPC, Websockets, or gRPC protocols. It verifies if a node ("upstream") is +Dshackle connects to several upstreams via JSON RPC, Websockets, or gRPC protocols. +It verifies if a node ("upstream") is fully synchronized (not in initial sync mode), has enough peers, and its height is not behind other nodes. If upstream lags behind others, lost peers, started to resync or went down, then Dshackle temporarily excludes it from requests and returns when the upstream problem is fixed. @@ -44,14 +44,14 @@ image::call-schema.png[alt="Call Schema",width=100%,align="center"] == Roadmap -- [ ] JSON RPC emulation, in addition to gRPC protocol +- [x] JSON RPC emulation, in addition to gRPC protocol - [ ] *Support Bitcoin RPC* +- [ ] External logging - [ ] Access to ERC-20 tokens on asset level - [ ] Subscription to bitcoind notification over gRPC (instead of ZeroMQ) - [ ] Prometheus monitoring - [ ] BIP-32 Pubkey - [ ] Lightweight sidecar node connector -- [ ] External logging - [ ] Configurable upstream roles == Quick Start @@ -59,23 +59,39 @@ image::call-schema.png[alt="Call Schema",width=100%,align="center"] === Configuration Create file `dshackle.yaml` with following content: + [source,yaml] ---- version: v1 port: 2449 tls: enabled: false + +proxy: + host: 0.0.0.0 + port: 8545 + routes: + - id: eth + blockchain: ethereum + - id: kovan + blockchain: kovan + upstreams: config: "upstreams.yaml" ---- Which sets the following: -- application listen on 0.0.0.0:2449 -- TLS security is disabled (_don't use in production!_) +- gRPC access through 0.0.0.0:2449 +** TLS security is disabled (_don't use in production!_) +- JSON RPC access through 0.0.0.0:8545 +** proxying requests to Ethereum and Kovan upstreams +** request path for Ethereum Mainnet is `/eth`, for Kovan is `/kovan` +** i.e. call Mainnet by `POST http://127.0.0.0:8545/eth` with JSON RPC payload - read upstreams configuration from file `upstreams.yaml` in the current directory Now create file `upstreams.yaml`: + [source,yaml] ---- version: v1 @@ -116,12 +132,32 @@ export INFURA_USER=... .Run Dshackle [source,bash] ---- -docker run -p 2449:2449 -v $(pwd):/etc/dshackle -e "INFURA_USER=$INFURA_USER" emeraldpay/dshackle +docker run -p 2449:2449 -p 8545:8545 -v $(pwd):/etc/dshackle -e "INFURA_USER=$INFURA_USER" emeraldpay/dshackle ---- Now it listen on port 2449 at the localhost and can be connected from any gRPC compatible client. -Tools such as https://github.com/fullstorydev/grpcurl[gRPCurl] can automatically parse protobuf definitions and connect -to it (actual Protobuf sources are located in a separate repository which you can find at https://github.com/emeraldpay/proto) +Tools such as https://github.com/fullstorydev/grpcurl[gRPCurl] can automatically parse protobuf definitions and connect to it (actual Protobuf sources are located in a separate repository which you can find at https://github.com/emeraldpay/proto) + +==== Access using JSON RPC + +Dshackle implements standard JSON RPC interface, providing additional caching layer, upstream readiness/liveness checks, retry and other features for building Fault Tolerant services. + +.Request using Curl +[source,bash] +---- +curl --request POST \ + --url http://localhost:8545/eth \ + --header 'content-type: application/json' \ + --data '{"jsonrpc":"2.0", "method":"eth_getBalance", "id":1, "params":["0x690b2bdf41f33f9f251ae0459e5898b856ed96be", "latest"]}' +---- + +.Output +[source,bash] +---- +{"jsonrpc":"2.0","id":1,"result":"0x72fa5e0181"} +---- + +==== Access using gRPC .Connect and listen for new blocks on Ethereum Mainnet [source,bash] @@ -149,7 +185,7 @@ grpcurl -import-path ./proto/ -proto blockchain.proto -d "{\"type\": 100}" -plai ... ---- -The output above is for a _streaming subscription_ to all new blocks on Ethereum Mainnet. It's one of services provided +The output above is for a _streaming subscription_ to all new blocks on Ethereum Mainnet.It's one of services provided by Dshackle, in additional to standard methods provided by RPC JSON of underlying nodes. == Documentation @@ -158,12 +194,17 @@ For detailed documentation see link:docs/[] directory. == Client Libraries +Dshackle should be compatible with all standard libraries that use Ethereum JSON RPC. + +But in addition to JSON RPC it provides gRPC API with many additional features and asynchronous access (please refer to the documentation: link:docs/06-methods.adoc[gRPC Methods]). +Below is the list of the libraries to use native gRPC API. + === Java gRPC Client + image:https://api.bintray.com/packages/emerald/emerald-grpc/emerald-grpc/images/download.svg[link="https://bintray.com/emerald/emerald-grpc/emerald-grpc/"] https://github.com/emeraldpay/emerald-java-client - [source,groovy] ---- repositories { diff --git a/build.gradle b/build.gradle index ac0a95e7..d3089adb 100644 --- a/build.gradle +++ b/build.gradle @@ -56,8 +56,8 @@ dependencies { compile "io.grpc:grpc-protobuf:${grpcVersion}" compile "io.grpc:grpc-stub:${grpcVersion}" compile "io.grpc:grpc-netty:${grpcVersion}" - compile "io.netty:netty-tcnative-boringssl-static:2.0.25.Final" - compile "io.netty:netty-all:4.1.36.Final" + compile "io.netty:netty-tcnative-boringssl-static:2.0.29.Final" + compile "io.netty:netty-all:4.1.48.Final" compile "org.jetbrains.kotlin:kotlin-stdlib-jdk8" compile "org.jetbrains.kotlin:kotlin-reflect" @@ -68,6 +68,7 @@ dependencies { compile "org.springframework.security:spring-security-web:$springVersion" compile "org.springframework.security:spring-security-config:$springVersion" compile "io.projectreactor:reactor-core:$reactorVersion" + compile "io.projectreactor.netty:reactor-netty:0.9.5.RELEASE" compile 'io.projectreactor.addons:reactor-extra:3.2.3.RELEASE' compile 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.2.RELEASE' compile 'com.salesforce.servicelibs:reactor-grpc:0.10.0' diff --git a/docs/01-architecture-intro.adoc b/docs/01-architecture-intro.adoc index 2645bd0b..cf0c334e 100644 --- a/docs/01-architecture-intro.adoc +++ b/docs/01-architecture-intro.adoc @@ -29,32 +29,41 @@ And for a request: - Is it for concrete data (_block #100_) or the latest (_get balance_)? - Is result a static value (_just block_) or may vary depending on network and node (_latest nonce_)? - Does it need to be repeated over multiple nodes (_broadcast transaction_)? -- Request can also specify which subset of nodes should be able to execute the request by selecting node _Labels_ -(see "link:08-quorum-and-selectors.adoc[Quorum and Selectors]") +- Request can also specify which subset of nodes should be able to execute the request by selecting node _Labels_ (see "link:08-quorum-and-selectors.adoc[Quorum and Selectors]") -Based on these factors, Dshackle executes the request on a most optimal node. For most of the simple requests, it just gets a node that is synchronized -to the point that must have a response for that particular request. If node failed, returned an invalid response, returned -an empty response when shouldn't then Dshackle tries again on another node, or on the same node after awhile (default is -200ms between failover repeats) +Based on these factors, Dshackle executes the request on a most optimal node. +For most of the simple requests, it just gets a node that is synchronized to the point that must have a response for that particular request. +If node failed, returned an invalid response, returned an empty response when shouldn't then Dshackle tries again on another node, or on the same node after awhile (default is 200ms between failover repeats) + +=== Proxy + +Dshackle provides access with standard JSON RPC protocol, functioning as a proxy to upstreams. +It provides: + +- Routes only realy/alive upstreams, i.e., synchronized and with enough peers +- Load Balancing +- Request Retry on upstream errors +- Local Caching (memory and Redis, see link:09-caching.adoc[Caching]) +- Broadcasting and Quorum for requests === gRPC protocol -Dshackle uses gRPC protocol for communications, because: +Dshackle native protocol is based on gRPC, which provides many additional features: - provides additional parameters on top of upstreams JSON RPC requests - based on HTTP/2 with low latency, compression, server push, pipelining and multiplexing - gRPC has binding and code generators for most of the languages and frameworks - easy to support TLS encryption and authentication -Dshackle has extra methods and functionality on top of standard APIs of upstreams, and it's more flexible to wrap -original APIs, such as JSON, into gRPC, and Protobuf + has additional data provided by Dshackle. It also allowed -to push new data from the server or send responses asynchronously, immediately after it gets executed on an upstream. +gRPC protocol is more flexible, and many of the Dshackle features are accessible mostly through that native protocol. + +Dshackle has extra methods and functionality on top of standard APIs of upstreams, and it's more flexible to wrap original APIs, such as JSON, into gRPC, and Protobuf + has additional data provided by Dshackle. +It also allowed to push new data from the server or send responses asynchronously, immediately after it gets executed on an upstream. === Distributed Load Balancing Dshackle servers can connect to each other through a secure encrypted and authenticated connection. -It allows to build a network of nodes deployed to different regions or run blockchain nodes outside of the main -network. +It allows to build a network of nodes deployed to different regions or run blockchain nodes outside of the main network. Later is especially important for blockchain nodes, as they require an open firewall with incoming connections for P2P, and to execute untrusted code ("smart contracts") at the same time. With Dshackle, it's possible to separate insecure nodes from diff --git a/docs/02-quick-start.adoc b/docs/02-quick-start.adoc index caf2fcd0..cfe5b9e7 100644 --- a/docs/02-quick-start.adoc +++ b/docs/02-quick-start.adoc @@ -20,6 +20,14 @@ version: v1 port: 2449 tls: enabled: false +proxy: + host: 0.0.0.0 + port: 8545 + routes: + - id: eth + blockchain: ethereum + - id: kovan + blockchain: kovan upstreams: config: "upstreams.yaml" ---- @@ -71,9 +79,30 @@ export INFURA_USER=... .Run Dshackle [source,bash] ---- -docker run -p 2449:2449 -v $(pwd):/etc/dshackle -e "INFURA_USER=$INFURA_USER" emeraldpay/dshackle +docker run -p 2449:2449 -p 8545:8545 -v $(pwd):/etc/dshackle -e "INFURA_USER=$INFURA_USER" emeraldpay/dshackle ---- +==== Access using JSON RPC + +Dshackle implements standard JSON RPC interface, providing additional caching layer, upstream readiness/liveness checks, retry and other features for building Fault Tolerant services. + +.Request using Curl +[source,bash] +---- +curl --request POST \ + --url http://localhost:8545/eth \ + --header 'content-type: application/json' \ + --data '{"jsonrpc":"2.0", "method":"eth_getBalance", "id":1, "params":["0x690b2bdf41f33f9f251ae0459e5898b856ed96be", "latest"]}' +---- + +.Output +[source,bash] +---- +{"jsonrpc":"2.0","id":1,"result":"0x72fa5e0181"} +---- + +==== Access using gRPC + .Connect and listen for new blocks on Ethereum Mainnet [source,bash] ---- @@ -100,5 +129,5 @@ grpcurl -import-path ./proto/ -proto blockchain.proto -d "{\"type\": 100}" -plai ... ---- -The output above is for a _streaming subscription_ to all new blocks on Ethereum Mainnet. It's a method provided -by Dshackle, provided in additional to methods provided by RPC JSON of underlying nodes. +The output above is for a _streaming subscription_ to all new blocks on Ethereum Mainnet. +It's a method provided by Dshackle, available in additional to methods provided by RPC JSON of underlying nodes. diff --git a/docs/03-server-config.adoc b/docs/03-server-config.adoc index d7c0b541..8b99ec36 100644 --- a/docs/03-server-config.adoc +++ b/docs/03-server-config.adoc @@ -3,8 +3,8 @@ Dshackle server tries to load its configuration from `/etc/dshackle/dshackle.yaml`, if it can't find a file at that path it tries to load file `dshackle.yaml` from current working directory. If none of them found server fails to run with an error. -[source,yaml] .Example dshackle.yaml configuration: +[source,yaml] ---- version: v1 port: 2449 @@ -22,8 +22,33 @@ upstreams: It configures following: -- server is listening on `0.0.0.0:2449` +- server is listening with gRCP API on `0.0.0.0:2449` - TLS is enabled - server certificate is located at `server.crt` with the key for it at `server.p8.key` - the server requires a client authentication by TLS client certificate signed by `ca.crt` certificate +- no JSON RPC is configured - upstreams configuration is configured in the file `upstreams.yaml` + +=== Enabling JSON RPC proxy + +.Example proxy: +[source,yaml] +---- +version: v1 +port: 2449 + +proxy: + host: 0.0.0.0 + port: 8080 + routes: + - id: eth + blockchain: ethereum + +upstreams: + config: "upstreams.yaml" +---- + +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 \ No newline at end of file diff --git a/docs/10-client-libraries.adoc b/docs/10-client-libraries.adoc index a8caf873..3dc69008 100644 --- a/docs/10-client-libraries.adoc +++ b/docs/10-client-libraries.adoc @@ -1,4 +1,4 @@ -== Client Libraries +== Client Libraries using native gRPC-based protocol === Protobuf diff --git a/gradle.properties b/gradle.properties index 278197c7..3c39a100 100644 --- a/gradle.properties +++ b/gradle.properties @@ -12,7 +12,7 @@ protobufVersion=3.7.1 # Core springBootVersion=2.1.4.RELEASE springVersion=5.1.4.RELEASE -reactorVersion=3.2.9.RELEASE +reactorVersion=3.3.3.RELEASE # Our Libs etherjarVersion=0.9.1 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/DshackleEnvironment.kt b/src/main/kotlin/io/emeraldpay/dshackle/DshackleEnvironment.kt index a5d782a3..f45bc055 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/DshackleEnvironment.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/DshackleEnvironment.kt @@ -15,6 +15,8 @@ */ package io.emeraldpay.dshackle +import io.emeraldpay.dshackle.config.ProxyConfig +import io.emeraldpay.dshackle.config.ProxyConfigReader import org.slf4j.LoggerFactory import org.springframework.beans.factory.config.YamlPropertiesFactoryBean import org.springframework.core.env.* @@ -23,7 +25,6 @@ import org.springframework.core.io.Resource import org.springframework.core.io.support.ResourcePropertySource import java.io.File import java.util.* -import javax.annotation.PostConstruct const val DEFAULT_CONFIG = "/etc/dshackle/dshackle.yaml" const val LOCAL_CONFIG = "./dshackle.yaml" @@ -40,18 +41,25 @@ open class DshackleEnvironment: StandardEnvironment() { propertySources.addLast(ResourcePropertySource("version.properties")) } - open fun mainConfig(): PropertySource<*> { + open fun getResource(): File? { var target = File(DEFAULT_CONFIG) if (!isAcceptedConfig(target)) { target = File(LOCAL_CONFIG) if (!isAcceptedConfig(target)) { log.error("Configuration is not found neither at $DEFAULT_CONFIG nor $LOCAL_CONFIG") - return PropertySource.named("mainConfig") + return null } } target = target.normalize() - val loadedProperties = this.loadYaml(FileSystemResource(target)) + return target + } + + open fun mainConfig(): PropertySource<*> { + val target = getResource() ?: return PropertySource.named("mainConfig") + val resource = FileSystemResource(target) + val loadedProperties = this.loadYaml(resource) loadedProperties["configPath"] = target.absolutePath + loadedProperties[ProxyConfig.CONFIG_ID] = extractYamlProxy(resource) return PropertiesPropertySource("mainConfig", loadedProperties) } @@ -62,6 +70,11 @@ open class DshackleEnvironment: StandardEnvironment() { return factory.getObject()!! } + protected fun extractYamlProxy(resource: Resource): ProxyConfig? { + val reader = ProxyConfigReader() + return reader.read(resource.inputStream) + } + protected fun isAcceptedConfig(target: File): Boolean { return target.exists() && target.isFile } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt index abea9539..6e07e69e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt @@ -45,7 +45,7 @@ open class GrpcServer( log.info("Starting GRPC Server...") log.debug("Running with DEBUG LOGGING") val port = env.getProperty("port", "2449").toInt() - log.info("Listening on 0.0.0.0:$port") + log.info("Listening Native on 0.0.0.0:$port") val serverBuilder = NettyServerBuilder.forPort(port) rpcs.forEach { serverBuilder.addService(it) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt b/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt new file mode 100644 index 00000000..0c0b6a54 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/ProxyStarter.kt @@ -0,0 +1,55 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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 + +import io.emeraldpay.dshackle.config.ProxyConfig +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 org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.core.env.Environment +import org.springframework.stereotype.Service +import javax.annotation.PostConstruct + +/** + * Starts HTTP proxy endpoint, if configured + */ +@Service +class ProxyStarter( + @Autowired private val env: Environment, + @Autowired private val readRpcJson: ReadRpcJson, + @Autowired private val writeRpcJson: WriteRpcJson, + @Autowired private val nativeCall: NativeCall +) { + + companion object { + private val log = LoggerFactory.getLogger(ProxyStarter::class.java) + } + + @PostConstruct + fun start() { + val config = env.getProperty(ProxyConfig.CONFIG_ID, ProxyConfig::class.java) + if (config == null) { + log.debug("Proxy server is not configured") + return + } + val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall) + server.start() + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/InvalidConfigException.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/InvalidConfigException.kt new file mode 100644 index 00000000..5eb2e9f5 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/InvalidConfigException.kt @@ -0,0 +1,28 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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.config + +import org.yaml.snakeyaml.error.Mark + +open class InvalidConfigException( + message: String +) : Exception(message) + +class InvalidConfigYamlException( + filename: String, + mark: Mark, + message: String +) : InvalidConfigException("Invalid YAML configuration ${message}, at ${filename}:${mark.line}") \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt new file mode 100644 index 00000000..b543a741 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt @@ -0,0 +1,56 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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.config + +import io.emeraldpay.grpc.Chain + +/** + * Configure HTTP Proxy to Upstreams + */ +class ProxyConfig { + + companion object { + public const val CONFIG_ID = "parsed.proxy" + } + + var enabled: Boolean = true + + /** + * Host to bind server. Default: 127.0.0.1 + */ + var host = "127.0.0.1" + + /** + * Port to bind. Default: 8080 + */ + var port: Int = 8080 + + /** + * List of available routes + */ + var routes: List = ArrayList() + + class Route( + /** + * URL binding for the route. http://$host:$port/$id + */ + val id: String, + /** + * Blockchain to dispatch requests + */ + val blockchain: Chain + ) +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt new file mode 100644 index 00000000..43436f21 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt @@ -0,0 +1,81 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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.config + +import io.emeraldpay.grpc.Chain +import org.apache.commons.lang3.StringUtils +import org.slf4j.LoggerFactory +import org.yaml.snakeyaml.Yaml +import org.yaml.snakeyaml.nodes.MappingNode +import java.io.InputStream +import java.io.InputStreamReader + +/** + * Read YAML config, part related to Proxy configuration + */ +class ProxyConfigReader : YamlConfigReader() { + + companion object { + private val log = LoggerFactory.getLogger(ProxyConfigReader::class.java) + } + + private var filename = "dshackle.yaml" + + fun read(input: InputStream): ProxyConfig? { + val yaml = Yaml() + val configNode = asMappingNode(yaml.compose(InputStreamReader(input))) + return read(getMapping(configNode, "proxy")) + } + + fun read(input: MappingNode?): ProxyConfig? { + if (input == null) { + return null + } + val config = ProxyConfig() + getValueAsString(input, "host")?.let { + config.host = it + } + getValueAsInt(input, "port")?.let { + config.port = it + } + getValueAsBool(input, "enabled")?.let { + config.enabled = it + } + val currentRoutes = HashSet() + getList(input, "routes")?.let { routes -> + config.routes = routes.value.map { route -> + val id = getValueAsString(route, "id") + if (id == null || StringUtils.isEmpty(id) || !StringUtils.isAlphanumeric(id)) { + throw InvalidConfigYamlException(filename, route.startMark, "Route id must be alphanumeric") + } + if (currentRoutes.contains(id)) { + throw InvalidConfigYamlException(filename, route.startMark, "Route id repeated: $id") + } + currentRoutes.add(id) + val blockchain = getValueAsString(route, "blockchain") + if (StringUtils.isEmpty(blockchain) || getBlockchain(blockchain!!) == Chain.UNSPECIFIED) { + throw InvalidConfigYamlException(filename, route.startMark, "Invalid blockchain or not specified") + } + ProxyConfig.Route(id, getBlockchain(blockchain)) + } + } + if (config.routes.isEmpty()) { + return null + } + return config + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 48a47a05..8ddf1f79 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -29,10 +29,9 @@ import java.lang.IllegalArgumentException import java.net.URI import java.time.Duration -class UpstreamsConfigReader { +class UpstreamsConfigReader : YamlConfigReader() { private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java) - private val envVariables = EnvVariables() fun read(input: InputStream): UpstreamsConfig { val yaml = Yaml() @@ -218,91 +217,4 @@ class UpstreamsConfigReader { } } - private fun hasAny(mappingNode: MappingNode?, key: String): Boolean { - if (mappingNode == null) { - return false - } - return mappingNode.value - .stream() - .filter { n -> n.keyNode is ScalarNode } - .filter { n -> - val sn = n.keyNode as ScalarNode - key == sn.value - }.count() > 0 - } - - private fun getValue(mappingNode: MappingNode?, key: String, type: Class): T? { - if (mappingNode == null) { - return null - } - return mappingNode.value - .stream() - .filter { n -> n.keyNode is ScalarNode && type.isAssignableFrom(n.valueNode.javaClass) } - .filter { n -> - val sn = n.keyNode as ScalarNode - key == sn.value - } - .map { n -> n.valueNode as T } - .findFirst().let { - if (it.isPresent) { - it.get() - } else { - null - } - } - } - - private fun getMapping(mappingNode: MappingNode?, key: String): MappingNode? { - return getValue(mappingNode, key, MappingNode::class.java) - } - - private fun getValue(mappingNode: MappingNode?, key: String): ScalarNode? { - return getValue(mappingNode, key, ScalarNode::class.java) - } - - private fun getList(mappingNode: MappingNode?, key: String): CollectionNode? { - val value = getValue(mappingNode, key, CollectionNode::class.java) ?: return null - return value as CollectionNode - } - - private fun getListOfString(mappingNode: MappingNode?, key: String): List? { - return getList(mappingNode, key)?.value - ?.map { it.value } - ?.map(envVariables::postProcess) - } - - private fun getValueAsString(mappingNode: MappingNode?, key: String): String? { - return getValue(mappingNode, key)?.let { - return@let it.value - }?.let(envVariables::postProcess) - } - - private fun getValueAsInt(mappingNode: MappingNode?, key: String): Int? { - return getValue(mappingNode, key)?.let { - return@let if (it.isPlain) { - it.value.toIntOrNull() - } else { - null - } - } - } - - private fun getValueAsBool(mappingNode: MappingNode?, key: String): Boolean? { - return getValue(mappingNode, key)?.let { - return@let if (it.isPlain) { - it.value?.toLowerCase() == "true" - } else { - null - } - } - } - - private fun asMappingNode(node: Node): MappingNode { - return if (MappingNode::class.java.isAssignableFrom(node.javaClass)) { - node as MappingNode - } else { - throw IllegalArgumentException("Not a map") - } - } - } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt new file mode 100644 index 00000000..8aff504d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt @@ -0,0 +1,123 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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.config + +import io.emeraldpay.grpc.Chain +import org.yaml.snakeyaml.nodes.CollectionNode +import org.yaml.snakeyaml.nodes.MappingNode +import org.yaml.snakeyaml.nodes.Node +import org.yaml.snakeyaml.nodes.ScalarNode + +open class YamlConfigReader { + private val envVariables = EnvVariables() + + protected fun hasAny(mappingNode: MappingNode?, key: String): Boolean { + if (mappingNode == null) { + return false + } + return mappingNode.value + .stream() + .filter { n -> n.keyNode is ScalarNode } + .filter { n -> + val sn = n.keyNode as ScalarNode + key == sn.value + }.count() > 0 + } + + private fun getValue(mappingNode: MappingNode?, key: String, type: Class): T? { + if (mappingNode == null) { + return null + } + return mappingNode.value + .stream() + .filter { n -> n.keyNode is ScalarNode && type.isAssignableFrom(n.valueNode.javaClass) } + .filter { n -> + val sn = n.keyNode as ScalarNode + key == sn.value + } + .map { n -> n.valueNode as T } + .findFirst().let { + if (it.isPresent) { + it.get() + } else { + null + } + } + } + + protected fun getMapping(mappingNode: MappingNode?, key: String): MappingNode? { + return getValue(mappingNode, key, MappingNode::class.java) + } + + private fun getValue(mappingNode: MappingNode?, key: String): ScalarNode? { + return getValue(mappingNode, key, ScalarNode::class.java) + } + + protected fun getList(mappingNode: MappingNode?, key: String): CollectionNode? { + val value = getValue(mappingNode, key, CollectionNode::class.java) ?: return null + return value as CollectionNode + } + + protected fun getListOfString(mappingNode: MappingNode?, key: String): List? { + return getList(mappingNode, key)?.value + ?.map { it.value } + ?.map(envVariables::postProcess) + } + + protected fun getValueAsString(mappingNode: MappingNode?, key: String): String? { + return getValue(mappingNode, key)?.let { + return@let it.value + }?.let(envVariables::postProcess) + } + + protected fun getValueAsInt(mappingNode: MappingNode?, key: String): Int? { + return getValue(mappingNode, key)?.let { + return@let if (it.isPlain) { + it.value.toIntOrNull() + } else { + null + } + } + } + + protected fun getValueAsBool(mappingNode: MappingNode?, key: String): Boolean? { + return getValue(mappingNode, key)?.let { + return@let if (it.isPlain) { + it.value?.toLowerCase() == "true" + } else { + null + } + } + } + + protected fun asMappingNode(node: Node): MappingNode { + return if (MappingNode::class.java.isAssignableFrom(node.javaClass)) { + node as MappingNode + } else { + throw IllegalArgumentException("Not a map") + } + } + + // ---- + + fun getBlockchain(id: String): Chain { + return Chain.values().find { chain -> + chain.name == id.toUpperCase() + || chain.chainCode.toUpperCase() == id.toUpperCase() + || chain.id.toString() == id + } ?: Chain.UNSPECIFIED + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt new file mode 100644 index 00000000..1e1ae373 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt @@ -0,0 +1,57 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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 org.slf4j.LoggerFactory + +/** + * JSON RPC call to the proxy + */ +class ProxyCall( + /** + * Type of the request. The response format depends on it + */ + val type: RpcType +) { + + companion object { + private val log = LoggerFactory.getLogger(ProxyCall::class.java) + } + + /** + * Mapping from our internal ids to user provided JSON RPC ids. + */ + val ids = HashMap() + + /** + * Content of the request + */ + val items: MutableList = ArrayList() + + enum class RpcType { + /** + * One item request passed as Object + */ + SINGLE, + + /** + * Batch passed as Array of Object. It may be one-element array, i.e., single request, though response + * must be formatted as an Array + */ + BATCH + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt new file mode 100644 index 00000000..da401a0f --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -0,0 +1,95 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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.config.ProxyConfig +import io.emeraldpay.dshackle.rpc.NativeCall +import io.netty.buffer.Unpooled +import org.reactivestreams.Publisher +import org.slf4j.LoggerFactory +import org.springframework.http.HttpHeaders +import reactor.core.publisher.Mono +import reactor.netty.DisposableServer +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.function.BiFunction + +/** + * HTTP Proxy Server + */ +class ProxyServer( + private var config: ProxyConfig, + private val readRpcJson: ReadRpcJson, + private val writeRpcJson: WriteRpcJson, + private val nativeCall: NativeCall +) { + + companion object { + private val log = LoggerFactory.getLogger(ProxyServer::class.java) + } + + fun start() { + if (!config.enabled) { + log.debug("Proxy server is not enabled") + return + } + log.info("Listening Proxy on ${config.host}:${config.port}") + val server: DisposableServer = HttpServer.create() + .host(config.host) + .port(config.port) + .route(this::setupRoutes) + .bindNow() + } + + fun setupRoutes(routes: HttpServerRoutes) { + config.routes.forEach { routeConfig -> + routes.post("/" + routeConfig.id, proxy(routeConfig)) + } + } + + fun execute(chain: Common.ChainRef, call: ProxyCall): Publisher { + val request = BlockchainOuterClass.NativeCallRequest.newBuilder() + .setChain(chain) + .addAllItems(call.items) + .build() + val jsons = nativeCall + .nativeCall(Mono.just(request)) + .transform(writeRpcJson.toJsons(call)) + return if (call.type == ProxyCall.RpcType.SINGLE) { + jsons.next() + } else { + jsons.transform(writeRpcJson.asArray()) + } + } + + fun proxy(routeConfig: ProxyConfig.Route): BiFunction> { + val chain = Common.ChainRef.forNumber(routeConfig.blockchain.id) + return BiFunction { req, resp -> + val results = req.receive() + .aggregate() + .asByteArray() + .map(readRpcJson) + .flatMapMany { call -> execute(chain, call) } + .map { Unpooled.wrappedBuffer(it.toByteArray()) } + resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json") + .send(results) + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt new file mode 100644 index 00000000..6c28ab16 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt @@ -0,0 +1,145 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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 com.fasterxml.jackson.databind.ObjectMapper +import com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.infinitape.etherjar.rpc.RpcException +import io.infinitape.etherjar.rpc.RpcResponseError +import io.infinitape.etherjar.rpc.json.RequestJson +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.stereotype.Service +import java.io.IOException +import java.util.* +import java.util.function.Function +import java.util.stream.Collectors + + +/** + * Reader for JSON RPC request + */ +@Service +open class ReadRpcJson( + @Autowired private val objectMapper: ObjectMapper +) : Function { + + companion object { + private val log = LoggerFactory.getLogger(ReadRpcJson::class.java) + private val spaces = " \n\t".toByteArray() + } + + private val jsonExtractor: Function, RequestJson> + + init { + jsonExtractor = Function { json -> + if ("2.0" != json["jsonrpc"]) { + throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Unsupported JSON RPC version") + } + if (json["id"] == null) { + throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID not set") + } + val id = json["id"] + if (!(json["method"] != null && json["method"] is String)) { + throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID not set") + } + if (json.containsKey("params") && json["params"] !is List<*>) { + throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Params must be an array") + } + RequestJson( + json["method"].toString(), + json["params"] as List<*>, + id + ) + } + } + + /** + * Read fist non-space character, which supposed to start actual JSON part of the request + */ + @Throws(IOException::class) + fun getStartOfJson(buf: ByteArray): Byte { + val count = buf.size + var i = 0 + //if cannot find anything in the first 255 bytes, just consider it as invalid + while (i < 256 && i < count) { + if (buf[i] != spaces[0] && buf[i] != spaces[1] && buf[i] != spaces[2]) { + return buf[i] + } + i++ + } + throw IllegalArgumentException("Invalid input") + } + + /** + * Check the type of the payload, based on the format (first character at this case) + */ + @Throws(IOException::class) + fun getType(data: ByteArray): ProxyCall.RpcType { + val first = try { + getStartOfJson(data) + } catch (e: IllegalArgumentException) { + throw RpcException(RpcResponseError.CODE_INVALID_JSON, "Empty JSON") + } + if (first == '{'.toByte()) { + return ProxyCall.RpcType.SINGLE + } else if (first == '['.toByte()) { + return ProxyCall.RpcType.BATCH + } + throw RpcException(RpcResponseError.CODE_INVALID_JSON, "Failed to parse JSON") + } + + /** + * Convert payload to the proxy call details + */ + override fun apply(data: ByteArray): ProxyCall { + val list: MutableList> + 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 + } catch (e: RpcException) { + throw e + } catch (e: Exception) { + log.error("Parse Error: " + e.message) + throw RpcException(RpcResponseError.CODE_INVALID_JSON, e.message) + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt new file mode 100644 index 00000000..f364eee3 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt @@ -0,0 +1,92 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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 com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.infinitape.etherjar.rpc.RpcResponseError +import io.infinitape.etherjar.rpc.json.ResponseJson +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.stereotype.Service +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import java.lang.StringBuilder +import java.time.Duration +import java.util.function.Function + +/** + * Writer for JSON RPC requests + */ +@Service +open class WriteRpcJson( + @Autowired private val objectMapper: ObjectMapper +) { + + companion object { + private val log = LoggerFactory.getLogger(WriteRpcJson::class.java) + } + + /** + * Convert Dshackle protobuf based responses to JSON RPC formatted as strings + */ + open fun toJsons(call: ProxyCall): Function, Flux> { + return Function { flux -> + flux.flatMap { response -> + val json = ResponseJson() + if (!call.ids.containsKey(response.id)) { + log.warn("ID wasn't requested: ${response.id}") + return@flatMap Flux.empty() + } + json.id = call.ids[response.id] + if (response.succeed) { + val payload = objectMapper.readValue(response.payload.toByteArray(), ResponseJson::class.java) + if (payload.error != null) { + json.error = payload.error + } else { + json.result = payload.result + } + } else { + json.error = RpcResponseError(-32002, response.errorMessage) + } + Flux.just(objectMapper.writeValueAsString(json)) + }.onErrorContinue { t, u -> + log.warn("Failed to convert to JSON", t) + } + } + } + + /** + * Format response as JSON Array, for Batch requests + */ + fun asArray(): Function, Flux> { + return Function { flux -> + val body = flux.zipWith(Flux.concat(Mono.just(false), Flux.just(true).repeat())) + .map { + if (it.t2) { + "," + it.t1 + } else { + it.t1 + } + } + Flux.concat( + Mono.just("["), + body, + Mono.just("]") + ) + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 8d40a79c..e94e61b1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -33,7 +33,7 @@ import reactor.util.function.Tuples import java.lang.Exception @Service -class NativeCall( +open class NativeCall( @Autowired private val upstreams: Upstreams, @Autowired private val objectMapper: ObjectMapper ) { @@ -42,8 +42,8 @@ class NativeCall( open fun nativeCall(requestMono: Mono): Flux { return requestMono.flatMapMany(this::prepareCall) - .map(this::setupCallParams) - .parallel() + .map(this::setupCallParams) + .parallel() .flatMap(this::fetch) .sequential() .map(this::buildResponse) diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy new file mode 100644 index 00000000..edf0c605 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy @@ -0,0 +1,67 @@ +package io.emeraldpay.dshackle.config + +import io.emeraldpay.grpc.Chain +import spock.lang.Specification + +class ProxyConfigReaderSpec extends Specification { + + ProxyConfigReader reader = new ProxyConfigReader() + + def "Read basic proxy config"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("dshackle-proxy-basic.yaml") + when: + def act = reader.read(config) + + then: + act.enabled + act.port == 8080 + act.host == '127.0.0.1' + act.routes.size() == 1 + with(act.routes[0]) { + id == "ethereum" + blockchain == Chain.ETHEREUM + } + } + + def "Read proxy config with two elements"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("dshackle-proxy-two.yaml") + when: + def act = reader.read(config) + + then: + act.enabled + act.port == 8080 + act.routes.size() == 2 + with(act.routes[0]) { + id == "ethereum" + blockchain == Chain.ETHEREUM + } + with(act.routes[1]) { + id == "classic" + blockchain == Chain.ETHEREUM_CLASSIC + } + } + + def "Read max proxy config"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("dshackle-proxy-max.yaml") + when: + def act = reader.read(config) + + then: + act.enabled + act.host == '0.0.0.0' + act.port == 8080 + act.routes.size() == 2 + with(act.routes[0]) { + id == "ethereum" + blockchain == Chain.ETHEREUM + } + with(act.routes[1]) { + id == "classic" + blockchain == Chain.ETHEREUM_CLASSIC + } + } +} 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..e8257952 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy @@ -0,0 +1,65 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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.config.ProxyConfig +import io.emeraldpay.dshackle.rpc.NativeCall +import io.emeraldpay.dshackle.test.TestingCommons +import reactor.core.publisher.Flux +import reactor.test.StepVerifier +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(TestingCommons.objectMapper()), + writeRpcJson, + nativeCall + ) + + 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(Common.ChainRef.CHAIN_ETHEREUM, call) + + then: + 1 * nativeCall.nativeCall(_) >> Flux.just(BlockchainOuterClass.NativeCallReplyItem.newBuilder().build()) + 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 new file mode 100644 index 00000000..34b121d0 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy @@ -0,0 +1,81 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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.test.TestingCommons +import io.infinitape.etherjar.rpc.RpcException +import spock.lang.Specification + +class ReadRpcJsonSpec extends Specification { + + ReadRpcJson reader = new ReadRpcJson(TestingCommons.objectMapper()) + + def "Get first symbol"() { + expect: + reader.getStartOfJson(input.bytes) == exp.bytes[0] + where: + exp | input + "{" | "{}" + "{" | " { }" + "{" | "\n\n { }" + "[" | " [ { } ] " + } + + def "Error for input with many spaces"() { + setup: + def empty = " " * 1000 + when: + reader.getStartOfJson((empty + "{}").bytes) + then: + thrown(IllegalArgumentException) + } + + def "Error for empty spaces"() { + when: + reader.getStartOfJson("".bytes) + then: + thrown(IllegalArgumentException) + } + + def "Get type"() { + expect: + reader.getType(input.bytes) == exp + where: + exp | input + ProxyCall.RpcType.SINGLE | "{}" + ProxyCall.RpcType.SINGLE | " { }" + ProxyCall.RpcType.SINGLE | "\n\n { }" + ProxyCall.RpcType.BATCH | " [ { } ] " + } + + def "Error type for invalid input"() { + when: + reader.getType("hello".bytes) + then: + thrown(RpcException) + + when: + reader.getType("1".bytes) + then: + thrown(RpcException) + + when: + reader.getType("".bytes) + then: + thrown(RpcException) + + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy new file mode 100644 index 00000000..a4e4427c --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy @@ -0,0 +1,178 @@ +/** + * Copyright (c) 2020 ETCDEV GmbH + * + * 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 com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.dshackle.test.TestingCommons +import reactor.core.publisher.Flux +import spock.lang.Specification + +import java.time.Duration + +class WriteRpcJsonSpec extends Specification { + + WriteRpcJson writer = new WriteRpcJson(TestingCommons.objectMapper()) + + def "Write empty array"() { + when: + def act = Flux.empty().transform(writer.asArray()) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == "[]" + } + + def "Write single item array"() { + when: + def act = Flux.just('{"id": 1}').transform(writer.asArray()) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '[{"id": 1}]' + } + + def "Write two item array"() { + setup: + def data = [ + '{"id": 1}', + '{"id": 2}', + ] + when: + def act = Flux.fromIterable(data).transform(writer.asArray()) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '[{"id": 1},{"id": 2}]' + } + + def "Write few items array"() { + setup: + def data = [ + '{"id": 1}', + '{"id": 2, "foo": "bar"}', + '{"id": 3, "foo": "baz"}', + '{"id": 4}', + '{"id": 5, "x": 5}', + ] + when: + def act = Flux.fromIterable(data).transform(writer.asArray()) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '[{"id": 1},{"id": 2, "foo": "bar"},{"id": 3, "foo": "baz"},{"id": 4},{"id": 5, "x": 5}]' + } + + def "Convert basic to JSON"() { + setup: + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.ids[1] = "aaa" + def data = [ + BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setId(1) + .setSucceed(true) + .setPayload(ByteString.copyFrom('{"jsonrpc": "2.0", "id": 1, "result": "0x98dbb1"}', 'UTF-8')) + .build() + ] + when: + def act = Flux.fromIterable(data) + .transform(writer.toJsons(call)) + .collectList() + .block(Duration.ofSeconds(1)) + then: + act == ['{"jsonrpc":"2.0","id":"aaa","result":"0x98dbb1"}'] + } + + def "Convert error to JSON"() { + setup: + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.ids[1] = 1 + def data = [ + BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setId(1) + .setSucceed(true) + .setPayload(ByteString.copyFrom('{"jsonrpc": "2.0", "id": 1, "error": {"code": -32001, "message": "oops"}}', 'UTF-8')) + .build() + ] + when: + def act = Flux.fromIterable(data) + .transform(writer.toJsons(call)) + .collectList() + .block(Duration.ofSeconds(1)) + then: + act == ['{"jsonrpc":"2.0","id":1,"error":{"code":-32001,"message":"oops"}}'] + } + + def "Convert gRPC error to JSON"() { + setup: + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.ids[1] = 1 + def data = [ + BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setId(1) + .setSucceed(false) + .setErrorMessage("Internal Error") + .build() + ] + when: + def act = Flux.fromIterable(data) + .transform(writer.toJsons(call)) + .collectList() + .block(Duration.ofSeconds(1)) + then: + act == ['{"jsonrpc":"2.0","id":1,"error":{"code":-32002,"message":"Internal Error"}}'] + } + + def "Convert few items to JSON"() { + setup: + def call = new ProxyCall(ProxyCall.RpcType.SINGLE) + call.ids[1] = 10 + call.ids[2] = 11 + call.ids[3] = 15 + def data = [ + BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setId(1) + .setSucceed(true) + .setPayload(ByteString.copyFrom('{"jsonrpc": "2.0", "id": 1, "result": "0x98dbb1"}', 'UTF-8')) + .build(), + BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setId(2) + .setSucceed(true) + .setPayload(ByteString.copyFrom('{"jsonrpc": "2.0", "id": 2, "error": {"code": -32001, "message": "oops"}}', 'UTF-8')) + .build(), + BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setId(3) + .setSucceed(true) + .setPayload(ByteString.copyFrom('{"jsonrpc": "2.0", "id": 3, "result": {"hash": "0x2484f459dc"}}', 'UTF-8')) + .build(), + ] + when: + def act = Flux.fromIterable(data) + .transform(writer.toJsons(call)) + .collectList() + .block(Duration.ofSeconds(1)) + then: + act == [ + '{"jsonrpc":"2.0","id":10,"result":"0x98dbb1"}', + '{"jsonrpc":"2.0","id":11,"error":{"code":-32001,"message":"oops"}}', + '{"jsonrpc":"2.0","id":15,"result":{"hash":"0x2484f459dc"}}' + ] + } +} diff --git a/src/test/resources/dshackle-proxy-basic.yaml b/src/test/resources/dshackle-proxy-basic.yaml new file mode 100644 index 00000000..65fb723e --- /dev/null +++ b/src/test/resources/dshackle-proxy-basic.yaml @@ -0,0 +1,5 @@ +proxy: + port: 8080 + routes: + - id: ethereum + blockchain: ethereum \ No newline at end of file diff --git a/src/test/resources/dshackle-proxy-max.yaml b/src/test/resources/dshackle-proxy-max.yaml new file mode 100644 index 00000000..1b9fa1a0 --- /dev/null +++ b/src/test/resources/dshackle-proxy-max.yaml @@ -0,0 +1,9 @@ +proxy: + enabled: true + host: 0.0.0.0 + port: 8080 + routes: + - id: ethereum + blockchain: ethereum + - id: classic + blockchain: etc \ No newline at end of file diff --git a/src/test/resources/dshackle-proxy-two.yaml b/src/test/resources/dshackle-proxy-two.yaml new file mode 100644 index 00000000..5ac98e86 --- /dev/null +++ b/src/test/resources/dshackle-proxy-two.yaml @@ -0,0 +1,7 @@ +proxy: + port: 8080 + routes: + - id: ethereum + blockchain: ethereum + - id: classic + blockchain: etc \ No newline at end of file