diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index 70ab0ae5..d8ff18ee 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt @@ -15,9 +15,10 @@ */ package io.emeraldpay.dshackle.upstream.rpcclient -import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.reader.Reader +import io.infinitape.etherjar.rpc.RpcException +import io.infinitape.etherjar.rpc.RpcResponseError import io.netty.buffer.Unpooled import io.netty.handler.codec.http.HttpHeaderNames import io.netty.handler.codec.http.HttpHeaders @@ -25,7 +26,6 @@ import io.netty.handler.ssl.SslContextBuilder import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import reactor.netty.http.client.HttpClient -import reactor.netty.tcp.SslProvider import java.io.ByteArrayInputStream import java.security.KeyStore import java.security.cert.CertificateFactory @@ -86,9 +86,13 @@ class JsonRpcHttpClient( .uri(target) .send(Mono.just(request).map { Unpooled.wrappedBuffer(it) }) - return response.responseContent() - .aggregate() - .asByteArray() + return response.response { header, bytes -> + if (header.status().code() != 200) { + Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "HTTP Code: ${header.status().code()}")) + } else { + bytes.aggregate().asByteArray() + } + }.single() } override fun read(key: JsonRpcRequest): Mono { @@ -96,5 +100,13 @@ class JsonRpcHttpClient( .map(JsonRpcRequest::toJson) .flatMap(this@JsonRpcHttpClient::execute) .map(parser::parse) + .onErrorResume { t -> + val err = if (t is RpcException) { + JsonRpcResponse.error(t.code, t.rpcMessage) + } else { + JsonRpcResponse.error(1, t.message ?: t.javaClass.name) + } + Mono.just(err) + } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt index 6d043ac7..b19894bd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt @@ -16,6 +16,7 @@ package io.emeraldpay.dshackle.upstream.rpcclient import com.fasterxml.jackson.core.JsonFactory +import com.fasterxml.jackson.core.JsonParseException import com.fasterxml.jackson.core.JsonParser import com.fasterxml.jackson.core.JsonToken import io.infinitape.etherjar.rpc.RpcResponseError @@ -30,48 +31,52 @@ class JsonRpcParser() { private val jsonFactory = JsonFactory() fun parse(json: ByteArray): JsonRpcResponse { - val parser: JsonParser = jsonFactory.createParser(json) - parser.nextToken() - if (parser.currentToken != JsonToken.START_OBJECT) { - return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON")) - } - var nullResponse: JsonRpcResponse? = null - while (parser.nextToken() != JsonToken.END_OBJECT) { - val field = parser.currentName - if (field == "jsonrpc" || field == "id") { - if (!parser.nextToken().isScalarValue) { - return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (id/type)")) - } - // just skip the field - } else if (field == "result") { - val value = parser.nextToken() - val start = parser.tokenLocation - if (value.isScalarValue) { - val text = parser.text - if (value == JsonToken.VALUE_STRING) { - return JsonRpcResponse(("\"" + text + "\"").toByteArray(), null) - } else if (value == JsonToken.VALUE_NULL) { - //if null we should check if error is present - nullResponse = JsonRpcResponse(text.toByteArray(), null) - } else { - return JsonRpcResponse(text.toByteArray(), null) + try { + val parser: JsonParser = jsonFactory.createParser(json) + parser.nextToken() + if (parser.currentToken != JsonToken.START_OBJECT) { + return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON")) + } + var nullResponse: JsonRpcResponse? = null + while (parser.nextToken() != JsonToken.END_OBJECT) { + val field = parser.currentName + if (field == "jsonrpc" || field == "id") { + if (!parser.nextToken().isScalarValue) { + return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (id/type)")) + } + // just skip the field + } else if (field == "result") { + val value = parser.nextToken() + val start = parser.tokenLocation + if (value.isScalarValue) { + val text = parser.text + if (value == JsonToken.VALUE_STRING) { + return JsonRpcResponse(("\"" + text + "\"").toByteArray(), null) + } else if (value == JsonToken.VALUE_NULL) { + //if null we should check if error is present + nullResponse = JsonRpcResponse(text.toByteArray(), null) + } else { + return JsonRpcResponse(text.toByteArray(), null) + } + } else if (value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY) { + parser.skipChildren() + val end = parser.currentLocation.byteOffset.toInt() + val copy = ByteArray((end - start.byteOffset).toInt()) + System.arraycopy(json, start.byteOffset.toInt(), copy, 0, copy.size) + return JsonRpcResponse(copy, null) + } + } else if (field == "error") { + val err = readError(parser) + if (err != null) { + return JsonRpcResponse(null, err) } - } else if (value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY) { - parser.skipChildren() - val end = parser.currentLocation.byteOffset.toInt() - val copy = ByteArray((end - start.byteOffset).toInt()) - System.arraycopy(json, start.byteOffset.toInt(), copy, 0, copy.size) - return JsonRpcResponse(copy, null) - } - } else if (field == "error") { - val err = readError(parser) - if (err != null) { - return JsonRpcResponse(null, err) } } - } - if (nullResponse != null) { - return nullResponse + if (nullResponse != null) { + return nullResponse + } + } catch (e: JsonParseException) { + log.warn("Failed to parse JSON from upstream: ${e.message}") } return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON structure")) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy index 9b019b5b..db9a92b2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy @@ -17,6 +17,8 @@ package io.emeraldpay.dshackle.upstream.rpcclient import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.test.TestingCommons +import io.infinitape.etherjar.rpc.RpcException +import io.infinitape.etherjar.rpc.RpcResponseError import org.mockserver.integration.ClientAndServer import org.mockserver.model.HttpRequest import org.mockserver.model.HttpResponse @@ -88,4 +90,25 @@ class JsonRpcHttpClientSpec extends Specification { ) } + def "Produces RPC Exception on error status code"() { + setup: + def client = new JsonRpcHttpClient("localhost:18332", null, null) + + mockServer.when( + HttpRequest.request() + ).respond( + HttpResponse.response() + .withStatusCode(500) + .withBody("pong") + ) + when: + def act = client.execute("ping".bytes).map { new String(it) } + then: + StepVerifier.create(act) + .expectErrorMatches { t -> + t instanceof RpcException && t.code == RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE + } + .verify(Duration.ofSeconds(1)) + } + } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy index 0fa9b81d..11fae907 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.rpcclient +import io.infinitape.etherjar.rpc.RpcResponseError import spock.lang.Specification class JsonRpcParserSpec extends Specification { @@ -177,4 +178,16 @@ class JsonRpcParserSpec extends Specification { !act.hasResult() } + def "Handle non-json with producing an error response"() { + setup: + def json = 'NOT JSON' + when: + def act = parser.parse(json.getBytes()) + then: + act.error != null + act.error.code == RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE + act.hasError() + !act.hasResult() + } + }