Add JWT client/server support for gRPC

pom.xml

@@ -63,6 +63,7 @@
     <module>vertx-grpc-common</module>
     <module>vertx-grpc-server</module>
     <module>vertx-grpc-client</module>
+    <module>vertx-grpc-jwt-auth</module>
     <module>vertx-grpc-context-storage</module>
     <module>vertx-grpc-protoc-plugin2</module>
     <module>vertx-grpc-it</module>

vertx-grpc-client/src/main/asciidoc/client.adoc

@@ -163,3 +163,12 @@ handle the message encoding:
 
 - when the message uses the response encoding, the message is sent as is
 - when the message uses a different encoding, it will be encoded, e.g. compressed or uncompressed
+
+=== Error Handling
+
+A `{@link io.vertx.grpc.common.GrpcException}` may be throw and accessed via a `onFailure` handler. It may grant access to the response and the gRPC status via `{@link io.vertx.grpc.common.GrpcStatus}`.
+
+[source,java]
+----
+{@link examples.GrpcClientExamples#errorHandling}
+----

vertx-grpc-client/src/main/java/examples/GrpcClientExamples.java

@@ -5,13 +5,16 @@
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClientResponse;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.docgen.Source;
 import io.vertx.grpc.client.GrpcClient;
 import io.vertx.grpc.client.GrpcClientChannel;
 import io.vertx.grpc.client.GrpcClientRequest;
 import io.vertx.grpc.client.GrpcClientResponse;
+import io.vertx.grpc.common.GrpcException;
 import io.vertx.grpc.common.GrpcMessage;
+import io.vertx.grpc.common.GrpcStatus;
 import io.vertx.grpc.common.ServiceName;
 
 @Source
@@ -189,4 +192,18 @@ public void messageLevelAPI(GrpcClient client, Buffer protoHello, SocketAddress
       });
     });
   }
+
+  public void errorHandling(GrpcClientRequest<HelloRequest, HelloReply> request) {
+    request.response().onSuccess(response -> {
+      Future<HelloReply> fut = response.last();
+      fut.onFailure(error -> {
+        if (error instanceof GrpcException) {
+          GrpcException grpcError = (GrpcException) error;
+          GrpcStatus status = grpcError.status();
+          HttpClientResponse httpResponse = grpcError.response();
+        }
+      });
+    });
+  }
+
 }

vertx-grpc-client/src/main/java/io/vertx/grpc/client/GrpcClientRequest.java

@@ -16,11 +16,8 @@
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
-import io.vertx.core.MultiMap;
-import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.HttpClientResponse;
+import io.vertx.core.http.HttpClientRequest;
 import io.vertx.core.http.HttpConnection;
-import io.vertx.core.http.HttpHeaders;
 import io.vertx.core.streams.ReadStream;
 import io.vertx.grpc.common.GrpcWriteStream;
 import io.vertx.grpc.common.ServiceName;
@@ -109,4 +106,6 @@ default Future<GrpcClientResponse<Req, Resp>> send(ReadStream<Req> body) {
     body.pipeTo(this);
     return this.response();
   }
+
+  HttpClientRequest delegate();
 }

vertx-grpc-client/src/main/java/io/vertx/grpc/client/impl/GrpcClientRequestImpl.java

@@ -258,4 +258,9 @@ public void cancel() {
   public HttpConnection connection() {
     return httpRequest.connection();
   }
+
+  @Override
+  public HttpClientRequest delegate() {
+    return httpRequest;
+  }
 }

vertx-grpc-client/src/main/java/io/vertx/grpc/client/impl/GrpcClientResponseImpl.java

@@ -14,15 +14,14 @@
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
-import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.HttpClientResponse;
-
 import io.vertx.grpc.client.GrpcClientResponse;
 import io.vertx.grpc.common.CodecException;
+import io.vertx.grpc.common.GrpcException;
 import io.vertx.grpc.common.GrpcMessageDecoder;
-import io.vertx.grpc.common.impl.GrpcReadStreamBase;
 import io.vertx.grpc.common.GrpcStatus;
