POC to replace Kafka's DefaultSslEngineFactory with sslcontext-kickstart SSLFactory

gistfile1.txt

package org.example.kafka;

import java.io.IOException;
import java.security.KeyStore;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLParameters;

import org.apache.kafka.common.config.SslClientAuth;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.security.auth.SslEngineFactory;
import org.apache.kafka.common.security.ssl.DefaultSslEngineFactory;

import nl.altindag.ssl.SSLFactory;

/**
 * POC to replace Kafka's {@link DefaultSslEngineFactory} with {@link SSLFactory} {@link https://sslcontext-kickstart.com/}
 */
public class KikStartSslEngineFactory implements SslEngineFactory {

    private SslClientAuth sslClientAuth;

    private SSLFactory sslFactory;

    /**
     * Use original Kafka configuration to configure SSLFactory, Actual implementation of sslBuilder is application dependent
     */
    @Override
    @SuppressWarnings("unchecked")
    public void configure(Map<String, ?> configs) {
        SSLFactory.Builder sslBuilder = SSLFactory.builder();
        sslBuilder.withDefaultTrustMaterial();

        if (configs.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG) != null) {
            sslBuilder.withTrustMaterial(
                    (String) configs.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG),
                    ((Password) configs.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)).value().toCharArray());
        }

        if (configs.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG) != null) {
            sslBuilder.withIdentityMaterial(
                    (String) configs.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG),
                    ((Password) configs.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG)).value().toCharArray(),
                    ((Password) configs.get(SslConfigs.SSL_KEY_PASSWORD_CONFIG)).value().toCharArray());
        }

        List<String> cipherSuitesList = (List<String>) configs.get(SslConfigs.SSL_CIPHER_SUITES_CONFIG);
        if ((cipherSuitesList != null) && !cipherSuitesList.isEmpty()) {
            sslBuilder.withCiphers(cipherSuitesList.toArray(new String[0]));
        }

        List<String> enabledProtocolsList = (List<String>) configs.get(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG);
        if ((enabledProtocolsList != null) && !enabledProtocolsList.isEmpty()) {
            sslBuilder.withProtocols(enabledProtocolsList.toArray(new String[0]));
        }

        sslFactory = sslBuilder.build();

        sslClientAuth = SslClientAuth.forConfig((String) configs.get(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG));
        if (sslClientAuth == null) {
            sslClientAuth = SslClientAuth.NONE;
        }
    }

    private SSLEngine createSslEngine(String peerHost, int peerPort) {
        SSLEngine sslEngine = sslFactory.getSslContext().createSSLEngine(peerHost, peerPort);
        sslEngine.setEnabledCipherSuites(sslFactory.getCiphers().toArray(new String[0]));
        sslEngine.setEnabledProtocols(sslFactory.getProtocols().toArray(new String[0]));
        return sslEngine;
    }

    @Override
    public SSLEngine createServerSslEngine(String peerHost, int peerPort) {
        SSLEngine sslEngine = createSslEngine(peerHost, peerPort);
        sslEngine.setUseClientMode(false);
        switch (sslClientAuth) {
        case REQUIRED:
            sslEngine.setNeedClientAuth(true);
            break;
        case REQUESTED:
            sslEngine.setWantClientAuth(true);
            break;
        case NONE:
            break;
        }
        sslEngine.setUseClientMode(false);

        return null;
    }

    @Override
    public SSLEngine createClientSslEngine(String peerHost, int peerPort, String endpointIdentification) {
        SSLEngine sslEngine = createSslEngine(peerHost, peerPort);

        sslEngine.setUseClientMode(true);
        SSLParameters sslParams = sslEngine.getSSLParameters();
        // SSLParameters#setEndpointIdentificationAlgorithm enables endpoint validation
        // only in client mode. Hence, validation is enabled only for clients.
        sslParams.setEndpointIdentificationAlgorithm(endpointIdentification);
        sslEngine.setSSLParameters(sslParams);

        return sslEngine;
    }

    @Override
    public KeyStore keystore() {
        return null;
    }

    @Override
    public KeyStore truststore() {
        return null;
    }

    @Override
    public boolean shouldBeRebuilt(Map<String, Object> nextConfigs) {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public Set<String> reconfigurableConfigs() {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public void close() throws IOException {
        // TODO Auto-generated method stub
    }

}