jwt auth on websockets (#4039)

* integration test covering websocket subscription without auth
* uses authenticated user on websocket handler when auth enabled and successful
* sonarlint fixes and copyright correction
* moved test specific class to test sources

Signed-off-by: Justin Florentine <justin+github@florentine.us>

Co-authored-by: garyschulte <garyschulte@gmail.com>

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/JsonRpcService.java

@@ -87,11 +87,13 @@ import io.vertx.core.http.HttpServerResponse;
 import io.vertx.core.http.ServerWebSocket;
 import io.vertx.core.net.PfxOptions;
 import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.User;
 import io.vertx.ext.web.Route;
 import io.vertx.ext.web.Router;
 import io.vertx.ext.web.RoutingContext;
 import io.vertx.ext.web.handler.BodyHandler;
 import io.vertx.ext.web.handler.CorsHandler;
+import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -331,26 +333,21 @@ public class JsonRpcService {
                 user -> {
                   if (user.isEmpty()) {
                     websocket.reject(403);
+                  } else {
+                    final Handler<Buffer> socketHandler =
+                        handlerForUser(socketAddress, websocket, user);
+                    websocket.textMessageHandler(text -> socketHandler.handle(Buffer.buffer(text)));
+                    websocket.binaryMessageHandler(socketHandler);
                   }
                 });
+      } else {
+        final Handler<Buffer> socketHandler =
+            handlerForUser(socketAddress, websocket, Optional.empty());
+        websocket.textMessageHandler(text -> socketHandler.handle(Buffer.buffer(text)));
+        websocket.binaryMessageHandler(socketHandler);
       }
