spring - 使用 Spring Data Cassandra 启用 Cassandra 客户端到节点加密

Question

我有一个使用 Spring 和 Cassadra 数据库构建的项目。实际上，将 cassandra.yml 中的客户端加密设置为 false 所有的工作。

工作设置(使用 spring XML bean 和客户端加密设置为 false)

<cassandra:cluster contact-points="${cassandra.contactpoints}"
                   port="${cassandra.port}"
                   username="${cassandra.username}"
                   password="${cassandra.password}"  />

<cassandra:session id="cassandraSession"
                   keyspace-name="${cassandra.keyspace}" />

<cassandra:mapping />
<cassandra:converter />
<cassandra:template id="cassandraTemplate" />

现在我想启用客户端到节点的加密(节点到节点已经设置好并且可以工作)。我在网上找到了一些教程，解释了如何创建所需的 keystore ，但我无法找到如何设置 Spring 项目来处理它。

我在 cassandra.yml 中启用了客户端到节点的加密并尝试设置 ssl-enabled="true"在集群 bean 中，但我无法从 Spring 连接到 cassandra。我知道有一个 ssl-options-ref cassandra cluster bean 中的属性，但我找不到任何关于如何使用它的教程。

还试图遵循这个: How to set System Properties on run time Spring 3 MVC在运行时设置系统属性以加载正确的信任库文件，但添加该代码没有区别。我总是得到这个错误:

2016-08-19 13:46:26 INFO  NettyUtil:83 - Did not find Netty's native epoll transport in the classpath, defaulting to NIO.
2016-08-19 13:46:26 WARN  DefaultPromise:151 - An exception was thrown by com.datastax.driver.core.Connection$10.operationComplete()
java.util.concurrent.RejectedExecutionException: Task com.datastax.driver.core.Connection$10$1@5c21d76e rejected from java.util.concurrent.ThreadPoolExecutor@78eaecc1[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    at com.google.common.util.concurrent.MoreExecutors$ListeningDecorator.execute(MoreExecutors.java:556)
    at com.datastax.driver.core.Connection$10.operationComplete(Connection.java:573)
    at com.datastax.driver.core.Connection$10.operationComplete(Connection.java:547)
    at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:514)
    at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:507)
    at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:486)
    at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:427)
    at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:129)
    at io.netty.channel.PendingWriteQueue.safeFail(PendingWriteQueue.java:286)
    at io.netty.channel.PendingWriteQueue.removeAndFailAll(PendingWriteQueue.java:132)
    at io.netty.handler.ssl.SslHandler.setHandshakeFailure(SslHandler.java:1231)
    at io.netty.handler.ssl.SslHandler.setHandshakeFailure(SslHandler.java:1205)
    at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1060)
    at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:900)
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:366)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:352)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:345)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1294)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:366)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:352)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:911)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:572)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:513)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:427)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:399)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140)
    at java.lang.Thread.run(Thread.java:745)