+import io.vertx.grpc.common.impl.GrpcReadStreamBase;
 
 import java.nio.charset.StandardCharsets;
 
@@ -101,7 +100,7 @@ public Future<Void> end() {
       if (status == GrpcStatus.OK) {
         return Future.succeededFuture();
       } else {
-        return Future.failedFuture("Invalid gRPC status " + status);
+        return Future.failedFuture(new GrpcException("Invalid gRPC status " + status, status, httpResponse));
       }
     });
   }

vertx-grpc-common/src/main/java/io/vertx/grpc/common/GrpcException.java

@@ -0,0 +1,40 @@
+package io.vertx.grpc.common;
+
+import io.vertx.core.http.HttpClientResponse;
+
+public class GrpcException extends RuntimeException {
+
+  private static final long serialVersionUID = -7838327176604697641L;
+
+  private GrpcStatus status;
+
+  private HttpClientResponse httpResponse;
+
+  public GrpcException(String msg, GrpcStatus status, HttpClientResponse httpResponse) {
+    super(msg);
+    this.status = status;
+    this.httpResponse = httpResponse;
+  }
+
+  public GrpcException(GrpcStatus status) {
+    this.status = status;
+  }
+
+  public GrpcException(GrpcStatus status, Throwable err) {
+    super(err);
+    this.status = status;
+  }
+
+  public GrpcException(GrpcStatus status, String msg) {
+    super(msg);
+    this.status = status;
+  }
+
+  public GrpcStatus status() {
+    return status;
+  }
+
+  public HttpClientResponse response() {
+    return httpResponse;
+  }
+}

vertx-grpc-it/pom.xml

@@ -43,6 +43,10 @@
       <groupId>io.vertx</groupId>
       <artifactId>vertx-grpc-server</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-grpc-jwt-auth</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>io.vertx</groupId>
@@ -51,6 +55,13 @@
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-grpc-server</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>io.vertx</groupId>
       <artifactId>vertx-unit</artifactId>

vertx-grpc-it/src/test/java/io/vertx/grpc/it/JWTAuthTest.java