-      LOG.debug("Websocket Connected ({})", socketAddressAsString(socketAddress));
+      String addr = socketAddressAsString(socketAddress);
-
+      LOG.debug("Websocket Connected ({})", addr);
-      final Handler<Buffer> socketHandler =
-          buffer -> {
-            LOG.debug(
-                "Received Websocket request (binary frame) {} ({})",
-                buffer.toString(),
-                socketAddressAsString(socketAddress));
-
-            if (webSocketMessageHandler.isPresent()) {
-              webSocketMessageHandler.get().handle(websocket, buffer, Optional.empty());
-            } else {
-              LOG.error("No socket request handler configured");
-            }
-          };
-      websocket.textMessageHandler(text -> socketHandler.handle(Buffer.buffer(text)));
-      websocket.binaryMessageHandler(socketHandler);
 
       websocket.closeHandler(
           v -> {
@@ -371,6 +368,24 @@ public class JsonRpcService {
     };
   }
 
+  @NotNull
+  private Handler<Buffer> handlerForUser(
+      final SocketAddress socketAddress,
+      final ServerWebSocket websocket,
+      final Optional<User> user) {
+    return buffer -> {
+      String addr = socketAddressAsString(socketAddress);
+      LOG.debug("Received Websocket request (binary frame) {} ({})", buffer, addr);
+
+      if (webSocketMessageHandler.isPresent()) {
+        // if auth enabled and user empty will always 401
+        webSocketMessageHandler.get().handle(websocket, buffer, user);
+      } else {
+        LOG.error("No socket request handler configured");
+      }
+    };
+  }
+
   private void validateConfig(final JsonRpcConfiguration config) {
     checkArgument(
         config.getPort() == 0 || NetworkUtility.isValidPort(config.getPort()),
@@ -588,20 +603,18 @@ public class JsonRpcService {
             : event.request().host();
     final Iterable<String> splitHostHeader = Splitter.on(':').split(hostname);
     final long hostPieces = stream(splitHostHeader).count();
-    if (hostPieces > 1) {
+    // If the host contains a colon, verify the host is correctly formed - host [ ":" port ]
-      // If the host contains a colon, verify the host is correctly formed - host [ ":" port ]
+    if (hostPieces > 2 || !Iterables.get(splitHostHeader, 1).matches("\\d{1,5}+")) {
-      if (hostPieces > 2 || !Iterables.get(splitHostHeader, 1).matches("\\d{1,5}+")) {
+      return Optional.empty();
-        return Optional.empty();
-      }
     }
+
     return Optional.ofNullable(Iterables.get(splitHostHeader, 0));
   }
 
   private boolean hostIsInAllowlist(final String hostHeader) {
     if (config.getHostsAllowlist().contains("*")
         || config.getHostsAllowlist().stream()
-            .anyMatch(
+            .anyMatch(allowlistEntry -> allowlistEntry.equalsIgnoreCase(hostHeader))) {
-                allowlistEntry -> allowlistEntry.toLowerCase().equals(hostHeader.toLowerCase()))) {
       return true;
     } else {
       LOG.trace("Host not in allowlist: '{}'", hostHeader);

ethereum/api/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/authentication/EngineAuthService.java

@@ -52,6 +52,12 @@ public class EngineAuthService implements AuthenticationService {
     this.jwtAuthProvider = JWTAuth.create(vertx, jwtAuthOptions);
   }
 
+  public String createToken() {
+    JsonObject claims = new JsonObject();
+    claims.put("iat", System.currentTimeMillis() / 1000);
+    return this.jwtAuthProvider.generateToken(claims);
+  }
+
   private JWTAuthOptions engineApiJWTOptions(
       final JwtAlgorithm jwtAlgorithm, final Optional<File> keyFile, final Path datadir) {
     byte[] signingKey = null;

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/response/MutableJsonRpcSuccessResponse.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright Hyperledger Besu contributors.
+ *
+ * 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.internal.response;
+
+import java.util.Objects;
+
+import com.fasterxml.jackson.annotation.JsonGetter;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.annotation.JsonSetter;
+
+@JsonPropertyOrder({"jsonrpc", "id", "result"})
+public class MutableJsonRpcSuccessResponse {
+
+  private Object id;
+  private Object result;
+  private Object version;
+
+  public MutableJsonRpcSuccessResponse() {
+    this.id = null;
+    this.result = null;
+  }
+
+  public MutableJsonRpcSuccessResponse(final Object id, final Object result) {
+    this.id = id;
+    this.result = result;
+  }
+
+  public MutableJsonRpcSuccessResponse(final Object id) {
+    this.id = id;
+    this.result = "Success";
+  }
+
+  @JsonGetter("id")
+  public Object getId() {
+    return id;
+  }
+
+  @JsonGetter("result")
+  public Object getResult() {
+    return result;
+  }
+
+  @JsonSetter("id")
+  public void setId(final Object id) {
+    this.id = id;
+  }
+
+  @JsonSetter("result")
+  public void setResult(final Object result) {
+    this.result = result;
+  }
+
+  @JsonGetter("jsonrpc")
+  public Object getVersion() {
+    return version;
+  }
+
+  @JsonSetter("jsonrpc")
+  public void setVersion(final Object version) {
+    this.version = version;
+  }
+
+  @JsonIgnore
+  public JsonRpcResponseType getType() {
+    return JsonRpcResponseType.SUCCESS;
+  }
+
+  @Override
+  public boolean equals(final Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final MutableJsonRpcSuccessResponse that = (MutableJsonRpcSuccessResponse) o;
+    return Objects.equals(id, that.id) && Objects.equals(result, that.result);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(id, result);
+  }
+}

ethereum/api/src/test/java/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/JsonRpcJWTTest.java

@@ -0,0 +1,297 @@
+/*
+ * Copyright Hyperledger Besu contributors.
+ *
+ * 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.websocket;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+
+import org.hyperledger.besu.ethereum.api.jsonrpc.JsonRpcConfiguration;
+import org.hyperledger.besu.ethereum.api.jsonrpc.JsonRpcService;
+import org.hyperledger.besu.ethereum.api.jsonrpc.authentication.AuthenticationService;
+import org.hyperledger.besu.ethereum.api.jsonrpc.authentication.EngineAuthService;
+import org.hyperledger.besu.ethereum.api.jsonrpc.health.HealthService;
+import org.hyperledger.besu.ethereum.api.jsonrpc.health.HealthService.HealthCheck;
+import org.hyperledger.besu.ethereum.api.jsonrpc.health.HealthService.ParamSource;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.JsonRpcRequest;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.methods.JsonRpcMethod;
+import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.MutableJsonRpcSuccessResponse;
+import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.methods.WebSocketMethodsFactory;
+import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.SubscriptionManager;
+import org.hyperledger.besu.ethereum.eth.manager.EthScheduler;
+import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem;
+import org.hyperledger.besu.nat.NatService;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URISyntaxException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import io.vertx.core.Vertx;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpClientOptions;
+import io.vertx.core.http.WebSocket;
+import io.vertx.core.http.WebSocketConnectOptions;
+import io.vertx.core.json.Json;
+import io.vertx.ext.unit.Async;
+import io.vertx.ext.unit.TestContext;
+import io.vertx.ext.unit.junit.VertxUnitRunner;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+@RunWith(VertxUnitRunner.class)
+public class JsonRpcJWTTest {
+
+  @ClassRule public static final TemporaryFolder folder = new TemporaryFolder();
+  public static final String HOSTNAME = "127.0.0.1";
+
+  protected static Vertx vertx;
+
+  private final JsonRpcConfiguration jsonRpcConfiguration =
+      JsonRpcConfiguration.createEngineDefault();
+  private HttpClient httpClient;
+  private Optional<AuthenticationService> jwtAuth;
+  private HealthService healthy;
+  private EthScheduler scheduler;
+  private Path bufferDir;
+  private Map<String, JsonRpcMethod> websocketMethods;
+
+  @Before
+  public void initServerAndClient() {
+    jsonRpcConfiguration.setPort(0);
+    jsonRpcConfiguration.setHostsAllowlist(List.of("*"));
+    try {
+      jsonRpcConfiguration.setAuthenticationPublicKeyFile(
+          new File(this.getClass().getResource("jwt.hex").toURI()));
+    } catch (URISyntaxException e) {
+      fail("couldn't load jwt key from jwt.hex in classpath");
+    }
+    vertx = Vertx.vertx();
+
+    websocketMethods =
+        new WebSocketMethodsFactory(
+                new SubscriptionManager(new NoOpMetricsSystem()), new HashMap<>())
+            .methods();
+
+    bufferDir = null;
+    try {
+      bufferDir = Files.createTempDirectory("JsonRpcJWTTest").toAbsolutePath();
+    } catch (IOException e) {
+      fail("can't create tempdir", e);
+    }
+
+    jwtAuth =
+        Optional.of(
+            new EngineAuthService(
+                vertx,
+                Optional.ofNullable(jsonRpcConfiguration.getAuthenticationPublicKeyFile()),
+                bufferDir));
+
+    healthy =
+        new HealthService(
+            new HealthCheck() {
+              @Override
+              public boolean isHealthy(final ParamSource paramSource) {
+                return true;
+              }
+            });
+
+    scheduler = new EthScheduler(1, 1, 1, new NoOpMetricsSystem());
+  }
+
+  @After
+  public void after() {}
+
+  @Test
+  public void unauthenticatedWebsocketAllowedWithoutJWTAuth(final TestContext context) {
+
+    JsonRpcService jsonRpcService =
+        new JsonRpcService(
+            vertx,
+            bufferDir,
+            jsonRpcConfiguration,
+            new NoOpMetricsSystem(),
+            new NatService(Optional.empty(), true),
+            websocketMethods,
+            Optional.empty(),
+            scheduler,
+            Optional.empty(),
+            healthy,
+            healthy);
+
+    jsonRpcService.start().join();
+
+    final InetSocketAddress inetSocketAddress = jsonRpcService.socketAddress();
+    int listenPort = inetSocketAddress.getPort();
+
+    final HttpClientOptions httpClientOptions =
+        new HttpClientOptions().setDefaultHost(HOSTNAME).setDefaultPort(listenPort);
+
+    httpClient = vertx.createHttpClient(httpClientOptions);
+
+    WebSocketConnectOptions wsOpts = new WebSocketConnectOptions();
+    wsOpts.setPort(listenPort);
+    wsOpts.setHost(HOSTNAME);
+    wsOpts.setURI("/");
+
+    final Async async = context.async();
+    httpClient.webSocket(
+        wsOpts,
+        connected -> {
+          if (connected.failed()) {
+            connected.cause().printStackTrace();
+          }
+          assertThat(connected.succeeded()).isTrue();
+          WebSocket ws = connected.result();
+
+          JsonRpcRequest req =
+              new JsonRpcRequest("2.0", "eth_subscribe", List.of("syncing").toArray());
+          ws.frameHandler(
+              resp -> {
+                assertThat(resp.isText()).isTrue();
+                MutableJsonRpcSuccessResponse messageReply =
+                    Json.decodeValue(resp.textData(), MutableJsonRpcSuccessResponse.class);
+                assertThat(messageReply.getResult()).isEqualTo("0x1");
+                async.complete();
+              });
+          ws.writeTextMessage(Json.encode(req));
+        });
+
+    async.awaitSuccess(10000);
+    jsonRpcService.stop();
+    httpClient.close();
+  }
+
+  @Test
+  public void httpRequestWithDefaultHeaderAndValidJWTIsAccepted(final TestContext context) {
+
+    JsonRpcService jsonRpcService =
+        new JsonRpcService(
+            vertx,
+            bufferDir,
+            jsonRpcConfiguration,
+            new NoOpMetricsSystem(),
+            new NatService(Optional.empty(), true),
+            websocketMethods,
+            Optional.empty(),
+            scheduler,
+            jwtAuth,
+            healthy,
+            healthy);
+
+    jsonRpcService.start().join();
+
+    final InetSocketAddress inetSocketAddress = jsonRpcService.socketAddress();
+    int listenPort = inetSocketAddress.getPort();
+
+    final HttpClientOptions httpClientOptions =
+        new HttpClientOptions().setDefaultHost(HOSTNAME).setDefaultPort(listenPort);
+
+    httpClient = vertx.createHttpClient(httpClientOptions);
+
+    WebSocketConnectOptions wsOpts = new WebSocketConnectOptions();
+    wsOpts.setPort(listenPort);
+    wsOpts.setHost(HOSTNAME);
+    wsOpts.setURI("/");
+    wsOpts.addHeader(
+        "Authorization", "Bearer " + ((EngineAuthService) jwtAuth.get()).createToken());
+
+    final Async async = context.async();
+    httpClient.webSocket(
+        wsOpts,
+        connected -> {
+          if (connected.failed()) {
+            connected.cause().printStackTrace();
+          }
+          assertThat(connected.succeeded()).isTrue();
+          WebSocket ws = connected.result();
+          JsonRpcRequest req =
+              new JsonRpcRequest("1", "eth_subscribe", List.of("syncing").toArray());
+          ws.frameHandler(
+              resp -> {
+                assertThat(resp.isText()).isTrue();
+                System.out.println(resp.textData());
+                assertThat(resp.textData()).doesNotContain("error");
+                MutableJsonRpcSuccessResponse messageReply =
+                    Json.decodeValue(resp.textData(), MutableJsonRpcSuccessResponse.class);
+                assertThat(messageReply.getResult()).isEqualTo("0x1");
+                async.complete();
+              });
+          ws.writeTextMessage(Json.encode(req));
+        });
+
+    async.awaitSuccess(10000);
+    jsonRpcService.stop();
+    httpClient.close();
+  }
+
+  @Test
+  public void httpRequestWithDefaultHeaderAndInvalidJWTIsDenied(final TestContext context) {
+
+    JsonRpcService jsonRpcService =
+        new JsonRpcService(
+            vertx,
+            bufferDir,
+            jsonRpcConfiguration,
+            new NoOpMetricsSystem(),
+            new NatService(Optional.empty(), true),
+            websocketMethods,
+            Optional.empty(),
+            scheduler,
+            jwtAuth,
+            healthy,
+            healthy);
+
+    jsonRpcService.start().join();
+
+    final InetSocketAddress inetSocketAddress = jsonRpcService.socketAddress();
+    int listenPort = inetSocketAddress.getPort();
+
+    final HttpClientOptions httpClientOptions =
+        new HttpClientOptions().setDefaultHost(HOSTNAME).setDefaultPort(listenPort);
+
+    httpClient = vertx.createHttpClient(httpClientOptions);
+
+    WebSocketConnectOptions wsOpts = new WebSocketConnectOptions();
+    wsOpts.setPort(listenPort);
+    wsOpts.setHost(HOSTNAME);
+    wsOpts.setURI("/");
+    wsOpts.addHeader("Authorization", "Bearer totallyunparseablenonsense");
+
+    final Async async = context.async();
+    httpClient.webSocket(
+        wsOpts,
+        connected -> {
+          if (connected.failed()) {
+            connected.cause().printStackTrace();
+          }
+          assertThat(connected.succeeded()).isFalse();
+          async.complete();
+        });
+
+    async.awaitSuccess(10000);
+    jsonRpcService.stop();
+    httpClient.close();
+  }
+}

ethereum/api/src/test/resources/org/hyperledger/besu/ethereum/api/jsonrpc/websocket/jwt.hex

@@ -0,0 +1 @@
+9465710175a93a3f2d67b0cb98d92d44ead4d1126a12233571884de92a8edc76