2016-08-19 13:46:26 INFO  DefaultListableBeanFactory:444 - Destroying singletons in org.springframework.beans.factory.support.DefaultListableBeanFactory@51cc1210: defining beans [org.springframework.context.support.PropertySourcesPlaceholderConfigurer#0,cassandraCluster,cassandraSession,cassandraMapping,cassandraConverter,cassandraTemplate,smsBehavior,mailSender,preConfiguredMessage,recordingBehavior,user,trustStore]; root of factory hierarchy
2016-08-19 13:46:28 ERROR ContextLoader:331 - Context initialization failed
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'cassandraSession': Invocation of init method failed; nested exception is com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (tried: ******** (com.datastax.driver.core.exceptions.TransportException: [*******] Channel has been closed))
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1512)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:521)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:458)
    at org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:296)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:223)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:293)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:194)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:610)
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:932)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:479)
    at org.springframework.web.context.ContextLoader.configureAndRefreshWebApplicationContext(ContextLoader.java:410)
    at org.springframework.web.context.ContextLoader.initWebApplicationContext(ContextLoader.java:306)
    at org.springframework.web.context.ContextLoaderListener.contextInitialized(ContextLoaderListener.java:112)
    at org.apache.catalina.core.StandardContext.listenerStart(StandardContext.java:4939)
    at org.apache.catalina.core.StandardContext.startInternal(StandardContext.java:5434)
    at org.apache.catalina.util.LifecycleBase.start(LifecycleBase.java:150)
    at org.apache.catalina.core.ContainerBase.addChildInternal(ContainerBase.java:901)
    at org.apache.catalina.core.ContainerBase.addChild(ContainerBase.java:877)
    at org.apache.catalina.core.StandardHost.addChild(StandardHost.java:633)
    at org.apache.catalina.startup.HostConfig.deployDescriptor(HostConfig.java:656)
    at org.apache.catalina.startup.HostConfig.deployApps(HostConfig.java:535)
    at org.apache.catalina.startup.HostConfig.check(HostConfig.java:1461)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:483)
    at org.apache.tomcat.util.modeler.BaseModelMBean.invoke(BaseModelMBean.java:301)
    at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.invoke(DefaultMBeanServerInterceptor.java:819)
    at com.sun.jmx.mbeanserver.JmxMBeanServer.invoke(JmxMBeanServer.java:801)
    at org.apache.catalina.manager.ManagerServlet.check(ManagerServlet.java:1445)
    at org.apache.catalina.manager.ManagerServlet.deploy(ManagerServlet.java:860)
    at org.apache.catalina.manager.ManagerServlet.doGet(ManagerServlet.java:357)
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:621)
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:728)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:305)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
    at org.apache.catalina.filters.SetCharacterEncodingFilter.doFilter(SetCharacterEncodingFilter.java:108)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:243)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:210)
    at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:222)
    at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:123)
    at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:611)
    at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:171)
    at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:99)
    at org.apache.catalina.valves.AccessLogValve.invoke(AccessLogValve.java:953)
    at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:118)
    at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:408)
    at org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:1023)
    at org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:589)
    at org.apache.tomcat.util.net.JIoEndpoint$SocketProcessor.run(JIoEndpoint.java:312)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (tried: ************* (com.datastax.driver.core.exceptions.TransportException: [*********] Channel has been closed))
    at com.datastax.driver.core.ControlConnection.reconnectInternal(ControlConnection.java:233)
    at com.datastax.driver.core.ControlConnection.connect(ControlConnection.java:79)
    at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1424)
    at com.datastax.driver.core.Cluster.init(Cluster.java:163)
    at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:334)
    at com.datastax.driver.core.Cluster.connect(Cluster.java:284)
    at org.springframework.cassandra.config.CassandraCqlSessionFactoryBean.connect(CassandraCqlSessionFactoryBean.java:100)
    at org.springframework.cassandra.config.CassandraCqlSessionFactoryBean.afterPropertiesSet(CassandraCqlSessionFactoryBean.java:94)
    at org.springframework.data.cassandra.config.CassandraSessionFactoryBean.afterPropertiesSet(CassandraSessionFactoryBean.java:60)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1571)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1509)
    ... 52 more

更新

假设有一个 SslOptionFactoryBean延伸 AbstractFactoryBean<SSLOptions>使用它并初始化集群的示例配置类可以是:

    @Configuration
    @PropertySource(value = {"file:./db.properties"})
    public class CassandraConfig extends SpringHttpSessionConfiguration {

        @Autowired
        private SSLOptions sslOption;

        @Bean
        @Lazy(false)
        public SslOptionFactoryBean sslOptions() throws URISyntaxException, IOException {

            Resource trustStore = new FileSystemResource(env.getProperty("db.truststorefilename", ""));
            Resource keyStore = new FileSystemResource(env.getProperty("db.keystorefilename", ""));
            String trustStorePassword = env.getProperty("db.truststorepassword", "");
            String keyStorePassword = env.getProperty("db.keystorepassword", "");

            SslOptionFactoryBean option = new SslOptionFactoryBean();
            option.setTrustStore(trustStore);
            option.setTrustStorePassword(trustStorePassword);
            option.setKeyStore(keyStore);
            option.setKeyStorePassword(keyStorePassword);

            return option;
        }

        @Bean
        public Cluster cluster() throws Exception {

            // load node address and port

            /* ... */

            return Cluster.builder()
                    .addContactPoint(node)
                    .withPort(port)
                    .withSSL(sslOption)
                    .build();
        }

    }

Answer 1

长话短说

使用 JVM 外部的系统属性设置信任库 (-Djavax.net.ssl.trustStore=...) 或添加对系统属性工厂 bean 的依赖以确保这些属性在 Cassandra 客户端初始化之前应用。使用 ssl-options-ref 需要更多努力。

说明

