feat: Add plugin API for selecting TLS client credentials (kroxylicious#3218)

* feat: Add TLS credential supplier plugin API for server connections

- Define TLS credential supplier plugin API
- Extend TargetCluster configuration for TLS credential suppliers
- Implement TLS credential supplier runtime integration
- Add comprehensive tests for TLS credential supplier plugin API
- Document TLS credential supplier configuration
- Address PR review feedback and CI failures
- Align tlsCredentials() API with design proposal (JDK types)
- Fix security and correctness issues in TLS credential supplier

See https://github.com/kroxylicious/design/blob/main/proposals/011-plugin-api-to-select-tls-credentials-for-server-connection.md

Assisted-by: AI (Pi/Claude Sonnet 4.6) <noreply@pi.dev>
Assisted-by: AI (Pi/GPT-5.5) <noreply@pi.dev>
Signed-off-by: Paco Viramontes <kidpollo@gmail.com>
Signed-off-by: Paco Viramontes <paco.viramontes@shopify.com>

Author: kidpollo

kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/tls/Tls.java

@@ -18,11 +18,43 @@
  * @param trust specifies a trust provider used by this peer to determine whether to trust the peer. If omitted platform trust is used instead.
  * @param cipherSuites specifies a custom object which contains details of allowed and denied cipher suites
  * @param protocols specifies a custom object which contains details of allowed and denied tls protocols
+ * @param credentialSupplier specifies a dynamic TLS credential supplier for per-client certificate selection (optional)
  */
 public record Tls(@Nullable KeyProvider key,
                   @Nullable TrustProvider trust,
                   @Nullable AllowDeny<String> cipherSuites,
-                  @Nullable AllowDeny<String> protocols) {
+                  @Nullable AllowDeny<String> protocols,
+                  @Nullable TlsCredentialSupplierConfig credentialSupplier) {
+
+    /**
+     * Compact constructor with validation.
+     */
+    public Tls {
+        if (key != null && credentialSupplier != null) {
+            throw new IllegalArgumentException(
+                    "Cannot configure both 'key' and 'credentialSupplier' - they are mutually exclusive. " +
+                            "Use 'key' for static TLS credentials or 'credentialSupplier' for dynamic credential selection.");
+        }
+    }
+
+    /**
+     * Creates a Tls configuration without a TLS credential supplier.
+     * This constructor is provided for backward compatibility with v0.18.0.
+     * <p>
+     * For configurations that require dynamic TLS credential selection, use the
+     * 5-parameter constructor that includes {@code credentialSupplier}.
+     *
+     * @param key specifies a key provider that provides the certificate/key used to identify this peer.
+     * @param trust specifies a trust provider used by this peer to determine whether to trust the peer.
+     * @param cipherSuites specifies allowed and denied cipher suites
+     * @param protocols specifies allowed and denied tls protocols
+     */
+    public Tls(@Nullable KeyProvider key,
+               @Nullable TrustProvider trust,
+               @Nullable AllowDeny<String> cipherSuites,
+               @Nullable AllowDeny<String> protocols) {
+        this(key, trust, cipherSuites, protocols, null);
+    }
 
     public static final String PEM = "PEM";
 

kroxylicious-api/src/main/java/io/kroxylicious/proxy/config/tls/TlsCredentialSupplierConfig.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.config.tls;
+
+import java.util.Objects;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import io.kroxylicious.proxy.plugin.PluginImplConfig;
+import io.kroxylicious.proxy.plugin.PluginImplName;
+import io.kroxylicious.proxy.tls.ServerTlsCredentialSupplierFactory;
+
+import edu.umd.cs.findbugs.annotations.Nullable;
+
+/**
+ * Configuration for a TLS credential supplier that dynamically provides TLS credentials.
+ * <p>
+ * This follows the same pattern as filter configuration, with a type specifying the
+ * {@link ServerTlsCredentialSupplierFactory} implementation and an optional config object
+ * for supplier-specific configuration.
+ * </p>
+ *
+ * @param type The type of TLS credential supplier (references a {@link ServerTlsCredentialSupplierFactory} implementation)
+ * @param config Supplier-specific configuration (optional)
+ */
+public record TlsCredentialSupplierConfig(
+                                          @PluginImplName(ServerTlsCredentialSupplierFactory.class) @JsonProperty(required = true) String type,
+                                          @Nullable @PluginImplConfig(implNameProperty = "type") Object config) {
+
+    @JsonCreator
+    public TlsCredentialSupplierConfig {
+        Objects.requireNonNull(type, "TLS credential supplier type must not be null");
+    }
+}

kroxylicious-api/src/main/java/io/kroxylicious/proxy/tls/ServerTlsCredentialSupplier.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.tls;
+
+import java.util.concurrent.CompletionStage;
+
+/**
+ * <p>Supplies TLS credentials for proxy-to-server (upstream) TLS connections.</p>
+ *
+ * <p>Instances of this interface are created by {@link ServerTlsCredentialSupplierFactory}
+ * and are responsible for providing TLS credentials (private keys and certificate chains)
+ * that the proxy uses when connecting to the target Kafka cluster.</p>
+ *
+ * <p>The supplier supports asynchronous credential retrieval, allowing implementations
+ * to load credentials from remote sources, perform cryptographic operations, or
+ * interact with external services without blocking the proxy runtime.</p>
+ *
+ * <h2>Thread Safety</h2>
+ * <p>Implementations must be thread-safe as the {@link #tlsCredentials(ServerTlsCredentialSupplierContext)}
+ * method may be called concurrently from multiple threads.</p>
+ *
+ * <h2>Non-Blocking Requirement</h2>
+ * <p>Implementations must not block the calling thread or perform heavy I/O operations synchronously.
+ * Long-running work such as network calls, file I/O, or key generation should be performed
+ * asynchronously, returning a {@link CompletionStage} that completes when the work is done.</p>
+ *
+ * <h2>Error Handling</h2>
+ * <p>If credential retrieval fails, implementations should return a {@link CompletionStage}
+ * that completes exceptionally. The runtime will handle the exception appropriately,
+ * typically by rejecting the connection attempt.</p>
+ *
+ * <h2>Usage Example: File-Based Credential Loading</h2>
+ * <pre>{@code
+ * public class FileBasedCredentialSupplier implements ServerTlsCredentialSupplier {
+ *     private final PrivateKey key;
+ *     private final X509Certificate[] chain;
+ *
+ *     public FileBasedCredentialSupplier(PrivateKey key, X509Certificate[] chain) {
+ *         this.key = key;
+ *         this.chain = chain;
+ *     }
+ *
+ *     @Override
+ *     public CompletionStage<TlsCredentials> tlsCredentials(ServerTlsCredentialSupplierContext context) {
+ *         // Plugin has already parsed the key and certificate chain (from PEM, PKCS12, etc.)
+ *         // Use context factory method to create validated TlsCredentials
+ *         TlsCredentials creds = context.tlsCredentials(key, chain);
+ *         return CompletableFuture.completedFuture(creds);
+ *     }
+ * }
+ * }</pre>
+ *
+ * <h2>Usage Example: Client-Specific Credentials</h2>
+ * <pre>{@code
+ * public class ClientSpecificSupplier implements ServerTlsCredentialSupplier {
+ *     private final Map<String, PrivateKey> clientKeys;
+ *     private final Map<String, X509Certificate[]> clientChains;
+ *     private final PrivateKey defaultKey;
+ *     private final X509Certificate[] defaultChain;
+ *
+ *     public ClientSpecificSupplier(Map<String, PrivateKey> clientKeys,
+ *                                   Map<String, X509Certificate[]> clientChains,
+ *                                   PrivateKey defaultKey,
+ *                                   X509Certificate[] defaultChain) {
+ *         this.clientKeys = clientKeys;
+ *         this.clientChains = clientChains;
+ *         this.defaultKey = defaultKey;
+ *         this.defaultChain = defaultChain;
+ *     }
+ *
+ *     @Override
+ *     public CompletionStage<TlsCredentials> tlsCredentials(ServerTlsCredentialSupplierContext context) {
+ *         Optional<ClientTlsContext> clientContext = context.clientTlsContext();
+ *
+ *         if (clientContext.isPresent() && clientContext.get().clientCertificate().isPresent()) {
+ *             String clientId = clientContext.get().clientCertificate().get()
+ *                 .getSubjectX500Principal().getName();
+ *
+ *             PrivateKey key = clientKeys.get(clientId);
+ *             X509Certificate[] chain = clientChains.get(clientId);
+ *
+ *             if (key != null && chain != null) {
+ *                 TlsCredentials creds = context.tlsCredentials(key, chain);
+ *                 return CompletableFuture.completedFuture(creds);
+ *             }
+ *         }
+ *
+ *         // Fall back to shared default credentials
+ *         TlsCredentials creds = context.tlsCredentials(defaultKey, defaultChain);
+ *         return CompletableFuture.completedFuture(creds);
+ *     }
+ * }
+ * }</pre>
+ *
+ * @see ServerTlsCredentialSupplierFactory
+ * @see TlsCredentials
+ */
+public interface ServerTlsCredentialSupplier {
+
+    /**
+     * <p>Asynchronously retrieves TLS credentials for the proxy to use when connecting
+     * to the target Kafka cluster.</p>
+     *
+     * <p>This method may be called multiple times and should return credentials
+     * appropriate for the current request context. Implementations may cache
+     * credentials, retrieve them from external sources, or generate them on-demand.</p>
+     *
+     * @param context The runtime context for this credential request
+     * @return A {@link CompletionStage} that completes with the TLS credentials
+     */
+    CompletionStage<TlsCredentials> tlsCredentials(ServerTlsCredentialSupplierContext context);
+}

kroxylicious-api/src/main/java/io/kroxylicious/proxy/tls/ServerTlsCredentialSupplierContext.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.tls;
+
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+import java.util.Optional;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * <p>Runtime context provided to {@link ServerTlsCredentialSupplier} instances when
+ * TLS credentials are requested.</p>
+ *
+ * <p>This context provides access to runtime information and a factory method for
+ * creating validated {@link TlsCredentials} instances from JDK cryptographic objects.
+ * The context is implemented by the Kroxylicious runtime and passed to the supplier's
+ * {@link ServerTlsCredentialSupplier#tlsCredentials(ServerTlsCredentialSupplierContext)} method.</p>
+ *
+ * <h2>Usage Example</h2>
+ * <pre>{@code
+ * public class MyCredentialSupplier implements ServerTlsCredentialSupplier {
+ *     private final PrivateKey defaultKey;
+ *     private final X509Certificate[] defaultChain;
+ *
+ *     @Override
+ *     public CompletionStage<TlsCredentials> tlsCredentials(ServerTlsCredentialSupplierContext context) {
+ *         TlsCredentials creds = context.tlsCredentials(defaultKey, defaultChain);
+ *         return CompletableFuture.completedFuture(creds);
+ *     }
+ * }
+ * }</pre>
+ */
+public interface ServerTlsCredentialSupplierContext {
+
+    /**
+     * <p>Returns TLS information about the client-to-proxy connection, if available.</p>
+     *
+     * <p>This provides access to the client's TLS certificate (if client authentication
+     * was performed) and the proxy's server certificate that was presented to the client.
+     * This information can be used to make credential selection decisions based on
+     * client identity or other TLS handshake data.</p>
+     *
+     * @return Optional containing the client TLS context, or empty if TLS is not in use
+     *         or if the handshake has not yet completed
+     */
+    @NonNull
+    Optional<ClientTlsContext> clientTlsContext();
+
+    /**
+     * <p>Creates a {@link TlsCredentials} instance from the given private key and certificate chain.</p>
+     *
+     * <p>This factory method validates the provided credentials before creating the
+     * {@link TlsCredentials} instance. The validation ensures that:</p>
+     * <ul>
+     *   <li>The certificate chain is structurally valid</li>
+     *   <li>The private key matches the leaf certificate's public key</li>
+     * </ul>
+     *
+     * <p>The plugin is responsible for loading and parsing the credentials from whatever
+     * source and format it uses (PEM files, PKCS12 keystores, HSMs, etc.).</p>
+     *
+     * @param key The private key corresponding to the leaf certificate.
+     * @param certificateChain The certificate chain, starting with the leaf certificate
+     *        and including any intermediate certificates up to (but not including) the root CA.
+     * @return Validated TlsCredentials instance
+     * @throws IllegalArgumentException if the key does not match the certificate or the chain is invalid
+     */
+    @NonNull
+    TlsCredentials tlsCredentials(@NonNull PrivateKey key, @NonNull X509Certificate[] certificateChain);
+}