problem: no response on invalid request to proxy

rel: #43

src/main/kotlin/io/emeraldpay/dshackle/TlsSetup.kt

@@ -28,7 +28,7 @@ import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 
 @Service
-class TlsSetup(
+open class TlsSetup(
         @Autowired val fileResolver: FileResolver
 ) {
 

src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt

@@ -21,7 +21,7 @@ import io.emeraldpay.grpc.Chain
 /**
  * Configure HTTP Proxy to Upstreams
  */
-class ProxyConfig {
+open class ProxyConfig {
 
     companion object {
         public const val CONFIG_ID = "parsed.proxy"

src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt

@@ -18,14 +18,19 @@ 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.rpc.NativeCall
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.infinitape.etherjar.rpc.RpcException
+import io.netty.buffer.ByteBuf
 import io.netty.buffer.Unpooled
 import io.netty.handler.ssl.SslContextBuilder
 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.DisposableServer
 import reactor.netty.http.server.HttpServer
@@ -90,15 +95,27 @@ class ProxyServer(
         }
     }
 
+    fun processRequest(chain: Common.ChainRef, request: Mono<ByteArray>): Flux<ByteBuf> {
+        return request.map(readRpcJson)
+                .flatMapMany { call -> execute(chain, call) }
+                .onErrorResume(RpcException::class.java) { err ->
+                    val id = err.details?.let {
+                        if (it is JsonRpcResponse.Id) it else JsonRpcResponse.IntId(-1)
+                    } ?: JsonRpcResponse.IntId(-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<HttpServerRequest, HttpServerResponse, Publisher<Void>> {
         val chain = Common.ChainRef.forNumber(routeConfig.blockchain.id)
         return BiFunction { req, resp ->
-            val results = req.receive()
+            val request = req.receive()
                     .aggregate()
                     .asByteArray()
-                    .map(readRpcJson)
-                    .flatMapMany { call -> execute(chain, call) }
-                    .map { Unpooled.wrappedBuffer(it.toByteArray()) }
+            val results = processRequest(chain, request)
             resp.addHeader(HttpHeaders.CONTENT_TYPE, "application/json")
                     .send(results)
         }

src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt

@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.RpcResponseError
 import io.infinitape.etherjar.rpc.json.RequestJson
@@ -36,8 +37,7 @@ import java.util.stream.Collectors
  * Reader for JSON RPC request
  */
 @Service
-open class ReadRpcJson(
-) : Function<ByteArray, ProxyCall> {
+open class ReadRpcJson() : Function<ByteArray, ProxyCall> {
 
     companion object {
         private val log = LoggerFactory.getLogger(ReadRpcJson::class.java)
@@ -49,18 +49,21 @@ open class ReadRpcJson(
 
     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")
+                throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID is not set")
             }
             val id = json["id"]
+            if ("2.0" != json["jsonrpc"]) {
+                if (json["jsonrpc"] == null) {
+                    throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "jsonrpc version is not set", id?.let { JsonRpcResponse.Id.from(it) })
+                }
+                throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Unsupported JSON RPC version: " + json["jsonrpc"].toString(), id?.let { JsonRpcResponse.Id.from(it) })
+            }
             if (!(json["method"] != null && json["method"] is String)) {
-                throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "ID not set")
+                throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Method is not set", id?.let { JsonRpcResponse.Id.from(it) })
             }
             if (json.containsKey("params") && json["params"] !is List<*>) {
-                throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Params must be an array")
+                throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Params must be an array", id?.let { JsonRpcResponse.Id.from(it) })
             }
             RequestJson<Any>(
                     json["method"].toString(),

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt

@@ -164,6 +164,19 @@ class JsonRpcResponse(
         override fun isInt(): Boolean {
             return true
         }
+
+        override fun equals(other: Any?): Boolean {
+            if (this === other) return true
+            if (other !is IntId) return false
+
+            if (id != other.id) return false
+
+            return true
+        }
+
+        override fun hashCode(): Int {
+            return id
+        }
     }
 
     class StringId(val id: String) : Id {
@@ -178,6 +191,20 @@ class JsonRpcResponse(
         override fun isInt(): Boolean {
             return false
         }
+
+        override fun equals(other: Any?): Boolean {
+            if (this === other) return true
+            if (other !is StringId) return false
+
+            if (id != other.id) return false
+
+            return true
+        }
+
+        override fun hashCode(): Int {
+            return id.hashCode()
+        }
+
     }
 
     class ResponseJsonSerializer : JsonSerializer<JsonRpcResponse>() {

src/test/groovy/io/emeraldpay/dshackle/proxy/ProxyServerSpec.groovy

@@ -22,7 +22,10 @@ import io.emeraldpay.dshackle.TlsSetup
 import io.emeraldpay.dshackle.config.ProxyConfig
 import io.emeraldpay.dshackle.rpc.NativeCall
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.infinitape.etherjar.rpc.RpcException
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 import reactor.test.StepVerifier
 import spock.lang.Specification
 
@@ -65,4 +68,24 @@ class ProxyServerSpec extends Specification {
                 .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.IntId(4)) }
+        }
+        def server = new ProxyServer(
+                Stub(ProxyConfig),
+                read,
+                Stub(WriteRpcJson), Stub(NativeCall), Stub(TlsSetup)
+        )
+        when:
+        def act = server.processRequest(Common.ChainRef.CHAIN_ETHEREUM, Mono.just("".bytes))
+                .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))
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/proxy/ReadRpcJsonSpec.groovy

@@ -17,6 +17,7 @@
 package io.emeraldpay.dshackle.proxy
 
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.infinitape.etherjar.rpc.RpcException
 import spock.lang.Specification
 
@@ -174,4 +175,53 @@ class ReadRpcJsonSpec extends Specification {
             payload.toStringUtf8() == '[143,false]'
         }
     }
+
+    def "Error if id is not set"() {
+        when:
+        reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "params":[]}'.bytes)
+        then:
+        def t = thrown(RpcException)
+        t.code == -32600
+        t.rpcMessage.toLowerCase() == "id is not set"
+    }
+
+    def "Error if jsonrpc is not set"() {
+        when:
+        reader.apply('{"id":2, "method":"net_peerCount", "params":[]}'.bytes)
+        then:
+        def t = thrown(RpcException)
+        t.code == -32600
+        t.rpcMessage.toLowerCase() == "jsonrpc version is not set"
+        t.details == new JsonRpcResponse.IntId(2)
+    }
+
+    def "Error if jsonrpc version is invalid"() {
+        when:
+        reader.apply('{"id":2, "jsonrpc":"3.0", "method":"net_peerCount", "params":[]}'.bytes)
+        then:
+        def t = thrown(RpcException)
+        t.code == -32600
+        t.rpcMessage.toLowerCase() == "unsupported json rpc version: 3.0"
+        t.details == new JsonRpcResponse.IntId(2)
+    }
+
+    def "Error if method is not set"() {
+        when:
+        reader.apply('{"id":2,  "jsonrpc":"2.0", "params":[]}'.bytes)
+        then:
+        def t = thrown(RpcException)
+        t.code == -32600
+        t.rpcMessage.toLowerCase() == "method is not set"
+        t.details == new JsonRpcResponse.IntId(2)
+    }
+
+    def "Error if params is not array"() {
+        when:
+        reader.apply('{"id":2, "jsonrpc":"2.0", "method":"test", "params":123}'.bytes)
+        then:
+        def t = thrown(RpcException)
+        t.code == -32600
+        t.rpcMessage.toLowerCase() == "params must be an array"
+        t.details == new JsonRpcResponse.IntId(2)
+    }
 }

testing/dshackle/dshackle.yaml

@@ -8,6 +8,9 @@ cluster:
   upstreams:
     - id: test-1
       chain: ethereum
+      methods:
+        enabled:
+          - name: debug_traceTransaction
       options:
         disable-validation: true
       connection:

testing/simple-upstream/src/main/groovy/testing/BlocksHandler.groovy

@@ -5,9 +5,11 @@ import com.fasterxml.jackson.databind.ObjectMapper
 class BlocksHandler implements CallHandler {
 
     ObjectMapper objectMapper
+    ResourceResponse resourceResponse
 
     BlocksHandler(ObjectMapper objectMapper) {
         this.objectMapper = objectMapper
+        this.resourceResponse = new ResourceResponse(objectMapper)
     }
 
     @Override
@@ -18,22 +20,13 @@ class BlocksHandler implements CallHandler {
         if (method == "eth_getBlockByNumber") {
             String blockId = params[0]
             println("get block $blockId")
-            def result = getResource("block-${blockId}.json")
-            return Result.ok(result)
+            return resourceResponse.respondWith("block-${blockId}.json")
         }
         if (method == "eth_getTransactionByHash") {
             String txId = params[0]
-            def result = getResource("tx-${txId}.json")
-            return Result.ok(result)
+            return resourceResponse.respondWith("tx-${txId}.json")
         }
         return null
     }
 
-    Object getResource(String name) {
-        String json = BlocksHandler.class.getResourceAsStream("/" + name)?.text
-        if (json == null) {
-            return null
-        }
-        return objectMapper.readValue(json, Map)
-    }
 }

testing/simple-upstream/src/main/groovy/testing/ResourceResponse.groovy

@@ -0,0 +1,25 @@
+package testing
+
+import com.fasterxml.jackson.databind.ObjectMapper
+
+class ResourceResponse {
+
+    ObjectMapper objectMapper
+
+    ResourceResponse(ObjectMapper objectMapper) {
+        this.objectMapper = objectMapper
+    }
+
+    Object getResource(String name) {
+        String json = BlocksHandler.class.getResourceAsStream("/" + name)?.text
+        if (json == null) {
+            return null
+        }
+        return objectMapper.readValue(json, Map)
+    }
+
+    CallHandler.Result respondWith(String name) {
+        return CallHandler.Result.ok(getResource(name))
+    }
+
+}

testing/simple-upstream/src/main/groovy/testing/SimpleUpstream.groovy

@@ -13,7 +13,7 @@ class SimpleUpstream {
     void prepare() {
         objectMapper = new ObjectMapper()
 
-        handlers << new TestcaseHandler()
+        handlers << new TestcaseHandler(objectMapper)
         handlers << new CommonHandlers()
         handlers << new BlocksHandler(objectMapper)
         handlers << new InvalidCallHandler()

testing/simple-upstream/src/main/groovy/testing/TestcaseHandler.groovy

@@ -1,14 +1,29 @@
 package testing
 
+import com.fasterxml.jackson.databind.ObjectMapper
+
 class TestcaseHandler implements CallHandler {
 
+    ObjectMapper objectMapper
+    ResourceResponse resourceResponse
+
+    TestcaseHandler(ObjectMapper objectMapper) {
+        this.objectMapper = objectMapper
+        this.resourceResponse = new ResourceResponse(objectMapper)
+    }
+
     @Override
     Result handle(String method, List<Object> params) {
         // https://github.com/emeraldpay/dshackle/issues/35
         if (method == "eth_call"
                 && params[0].to?.toLowerCase() == "0x542156d51D10Db5acCB99f9Db7e7C91B74E80a2c".toLowerCase()) {
             return Result.error(-32015, "VM execution error.")
         }
+        // https://github.com/emeraldpay/dshackle/issues/43
+        if (method == "debug_traceTransaction"
+                && params[0].toLowerCase() == "0xd949bc0fe1a5d16f4522bc47933554dcc4ada0493ff71ee1973b2410257af9fe".toLowerCase()) {
+            return resourceResponse.respondWith("trace-0xd949bc.json")
+        }
         return null
     }
 }