需要在初始化 Datastax 客户端之前应用基于系统属性的 SSL 配置。事实上，应尽早应用属性以确保没有其他类初始化默认 SSL 上下文，因为默认 SSL 上下文已缓存。在任何组件初始化默认 SSL 上下文后应用 javax.net.ssl.trustStore 将不会应用您的设置。

您可以使用 ssl-options-ref 为配置的 SSL 上下文提供专用的 SSL 选项，但这需要额外的代码。 com.datastax.driver.core.SSLOptions 不能只配置属性值。另请注意，将 Cassandra 驱动程序升级到 3.0 需要不同的 SSL 上下文初始化，因为驱动程序 API 已更改。

示例 SslOptionsFactoryBean 可能如下所示:

public class SslOptionsFactoryBean extends AbstractFactoryBean<SSLOptions> {

    private Resource keyStore;
    private String keyStorePassword;
    private Resource trustStore;
    private String trustStorePassword;

    @Override
    public Class<?> getObjectType() {
        return SSLOptions.class;
    }

    @Override
    protected SSLOptions createInstance() throws Exception {

        KeyManager[] keyManagers = getKeyStore() != null
                ? createKeyManagerFactory(getKeyStore(), getKeyStorePassword()).getKeyManagers() : null;

        TrustManager[] trustManagers = getTrustStore() != null
                ? createTrustManagerFactory(getTrustStore(), getTrustStorePassword()).getTrustManagers() : null;

        SSLContext sslContext = SSLContext.getInstance("TLS");
        sslContext.init(keyManagers, trustManagers, null);

        return new SSLOptions(sslContext, SSLOptions.DEFAULT_SSL_CIPHER_SUITES);
    }

    private static KeyManagerFactory createKeyManagerFactory(Resource keystoreFile, String storePassword)
            throws GeneralSecurityException, IOException {

        KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());

        try (InputStream inputStream = keystoreFile.getInputStream()) {
            keyStore.load(inputStream, StringUtils.hasText(storePassword) ? storePassword.toCharArray() : null);
        }

        KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        keyManagerFactory.init(keyStore, StringUtils.hasText(storePassword) ? storePassword.toCharArray() : new char[0]);

        return keyManagerFactory;
    }

    private static TrustManagerFactory createTrustManagerFactory(Resource trustFile, String storePassword)
            throws GeneralSecurityException, IOException {

        KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());

        try (InputStream inputStream = trustFile.getInputStream()) {
            trustStore.load(inputStream, StringUtils.hasText(storePassword) ? storePassword.toCharArray() : null);
        }

        TrustManagerFactory trustManagerFactory = TrustManagerFactory
                .getInstance(TrustManagerFactory.getDefaultAlgorithm());
        trustManagerFactory.init(trustStore);

        return trustManagerFactory;
    }

    public Resource getKeyStore() {
        return keyStore;
    }

    public void setKeyStore(Resource keyStore) {
        this.keyStore = keyStore;
    }

    public String getKeyStorePassword() {
        return keyStorePassword;
    }

    public void setKeyStorePassword(String keyStorePassword) {
        this.keyStorePassword = keyStorePassword;
    }

    public Resource getTrustStore() {
        return trustStore;
    }

    public void setTrustStore(Resource trustStore) {
        this.trustStore = trustStore;
    }

    public String getTrustStorePassword() {
        return trustStorePassword;
    }

    public void setTrustStorePassword(String trustStorePassword) {
        this.trustStorePassword = trustStorePassword;
    }
}

XML 配置部分如下所示:

<bean id="sslOptions" class="x.y.SslOptionsFactoryBean" lazy-init="false">
    <property name="trustStore" value="file:truststore.jks"/>
</bean>

<cassandra:cluster contact-points="localhost"
               port="9042"
               username="user"
               password="pass"
               ssl-enabled="true"
               ssl-options-ref="sslOptions"
/>

注意:SslOptionsFactoryBean 包含用于完整 SSLContext 初始化的信任管理器和 key 管理器。

堆栈跟踪

堆栈跟踪显示 SSL 是使用 Datastax 驱动程序配置的。它还显示握手失败，失败通知事件失败:

Task com.datastax.driver.core.Connection$10$1@5c21d76e rejected from java.util.concurrent.ThreadPoolExecutor@78eaecc1[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]

该消息表示 Netty EventLoopGroup 在 SSL 握手(失败)完成期间关闭。 Spring 容器应该运行更长时间才能看到握手失败消息。