@@ -0,0 +1,152 @@
+package io.vertx.grpc.it;
+
+import org.junit.Test;
+
+import io.grpc.examples.helloworld.GreeterGrpc;
+import io.grpc.examples.helloworld.HelloReply;
+import io.grpc.examples.helloworld.HelloRequest;
+import io.vertx.core.Future;
+import io.vertx.core.json.JsonObject;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.JWTOptions;
+import io.vertx.ext.auth.KeyStoreOptions;
+import io.vertx.ext.auth.User;
+import io.vertx.ext.auth.authentication.TokenCredentials;
+import io.vertx.ext.auth.jwt.JWTAuth;
+import io.vertx.ext.auth.jwt.JWTAuthOptions;
+import io.vertx.ext.unit.Async;
+import io.vertx.ext.unit.TestContext;
+import io.vertx.grpc.client.GrpcClientResponse;
+import io.vertx.grpc.common.GrpcException;
+import io.vertx.grpc.common.GrpcStatus;
+import io.vertx.grpc.server.GrpcServerResponse;
+import io.vertx.grpc.server.ServerTestBase;
+import io.vertx.grpc.server.auth.jwt.JWTGrpcClient;
+import io.vertx.grpc.server.auth.jwt.JWTGrpcServer;
+
+public class JWTAuthTest extends ServerTestBase {
+
+  private String validToken;
+  private JWTGrpcServer jwtServer;
+  private String expiredToken;
+  private static final String BROKEN_TOKEN = "this-token-value-is-bogus";
+  private static final String INVALID_TOKEN = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJqb2hhbm5lcyIsImlhdCI6MTY3ODgwNDgwN30";
+
+  public void setupClientServer(TestContext should, boolean expectUser) {
+
+    // Prepare JWT auth and generate token to be used for the client
+    JWTAuthOptions config = new JWTAuthOptions()
+      .setKeyStore(new KeyStoreOptions()
+        .setPath("keystore.jceks")
+        .setPassword("secret")
+        .setType("jceks"));
+    JWTAuth authProvider = JWTAuth.create(vertx, config);
+    validToken = authProvider.generateToken(new JsonObject().put("sub", "johannes"), new JWTOptions().setIgnoreExpiration(true));
+    expiredToken = authProvider.generateToken(new JsonObject().put("sub", "johannes"), new JWTOptions().setExpiresInSeconds(1));
+
+    jwtServer = JWTGrpcServer.create(vertx, authProvider);
+    jwtServer.callHandler(GreeterGrpc.getSayHelloMethod(), true, request -> {
+      request.handler(hello -> {
+        GrpcServerResponse<HelloRequest, HelloReply> response = request.response();
+        HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + hello.getName()).build();
+        response.end(reply);
+        User user = request.user();
+        if (expectUser) {
+          should.assertNotNull(user);
+          should.assertEquals("johannes", user.subject());
+        } else {
+          should.assertNull(user);
+        }
+      }).errorHandler(error -> {
+        should.fail("Error should not happen " + error);
+      });
+    });
+
+    startServer(jwtServer);
+  }
+
+  @Test
+  public void testJWTBrokenTokenAuthentication(TestContext should) {
+    setupClientServer(should, false);
+
+    JWTGrpcClient jwtClient = JWTGrpcClient.create(vertx).withCredentials(new TokenCredentials(BROKEN_TOKEN));
+    Future<HelloReply> clientReply = sayHello(jwtClient);
+
+    Async test = should.async();
+    clientReply
+      .onFailure(error -> {
+        if (error instanceof GrpcException) {
+          GrpcException grpcError = (GrpcException) error;
+          should.assertEquals(GrpcStatus.UNAUTHENTICATED, grpcError.status(), "The status code did not match. Got: " + grpcError.status().name());
+          should.assertNotNull(grpcError.response());
+        } else {
+          should.fail(error);
+        }
+        test.complete();
+      })
+      .onSuccess(reply -> {
+        should.fail("The test should fail with an error due to the usage of the invalid token");
+      });
+  }
+
+  @Test
+  public void testJWTExpiredTokenAuthentication(TestContext should) {
+    setupClientServer(should, false);
+    // Let the token expire
+    try {
+      Thread.sleep(2000);
+    } catch (InterruptedException e) {
+      // TODO Auto-generated catch block
+      e.printStackTrace();
+    }
+    JWTGrpcClient jwtClient = JWTGrpcClient.create(vertx).withCredentials(new TokenCredentials(expiredToken));
+    Future<HelloReply> clientReply = sayHello(jwtClient);
+
+    Async test = should.async();
+    clientReply
+      .onFailure(error -> {
+        if (error instanceof GrpcException) {
+          GrpcException grpcError = (GrpcException) error;
+          should.assertEquals(GrpcStatus.UNAUTHENTICATED, grpcError.status(), "The status code did not match. Got: " + grpcError.status().name());
+          should.assertNotNull(grpcError.response());
+        } else {
+          should.fail(error);
+        }
+        test.complete();
+      })
+      .onSuccess(reply -> {
+        should.fail("The test should fail with an error due to the usage of the invalid token");
+      });
+  }
+
+  @Test
+  public void testJWTValidAuthentication(TestContext should) {
+    setupClientServer(should, true);
+
+    JWTGrpcClient jwtClient = JWTGrpcClient.create(vertx).withCredentials(new TokenCredentials(validToken));
+    Future<HelloReply> clientReply = sayHello(jwtClient);
+
+    Async test = should.async();
+    clientReply
+      .onFailure(should::fail)
+      .onSuccess(reply -> {
+        System.out.println("Reply: " + reply.getMessage());
+        test.complete();
+      });
+
+  }
+
+  private Future<HelloReply> sayHello(JWTGrpcClient jwtClient) {
+    return jwtClient
+      .request(SocketAddress.inetSocketAddress(8080, "localhost"), GreeterGrpc.getSayHelloMethod())
+      .compose(request -> {
+        request.end(HelloRequest
+          .newBuilder()
+          .setName("Johannes")
+          .build());
+
+        return request.response()
+          .compose(GrpcClientResponse::last);
+      });
+  }
+}