diff --git a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java b/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java index cbaba8d63a..64864f0b72 100644 --- a/kroxylicious-api/src/main/java/io/kroxylicious/proxy/filter/FilterContext.java +++ b/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. + *

+ * 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 diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java index 5c851e72a2..5415cbce0b 100644 --- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/FilterHandler.java +++ b/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(); } diff --git a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java b/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java index 7b1f3066d6..fb7ce68d74 100644 --- a/kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java +++ b/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)); } diff --git a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java index 53d6d3d53a..5bb348e79c 100644 --- a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/FilterHarness.java +++ b/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); diff --git a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java index 44387aae42..68423fbcbf 100644 --- a/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java +++ b/kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandlerTest.java @@ -5,6 +5,7 @@ */ package io.kroxylicious.proxy.internal; +import java.security.Principal; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -12,6 +13,10 @@ 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 connectContext = new AtomicReference<>(); @@ -123,12 +133,14 @@ public void closeChannel() { public static List provideArgsForExpectedFlow() { var result = new ArrayList(); 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) {