mirror of https://github.com/hyperledger/besu
Add IPC transport for JSON-RPC APIs (#3695)
* Fix json-rpc HTTP tests [#535] The `o.h.b.e.a.j.JsonRpcHttpServiceTest#exceptionallyHandleJsonSingleRequest` and `o.h.b.e.a.j.JsonRpcHttpServiceTest#exceptionallyHandleJsonBatchRequest` tests were throwing ClassCastException in `o.h.b.e.a.j.JsonRpcHttpService#validateMethodAvailability` which wasn't ever catched, returning status 500 by default, but that wasn't the use case aimed to test. Another test running an exceptional method is `o.h.b.t.a.p.EnclaveErrorAcceptanceTest#whenEnclaveIsDisconnectedGetReceiptReturnsInternalError` which validates an "Internal Error" proper json-rpc response. I changed the first two tests to be consistent with the later one. * Extract json-rpc HTTP authentication to a handler [#535] * Replace TimeoutHandler in GraphQLHttpService with Vert.x's impl [#535] * Extract json-rpc HTTP parser to a handler [#535] * Refactor json-rpc WS handler [#535] * Add json-rpc IPC support [#535] Signed-off-by: Diego López León <dieguitoll@gmail.com> Signed-off-by: Diego López León <dieguitoll@gmail.com>pull/3768/head
Diego López León
3 years ago
committed by
GitHub
parent
bf349a4520
commit
28845823a4
@ -0,0 +1,74 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.tests.acceptance.jsonrpc.ipc; |
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
import static org.assertj.core.api.Assertions.fail; |
||||
import static org.junit.Assume.assumeTrue; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.RpcApis; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.ipc.JsonRpcIpcConfiguration; |
||||
import org.hyperledger.besu.tests.acceptance.dsl.AcceptanceTestBase; |
||||
import org.hyperledger.besu.tests.acceptance.dsl.node.Node; |
||||
|
||||
import java.io.IOException; |
||||
import java.nio.file.Files; |
||||
import java.nio.file.Path; |
||||
import java.util.Collections; |
||||
import java.util.Locale; |
||||
|
||||
import org.junit.Before; |
||||
import org.junit.Test; |
||||
import org.web3j.protocol.Web3j; |
||||
import org.web3j.protocol.core.Request; |
||||
import org.web3j.protocol.core.methods.response.NetVersion; |
||||
import org.web3j.protocol.ipc.UnixIpcService; |
||||
|
||||
public class Web3JSupportAcceptanceTest extends AcceptanceTestBase { |
||||
|
||||
private Node node; |
||||
private Path socketPath; |
||||
|
||||
@Before |
||||
public void setUp() throws Exception { |
||||
socketPath = Files.createTempFile("besu-test-", ".ipc"); |
||||
node = |
||||
besu.createNode( |
||||
"node1", |
||||
(configurationBuilder) -> |
||||
configurationBuilder.jsonRpcIpcConfiguration( |
||||
new JsonRpcIpcConfiguration( |
||||
true, socketPath, Collections.singletonList(RpcApis.NET.name())))); |
||||
cluster.start(node); |
||||
} |
||||
|
||||
@Test |
||||
public void netVersionCall() { |
||||
final String osName = System.getProperty("os.name").toLowerCase(Locale.ENGLISH); |
||||
assumeTrue(osName.contains("mac") || osName.contains("linux")); |
||||
|
||||
final Web3j web3 = Web3j.build(new UnixIpcService(socketPath.toString())); |
||||
final Request<?, NetVersion> ethBlockNumberRequest = web3.netVersion(); |
||||
node.verify( |
||||
node -> { |
||||
try { |
||||
assertThat(ethBlockNumberRequest.send().getNetVersion()) |
||||
.isEqualTo(String.valueOf(2018)); |
||||
} catch (IOException e) { |
||||
fail("Web3J net_version call failed", e); |
||||
} |
||||
}); |
||||
} |
||||
} |
||||
@ -0,0 +1,71 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.cli.options.unstable; |
||||
|
||||
import static org.hyperledger.besu.ethereum.api.jsonrpc.RpcApis.DEFAULT_RPC_APIS; |
||||
|
||||
import java.nio.file.Path; |
||||
import java.util.List; |
||||
|
||||
import picocli.CommandLine; |
||||
|
||||
public class IpcOptions { |
||||
private static final String DEFAULT_IPC_FILE = "besu.ipc"; |
||||
|
||||
public static IpcOptions create() { |
||||
return new IpcOptions(); |
||||
} |
||||
|
||||
public static Path getDefaultPath(final Path dataDir) { |
||||
return dataDir.resolve(DEFAULT_IPC_FILE); |
||||
} |
||||
|
||||
@CommandLine.Option( |
||||
names = {"--Xrpc-ipc-enabled"}, |
||||
hidden = true, |
||||
description = "Set to start the JSON-RPC IPC service (default: ${DEFAULT-VALUE})") |
||||
private final Boolean enabled = false; |
||||
|
||||
@CommandLine.Option( |
||||
names = {"--Xrpc-ipc-path"}, |
||||
hidden = true, |
||||
description = |
||||
"IPC socket/pipe file (default: a file named \"" |
||||
+ DEFAULT_IPC_FILE |
||||
+ "\" in the Besu data directory)") |
||||
private Path ipcPath; |
||||
|
||||
@CommandLine.Option( |
||||
names = {"--Xrpc-ipc-api", "--Xrpc-ipc-apis"}, |
||||
hidden = true, |
||||
paramLabel = "<api name>", |
||||
split = " {0,1}, {0,1}", |
||||
arity = "1..*", |
||||
description = |
||||
"Comma separated list of APIs to enable on JSON-RPC IPC service (default: ${DEFAULT-VALUE})") |
||||
private final List<String> rpcIpcApis = DEFAULT_RPC_APIS; |
||||
|
||||
public Boolean isEnabled() { |
||||
return enabled; |
||||
} |
||||
|
||||
public Path getIpcPath() { |
||||
return ipcPath; |
||||
} |
||||
|
||||
public List<String> getRpcIpcApis() { |
||||
return rpcIpcApis; |
||||
} |
||||
} |
||||
@ -0,0 +1,64 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.handlers; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.authentication.AuthenticationService; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.authentication.AuthenticationUtils; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.context.ContextKey; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcErrorResponse; |
||||
|
||||
import java.util.Collection; |
||||
|
||||
import io.netty.handler.codec.http.HttpResponseStatus; |
||||
import io.vertx.core.Handler; |
||||
import io.vertx.core.http.HttpServerResponse; |
||||
import io.vertx.core.json.Json; |
||||
import io.vertx.ext.web.RoutingContext; |
||||
|
||||
public class AuthenticationHandler { |
||||
|
||||
private AuthenticationHandler() {} |
||||
|
||||
public static Handler<RoutingContext> handler( |
||||
final AuthenticationService authenticationService, final Collection<String> noAuthRpcApis) { |
||||
return ctx -> { |
||||
// first check token if authentication is required
|
||||
final String token = getAuthToken(ctx); |
||||
if (token == null && noAuthRpcApis.isEmpty()) { |
||||
// no auth token when auth required
|
||||
handleJsonRpcUnauthorizedError(ctx); |
||||
} else { |
||||
authenticationService.authenticate( |
||||
token, user -> ctx.put(ContextKey.AUTHENTICATED_USER.name(), user)); |
||||
ctx.next(); |
||||
} |
||||
}; |
||||
} |
||||
|
||||
private static String getAuthToken(final RoutingContext routingContext) { |
||||
return AuthenticationUtils.getJwtTokenFromAuthorizationHeaderValue( |
||||
routingContext.request().getHeader("Authorization")); |
||||
} |
||||
|
||||
private static void handleJsonRpcUnauthorizedError(final RoutingContext routingContext) { |
||||
final HttpServerResponse response = routingContext.response(); |
||||
if (!response.closed()) { |
||||
response |
||||
.setStatusCode(HttpResponseStatus.UNAUTHORIZED.code()) |
||||
.end(Json.encode(new JsonRpcErrorResponse(null, JsonRpcError.UNAUTHORIZED))); |
||||
} |
||||
} |
||||
} |
||||
@ -1,19 +0,0 @@ |
||||
/* |
||||
* Copyright ConsenSys AG. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.handlers; |
||||
|
||||
public enum HandlerName { |
||||
TIMEOUT |
||||
} |
||||
@ -0,0 +1,173 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.handlers; |
||||
|
||||
import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError.INVALID_REQUEST; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.JsonResponseStreamer; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.context.ContextKey; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.execution.JsonRpcExecutor; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequest; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcErrorResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponseType; |
||||
|
||||
import java.io.IOException; |
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
import java.util.Optional; |
||||
|
||||
import com.fasterxml.jackson.core.JsonGenerator; |
||||
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
import com.fasterxml.jackson.databind.ObjectWriter; |
||||
import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; |
||||
import io.netty.handler.codec.http.HttpResponseStatus; |
||||
import io.opentelemetry.api.trace.Tracer; |
||||
import io.opentelemetry.context.Context; |
||||
import io.vertx.core.Handler; |
||||
import io.vertx.core.http.HttpServerResponse; |
||||
import io.vertx.core.json.Json; |
||||
import io.vertx.core.json.JsonArray; |
||||
import io.vertx.core.json.JsonObject; |
||||
import io.vertx.ext.auth.User; |
||||
import io.vertx.ext.web.RoutingContext; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
public class JsonRpcExecutorHandler { |
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(JsonRpcExecutorHandler.class); |
||||
private static final String SPAN_CONTEXT = "span_context"; |
||||
private static final String APPLICATION_JSON = "application/json"; |
||||
private static final ObjectWriter JSON_OBJECT_WRITER = |
||||
new ObjectMapper() |
||||
.registerModule(new Jdk8Module()) // Handle JDK8 Optionals (de)serialization
|
||||
.writerWithDefaultPrettyPrinter() |
||||
.without(JsonGenerator.Feature.FLUSH_PASSED_TO_STREAM) |
||||
.with(JsonGenerator.Feature.AUTO_CLOSE_TARGET); |
||||
|
||||
private JsonRpcExecutorHandler() {} |
||||
|
||||
public static Handler<RoutingContext> handler( |
||||
final JsonRpcExecutor jsonRpcExecutor, final Tracer tracer) { |
||||
return ctx -> { |
||||
HttpServerResponse response = ctx.response(); |
||||
try { |
||||
Optional<User> user = ContextKey.AUTHENTICATED_USER.extractFrom(ctx, Optional::empty); |
||||
Context spanContext = ctx.get(SPAN_CONTEXT); |
||||
response = response.putHeader("Content-Type", APPLICATION_JSON); |
||||
|
||||
if (ctx.data().containsKey(ContextKey.REQUEST_BODY_AS_JSON_OBJECT.name())) { |
||||
JsonObject jsonRequest = ctx.get(ContextKey.REQUEST_BODY_AS_JSON_OBJECT.name()); |
||||
JsonRpcResponse jsonRpcResponse = |
||||
jsonRpcExecutor.execute( |
||||
user, |
||||
tracer, |
||||
spanContext, |
||||
() -> !ctx.response().closed(), |
||||
jsonRequest, |
||||
req -> req.mapTo(JsonRpcRequest.class)); |
||||
response.setStatusCode(status(jsonRpcResponse).code()); |
||||
if (jsonRpcResponse.getType() == JsonRpcResponseType.NONE) { |
||||
response.end(); |
||||
} else { |
||||
try (final JsonResponseStreamer streamer = |
||||
new JsonResponseStreamer(response, ctx.request().remoteAddress())) { |
||||
// underlying output stream lifecycle is managed by the json object writer
|
||||
JSON_OBJECT_WRITER.writeValue(streamer, jsonRpcResponse); |
||||
} |
||||
} |
||||
} else if (ctx.data().containsKey(ContextKey.REQUEST_BODY_AS_JSON_ARRAY.name())) { |
||||
JsonArray batchJsonRequest = ctx.get(ContextKey.REQUEST_BODY_AS_JSON_ARRAY.name()); |
||||
List<JsonRpcResponse> jsonRpcBatchResponse; |
||||
try { |
||||
List<JsonRpcResponse> responses = new ArrayList<>(); |
||||
for (int i = 0; i < batchJsonRequest.size(); i++) { |
||||
final JsonObject jsonRequest; |
||||
try { |
||||
jsonRequest = batchJsonRequest.getJsonObject(i); |
||||
} catch (ClassCastException e) { |
||||
responses.add(new JsonRpcErrorResponse(null, INVALID_REQUEST)); |
||||
continue; |
||||
} |
||||
responses.add( |
||||
jsonRpcExecutor.execute( |
||||
user, |
||||
tracer, |
||||
spanContext, |
||||
() -> !ctx.response().closed(), |
||||
jsonRequest, |
||||
req -> req.mapTo(JsonRpcRequest.class))); |
||||
} |
||||
jsonRpcBatchResponse = responses; |
||||
} catch (RuntimeException e) { |
||||
response.setStatusCode(HttpResponseStatus.BAD_REQUEST.code()).end(); |
||||
return; |
||||
} |
||||
final JsonRpcResponse[] completed = |
||||
jsonRpcBatchResponse.stream() |
||||
.filter(jsonRpcResponse -> jsonRpcResponse.getType() != JsonRpcResponseType.NONE) |
||||
.toArray(JsonRpcResponse[]::new); |
||||
try (final JsonResponseStreamer streamer = |
||||
new JsonResponseStreamer(response, ctx.request().remoteAddress())) { |
||||
// underlying output stream lifecycle is managed by the json object writer
|
||||
JSON_OBJECT_WRITER.writeValue(streamer, completed); |
||||
} |
||||
} else { |
||||
handleJsonRpcError(ctx, null, JsonRpcError.PARSE_ERROR); |
||||
} |
||||
} catch (IOException ex) { |
||||
LOG.error("Error streaming JSON-RPC response", ex); |
||||
} catch (RuntimeException e) { |
||||
handleJsonRpcError(ctx, null, JsonRpcError.INTERNAL_ERROR); |
||||
} |
||||
}; |
||||
} |
||||
|
||||
private static void handleJsonRpcError( |
||||
final RoutingContext routingContext, final Object id, final JsonRpcError error) { |
||||
final HttpServerResponse response = routingContext.response(); |
||||
if (!response.closed()) { |
||||
response |
||||
.setStatusCode(statusCodeFromError(error).code()) |
||||
.end(Json.encode(new JsonRpcErrorResponse(id, error))); |
||||
} |
||||
} |
||||
|
||||
private static HttpResponseStatus status(final JsonRpcResponse response) { |
||||
switch (response.getType()) { |
||||
case UNAUTHORIZED: |
||||
return HttpResponseStatus.UNAUTHORIZED; |
||||
case ERROR: |
||||
return statusCodeFromError(((JsonRpcErrorResponse) response).getError()); |
||||
case SUCCESS: |
||||
case NONE: |
||||
default: |
||||
return HttpResponseStatus.OK; |
||||
} |
||||
} |
||||
|
||||
private static HttpResponseStatus statusCodeFromError(final JsonRpcError error) { |
||||
switch (error) { |
||||
case INVALID_REQUEST: |
||||
case INVALID_PARAMS: |
||||
case PARSE_ERROR: |
||||
return HttpResponseStatus.BAD_REQUEST; |
||||
default: |
||||
return HttpResponseStatus.OK; |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,68 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.handlers; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.context.ContextKey; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcErrorResponse; |
||||
|
||||
import io.netty.handler.codec.http.HttpResponseStatus; |
||||
import io.vertx.core.Handler; |
||||
import io.vertx.core.http.HttpServerResponse; |
||||
import io.vertx.core.json.DecodeException; |
||||
import io.vertx.core.json.Json; |
||||
import io.vertx.core.json.JsonArray; |
||||
import io.vertx.ext.web.RoutingContext; |
||||
|
||||
public class JsonRpcParserHandler { |
||||
|
||||
private JsonRpcParserHandler() {} |
||||
|
||||
public static Handler<RoutingContext> handler() { |
||||
return ctx -> { |
||||
final HttpServerResponse response = ctx.response(); |
||||
if (ctx.getBody() == null) { |
||||
errorResponse(response, JsonRpcError.PARSE_ERROR); |
||||
} else { |
||||
try { |
||||
ctx.put(ContextKey.REQUEST_BODY_AS_JSON_OBJECT.name(), ctx.getBodyAsJson()); |
||||
} catch (DecodeException jsonObjectDecodeException) { |
||||
try { |
||||
final JsonArray batchRequest = ctx.getBodyAsJsonArray(); |
||||
if (batchRequest.isEmpty()) { |
||||
errorResponse(response, JsonRpcError.INVALID_REQUEST); |
||||
return; |
||||
} else { |
||||
ctx.put(ContextKey.REQUEST_BODY_AS_JSON_ARRAY.name(), batchRequest); |
||||
} |
||||
} catch (DecodeException jsonArrayDecodeException) { |
||||
errorResponse(response, JsonRpcError.PARSE_ERROR); |
||||
return; |
||||
} |
||||
} |
||||
ctx.next(); |
||||
} |
||||
}; |
||||
} |
||||
|
||||
private static void errorResponse( |
||||
final HttpServerResponse response, final JsonRpcError rpcError) { |
||||
if (!response.closed()) { |
||||
response |
||||
.setStatusCode(HttpResponseStatus.BAD_REQUEST.code()) |
||||
.end(Json.encode(new JsonRpcErrorResponse(null, rpcError))); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,55 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.execution; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.authentication.AuthenticationService; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestId; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcUnauthorizedResponse; |
||||
|
||||
import java.util.Collection; |
||||
|
||||
import io.opentelemetry.api.trace.Span; |
||||
|
||||
public class AuthenticatedJsonRpcProcessor implements JsonRpcProcessor { |
||||
|
||||
private final JsonRpcProcessor rpcProcessor; |
||||
private final AuthenticationService authenticationService; |
||||
private final Collection<String> noAuthRpcApis; |
||||
|
||||
public AuthenticatedJsonRpcProcessor( |
||||
final JsonRpcProcessor rpcProcessor, |
||||
final AuthenticationService authenticationService, |
||||
final Collection<String> noAuthRpcApis) { |
||||
this.rpcProcessor = rpcProcessor; |
||||
this.authenticationService = authenticationService; |
||||
this.noAuthRpcApis = noAuthRpcApis; |
||||
} |
||||
|
||||
@Override |
||||
public JsonRpcResponse process( |
||||
final JsonRpcRequestId id, |
||||
final JsonRpcMethod method, |
||||
final Span metricSpan, |
||||
final JsonRpcRequestContext request) { |
||||
if (authenticationService.isPermitted(request.getUser(), method, noAuthRpcApis)) { |
||||
return rpcProcessor.process(id, method, metricSpan, request); |
||||
} |
||||
return new JsonRpcUnauthorizedResponse(id, JsonRpcError.UNAUTHORIZED); |
||||
} |
||||
} |
||||
@ -0,0 +1,52 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.execution; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestId; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.exception.InvalidJsonRpcParameters; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcErrorResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcUnauthorizedResponse; |
||||
import org.hyperledger.besu.ethereum.privacy.MultiTenancyValidationException; |
||||
|
||||
import io.opentelemetry.api.trace.Span; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
public class BaseJsonRpcProcessor implements JsonRpcProcessor { |
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(BaseJsonRpcProcessor.class); |
||||
|
||||
@Override |
||||
public JsonRpcResponse process( |
||||
final JsonRpcRequestId id, |
||||
final JsonRpcMethod method, |
||||
final Span metricSpan, |
||||
final JsonRpcRequestContext request) { |
||||
try { |
||||
return method.response(request); |
||||
} catch (final InvalidJsonRpcParameters e) { |
||||
LOG.debug("Invalid Params for method: {}", method.getName(), e); |
||||
return new JsonRpcErrorResponse(id, JsonRpcError.INVALID_PARAMS); |
||||
} catch (final MultiTenancyValidationException e) { |
||||
return new JsonRpcUnauthorizedResponse(id, JsonRpcError.UNAUTHORIZED); |
||||
} catch (final RuntimeException e) { |
||||
return new JsonRpcErrorResponse(id, JsonRpcError.INTERNAL_ERROR); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,120 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.execution; |
||||
|
||||
import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError.INVALID_REQUEST; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.RpcMethod; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequest; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestId; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcErrorResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcNoResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse; |
||||
|
||||
import java.util.Map; |
||||
import java.util.Optional; |
||||
import java.util.function.Function; |
||||
import java.util.function.Supplier; |
||||
|
||||
import io.opentelemetry.api.trace.Span; |
||||
import io.opentelemetry.api.trace.SpanKind; |
||||
import io.opentelemetry.api.trace.StatusCode; |
||||
import io.opentelemetry.api.trace.Tracer; |
||||
import io.opentelemetry.context.Context; |
||||
import io.vertx.core.json.JsonObject; |
||||
import io.vertx.ext.auth.User; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
public class JsonRpcExecutor { |
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(JsonRpcExecutor.class); |
||||
|
||||
private final JsonRpcProcessor rpcProcessor; |
||||
private final Map<String, JsonRpcMethod> rpcMethods; |
||||
|
||||
public JsonRpcExecutor( |
||||
final JsonRpcProcessor rpcProcessor, final Map<String, JsonRpcMethod> rpcMethods) { |
||||
this.rpcProcessor = rpcProcessor; |
||||
this.rpcMethods = rpcMethods; |
||||
} |
||||
|
||||
public JsonRpcResponse execute( |
||||
final Optional<User> optionalUser, |
||||
final Tracer tracer, |
||||
final Context spanContext, |
||||
final Supplier<Boolean> alive, |
||||
final JsonObject jsonRpcRequest, |
||||
final Function<JsonObject, JsonRpcRequest> requestBodyProvider) { |
||||
try { |
||||
final JsonRpcRequest requestBody = requestBodyProvider.apply(jsonRpcRequest); |
||||
final JsonRpcRequestId id = new JsonRpcRequestId(requestBody.getId()); |
||||
// Handle notifications
|
||||
if (requestBody.isNotification()) { |
||||
// Notifications aren't handled so create empty result for now.
|
||||
return new JsonRpcNoResponse(); |
||||
} |
||||
final Span span; |
||||
if (tracer != null) { |
||||
span = |
||||
tracer |
||||
.spanBuilder(requestBody.getMethod()) |
||||
.setSpanKind(SpanKind.INTERNAL) |
||||
.setParent(spanContext) |
||||
.startSpan(); |
||||
} else { |
||||
span = Span.getInvalid(); |
||||
} |
||||
final Optional<JsonRpcError> unavailableMethod = validateMethodAvailability(requestBody); |
||||
if (unavailableMethod.isPresent()) { |
||||
span.setStatus(StatusCode.ERROR, "method unavailable"); |
||||
return new JsonRpcErrorResponse(id, unavailableMethod.get()); |
||||
} |
||||
|
||||
final JsonRpcMethod method = rpcMethods.get(requestBody.getMethod()); |
||||
|
||||
return rpcProcessor.process( |
||||
id, method, span, new JsonRpcRequestContext(requestBody, optionalUser, alive)); |
||||
} catch (IllegalArgumentException e) { |
||||
try { |
||||
final Integer id = jsonRpcRequest.getInteger("id", null); |
||||
return new JsonRpcErrorResponse(id, INVALID_REQUEST); |
||||
} catch (ClassCastException idNotIntegerException) { |
||||
return new JsonRpcErrorResponse(null, INVALID_REQUEST); |
||||
} |
||||
} |
||||
} |
||||
|
||||
private Optional<JsonRpcError> validateMethodAvailability(final JsonRpcRequest request) { |
||||
final String name = request.getMethod(); |
||||
LOG.debug("JSON-RPC request -> {} {}", name, request.getParams()); |
||||
|
||||
final JsonRpcMethod method = rpcMethods.get(name); |
||||
|
||||
if (method == null) { |
||||
if (!RpcMethod.rpcMethodExists(name)) { |
||||
return Optional.of(JsonRpcError.METHOD_NOT_FOUND); |
||||
} |
||||
if (!rpcMethods.containsKey(name)) { |
||||
return Optional.of(JsonRpcError.METHOD_NOT_ENABLED); |
||||
} |
||||
} |
||||
|
||||
return Optional.empty(); |
||||
} |
||||
} |
||||
@ -0,0 +1,30 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.execution; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestId; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse; |
||||
|
||||
import io.opentelemetry.api.trace.Span; |
||||
|
||||
public interface JsonRpcProcessor { |
||||
JsonRpcResponse process( |
||||
final JsonRpcRequestId id, |
||||
final JsonRpcMethod method, |
||||
final Span metricSpan, |
||||
final JsonRpcRequestContext request); |
||||
} |
||||
@ -0,0 +1,48 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.execution; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestId; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse; |
||||
import org.hyperledger.besu.plugin.services.metrics.LabelledMetric; |
||||
import org.hyperledger.besu.plugin.services.metrics.OperationTimer; |
||||
|
||||
import io.opentelemetry.api.trace.Span; |
||||
|
||||
public class TimedJsonRpcProcessor implements JsonRpcProcessor { |
||||
|
||||
private final JsonRpcProcessor rpcProcessor; |
||||
private final LabelledMetric<OperationTimer> requestTimer; |
||||
|
||||
public TimedJsonRpcProcessor( |
||||
final JsonRpcProcessor rpcProcessor, final LabelledMetric<OperationTimer> requestTimer) { |
||||
this.rpcProcessor = rpcProcessor; |
||||
this.requestTimer = requestTimer; |
||||
} |
||||
|
||||
@Override |
||||
public JsonRpcResponse process( |
||||
final JsonRpcRequestId id, |
||||
final JsonRpcMethod method, |
||||
final Span metricSpan, |
||||
final JsonRpcRequestContext request) { |
||||
try (final OperationTimer.TimingContext ignored = |
||||
requestTimer.labels(request.getRequest().getMethod()).startTimer()) { |
||||
return rpcProcessor.process(id, method, metricSpan, request); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,61 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.execution; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestContext; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequestId; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcErrorResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponseType; |
||||
|
||||
import io.opentelemetry.api.trace.Span; |
||||
import io.opentelemetry.api.trace.StatusCode; |
||||
|
||||
public class TracedJsonRpcProcessor implements JsonRpcProcessor { |
||||
|
||||
private final JsonRpcProcessor rpcProcessor; |
||||
|
||||
public TracedJsonRpcProcessor(final JsonRpcProcessor rpcProcessor) { |
||||
this.rpcProcessor = rpcProcessor; |
||||
} |
||||
|
||||
@Override |
||||
public JsonRpcResponse process( |
||||
final JsonRpcRequestId id, |
||||
final JsonRpcMethod method, |
||||
final Span metricSpan, |
||||
final JsonRpcRequestContext request) { |
||||
JsonRpcResponse jsonRpcResponse = rpcProcessor.process(id, method, metricSpan, request); |
||||
if (JsonRpcResponseType.ERROR == jsonRpcResponse.getType()) { |
||||
JsonRpcErrorResponse errorResponse = (JsonRpcErrorResponse) jsonRpcResponse; |
||||
switch (errorResponse.getError()) { |
||||
case INVALID_PARAMS: |
||||
metricSpan.setStatus(StatusCode.ERROR, "Invalid Params"); |
||||
break; |
||||
case UNAUTHORIZED: |
||||
metricSpan.setStatus(StatusCode.ERROR, "Unauthorized"); |
||||
break; |
||||
case INTERNAL_ERROR: |
||||
metricSpan.setStatus(StatusCode.ERROR, "Error processing JSON-RPC requestBody"); |
||||
break; |
||||
default: |
||||
metricSpan.setStatus(StatusCode.ERROR, "Unexpected error"); |
||||
} |
||||
} |
||||
metricSpan.end(); |
||||
return jsonRpcResponse; |
||||
} |
||||
} |
||||
@ -0,0 +1,51 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.ipc; |
||||
|
||||
import java.nio.file.Path; |
||||
import java.util.Collection; |
||||
import java.util.Collections; |
||||
|
||||
public class JsonRpcIpcConfiguration { |
||||
|
||||
private final boolean enabled; |
||||
private final Path ipcPath; |
||||
private final Collection<String> enabledApis; |
||||
|
||||
public JsonRpcIpcConfiguration() { |
||||
enabled = false; |
||||
ipcPath = null; |
||||
enabledApis = Collections.emptyList(); |
||||
} |
||||
|
||||
public JsonRpcIpcConfiguration( |
||||
final boolean enabled, final Path ipcPath, final Collection<String> enabledApis) { |
||||
this.enabled = enabled; |
||||
this.ipcPath = ipcPath; |
||||
this.enabledApis = enabledApis; |
||||
} |
||||
|
||||
public boolean isEnabled() { |
||||
return enabled; |
||||
} |
||||
|
||||
public Path getPath() { |
||||
return ipcPath; |
||||
} |
||||
|
||||
public Collection<String> getEnabledApis() { |
||||
return enabledApis; |
||||
} |
||||
} |
||||
@ -0,0 +1,210 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.ipc; |
||||
|
||||
import static org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError.INVALID_REQUEST; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.execution.JsonRpcExecutor; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequest; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcError; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcErrorResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponseType; |
||||
|
||||
import java.io.IOException; |
||||
import java.nio.file.Files; |
||||
import java.nio.file.Path; |
||||
import java.util.ArrayList; |
||||
import java.util.List; |
||||
import java.util.Optional; |
||||
import java.util.concurrent.atomic.AtomicBoolean; |
||||
|
||||
import com.fasterxml.jackson.core.JsonGenerator; |
||||
import com.fasterxml.jackson.core.JsonProcessingException; |
||||
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
import com.fasterxml.jackson.databind.ObjectWriter; |
||||
import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; |
||||
import io.vertx.core.Future; |
||||
import io.vertx.core.Vertx; |
||||
import io.vertx.core.buffer.Buffer; |
||||
import io.vertx.core.json.DecodeException; |
||||
import io.vertx.core.json.Json; |
||||
import io.vertx.core.json.JsonArray; |
||||
import io.vertx.core.json.JsonObject; |
||||
import io.vertx.core.net.NetServer; |
||||
import io.vertx.core.net.NetServerOptions; |
||||
import io.vertx.core.net.NetSocket; |
||||
import io.vertx.core.net.SocketAddress; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
public class JsonRpcIpcService { |
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(JsonRpcIpcService.class); |
||||
private static final ObjectWriter JSON_OBJECT_WRITER = |
||||
new ObjectMapper() |
||||
.registerModule(new Jdk8Module()) // Handle JDK8 Optionals (de)serialization
|
||||
.writer() |
||||
.without(JsonGenerator.Feature.FLUSH_PASSED_TO_STREAM) |
||||
.with(JsonGenerator.Feature.AUTO_CLOSE_TARGET); |
||||
|
||||
private final Vertx vertx; |
||||
private final Path path; |
||||
private final JsonRpcExecutor jsonRpcExecutor; |
||||
private NetServer netServer; |
||||
|
||||
public JsonRpcIpcService(final Vertx vertx, final Path path, final JsonRpcExecutor rpcExecutor) { |
||||
this.vertx = vertx; |
||||
this.path = path; |
||||
this.jsonRpcExecutor = rpcExecutor; |
||||
} |
||||
|
||||
public Future<NetServer> start() { |
||||
netServer = vertx.createNetServer(buildNetServerOptions()); |
||||
netServer.connectHandler( |
||||
socket -> { |
||||
AtomicBoolean closedSocket = new AtomicBoolean(false); |
||||
socket |
||||
.closeHandler(unused -> closedSocket.set(true)) |
||||
.handler( |
||||
buffer -> { |
||||
if (buffer.length() == 0) { |
||||
errorReturn(socket, null, JsonRpcError.INVALID_REQUEST); |
||||
} else { |
||||
try { |
||||
final JsonObject jsonRpcRequest = buffer.toJsonObject(); |
||||
vertx |
||||
.<JsonRpcResponse>executeBlocking( |
||||
promise -> { |
||||
final JsonRpcResponse jsonRpcResponse = |
||||
jsonRpcExecutor.execute( |
||||
Optional.empty(), |
||||
null, |
||||
null, |
||||
closedSocket::get, |
||||
jsonRpcRequest, |
||||
req -> req.mapTo(JsonRpcRequest.class)); |
||||
promise.complete(jsonRpcResponse); |
||||
}) |
||||
.onSuccess( |
||||
jsonRpcResponse -> { |
||||
try { |
||||
socket.write( |
||||
JSON_OBJECT_WRITER.writeValueAsString(jsonRpcResponse) |
||||
+ '\n'); |
||||
} catch (JsonProcessingException e) { |
||||
LOG.error("Error streaming JSON-RPC response", e); |
||||
} |
||||
}) |
||||
.onFailure( |
||||
throwable -> { |
||||
try { |
||||
final Integer id = jsonRpcRequest.getInteger("id", null); |
||||
errorReturn(socket, id, JsonRpcError.INTERNAL_ERROR); |
||||
} catch (ClassCastException idNotIntegerException) { |
||||
errorReturn(socket, null, JsonRpcError.INTERNAL_ERROR); |
||||
} |
||||
}); |
||||
} catch (DecodeException jsonObjectDecodeException) { |
||||
try { |
||||
final JsonArray batchJsonRpcRequest = buffer.toJsonArray(); |
||||
if (batchJsonRpcRequest.isEmpty()) { |
||||
errorReturn(socket, null, JsonRpcError.INVALID_REQUEST); |
||||
} else { |
||||
vertx |
||||
.<List<JsonRpcResponse>>executeBlocking( |
||||
promise -> { |
||||
List<JsonRpcResponse> responses = new ArrayList<>(); |
||||
for (int i = 0; i < batchJsonRpcRequest.size(); i++) { |
||||
final JsonObject jsonRequest; |
||||
try { |
||||
jsonRequest = batchJsonRpcRequest.getJsonObject(i); |
||||
} catch (ClassCastException e) { |
||||
responses.add( |
||||
new JsonRpcErrorResponse(null, INVALID_REQUEST)); |
||||
continue; |
||||
} |
||||
responses.add( |
||||
jsonRpcExecutor.execute( |
||||
Optional.empty(), |
||||
null, |
||||
null, |
||||
closedSocket::get, |
||||
jsonRequest, |
||||
req -> req.mapTo(JsonRpcRequest.class))); |
||||
} |
||||
promise.complete(responses); |
||||
}) |
||||
.onSuccess( |
||||
jsonRpcBatchResponse -> { |
||||
try { |
||||
final JsonRpcResponse[] completed = |
||||
jsonRpcBatchResponse.stream() |
||||
.filter( |
||||
jsonRpcResponse -> |
||||
jsonRpcResponse.getType() |
||||
!= JsonRpcResponseType.NONE) |
||||
.toArray(JsonRpcResponse[]::new); |
||||
|
||||
socket.write( |
||||
JSON_OBJECT_WRITER.writeValueAsString(completed) |
||||
+ '\n'); |
||||
} catch (JsonProcessingException e) { |
||||
LOG.error("Error streaming JSON-RPC response", e); |
||||
} |
||||
}) |
||||
.onFailure( |
||||
throwable -> |
||||
errorReturn(socket, null, JsonRpcError.INTERNAL_ERROR)); |
||||
} |
||||
} catch (DecodeException jsonArrayDecodeException) { |
||||
errorReturn(socket, null, JsonRpcError.PARSE_ERROR); |
||||
} |
||||
} |
||||
} |
||||
}); |
||||
}); |
||||
return netServer |
||||
.listen(SocketAddress.domainSocketAddress(path.toString())) |
||||
.onSuccess(successServer -> LOG.info("IPC endpoint opened: {}", path)) |
||||
.onFailure(throwable -> LOG.error("Unable to open IPC endpoint", throwable)); |
||||
} |
||||
|
||||
public Future<Void> stop() { |
||||
if (netServer == null) { |
||||
return Future.succeededFuture(); |
||||
} else { |
||||
return netServer |
||||
.close() |
||||
.onComplete( |
||||
closeResult -> { |
||||
try { |
||||
Files.deleteIfExists(path); |
||||
} catch (IOException e) { |
||||
LOG.error("Unable to delete IPC file", e); |
||||
} |
||||
}); |
||||
} |
||||
} |
||||
|
||||
private Future<Void> errorReturn( |
||||
final NetSocket socket, final Integer id, final JsonRpcError rpcError) { |
||||
return socket.write(Buffer.buffer(Json.encode(new JsonRpcErrorResponse(id, rpcError)) + '\n')); |
||||
} |
||||
|
||||
private NetServerOptions buildNetServerOptions() { |
||||
return new NetServerOptions(); |
||||
} |
||||
} |
||||
@ -0,0 +1,195 @@ |
||||
/* |
||||
* Copyright contributors to Hyperledger Besu. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.api.jsonrpc.ipc; |
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
import static org.mockito.ArgumentMatchers.any; |
||||
import static org.mockito.Mockito.mock; |
||||
import static org.mockito.Mockito.when; |
||||
|
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.execution.BaseJsonRpcProcessor; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.execution.JsonRpcExecutor; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod; |
||||
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcSuccessResponse; |
||||
|
||||
import java.nio.file.Path; |
||||
import java.util.Arrays; |
||||
import java.util.Collections; |
||||
import java.util.Map; |
||||
import java.util.concurrent.TimeUnit; |
||||
|
||||
import io.vertx.core.Vertx; |
||||
import io.vertx.core.VertxOptions; |
||||
import io.vertx.core.buffer.Buffer; |
||||
import io.vertx.core.json.JsonArray; |
||||
import io.vertx.core.json.JsonObject; |
||||
import io.vertx.core.net.SocketAddress; |
||||
import io.vertx.junit5.VertxExtension; |
||||
import io.vertx.junit5.VertxTestContext; |
||||
import org.junit.jupiter.api.AfterEach; |
||||
import org.junit.jupiter.api.BeforeEach; |
||||
import org.junit.jupiter.api.Test; |
||||
import org.junit.jupiter.api.condition.EnabledOnOs; |
||||
import org.junit.jupiter.api.condition.OS; |
||||
import org.junit.jupiter.api.extension.ExtendWith; |
||||
import org.junit.jupiter.api.io.TempDir; |
||||
|
||||
@EnabledOnOs({OS.LINUX, OS.MAC}) |
||||
@ExtendWith(VertxExtension.class) |
||||
class JsonRpcIpcServiceTest { |
||||
|
||||
@TempDir private Path tempDir; |
||||
private Vertx vertx; |
||||
private VertxTestContext testContext; |
||||
|
||||
@BeforeEach |
||||
public void setUp() { |
||||
vertx = Vertx.vertx(new VertxOptions().setPreferNativeTransport(true)); |
||||
testContext = new VertxTestContext(); |
||||
} |
||||
|
||||
@AfterEach |
||||
public void after() throws Throwable { |
||||
assertThat(testContext.awaitCompletion(5, TimeUnit.SECONDS)) |
||||
.describedAs("Test completed on time") |
||||
.isTrue(); |
||||
if (testContext.failed()) { |
||||
throw testContext.causeOfFailure(); |
||||
} |
||||
} |
||||
|
||||
@Test |
||||
void successfulExecution() { |
||||
final Path socketPath = tempDir.resolve("besu-test.ipc"); |
||||
final JsonRpcMethod testMethod = mock(JsonRpcMethod.class); |
||||
when(testMethod.response(any())).thenReturn(new JsonRpcSuccessResponse(1, "TEST OK")); |
||||
final JsonRpcIpcService service = |
||||
new JsonRpcIpcService( |
||||
vertx, |
||||
socketPath, |
||||
new JsonRpcExecutor(new BaseJsonRpcProcessor(), Map.of("test_method", testMethod))); |
||||
final String expectedResponse = "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":\"TEST OK\"}\n"; |
||||
|
||||
assertSocketCall( |
||||
service, |
||||
socketPath, |
||||
expectedResponse, |
||||
new JsonObject().put("id", 1).put("method", "test_method").toBuffer()); |
||||
} |
||||
|
||||
@Test |
||||
void successfulBatchExecution() { |
||||
final Path socketPath = tempDir.resolve("besu-test.ipc"); |
||||
final JsonRpcMethod fooMethod = mock(JsonRpcMethod.class); |
||||
when(fooMethod.response(any())).thenReturn(new JsonRpcSuccessResponse(1, "FOO OK")); |
||||
final JsonRpcMethod barMethod = mock(JsonRpcMethod.class); |
||||
when(barMethod.response(any())).thenReturn(new JsonRpcSuccessResponse(2, "BAR OK")); |
||||
final JsonRpcIpcService service = |
||||
new JsonRpcIpcService( |
||||
vertx, |
||||
socketPath, |
||||
new JsonRpcExecutor( |
||||
new BaseJsonRpcProcessor(), |
||||
Map.of("foo_method", fooMethod, "bar_method", barMethod))); |
||||
|
||||
assertSocketCall( |
||||
service, |
||||
socketPath, |
||||
"[{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":\"FOO OK\"},{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":\"BAR OK\"}]\n", |
||||
new JsonArray( |
||||
Arrays.asList( |
||||
new JsonObject().put("id", 1).put("method", "foo_method"), |
||||
new JsonObject().put("id", 2).put("method", "bar_method"))) |
||||
.toBuffer()); |
||||
} |
||||
|
||||
@Test |
||||
void validJsonButNotRpcShouldReturnInvalidRequest() { |
||||
final Path socketPath = tempDir.resolve("besu-test.ipc"); |
||||
final JsonRpcIpcService service = |
||||
new JsonRpcIpcService( |
||||
vertx, |
||||
socketPath, |
||||
new JsonRpcExecutor(new BaseJsonRpcProcessor(), Collections.emptyMap())); |
||||
final String expectedResponse = |
||||
"{\"jsonrpc\":\"2.0\",\"id\":null,\"error\":{\"code\":-32600,\"message\":\"Invalid Request\"}}\n"; |
||||
|
||||
assertSocketCall(service, socketPath, expectedResponse, Buffer.buffer("{\"foo\":\"bar\"}")); |
||||
} |
||||
|
||||
@Test |
||||
void nonJsonRequestShouldReturnParseError() { |
||||
final Path socketPath = tempDir.resolve("besu-test.ipc"); |
||||
final JsonRpcIpcService service = |
||||
new JsonRpcIpcService(vertx, socketPath, mock(JsonRpcExecutor.class)); |
||||
final String expectedResponse = |
||||
"{\"jsonrpc\":\"2.0\",\"id\":null,\"error\":{\"code\":-32700,\"message\":\"Parse error\"}}\n"; |
||||
|
||||
assertSocketCall(service, socketPath, expectedResponse, Buffer.buffer("bad request")); |
||||
} |
||||
|
||||
@Test |
||||
void shouldDeleteSocketFileOnStop() { |
||||
final Path socketPath = tempDir.resolve("besu-test.ipc"); |
||||
final JsonRpcIpcService service = |
||||
new JsonRpcIpcService(vertx, socketPath, mock(JsonRpcExecutor.class)); |
||||
service |
||||
.start() |
||||
.onComplete( |
||||
testContext.succeeding( |
||||
server -> |
||||
service |
||||
.stop() |
||||
.onComplete( |
||||
testContext.succeeding( |
||||
handler -> |
||||
testContext.verify( |
||||
() -> { |
||||
assertThat(socketPath).doesNotExist(); |
||||
testContext.completeNow(); |
||||
}))))); |
||||
} |
||||
|
||||
private void assertSocketCall( |
||||
final JsonRpcIpcService service, |
||||
final Path socketPath, |
||||
final String expectedResponse, |
||||
final Buffer request) { |
||||
service |
||||
.start() |
||||
.onComplete( |
||||
testContext.succeeding( |
||||
server -> |
||||
vertx |
||||
.createNetClient() |
||||
.connect(SocketAddress.domainSocketAddress(socketPath.toString())) |
||||
.onComplete( |
||||
testContext.succeeding( |
||||
socket -> |
||||
socket |
||||
.handler( |
||||
buffer -> |
||||
testContext.verify( |
||||
() -> { |
||||
assertThat(buffer) |
||||
.hasToString(expectedResponse); |
||||
service |
||||
.stop() |
||||
.onComplete( |
||||
testContext.succeedingThenComplete()); |
||||
})) |
||||
.write(request))))); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue