[FLINK-25866][statefun] Support additional TLS configuration

Author: FilKarnicki

docs/content/docs/modules/http-endpoint.md

@@ -191,13 +191,22 @@ spec:
     pool_ttl: 15s
     pool_size: 1024
     payload_max_bytes: 33554432
+    trust_cacerts: ~/trustedCAs.pem
+    client_cert: classpath:clientPublic.crt
+    client_key: ~/clientPrivate.key
+    client_key_password: /tmp/password.txt
 ```
 
 * `call`: total duration of a single request (including retries, and backoffs). After this duration, the call is considered failed.
 * `connect`: the total amount of time to wait for a successful TCP connection. After that amount of time, an attempt is considered failed, and if the total call time has not elapsed, an additional attempt will be scheduled (after a backoff).
 * `pool_ttl`: the amount of time a connection will live in the connection pool. Set to 0 to disable, otherwise the connection will be evicted from the pool after (approximately) that time. If a connection is evicted while it is serving a request, that connection will be only marked for eviction and will be dropped from the pool once the request returns.
 * `pool_size`: the maximum pool size.
 * `payload_max_bytes`: the maximum size for a request or response payload size. The default is set to 32MB.
+* `trust_cacerts`: Trusted public certificate authority certificates in a pem format. If none are provided, but the function uses https, the default jre truststore will be used. If you need to provide more than one CA cert, concat them with a newline in between. This can be taken from a classpath (e.g. classpath:file.pem) or a path.
+* `client_cert`: Client public certificate used for mutual tls authentication. This can be taken from a classpath (e.g. classpath:file.crt) or a path
+* `client_key`: PKCS8 client private key used for mutual tls authentication. This can be taken from a classpath (e.g. classpath:file.key) or a path
+* `client_key_password`: The location of a file containing the client key password (if required). This can be taken from a classpath (e.g. classpath:file.key) or a path
+
 
 {{< hint info >}}
 We highly recommend setting `statefun.async.max-per-task` to a much higher value (see [Configurations]({{< ref "docs/deployment/configurations">}}))

statefun-e2e-tests/statefun-smoke-e2e-java/pom.xml

@@ -28,23 +28,28 @@ under the License.
 
     <artifactId>statefun-smoke-e2e-java</artifactId>
 
+    <properties>
+        <netty.shaded.version>4.1.65.Final-14.0</netty.shaded.version>
+    </properties>
+
     <dependencies>
         <!-- Remote Java function dependencies -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>statefun-sdk-java</artifactId>
             <version>${project.version}</version>
         </dependency>
-        <dependency>
-            <groupId>io.undertow</groupId>
-            <artifactId>undertow-core</artifactId>
-            <version>1.4.18.Final</version>
-        </dependency>
         <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
             <version>${protobuf.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-shaded-netty</artifactId>
+            <version>${netty.shaded.version}</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

statefun-e2e-tests/statefun-smoke-e2e-java/src/main/java/org/apache/flink/statefun/e2e/smoke/java/CommandInterpreterAppServer.java

@@ -20,33 +20,120 @@
 
 import static org.apache.flink.statefun.e2e.smoke.java.Constants.CMD_INTERPRETER_FN;
 
-import io.undertow.Undertow;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import org.apache.flink.shaded.netty4.io.netty.bootstrap.ServerBootstrap;
+import org.apache.flink.shaded.netty4.io.netty.buffer.Unpooled;
+import org.apache.flink.shaded.netty4.io.netty.channel.*;
+import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup;
+import org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioServerSocketChannel;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.*;
+import org.apache.flink.shaded.netty4.io.netty.handler.ssl.ClientAuth;
+import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslContextBuilder;
+import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslProvider;
 import org.apache.flink.statefun.sdk.java.StatefulFunctionSpec;
 import org.apache.flink.statefun.sdk.java.StatefulFunctions;
-import org.apache.flink.statefun.sdk.java.handler.RequestReplyHandler;
+import org.apache.flink.statefun.sdk.java.slice.Slice;
+import org.apache.flink.statefun.sdk.java.slice.Slices;
 
 public class CommandInterpreterAppServer {
-  public static final int PORT = 8000;
-
-  public static void main(String[] args) {
-    final CommandInterpreter interpreter = new CommandInterpreter();
-    final StatefulFunctionSpec FN_SPEC =
-        StatefulFunctionSpec.builder(CMD_INTERPRETER_FN)
-            .withSupplier(() -> new CommandInterpreterFn(interpreter))
-            .withValueSpec(CommandInterpreterFn.STATE)
-            .build();
-    final StatefulFunctions functions = new StatefulFunctions();
+  private static final int PORT = 8000;
+  private static final String A_SERVER_KEY_PASSWORD = "test";
+  private static final CommandInterpreter commandInterpreter = new CommandInterpreter();
+  private static final StatefulFunctionSpec FN_SPEC =
+      StatefulFunctionSpec.builder(CMD_INTERPRETER_FN)
+          .withSupplier(() -> new CommandInterpreterFn(commandInterpreter))
+          .withValueSpec(CommandInterpreterFn.STATE)
+          .build();
+
+  public static void main(String[] args) throws IOException, InterruptedException {
+    final InputStream trustCaCerts =
+        Objects.requireNonNull(
+                CommandInterpreter.class.getClassLoader().getResource("certs/a_ca.pem"))
+            .openStream();
+    final InputStream aServerCert =
+        Objects.requireNonNull(
+                CommandInterpreter.class.getClassLoader().getResource("certs/a_server.crt"))
+            .openStream();
+    final InputStream aServerKey =
+        Objects.requireNonNull(
+                CommandInterpreter.class.getClassLoader().getResource("certs/a_server.key.p8"))
+            .openStream();
+
+    ServerBootstrap httpsMutualTlsBootstrap =
+        getServerBootstrap(getChannelInitializer(trustCaCerts, aServerCert, aServerKey));
+
+    httpsMutualTlsBootstrap.bind(PORT).sync();
+  }
+
+  private static ChannelInitializer<Channel> getChannelInitializer(
+      InputStream trustInputStream, InputStream certInputStream, InputStream keyInputStream) {
+    return getTlsEnabledInitializer(
+        SslContextBuilder.forServer(certInputStream, keyInputStream, A_SERVER_KEY_PASSWORD)
+            .trustManager(trustInputStream));
+  }
+
+  private static ChannelInitializer<Channel> getTlsEnabledInitializer(
+      SslContextBuilder sslContextBuilder) {
+    return new ChannelInitializer<Channel>() {
+      @Override
+      protected void initChannel(Channel channel) throws IOException {
+        ChannelPipeline pipeline = channel.pipeline();
+        SslContext sslContext =
+            sslContextBuilder.sslProvider(SslProvider.JDK).clientAuth(ClientAuth.REQUIRE).build();
+        pipeline.addLast(sslContext.newHandler(channel.alloc()));
+        addResponseHandlerToPipeline(pipeline);
+      }
+    };
+  }
+
+  private static ServerBootstrap getServerBootstrap(ChannelInitializer<Channel> childHandler) {
+    NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
+    NioEventLoopGroup workerGroup = new NioEventLoopGroup();
+
+    return new ServerBootstrap()
+        .group(eventLoopGroup, workerGroup)
+        .channel(NioServerSocketChannel.class)
+        .childHandler(childHandler)
+        .option(ChannelOption.SO_BACKLOG, 128)
+        .childOption(ChannelOption.SO_KEEPALIVE, true);
+  }
+
+  private static void addResponseHandlerToPipeline(ChannelPipeline pipeline) {
+    pipeline.addLast(new HttpServerCodec());
+    pipeline.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
+    pipeline.addLast(getStatefunInboundHandler());
+  }
+
+  private static SimpleChannelInboundHandler<FullHttpRequest> getStatefunInboundHandler() {
+    StatefulFunctions functions = new StatefulFunctions();
     functions.withStatefulFunction(FN_SPEC);
 
-    final RequestReplyHandler requestReplyHandler = functions.requestReplyHandler();
+    return new SimpleChannelInboundHandler<FullHttpRequest>() {
+      @Override
+      protected void channelRead0(
+          ChannelHandlerContext channelHandlerContext, FullHttpRequest fullHttpRequest) {
+        CompletableFuture<Slice> res =
+            functions
+                .requestReplyHandler()
+                .handle(Slices.wrap(fullHttpRequest.content().nioBuffer()));
+        res.whenComplete(
+            (r, e) -> {
+              FullHttpResponse response =
+                  new DefaultFullHttpResponse(
+                      HttpVersion.HTTP_1_1,
+                      HttpResponseStatus.OK,
+                      Unpooled.copiedBuffer(r.toByteArray()));
+              response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/octet-stream");
+              response.headers().set(HttpHeaderNames.CONTENT_LENGTH, r.readableBytes());
 
-    // Use the request-reply handler along with your favorite HTTP web server framework
-    // to serve the functions!
-    final Undertow httpServer =
-        Undertow.builder()
-            .addHttpListener(PORT, "0.0.0.0")
-            .setHandler(new UndertowHttpHandler(requestReplyHandler))
-            .build();
-    httpServer.start();
+              channelHandlerContext.write(response);
+              channelHandlerContext.flush();
+            });
+      }
+    };
   }
 }

statefun-e2e-tests/statefun-smoke-e2e-java/src/main/java/org/apache/flink/statefun/e2e/smoke/java/UndertowHttpHandler.java

@@ -0,0 +1 @@
+For instructions on how these cert/key files got created, see statefun-flink/statefun-flink-core/src/test/resources/certs/README.md

statefun-e2e-tests/statefun-smoke-e2e-java/src/main/resources/certs/README.md

@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDhTCCAm2gAwIBAgIUTRXcSpygsZsWmuX4QMz9ey0rPBYwDQYJKoZIhvcNAQEL
+BQAwUTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDEKMAgGA1UEAwwBYTAgFw0yMjAyMjIx
+MzA2MDJaGA8yMTIyMDEyOTEzMDYwMlowUTELMAkGA1UEBhMCQVUxEzARBgNVBAgM
+ClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDEK
+MAgGA1UEAwwBYTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAKBR0E/f
+fcTFxs8Oo4ItZm9HYuPR6U52OuBDN8TdzjMfA+FKoTvqvt/RUg7y6QFCdVEFwMS1
+uyjxoMlGDX1MF9SfDTH9oZlND+DL4YatsqpEXGRfjYSheBPUWusoQ6BbXvjiUNf4
+2bEdTFHnb08pdt0ixtymPp3vLsoUWAqscqYKOj4DW0AMRq2fAMdt0NlJsvW9hUQr
+I8wvU90SJlZ/YNOMv4uae8q67YQKZziRpFaLTM5aFjXnsIYAA6Z+PByFjpLTofgc
+BNuGM5WIk7O88UdZOwd5OCZhkaXrPCkkCtyyPDO8kWiiSmdTOnQ2sjkOSKgNTXkd
++6K7REk5gZ8WVv8CAwEAAaNTMFEwHQYDVR0OBBYEFFvccwv5Cufwh+0kLG0vmQCF
+6JbLMB8GA1UdIwQYMBaAFFvccwv5Cufwh+0kLG0vmQCF6JbLMA8GA1UdEwEB/wQF
+MAMBAf8wDQYJKoZIhvcNAQELBQADggEBADfgGnHZ42Cxoe5gNoWIUdgFjBqeiWOe
+NHNDLgXkx/jcOsbAO6H3+tz3Dz8QL3JIYTrwRKv0vX+5GAkkRz2R2ZaN4xHz73Co
+SuFfIqq4MjjrymVcnNA347vk50FjOMgfHrxpS3UQeTvlb1iuA88Zk8ewzhY376+t
+MoFmT1/ocb2E7jvxR0kDNCK5XsJIGzJmCBq8nIAD6wxrPXU3HJ/GLBlL5sL5kRrN
+x7l/DnL2oqN2DuyFmf4g03+DVmuu87XrbDrGHnn57CVnUe7z4jCFE82vA4u/tppK
+VgI17uYA714s3Jw6Iw+u38cmmSb79AVG05D2b2+TsqICVet3bc9MkQ8=
+-----END CERTIFICATE-----

statefun-e2e-tests/statefun-smoke-e2e-java/src/main/resources/certs/a_ca.pem

@@ -0,0 +1,22 @@
+-----BEGIN CERTIFICATE-----
+MIIDjTCCAnWgAwIBAgIUOfy6Qa7zTjHWOwUTdNZrzoPVVjIwDQYJKoZIhvcNAQEL
+BQAwUTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDEKMAgGA1UEAwwBYTAgFw0yMjAzMDIx
+MjE4MTRaGA8yMTIyMDIwNjEyMTgxNFowRTELMAkGA1UEBhMCQVUxEzARBgNVBAgM
+ClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDCC
+ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBANj0NXPAatJmpQZSqdoNU/aI
+bs+b7WXHJ7if2pRU+sdbAdIqU9+eGK+9lzkyin8YExvZokDXV2cHVQCW2kO8qnZE
+pksKk9ZtHMe8RoksbRN2uNq174GXLlku52+IleJE6SiuHDWZXU0s0tZWl3LL5yei
+oKaWLrF4bKxdZqm2bnWe3VZJVgiIJOuHMMoSQPE8BYnUe6n0YjpZR7vHifBLf0wM
+w70PuZrzcADRiokqVFS5dwZYhW1xErbNA0/pYg3MsnQeNuWyJ517KkYSUxCAnb8q
+LHWGBgcqJ7CSmGGpcLDIDEXJZg4lT/SQp08n99+EoTy7rkncoG+pG5IBmyJKLtUC
+AwEAAaNnMGUwHwYDVR0jBBgwFoAUW9xzC/kK5/CH7SQsbS+ZAIXolsswCQYDVR0T
+BAIwADALBgNVHQ8EBAMCBPAwKgYDVR0RBCMwIYIJbG9jYWxob3N0ghRyZW1vdGUt
+ZnVuY3Rpb24taG9zdDANBgkqhkiG9w0BAQsFAAOCAQEAVo2WzynblomqTGqUss9S
+eCOTrKkySETvGLK/GLLKaxn1Du/JRgEpIXCo5Ri3MQ8MBtpi748kN3z6Jh7ouoeu
+tG1I/hll0naxUzBDBzx25ZxSxGCQG/eoOFgRLF5OQ9L5BKRMW1T4/XNJqLtUVoiw
+qiulkNYJs2tKD/swAkM3LwAGuXl/p9KbGoYmxozwDYrW9PohL2zS0qzBRriJQGhM
+x65SeX50KhiDXx6mXTVUfFU91lh5H6er8SOHbS103jt0B7Y3403mhDmq2NcmY2Ln
+P3d6AZlhXwzsFDxD45U4lxYxMWTy6rOZkcCqrV9LGHYbmZYQPVgmiseWWo6qjLiI
+pw==
+-----END CERTIFICATE-----

statefun-e2e-tests/statefun-smoke-e2e-java/src/main/resources/certs/a_server.crt

@@ -0,0 +1,29 @@
+-----BEGIN ENCRYPTED PRIVATE KEY-----
+MIIE4zAcBgoqhkiG9w0BDAEBMA4ECMk3nj4DFVXtAgIIAASCBMFgWecNkVuMByRI
+P3qAOsQxC+Tmf1eamSe4jn+WS022jUTaqO8c+T/9yyj+UJE9w1VKb1+RYAcnijAY
+JJghl82TTIZgzQlIcD1MwiV2E6Fiz/HolaCeEIOHhNEZDdI5INke92uon++TG/ue
+aW2dhpy7AFL138BUr1GfeYZgnpRWh+VuR9tgR4CRbINFqohD0nein5XUMHUZr+cY
+pFho0f6P5+AGhkNJJK7b/NMQULUimehnFU+wiKavEkCPcLQ3ZOd6WwadOLJMRpiK
+jrqP5KB6YoB/W83L9qMscZnA6gaLnfBwP42EQOtKVaPVLF3p0vb19SjQ30XNJgwG
+O5snRRuFQJ+IeWDklZMlEGcDrnZrRh6CtOtszBPxwmX2TNjh7XxP3KRUDj2a1dg0
+WHLp9osJuBkHmCa5Yb7EnX/eMWfrGRnJ8h3gIPP237rgyGkkdgC2uh4KSzZ1m0f7
+hUow3sArKzg3zdMbNIaV7IP/9GjJbSIHVSta7LinWgygHUtgApjKsGn4AMdZsTRO
+NE5cN7M4S7ersvufRF2/1EH485A5HPP4UpD4xtq6v6dEX6hcN+mBoKCwVoeClxSC
+avvWbI6z1LWg+m3vwNywFWJ07tDV+iLQdBk197xeNZ687qjgaxHEG9hgP2neJ7Jg
+lhAtyW0SbgJG/Ux5D7/wc2beQ8VhgsNUuX5XDZd09RsgGtSh28qUb2lhP6jCHH6v
+FDRkWtZDmQDwvofyY9cRWvRxph2GE8P+/KVV36xs1c/SJmaPevM0rI8aRP1ocLXo
+yx+eeQqPPg7gxSx0v9yLQ6j6pj1yhq3CPOnG2vp8MKSjc3U1sTNYGn1mc6Et5bbG
+cLiTIiJv566g2je2xub74Ylcj0dZdispp20M3AYHOLNsO/JNY5rKJgfiuJNbhbNV
+VLMIcB5yr8hhZIE+AoE9FWxFGLhoPwmGoQbvXqlvj8YvL7uqRduMxxwYSULTnINn
+ghG9Gt8E5ha6FSk+neOd2upj9kJng+RT2xlKLTdeaQ4T/w9Uq5Av9tqD9Q8xuUFQ
+Dh1weFH8HjgiJD67ozqZa0u2AbE60QXbTj8ygSOADTEdTNox97NeDRjb9hdyFCbn
+7T5ISYB3ZnughKN1CdwNAz+OEnID0ydidcDvH5DHMkh2Cey4ghMd1LdHH61cswcB
+awWjBCC8MB6tEPeWO49q0Jdq2nXeCWIBvi+0Bsde4BY0yk+S283pu78r6/NZC9y+
+fyNemrFZKmnBIwLQ5TLAxiK6fSo+fZPXRESGzsCCezLp5y1WoMmkEQpyYbw/XWZs
+9WmbldJuvNW2YHfmZ3n8ijtpmyzUlzZrXIPLB9TVqNGINj/bgSG4OjyMT0jmSQ55
+uGOceaG02fQjl1culVIe+axyox0UFQP3g/B5bnCDbOv9J8Nohw3OZo8Me6ThnwsB
+I55krAmytF6vr9FtlOqv+eF9Ahf/RRnr4IrklkgoZpcK21itk5SLPoW/hhfZnvgg
+Lraeycf28uaVknxVZ9gEVGcxa2Ldh9sF4u7+lhTZ/ax22X9QN0fl76pIY8zoRYSL
+o2maqbjH9q8bkOpX7MzXJ349k4h3QrduF+7bC//zoUTLo/rV/7b+LX2s2nWLvBhy
+T4+PpNAkuA==
+-----END ENCRYPTED PRIVATE KEY-----

statefun-e2e-tests/statefun-smoke-e2e-java/src/main/resources/certs/a_server.key.p8

@@ -47,6 +47,7 @@ public void runWith() throws Throwable {
     StatefulFunctionsAppContainers.Builder builder =
         StatefulFunctionsAppContainers.builder("flink-statefun-cluster", NUM_WORKERS)
             .withBuildContextFileFromClasspath("remote-module", "/remote-module/")
+            .withBuildContextFileFromClasspath("certs", "/certs/")
             .dependsOn(remoteFunction);
 
     SmokeRunner.run(parameters, builder);

statefun-e2e-tests/statefun-smoke-e2e-java/src/test/java/org/apache/flink/statefun/e2e/smoke/java/SmokeVerificationJavaE2E.java

@@ -18,4 +18,5 @@ FROM flink-statefun:3.3-SNAPSHOT
 RUN mkdir -p /opt/statefun/modules/statefun-smoke-e2e
 COPY statefun-smoke-e2e-driver.jar /opt/statefun/modules/statefun-smoke-e2e/
 COPY remote-module/ /opt/statefun/modules/statefun-smoke-e2e/
+COPY certs/ /opt/statefun/modules/statefun-smoke-e2e/certs/
 COPY flink-conf.yaml $FLINK_HOME/conf/flink-conf.yaml