import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public Socket createConnection(InetAddress peer) throws IOException
{
    // Connect to secure port for all peers if ServerEncryptionOptions is configured other than 'none'
    // When 'all', 'dc' and 'rack', server nodes always have SSL port open, and since thin client like sstableloader
    // does not know which node is in which dc/rack, connecting to SSL port is always the option.
    if (encryptionOptions != null && encryptionOptions.internode_encryption != EncryptionOptions.ServerEncryptionOptions.InternodeEncryption.none)
    {
        if (outboundBindAny)
            return SSLFactory.getSocket(encryptionOptions, peer, secureStoragePort);
        else
            return SSLFactory.getSocket(encryptionOptions, peer, secureStoragePort, FBUtilities.getLocalAddress(), 0);
    }
    else
    {
        Socket socket = SocketChannel.open(new InetSocketAddress(peer, storagePort)).socket();
        if (outboundBindAny && !socket.isBound())
            socket.bind(new InetSocketAddress(FBUtilities.getLocalAddress(), 0));
        return socket;
    }
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public EncryptionOptions.ClientEncryptionOptions getEncryptionOptions()
{
    EncryptionOptions.ClientEncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
    if (options.trustStore.present())
    {
        encOptions.enabled = true;
        encOptions.truststore = options.trustStore.value();
        encOptions.truststore_password = options.trustStorePw.value();
        if (options.keyStore.present())
        {
            encOptions.keystore = options.keyStore.value();
            encOptions.keystore_password = options.keyStorePw.value();
        }
        else
        {
            // mandatory for SSLFactory.createSSLContext(), see CASSANDRA-9325
            encOptions.keystore = encOptions.truststore;
        }
        encOptions.algorithm = options.alg.value();
        encOptions.protocol = options.protocol.value();
        encOptions.cipher_suites = options.ciphers.value().split(",");
    }
    return encOptions;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
@Test
public void testServerSocketCiphers() throws IOException
{
    ServerEncryptionOptions options = new EncryptionOptions.ServerEncryptionOptions();
    options.keystore = "test/conf/keystore.jks";
    options.keystore_password = "cassandra";
    options.truststore = options.keystore;
    options.truststore_password = options.keystore_password;
    options.cipher_suites = new String[] {
        "TLS_RSA_WITH_AES_128_CBC_SHA", "TLS_RSA_WITH_AES_256_CBC_SHA",
        "TLS_DHE_RSA_WITH_AES_128_CBC_SHA", "TLS_DHE_RSA_WITH_AES_256_CBC_SHA",
        "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA", "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA"
    };

    // enabled ciphers must be a subset of configured ciphers with identical order
    try (SSLServerSocket socket = SSLFactory.getServerSocket(options, InetAddress.getLocalHost(), 55123))
    {
        String[] enabled = socket.getEnabledCipherSuites();
        String[] wanted = Iterables.toArray(Iterables.filter(Lists.newArrayList(options.cipher_suites),
                                                             Predicates.in(Lists.newArrayList(enabled))),
                                            String.class);
        assertArrayEquals(wanted, enabled);
    }
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
private void run()
{
    // Configure the server.
    executionHandler = new ExecutionHandler(new RequestThreadPoolExecutor());
    factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
    ServerBootstrap bootstrap = new ServerBootstrap(factory);

    bootstrap.setOption("child.tcpNoDelay", true);

    // Set up the event pipeline factory.
    final EncryptionOptions.ClientEncryptionOptions clientEnc = DatabaseDescriptor.getClientEncryptionOptions();
    if (clientEnc.enabled)
    {
        logger.info("Enabling encrypted CQL connections between client and server");
        bootstrap.setPipelineFactory(new SecurePipelineFactory(this, clientEnc));
    }
    else
    {
        bootstrap.setPipelineFactory(new PipelineFactory(this));
    }

    // Bind and start to accept incoming connections.
    logger.info("Starting listening for CQL clients on {}...", socket);
    Channel channel = bootstrap.bind(socket);
    connectionTracker.allChannels.add(channel);
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public JavaDriverClient getJavaDriverClient()
{
    if (client != null)
        return client;

    try
    {
        synchronized (this)
        {
            String currentNode = node.randomNode();
            if (client != null)
                return client;

            EncryptionOptions.ClientEncryptionOptions encOptions = transport.getEncryptionOptions();
            JavaDriverClient c = new JavaDriverClient(currentNode, port.nativePort, encOptions);
            c.connect(mode.compression());
            c.execute("USE \"" + schema.keyspace + "\";", org.apache.cassandra.db.ConsistencyLevel.ONE);
            return client = c;
        }
    }
    catch (Exception e)
    {
        throw new RuntimeException(e);
    }
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
private static SSLContext createSSLContext(EncryptionOptions options) throws IOException {
    SSLContext ctx;
    try {
        ctx = SSLContext.getInstance(PROTOCOL);
        TrustManagerFactory tmf = null;
        KeyManagerFactory kmf = null;

        tmf = TrustManagerFactory.getInstance(ALGORITHM);
        KeyStore ts = KeyStore.getInstance(STORE_TYPE);
        ts.load(new FileInputStream(options.truststore), options.truststore_password.toCharArray());
        tmf.init(ts);

        kmf = KeyManagerFactory.getInstance(ALGORITHM);
        KeyStore ks = KeyStore.getInstance(STORE_TYPE);
        ks.load(new FileInputStream(options.keystore), options.keystore_password.toCharArray());
        kmf.init(ks, options.keystore_password.toCharArray());

        ctx.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

    } catch (Exception e) {
        throw new IOException("Error creating the initializing the SSL Context", e);
    }
    return ctx;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public static SSLServerSocket getServerSocket(EncryptionOptions options, InetAddress address, int port) throws IOException
{
    SSLContext ctx = createSSLContext(options, true);
    SSLServerSocket serverSocket = (SSLServerSocket)ctx.getServerSocketFactory().createServerSocket();
    serverSocket.setReuseAddress(true);
    String[] suits = filterCipherSuites(serverSocket.getSupportedCipherSuites(), options.cipher_suites);
    serverSocket.setEnabledCipherSuites(suits);
    serverSocket.setNeedClientAuth(options.require_client_auth);
    serverSocket.setEnabledProtocols(ACCEPTED_PROTOCOLS);
    serverSocket.bind(new InetSocketAddress(address, port), 500);
    return serverSocket;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
/** Create a socket and connect */
public static SSLSocket getSocket(EncryptionOptions options, InetAddress address, int port, InetAddress localAddress, int localPort) throws IOException
{
    SSLContext ctx = createSSLContext(options, true);
    SSLSocket socket = (SSLSocket) ctx.getSocketFactory().createSocket(address, port, localAddress, localPort);
    String[] suits = filterCipherSuites(socket.getSupportedCipherSuites(), options.cipher_suites);
    socket.setEnabledCipherSuites(suits);
    socket.setEnabledProtocols(ACCEPTED_PROTOCOLS);
    return socket;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public SecureInitializer(Server server, EncryptionOptions encryptionOptions)
{
    super(server);
    this.encryptionOptions = encryptionOptions;
    try
    {
        this.sslContext = SSLFactory.createSSLContext(encryptionOptions, encryptionOptions.require_client_auth);
    }
    catch (IOException e)
    {
        throw new RuntimeException("Failed to setup secure pipeline", e);
    }
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public BulkLoadConnectionFactory(int storagePort, int secureStoragePort, EncryptionOptions.ServerEncryptionOptions encryptionOptions, boolean outboundBindAny)
{
    this.storagePort = storagePort;
    this.secureStoragePort = secureStoragePort;
    this.encryptionOptions = encryptionOptions;
    this.outboundBindAny = outboundBindAny;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
private static void configureTransportFactory(ITransportFactory transportFactory, EncryptionOptions encOptions)
{
    Map<String, String> options = new HashMap<>();
    // If the supplied factory supports the same set of options as our SSL impl, set those
    if (transportFactory.supportedOptions().contains(SSLTransportFactory.TRUSTSTORE))
        options.put(SSLTransportFactory.TRUSTSTORE, encOptions.truststore);
    if (transportFactory.supportedOptions().contains(SSLTransportFactory.TRUSTSTORE_PASSWORD))
        options.put(SSLTransportFactory.TRUSTSTORE_PASSWORD, encOptions.truststore_password);
    if (transportFactory.supportedOptions().contains(SSLTransportFactory.PROTOCOL))
        options.put(SSLTransportFactory.PROTOCOL, encOptions.protocol);
    if (transportFactory.supportedOptions().contains(SSLTransportFactory.CIPHER_SUITES))
        options.put(SSLTransportFactory.CIPHER_SUITES, Joiner.on(',').join(encOptions.cipher_suites));

    if (transportFactory.supportedOptions().contains(SSLTransportFactory.KEYSTORE)
            && encOptions.require_client_auth)
        options.put(SSLTransportFactory.KEYSTORE, encOptions.keystore);
    if (transportFactory.supportedOptions().contains(SSLTransportFactory.KEYSTORE_PASSWORD)
            && encOptions.require_client_auth)
        options.put(SSLTransportFactory.KEYSTORE_PASSWORD, encOptions.keystore_password);

    // Now check if any of the factory's supported options are set as system properties
    for (String optionKey : transportFactory.supportedOptions())
        if (System.getProperty(optionKey) != null)
            options.put(optionKey, System.getProperty(optionKey));

    transportFactory.setOptions(options);
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public JavaDriverClient getJavaDriverClient(boolean setKeyspace)
{
    if (client != null)
        return client;

    try
    {
        synchronized (this)
        {
            String currentNode = node.randomNode();
            if (client != null)
                return client;

            EncryptionOptions.ClientEncryptionOptions encOptions = transport.getEncryptionOptions();
            JavaDriverClient c = new JavaDriverClient(this, currentNode, port.nativePort, encOptions);
            c.connect(mode.compression());
            if (setKeyspace)
                c.execute("USE \"" + schema.keyspace + "\";", org.apache.cassandra.db.ConsistencyLevel.ONE);

            return client = c;
        }
    }
    catch (Exception e)
    {
        throw new RuntimeException(e);
    }
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public EncryptionOptions.ClientEncryptionOptions getEncryptionOptions()
{
    EncryptionOptions.ClientEncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
    if (options.trustStore.present())
    {
        encOptions.enabled = true;
        encOptions.truststore = options.trustStore.value();
        encOptions.truststore_password = options.trustStorePw.value();
        encOptions.algorithm = options.alg.value();
        encOptions.protocol = options.protocol.value();
        encOptions.cipher_suites = options.ciphers.value().split(",");
    }
    return encOptions;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public JavaDriverClient(StressSettings settings, String host, int port, EncryptionOptions.ClientEncryptionOptions encryptionOptions)
{
    this.host = host;
    this.port = port;
    this.username = settings.mode.username;
    this.password = settings.mode.password;
    this.authProvider = settings.mode.authProvider;
    this.encryptionOptions = encryptionOptions;
    if (settings.node.isWhiteList)
        whitelist = new WhiteListPolicy(new DCAwareRoundRobinPolicy(), settings.node.resolveAll(settings.port.nativePort));
    else
        whitelist = null;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public static SSLServerSocket getServerSocket(EncryptionOptions options, InetAddress address, int port) throws IOException
{
    SSLContext ctx = createSSLContext(options, true);
    SSLServerSocket serverSocket = (SSLServerSocket)ctx.getServerSocketFactory().createServerSocket();
    serverSocket.setReuseAddress(true);
    String[] suits = filterCipherSuites(serverSocket.getSupportedCipherSuites(), options.cipher_suites);
    serverSocket.setEnabledCipherSuites(suits);
    serverSocket.setNeedClientAuth(options.require_client_auth);
    serverSocket.bind(new InetSocketAddress(address, port), 100);
    return serverSocket;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
/** Create a socket and connect */
public static SSLSocket getSocket(EncryptionOptions options, InetAddress address, int port, InetAddress localAddress, int localPort) throws IOException
{
    SSLContext ctx = createSSLContext(options, true);
    SSLSocket socket = (SSLSocket) ctx.getSocketFactory().createSocket(address, port, localAddress, localPort);
    String[] suits = filterCipherSuites(socket.getSupportedCipherSuites(), options.cipher_suites);
    socket.setEnabledCipherSuites(suits);
    return socket;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public static SSLContext createSSLContext(EncryptionOptions options, boolean buildTruststore) throws IOException
{
    FileInputStream tsf = null;
    FileInputStream ksf = null;
    SSLContext ctx;
    try
    {
        ctx = SSLContext.getInstance(options.protocol);
        TrustManager[] trustManagers = null;

        if(buildTruststore)
        {
            tsf = new FileInputStream(options.truststore);
            TrustManagerFactory tmf = TrustManagerFactory.getInstance(options.algorithm);
            KeyStore ts = KeyStore.getInstance(options.store_type);
            ts.load(tsf, options.truststore_password.toCharArray());
            tmf.init(ts);
            trustManagers = tmf.getTrustManagers();
        }

        ksf = new FileInputStream(options.keystore);
        KeyManagerFactory kmf = KeyManagerFactory.getInstance(options.algorithm);
        KeyStore ks = KeyStore.getInstance(options.store_type);
        ks.load(ksf, options.keystore_password.toCharArray());
        kmf.init(ks, options.keystore_password.toCharArray());

        ctx.init(kmf.getKeyManagers(), trustManagers, null);

    }
    catch (Exception e)
    {
        throw new IOException("Error creating the initializing the SSL Context", e);
    }
    finally
    {
        FileUtils.closeQuietly(tsf);
        FileUtils.closeQuietly(ksf);
    }
    return ctx;
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
private void run()
{
    // Check that a SaslAuthenticator can be provided by the configured
    // IAuthenticator. If not, don't start the server.
    IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
    if (authenticator.requireAuthentication() && !(authenticator instanceof ISaslAwareAuthenticator))
    {
        logger.error("Not starting native transport as the configured IAuthenticator is not capable of SASL authentication");
        isRunning.compareAndSet(true, false);
        return;
    }

    // Configure the server.
    executionHandler = new ExecutionHandler(new RequestThreadPoolExecutor());
    factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
    ServerBootstrap bootstrap = new ServerBootstrap(factory);

    bootstrap.setOption("child.tcpNoDelay", true);

    // Set up the event pipeline factory.
    final EncryptionOptions.ClientEncryptionOptions clientEnc = DatabaseDescriptor.getClientEncryptionOptions();
    if (clientEnc.enabled)
    {
        logger.info("Enabling encrypted CQL connections between client and server");
        bootstrap.setPipelineFactory(new SecurePipelineFactory(this, clientEnc));
    }
    else
    {
        bootstrap.setPipelineFactory(new PipelineFactory(this));
    }

    // Bind and start to accept incoming connections.
    logger.info("Starting listening for CQL clients on {}...", socket);
    Channel channel = bootstrap.bind(socket);
    connectionTracker.allChannels.add(channel);
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public SecurePipelineFactory(Server server, EncryptionOptions encryptionOptions)
{
    super(server);
    this.encryptionOptions = encryptionOptions;
    try
    {
        this.sslContext = SSLFactory.createSSLContext(encryptionOptions, false);
    }
    catch (IOException e)
    {
        throw new RuntimeException("Failed to setup secure pipeline", e);
    }
}
 

import org.apache.cassandra.config.EncryptionOptions; //导入依赖的package包/类
public static SSLServerSocket getServerSocket(EncryptionOptions options, InetAddress address, int port) throws IOException
{
    SSLContext ctx = createSSLContext(options, true);
    SSLServerSocket serverSocket = (SSLServerSocket)ctx.getServerSocketFactory().createServerSocket();
    serverSocket.setReuseAddress(true);
    String[] suites = filterCipherSuites(serverSocket.getSupportedCipherSuites(), options.cipher_suites);
    serverSocket.setEnabledCipherSuites(suites);
    serverSocket.setNeedClientAuth(options.require_client_auth);
    serverSocket.setEnabledProtocols(ACCEPTED_PROTOCOLS);
    serverSocket.bind(new InetSocketAddress(address, port), 500);
    return serverSocket;
}