ssl client principal propagation

kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java

@@ -42,6 +42,20 @@ public interface FilterContext {
     @Nullable
     String sniHostname();
 
+    /**
+     * Returns the principal name from the client's TLS certificate.
+     * <p>
+     * For X.509 certificates, this is typically the Distinguished Name (DN) in format
+     * such as "CN=clientName, O=organizationName, C=US". This information can be used
+     * by filters for client authentication, authorization decisions, or audit logging.
+     *
+     * @return the client certificate principal as a string
+     */
+    @Nullable
+    default String downstreamCertificatePrincipal() {
+        return null;
+    }
+
     /**
      * Creates a builder for a request filter result objects.  This object encapsulates
      * the request to forward and optionally orders for actions such as closing

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java

@@ -59,6 +59,7 @@ public class FilterHandler extends ChannelDuplexHandler {
     private static final Logger LOGGER = LoggerFactory.getLogger(FilterHandler.class);
     private final long timeoutMs;
     private final String sniHostname;
+    private final String downstreamCertificatePrincipal;
     private final VirtualClusterModel virtualClusterModel;
     private final Channel inboundChannel;
     private final FilterAndInvoker filterAndInvoker;
@@ -67,10 +68,12 @@ public class FilterHandler extends ChannelDuplexHandler {
     private ChannelHandlerContext ctx;
     private PromiseFactory promiseFactory;
 
-    public FilterHandler(FilterAndInvoker filterAndInvoker, long timeoutMs, String sniHostname, VirtualClusterModel virtualClusterModel, Channel inboundChannel) {
+    public FilterHandler(FilterAndInvoker filterAndInvoker, long timeoutMs, String sniHostname, String downstreamCertificatePrincipal,
+                         VirtualClusterModel virtualClusterModel, Channel inboundChannel) {
         this.filterAndInvoker = Objects.requireNonNull(filterAndInvoker);
         this.timeoutMs = Assertions.requireStrictlyPositive(timeoutMs, "timeout");
         this.sniHostname = sniHostname;
+        this.downstreamCertificatePrincipal = downstreamCertificatePrincipal;
         this.virtualClusterModel = virtualClusterModel;
         this.inboundChannel = inboundChannel;
     }
@@ -475,6 +478,12 @@ public String sniHostname() {
             return sniHostname;
         }
 
+        @Nullable
+        @Override
+        public String downstreamCertificatePrincipal() {
+            return downstreamCertificatePrincipal;
+        }
+
         public String getVirtualClusterName() {
             return virtualClusterModel.getClusterName();
         }

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java

@@ -13,6 +13,8 @@
 import java.util.List;
 import java.util.Objects;
 
+import javax.net.ssl.SSLPeerUnverifiedException;
+
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.ApiVersionsResponseData;
 import org.apache.kafka.common.message.ApiVersionsResponseDataJsonConverter;
@@ -34,6 +36,7 @@
 import io.netty.handler.logging.LoggingHandler;
 import io.netty.handler.ssl.SniCompletionEvent;
 import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 
 import io.kroxylicious.proxy.filter.FilterAndInvoker;
 import io.kroxylicious.proxy.filter.NetFilter;
@@ -68,6 +71,7 @@ public class KafkaProxyFrontendHandler
 
     private static final String NET_FILTER_INVOKED_IN_WRONG_STATE = "NetFilterContext invoked in wrong session state";
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProxyFrontendHandler.class);
+    private static final String ANONYMOUS = "ANONYMOUS";
 
     /** Cache ApiVersions response which we use when returning ApiVersions ourselves */
     private static final ApiVersionsResponseData API_VERSIONS_RESPONSE;
@@ -87,6 +91,21 @@ public class KafkaProxyFrontendHandler
     private boolean pendingClientFlushes;
     private @Nullable AuthenticationEvent authentication;
     private @Nullable String sniHostname;
+    private @Nullable String downstreamCertificatePrincipal;
+
+    @Nullable
+    public String getDownstreamCertificatePrincipal() {
+        return downstreamCertificatePrincipal;
+    }
+
+    @VisibleForTesting
+    SslHandler getSslHandler(ChannelHandlerContext ctx) {
+        if (ctx == null) {
+            throw new IllegalStateException("No context available");
+        }
+        SslHandler sslHandler = ctx.pipeline().get(SslHandler.class);
+        return sslHandler;
+    }
 
     // Flag if we receive a channelReadComplete() prior to outbound connection activation
     // so we can perform the channelReadComplete()/outbound flush & auto_read
@@ -171,6 +190,18 @@ public void userEventTriggered(
         else if (event instanceof AuthenticationEvent authenticationEvent) {
             this.authentication = authenticationEvent;
         }
+        else if (event instanceof SslHandshakeCompletionEvent sslHandshakeCompletionEvent) {
+            if (sslHandshakeCompletionEvent.isSuccess()) {
+                SslHandler sslHandler = getSslHandler(ctx);
+                try {
+                    downstreamCertificatePrincipal = sslHandler.engine().getSession().getPeerPrincipal().toString();
+                }
+                catch (SSLPeerUnverifiedException e) {
+                    LOGGER.debug("No client principal received, setting principal as ANONYMOUS");
+                    downstreamCertificatePrincipal = ANONYMOUS;
+                }
+            }
+        }
         super.userEventTriggered(ctx, event);
     }
 
@@ -640,6 +671,7 @@ private void addFiltersToPipeline(
                             protocolFilter,
                             20000,
                             sniHostname,
+                            downstreamCertificatePrincipal,
                             virtualClusterModel,
                             inboundChannel));
         }

kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java

@@ -91,7 +91,7 @@ protected void buildChannel(Filter... filters) {
                     return d2;
                 })) // reverses order
                 .stream()
-                .map(f -> new FilterHandler(getOnlyElement(FilterAndInvoker.build(f.getClass().getSimpleName(), f)), timeoutMs, null, testVirtualCluster, inboundChannel))
+                .map(f -> new FilterHandler(getOnlyElement(FilterAndInvoker.build(f.getClass().getSimpleName(), f)), timeoutMs, null, null, testVirtualCluster, inboundChannel))
                 .map(ChannelHandler.class::cast);
         var handlers = Stream.concat(channelProcessors, filterHandlers);
 

kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java

@@ -5,13 +5,18 @@
  */
 package io.kroxylicious.proxy.internal;
 
+import java.security.Principal;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Stream;
 
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLSession;
+
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.MetadataRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
@@ -43,6 +48,8 @@
 import io.netty.handler.codec.haproxy.HAProxyProxiedProtocol;
 import io.netty.handler.ssl.SniCompletionEvent;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 
 import io.kroxylicious.proxy.filter.NetFilter;
 import io.kroxylicious.proxy.frame.DecodedFrame;
@@ -76,9 +83,12 @@ class KafkaProxyFrontendHandlerTest {
 
     public static final String SNI_HOSTNAME = "external.example.com";
     public static final String CLUSTER_HOST = "internal.example.org";
+    public static final String CLIENT_PRINCIPAL = "CN=client,O=example";
     public static final int CLUSTER_PORT = 9092;
     EmbeddedChannel inboundChannel;
     EmbeddedChannel outboundChannel;
+    SSLSession sslSession = mock(SSLSession.class);
+    Principal mockPrincipal = mock(Principal.class);
 
     int corrId = 0;
     private final AtomicReference<NetFilter.NetFilterContext> connectContext = new AtomicReference<>();
@@ -123,12 +133,14 @@ public void closeChannel() {
     public static List<Arguments> provideArgsForExpectedFlow() {
         var result = new ArrayList<Arguments>();
         boolean[] tf = { true, false };
-        for (boolean sslConfigured : tf) {
-            for (boolean haProxyConfigured : tf) {
-                for (boolean saslOffloadConfigured : tf) {
-                    for (boolean sendApiVersions : tf) {
-                        for (boolean sendSasl : tf) {
-                            result.add(Arguments.of(sslConfigured, haProxyConfigured, saslOffloadConfigured, sendApiVersions, sendSasl));
+        for (boolean clientAuthConfigured : tf) {
+            for (boolean sslConfigured : tf) {
+                for (boolean haProxyConfigured : tf) {
+                    for (boolean saslOffloadConfigured : tf) {
+                        for (boolean sendApiVersions : tf) {
+                            for (boolean sendSasl : tf) {
+                                result.add(Arguments.of(clientAuthConfigured, sslConfigured, haProxyConfigured, saslOffloadConfigured, sendApiVersions, sendSasl));
+                            }
                         }
                     }
                 }
@@ -310,12 +322,29 @@ ChannelFuture initConnection(String remoteHost, int remotePort, Bootstrap bootst
                 outboundChannel.pipeline().fireChannelRegistered();
                 return outboundChannel.newPromise();
             }
+
+            @Override
+            SslHandler getSslHandler(ChannelHandlerContext ctx) {
+                // We don't need to test SSL here, so just return a mock
+                SSLEngine sslEngine = mock(SSLEngine.class);
+                when(sslEngine.getSession()).thenReturn(sslSession);
+                when(mockPrincipal.toString()).thenReturn(CLIENT_PRINCIPAL);
+                return new SslHandler(sslEngine) {
+                    @Override
+                    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
+                        if (evt instanceof SniCompletionEvent) {
+                            ctx.fireUserEventTriggered(evt);
+                        }
+                    }
+                };
+            }
         };
     }
 
     /**
      * Test the normal flow, in a number of configurations.
      *
+     * @param clientAuthConfigured  whether mTLS is configured
      * @param sslConfigured         Whether SSL is configured
      * @param haProxyConfigured
      * @param saslOffloadConfigured
@@ -324,7 +353,8 @@ ChannelFuture initConnection(String remoteHost, int remotePort, Bootstrap bootst
      */
     @ParameterizedTest
     @MethodSource("provideArgsForExpectedFlow")
-    void expectedFlow(boolean sslConfigured,
+    void expectedFlow(boolean clientAuthConfigured,
+                      boolean sslConfigured,
                       boolean haProxyConfigured,
                       boolean saslOffloadConfigured,
                       boolean sendApiVersions,
@@ -379,11 +409,37 @@ void expectedFlow(boolean sslConfigured,
         var handler = handler(filter, dp, endpointBinding);
         initialiseInboundChannel(handler);
 
+        if (clientAuthConfigured) {
+            try {
+                when(sslSession.getPeerPrincipal()).thenReturn(mockPrincipal);
+            }
+            catch (SSLPeerUnverifiedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+        else {
+            try {
+                when(sslSession.getPeerPrincipal()).thenThrow(new SSLPeerUnverifiedException("No peer certificate"));
+            }
+            catch (SSLPeerUnverifiedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
         if (sslConfigured) {
             // Simulate the SSL handler
             inboundChannel.pipeline().fireUserEventTriggered(new SniCompletionEvent(SNI_HOSTNAME));
+            inboundChannel.pipeline().fireUserEventTriggered(SslHandshakeCompletionEvent.SUCCESS);
         }
 
+        if (sslConfigured) {
+            if (clientAuthConfigured) {
+                assertEquals(CLIENT_PRINCIPAL, handler.getDownstreamCertificatePrincipal());
+            }
+            else {
+                assertEquals("ANONYMOUS", handler.getDownstreamCertificatePrincipal());
+            }
+        }
         assertThat(proxyChannelStateMachine.state()).isExactlyInstanceOf(ProxyChannelState.ClientActive.class);
 
         if (haProxyConfigured) {