mirror of
https://github.com/corda/corda.git
synced 2025-06-19 15:43:52 +00:00
[ENT-2624] Disable switch for SNI functionality (#1487)
* [ENT-2624] Disable switch for SNI functionality * * Add SNI switch to driver * Make BridgeRestartTest test for both enableSNI = true and false
This commit is contained in:
Patrick Kuo
GitHub
@ -35,9 +35,12 @@ import kotlin.concurrent.withLock
|
||||
* The Netty thread pool used by the AMQPBridges is also shared and managed by the AMQPBridgeManager.
|
||||
*/
|
||||
@VisibleForTesting
|
||||
class AMQPBridgeManager(config: MutualSslConfiguration, socksProxyConfig: SocksProxyConfig? = null, maxMessageSize: Int,
|
||||
private val artemisMessageClientFactory: () -> ArtemisSessionProvider,
|
||||
private val bridgeMetricsService: BridgeMetricsService? = null) : BridgeManager {
|
||||
open class AMQPBridgeManager(config: MutualSslConfiguration,
|
||||
socksProxyConfig: SocksProxyConfig? = null,
|
||||
maxMessageSize: Int,
|
||||
enableSNI: Boolean,
|
||||
private val artemisMessageClientFactory: () -> ArtemisSessionProvider,
|
||||
private val bridgeMetricsService: BridgeMetricsService? = null) : BridgeManager {
|
||||
|
||||
private val lock = ReentrantLock()
|
||||
private val queueNamesToBridgesMap = mutableMapOf<String, MutableList<AMQPBridge>>()
|
||||
@ -47,19 +50,21 @@ class AMQPBridgeManager(config: MutualSslConfiguration, socksProxyConfig: SocksP
|
||||
override val socksProxyConfig: SocksProxyConfig?,
|
||||
override val maxMessageSize: Int,
|
||||
override val useOpenSsl: Boolean,
|
||||
override val enableSNI: Boolean,
|
||||
override val sourceX500Name: String? = null) : AMQPConfiguration {
|
||||
constructor(config: MutualSslConfiguration, socksProxyConfig: SocksProxyConfig?, maxMessageSize: Int) : this(config.keyStore.get(),
|
||||
constructor(config: MutualSslConfiguration, socksProxyConfig: SocksProxyConfig?, maxMessageSize: Int, enableSNI: Boolean) : this(config.keyStore.get(),
|
||||
config.trustStore.get(),
|
||||
socksProxyConfig,
|
||||
maxMessageSize,
|
||||
config.useOpenSsl)
|
||||
config.useOpenSsl,
|
||||
enableSNI)
|
||||
}
|
||||
|
||||
private val amqpConfig: AMQPConfiguration = AMQPConfigurationImpl(config, socksProxyConfig, maxMessageSize)
|
||||
private val amqpConfig: AMQPConfiguration = AMQPConfigurationImpl(config, socksProxyConfig, maxMessageSize, enableSNI)
|
||||
private var sharedEventLoopGroup: EventLoopGroup? = null
|
||||
private var artemis: ArtemisSessionProvider? = null
|
||||
|
||||
constructor(config: MutualSslConfiguration, p2pAddress: NetworkHostAndPort, maxMessageSize: Int, socksProxyConfig: SocksProxyConfig? = null) : this(config, socksProxyConfig, maxMessageSize, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
|
||||
constructor(config: MutualSslConfiguration, p2pAddress: NetworkHostAndPort, maxMessageSize: Int, enableSNI: Boolean, socksProxyConfig: SocksProxyConfig? = null) : this(config, socksProxyConfig, maxMessageSize, enableSNI, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
|
||||
|
||||
companion object {
|
||||
private const val NUM_BRIDGE_THREADS = 0 // Default sized pool
|
||||
@ -154,7 +159,11 @@ class AMQPBridgeManager(config: MutualSslConfiguration, socksProxyConfig: SocksP
|
||||
val session = sessionFactory.createSession(NODE_P2P_USER, NODE_P2P_USER, false, true, true, false, DEFAULT_ACK_BATCH_SIZE)
|
||||
this.session = session
|
||||
// Several producers (in the case of shared bridge) can put messages in the same outbound p2p queue. The consumers are created using the source x500 name as a filter
|
||||
val consumer = session.createConsumer(queueName, "hyphenated_props:sender-subject-name = '${amqpConfig.sourceX500Name}'")
|
||||
val consumer = if (amqpConfig.enableSNI) {
|
||||
session.createConsumer(queueName, "hyphenated_props:sender-subject-name = '${amqpConfig.sourceX500Name}'")
|
||||
} else {
|
||||
session.createConsumer(queueName)
|
||||
}
|
||||
this.consumer = consumer
|
||||
consumer.setMessageHandler(this@AMQPBridge::clientArtemisMessageHandler)
|
||||
session.start()
|
||||
@ -230,7 +239,7 @@ class AMQPBridgeManager(config: MutualSslConfiguration, socksProxyConfig: SocksP
|
||||
return
|
||||
}
|
||||
}
|
||||
val newAMQPConfig = AMQPConfigurationImpl(amqpConfig.keyStore, amqpConfig.trustStore, amqpConfig.socksProxyConfig, amqpConfig.maxMessageSize, amqpConfig.useOpenSsl, sourceX500Name)
|
||||
val newAMQPConfig = with(amqpConfig) { AMQPConfigurationImpl(keyStore, trustStore, socksProxyConfig, maxMessageSize, useOpenSsl, enableSNI, sourceX500Name) }
|
||||
val newBridge = AMQPBridge(sourceX500Name, queueName, targets, legalNames, newAMQPConfig, sharedEventLoopGroup!!, artemis!!, bridgeMetricsService)
|
||||
bridges += newBridge
|
||||
bridgeMetricsService?.bridgeCreated(targets, legalNames)
|
||||
|
||||
@ -1,6 +1,5 @@
|
||||
package net.corda.nodeapi.internal.bridging
|
||||
|
||||
import net.corda.core.identity.CordaX500Name
|
||||
import net.corda.core.serialization.SerializationDefaults
|
||||
import net.corda.core.serialization.deserialize
|
||||
import net.corda.core.serialization.serialize
|
||||
@ -11,7 +10,6 @@ import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_CON
|
||||
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_NOTIFY
|
||||
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2P_PREFIX
|
||||
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEERS_PREFIX
|
||||
import net.corda.nodeapi.internal.ArtemisMessagingComponent.RemoteInboxAddress.Companion.translateLocalQueueToInboxAddress
|
||||
import net.corda.nodeapi.internal.ArtemisSessionProvider
|
||||
import net.corda.nodeapi.internal.config.MutualSslConfiguration
|
||||
import net.corda.nodeapi.internal.protonwrapper.netty.SocksProxyConfig
|
||||
@ -28,13 +26,18 @@ import java.util.*
|
||||
class BridgeControlListener(val config: MutualSslConfiguration,
|
||||
socksProxyConfig: SocksProxyConfig? = null,
|
||||
maxMessageSize: Int,
|
||||
enableSNI: Boolean,
|
||||
private val artemisMessageClientFactory: () -> ArtemisSessionProvider,
|
||||
bridgeMetricsService: BridgeMetricsService? = null) : AutoCloseable {
|
||||
private val bridgeId: String = UUID.randomUUID().toString()
|
||||
private val bridgeControlQueue = "$BRIDGE_CONTROL.$bridgeId"
|
||||
private val bridgeNotifyQueue = "$BRIDGE_NOTIFY.$bridgeId"
|
||||
private val validInboundQueues = mutableSetOf<String>()
|
||||
private val bridgeManager = LoopbackBridgeManagerWrapper(config, socksProxyConfig, maxMessageSize, artemisMessageClientFactory, bridgeMetricsService, this::validateReceiveTopic)
|
||||
private val bridgeManager = if (enableSNI) {
|
||||
LoopbackBridgeManager(config, socksProxyConfig, maxMessageSize, enableSNI, artemisMessageClientFactory, bridgeMetricsService, this::validateReceiveTopic)
|
||||
} else {
|
||||
AMQPBridgeManager(config, socksProxyConfig, maxMessageSize, enableSNI, artemisMessageClientFactory, bridgeMetricsService)
|
||||
}
|
||||
private var artemis: ArtemisSessionProvider? = null
|
||||
private var controlConsumer: ClientConsumer? = null
|
||||
private var notifyConsumer: ClientConsumer? = null
|
||||
@ -42,7 +45,8 @@ class BridgeControlListener(val config: MutualSslConfiguration,
|
||||
constructor(config: MutualSslConfiguration,
|
||||
p2pAddress: NetworkHostAndPort,
|
||||
maxMessageSize: Int,
|
||||
socksProxy: SocksProxyConfig? = null) : this(config, socksProxy, maxMessageSize, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
|
||||
enableSNI: Boolean,
|
||||
socksProxy: SocksProxyConfig? = null) : this(config, socksProxy, maxMessageSize, enableSNI, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
|
||||
|
||||
companion object {
|
||||
private val log = contextLogger()
|
||||
@ -163,7 +167,10 @@ class BridgeControlListener(val config: MutualSslConfiguration,
|
||||
val wasActive = active
|
||||
validInboundQueues.addAll(controlMessage.inboxQueues)
|
||||
log.info("Added inbox: ${controlMessage.inboxQueues}")
|
||||
bridgeManager.inboxesAdded(controlMessage.inboxQueues)
|
||||
if (bridgeManager is LoopbackBridgeManager) {
|
||||
// Notify loopback bridge manager inboxes has changed.
|
||||
bridgeManager.inboxesAdded(controlMessage.inboxQueues)
|
||||
}
|
||||
if (!wasActive && active) {
|
||||
_activeChange.onNext(true)
|
||||
}
|
||||
@ -187,54 +194,4 @@ class BridgeControlListener(val config: MutualSslConfiguration,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class LoopbackBridgeManagerWrapper(config: MutualSslConfiguration,
|
||||
socksProxyConfig: SocksProxyConfig? = null,
|
||||
maxMessageSize: Int,
|
||||
artemisMessageClientFactory: () -> ArtemisSessionProvider,
|
||||
bridgeMetricsService: BridgeMetricsService? = null,
|
||||
private val isLocalInbox: (String) -> Boolean) : BridgeManager {
|
||||
|
||||
private val bridgeManager = AMQPBridgeManager(config, socksProxyConfig, maxMessageSize, artemisMessageClientFactory, bridgeMetricsService)
|
||||
private val loopbackBridgeManager = LoopbackBridgeManager(artemisMessageClientFactory, bridgeMetricsService)
|
||||
|
||||
override fun deployBridge(sourceX500Name: String, queueName: String, targets: List<NetworkHostAndPort>, legalNames: Set<CordaX500Name>) {
|
||||
val inboxAddress = translateLocalQueueToInboxAddress(queueName)
|
||||
if (isLocalInbox(inboxAddress)) {
|
||||
log.info("Deploying loopback bridge for $queueName, source $sourceX500Name")
|
||||
loopbackBridgeManager.deployBridge(sourceX500Name, queueName, targets, legalNames)
|
||||
} else {
|
||||
log.info("Deploying AMQP bridge for $queueName, source $sourceX500Name")
|
||||
bridgeManager.deployBridge(sourceX500Name, queueName, targets, legalNames)
|
||||
}
|
||||
}
|
||||
|
||||
override fun destroyBridge(queueName: String, targets: List<NetworkHostAndPort>) {
|
||||
bridgeManager.destroyBridge(queueName, targets)
|
||||
loopbackBridgeManager.destroyBridge(queueName, targets)
|
||||
}
|
||||
|
||||
override fun start() {
|
||||
bridgeManager.start()
|
||||
loopbackBridgeManager.start()
|
||||
}
|
||||
|
||||
override fun stop() {
|
||||
bridgeManager.stop()
|
||||
loopbackBridgeManager.stop()
|
||||
}
|
||||
|
||||
override fun close() = stop()
|
||||
|
||||
/**
|
||||
* Remove any AMQP bridge for the local inbox and create a loopback bridge for that queue.
|
||||
*/
|
||||
fun inboxesAdded(inboxes: List<String>) {
|
||||
for (inbox in inboxes) {
|
||||
bridgeManager.destroyAllBridge(inbox).forEach { source, bridgeEntry ->
|
||||
loopbackBridgeManager.deployBridge(source, bridgeEntry.queueName, bridgeEntry.targets, bridgeEntry.legalNames.toSet())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -9,7 +9,9 @@ import net.corda.nodeapi.internal.ArtemisMessagingComponent
|
||||
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_P2P_USER
|
||||
import net.corda.nodeapi.internal.ArtemisMessagingComponent.RemoteInboxAddress.Companion.translateLocalQueueToInboxAddress
|
||||
import net.corda.nodeapi.internal.ArtemisSessionProvider
|
||||
import net.corda.nodeapi.internal.config.MutualSslConfiguration
|
||||
import net.corda.nodeapi.internal.protonwrapper.messages.impl.SendableMessageImpl
|
||||
import net.corda.nodeapi.internal.protonwrapper.netty.SocksProxyConfig
|
||||
import org.apache.activemq.artemis.api.core.SimpleString
|
||||
import org.apache.activemq.artemis.api.core.client.ActiveMQClient.DEFAULT_ACK_BATCH_SIZE
|
||||
import org.apache.activemq.artemis.api.core.client.ClientConsumer
|
||||
@ -23,8 +25,17 @@ import org.slf4j.MDC
|
||||
* inboxes.
|
||||
*/
|
||||
@VisibleForTesting
|
||||
class LoopbackBridgeManager(private val artemisMessageClientFactory: () -> ArtemisSessionProvider,
|
||||
private val bridgeMetricsService: BridgeMetricsService? = null) : BridgeManager {
|
||||
class LoopbackBridgeManager(config: MutualSslConfiguration,
|
||||
socksProxyConfig: SocksProxyConfig? = null,
|
||||
maxMessageSize: Int,
|
||||
enableSNI: Boolean,
|
||||
private val artemisMessageClientFactory: () -> ArtemisSessionProvider,
|
||||
private val bridgeMetricsService: BridgeMetricsService? = null,
|
||||
private val isLocalInbox: (String) -> Boolean) : AMQPBridgeManager(config, socksProxyConfig, maxMessageSize, enableSNI, artemisMessageClientFactory, bridgeMetricsService) {
|
||||
|
||||
companion object {
|
||||
private val log = contextLogger()
|
||||
}
|
||||
|
||||
private val queueNamesToBridgesMap = ConcurrentBox(mutableMapOf<String, MutableList<LoopbackBridge>>())
|
||||
private var artemis: ArtemisSessionProvider? = null
|
||||
@ -118,21 +129,29 @@ class LoopbackBridgeManager(private val artemisMessageClientFactory: () -> Artem
|
||||
}
|
||||
|
||||
override fun deployBridge(sourceX500Name: String, queueName: String, targets: List<NetworkHostAndPort>, legalNames: Set<CordaX500Name>) {
|
||||
queueNamesToBridgesMap.exclusive {
|
||||
val bridges = getOrPut(queueName) { mutableListOf() }
|
||||
for (target in targets) {
|
||||
if (bridges.any { it.targets.contains(target) && it.sourceX500Name == sourceX500Name }) {
|
||||
return
|
||||
val inboxAddress = translateLocalQueueToInboxAddress(queueName)
|
||||
if (isLocalInbox(inboxAddress)) {
|
||||
log.info("Deploying loopback bridge for $queueName, source $sourceX500Name")
|
||||
queueNamesToBridgesMap.exclusive {
|
||||
val bridges = getOrPut(queueName) { mutableListOf() }
|
||||
for (target in targets) {
|
||||
if (bridges.any { it.targets.contains(target) && it.sourceX500Name == sourceX500Name }) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
val newBridge = LoopbackBridge(sourceX500Name, queueName, targets, legalNames, artemis!!, bridgeMetricsService)
|
||||
bridges += newBridge
|
||||
bridgeMetricsService?.bridgeCreated(targets, legalNames)
|
||||
newBridge
|
||||
}.start()
|
||||
val newBridge = LoopbackBridge(sourceX500Name, queueName, targets, legalNames, artemis!!, bridgeMetricsService)
|
||||
bridges += newBridge
|
||||
bridgeMetricsService?.bridgeCreated(targets, legalNames)
|
||||
newBridge
|
||||
}.start()
|
||||
} else {
|
||||
log.info("Deploying AMQP bridge for $queueName, source $sourceX500Name")
|
||||
super.deployBridge(sourceX500Name, queueName, targets, legalNames)
|
||||
}
|
||||
}
|
||||
|
||||
override fun destroyBridge(queueName: String, targets: List<NetworkHostAndPort>) {
|
||||
super.destroyBridge(queueName, targets)
|
||||
queueNamesToBridgesMap.exclusive {
|
||||
val bridges = this[queueName] ?: mutableListOf()
|
||||
for (target in targets) {
|
||||
@ -149,7 +168,19 @@ class LoopbackBridgeManager(private val artemisMessageClientFactory: () -> Artem
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove any AMQP bridge for the local inbox and create a loopback bridge for that queue.
|
||||
*/
|
||||
fun inboxesAdded(inboxes: List<String>) {
|
||||
for (inbox in inboxes) {
|
||||
super.destroyAllBridge(inbox).forEach { source, bridgeEntry ->
|
||||
deployBridge(source, bridgeEntry.queueName, bridgeEntry.targets, bridgeEntry.legalNames.toSet())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun start() {
|
||||
super.start()
|
||||
val artemis = artemisMessageClientFactory()
|
||||
this.artemis = artemis
|
||||
artemis.start()
|
||||
@ -158,6 +189,7 @@ class LoopbackBridgeManager(private val artemisMessageClientFactory: () -> Artem
|
||||
override fun stop() = close()
|
||||
|
||||
override fun close() {
|
||||
super.close()
|
||||
queueNamesToBridgesMap.exclusive {
|
||||
for (bridge in values.flatten()) {
|
||||
bridge.stop()
|
||||
|
||||
@ -41,8 +41,8 @@ internal class AMQPChannelHandler(private val serverMode: Boolean,
|
||||
private val userName: String?,
|
||||
private val password: String?,
|
||||
private val trace: Boolean,
|
||||
private val onOpen: (Pair<SocketChannel, ConnectionChange>) -> Unit,
|
||||
private val onClose: (Pair<SocketChannel, ConnectionChange>) -> Unit,
|
||||
private val onOpen: (SocketChannel, ConnectionChange) -> Unit,
|
||||
private val onClose: (SocketChannel, ConnectionChange) -> Unit,
|
||||
private val onReceive: (ReceivedMessage) -> Unit) : ChannelDuplexHandler() {
|
||||
companion object {
|
||||
private val log = contextLogger()
|
||||
@ -114,7 +114,7 @@ internal class AMQPChannelHandler(private val serverMode: Boolean,
|
||||
val ch = ctx.channel()
|
||||
logInfoWithMDC("Closed client connection ${ch.id()} from $remoteAddress to ${ch.localAddress()}")
|
||||
if (!suppressClose) {
|
||||
onClose(Pair(ch as SocketChannel, ConnectionChange(remoteAddress, remoteCert, false, badCert)))
|
||||
onClose(ch as SocketChannel, ConnectionChange(remoteAddress, remoteCert, false, badCert))
|
||||
}
|
||||
eventProcessor?.close()
|
||||
ctx.fireChannelInactive()
|
||||
@ -263,7 +263,7 @@ internal class AMQPChannelHandler(private val serverMode: Boolean,
|
||||
|
||||
logInfoWithMDC("Handshake completed with subject: $remoteX500Name, requested server name: ${sslHandler.getRequestedServerName()}.")
|
||||
createAMQPEngine(ctx)
|
||||
onOpen(Pair(ctx.channel() as SocketChannel, ConnectionChange(remoteAddress, remoteCert, true, false)))
|
||||
onOpen(ctx.channel() as SocketChannel, ConnectionChange(remoteAddress, remoteCert, true, false))
|
||||
}
|
||||
|
||||
private fun handleFailedHandshake(ctx: ChannelHandlerContext, evt: SslHandshakeCompletionEvent) {
|
||||
|
||||
@ -14,19 +14,14 @@ import io.netty.util.internal.logging.Slf4JLoggerFactory
|
||||
import net.corda.core.identity.CordaX500Name
|
||||
import net.corda.core.utilities.NetworkHostAndPort
|
||||
import net.corda.core.utilities.contextLogger
|
||||
import net.corda.nodeapi.internal.config.CertificateStore
|
||||
import net.corda.nodeapi.internal.crypto.x509
|
||||
import net.corda.nodeapi.internal.protonwrapper.messages.ReceivedMessage
|
||||
import net.corda.nodeapi.internal.protonwrapper.messages.SendableMessage
|
||||
import net.corda.nodeapi.internal.protonwrapper.messages.impl.SendableMessageImpl
|
||||
import net.corda.nodeapi.internal.requireMessageSize
|
||||
import rx.Observable
|
||||
import rx.subjects.PublishSubject
|
||||
import sun.security.x509.X500Name
|
||||
import java.lang.Long.min
|
||||
import java.net.InetSocketAddress
|
||||
import java.security.KeyStore
|
||||
import java.security.cert.X509Certificate
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.locks.ReentrantLock
|
||||
import javax.net.ssl.KeyManagerFactory
|
||||
@ -99,23 +94,21 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
|
||||
retryInterval = min(MAX_RETRY_INTERVAL, retryInterval * BACKOFF_MULTIPLIER)
|
||||
}
|
||||
|
||||
private val connectListener = object : ChannelFutureListener {
|
||||
override fun operationComplete(future: ChannelFuture) {
|
||||
if (!future.isSuccess) {
|
||||
log.info("Failed to connect to $currentTarget")
|
||||
private val connectListener = ChannelFutureListener { future ->
|
||||
if (!future.isSuccess) {
|
||||
log.info("Failed to connect to $currentTarget")
|
||||
|
||||
if (!stopping) {
|
||||
workerGroup?.schedule({
|
||||
nextTarget()
|
||||
restart()
|
||||
}, retryInterval, TimeUnit.MILLISECONDS)
|
||||
}
|
||||
} else {
|
||||
log.info("Connected to $currentTarget")
|
||||
// Connection established successfully
|
||||
clientChannel = future.channel()
|
||||
clientChannel?.closeFuture()?.addListener(closeListener)
|
||||
if (!stopping) {
|
||||
workerGroup?.schedule({
|
||||
nextTarget()
|
||||
restart()
|
||||
}, retryInterval, TimeUnit.MILLISECONDS)
|
||||
}
|
||||
} else {
|
||||
log.info("Connected to $currentTarget")
|
||||
// Connection established successfully
|
||||
clientChannel = future.channel()
|
||||
clientChannel?.closeFuture()?.addListener(closeListener)
|
||||
}
|
||||
}
|
||||
|
||||
@ -164,7 +157,7 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
|
||||
|
||||
val wrappedKeyManagerFactory = CertHoldingKeyManagerFactoryWrapper(keyManagerFactory, parent.configuration)
|
||||
val target = parent.currentTarget
|
||||
val handler = if (parent.configuration.useOpenSsl){
|
||||
val handler = if (parent.configuration.useOpenSsl) {
|
||||
createClientOpenSslHandler(target, parent.allowedRemoteLegalNames, wrappedKeyManagerFactory, trustManagerFactory, ch.alloc())
|
||||
} else {
|
||||
createClientSslHelper(target, parent.allowedRemoteLegalNames, wrappedKeyManagerFactory, trustManagerFactory)
|
||||
@ -178,13 +171,13 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
|
||||
conf.userName,
|
||||
conf.password,
|
||||
conf.trace,
|
||||
{
|
||||
{ _, change ->
|
||||
parent.retryInterval = MIN_RETRY_INTERVAL // reset to fast reconnect if we connect properly
|
||||
parent._onConnection.onNext(it.second)
|
||||
parent._onConnection.onNext(change)
|
||||
},
|
||||
{
|
||||
parent._onConnection.onNext(it.second)
|
||||
if (it.second.badCert) {
|
||||
{ _, change ->
|
||||
parent._onConnection.onNext(change)
|
||||
if (change.badCert) {
|
||||
log.error("Blocking future connection attempts to $target due to bad certificate on endpoint")
|
||||
parent.badCertTargets += target
|
||||
}
|
||||
|
||||
@ -71,5 +71,9 @@ interface AMQPConfiguration {
|
||||
*/
|
||||
val healthCheckPhrase: String?
|
||||
get() = null
|
||||
|
||||
@JvmDefault
|
||||
val enableSNI: Boolean
|
||||
get() = true
|
||||
}
|
||||
|
||||
|
||||
@ -1,10 +1,7 @@
|
||||
package net.corda.nodeapi.internal.protonwrapper.netty
|
||||
|
||||
import io.netty.bootstrap.ServerBootstrap
|
||||
import io.netty.channel.Channel
|
||||
import io.netty.channel.ChannelInitializer
|
||||
import io.netty.channel.ChannelOption
|
||||
import io.netty.channel.EventLoopGroup
|
||||
import io.netty.channel.*
|
||||
import io.netty.channel.nio.NioEventLoopGroup
|
||||
import io.netty.channel.socket.SocketChannel
|
||||
import io.netty.channel.socket.nio.NioServerSocketChannel
|
||||
@ -67,52 +64,45 @@ class AMQPServer(val hostName: String,
|
||||
override fun initChannel(ch: SocketChannel) {
|
||||
val amqpConfiguration = parent.configuration
|
||||
val pipeline = ch.pipeline()
|
||||
|
||||
amqpConfiguration.healthCheckPhrase?.let { pipeline.addLast(ModeSelectingChannel.NAME, ModeSelectingChannel(it)) }
|
||||
|
||||
val keyStore = amqpConfiguration.keyStore
|
||||
|
||||
// Used for SNI matching with javaSSL.
|
||||
val wrappedKeyManagerFactory = CertHoldingKeyManagerFactoryWrapper(keyManagerFactory, amqpConfiguration)
|
||||
// Used to create a mapping for SNI matching with openSSL.
|
||||
val keyManagerFactoriesMap = splitKeystore(amqpConfiguration)
|
||||
val handler = if (amqpConfiguration.useOpenSsl){
|
||||
// SNI matching needed only when multiple nodes exist behind the server.
|
||||
if (keyStore.aliases().size > 1) {
|
||||
createServerSNIOpenSslHandler(keyManagerFactoriesMap, trustManagerFactory)
|
||||
} else {
|
||||
createServerOpenSslHandler(wrappedKeyManagerFactory, trustManagerFactory, ch.alloc())
|
||||
}
|
||||
} else {
|
||||
// For javaSSL, SNI matching is handled at key manager level.
|
||||
createServerSslHelper(keyStore, wrappedKeyManagerFactory, trustManagerFactory)
|
||||
}
|
||||
|
||||
pipeline.addLast("sslHandler", handler)
|
||||
|
||||
val (sslHandler, keyManagerFactoriesMap) = createSSLHandler(amqpConfiguration, ch)
|
||||
pipeline.addLast("sslHandler", sslHandler)
|
||||
if (conf.trace) pipeline.addLast("logger", LoggingHandler(LogLevel.INFO))
|
||||
pipeline.addLast(AMQPChannelHandler(true,
|
||||
null,
|
||||
// Passing a mapping of legal names to key managers to be able to pick the correct one after
|
||||
// SNI completion event is fired up.
|
||||
if (keyStore.aliases().size > 1 && amqpConfiguration.useOpenSsl)
|
||||
keyManagerFactoriesMap
|
||||
else
|
||||
// Single entry, key can be anything.
|
||||
mapOf(DEFAULT to wrappedKeyManagerFactory),
|
||||
keyManagerFactoriesMap,
|
||||
conf.userName,
|
||||
conf.password,
|
||||
conf.trace,
|
||||
{
|
||||
parent.clientChannels[it.first.remoteAddress()] = it.first
|
||||
parent._onConnection.onNext(it.second)
|
||||
{ channel, change ->
|
||||
parent.clientChannels[channel.remoteAddress()] = channel
|
||||
parent._onConnection.onNext(change)
|
||||
},
|
||||
{
|
||||
parent.clientChannels.remove(it.first.remoteAddress())
|
||||
parent._onConnection.onNext(it.second)
|
||||
{ channel, change ->
|
||||
parent.clientChannels.remove(channel.remoteAddress())
|
||||
parent._onConnection.onNext(change)
|
||||
},
|
||||
{ rcv -> parent._onReceive.onNext(rcv) }))
|
||||
}
|
||||
|
||||
private fun createSSLHandler(amqpConfig: AMQPConfiguration, ch: SocketChannel): Pair<ChannelHandler, Map<String, CertHoldingKeyManagerFactoryWrapper>> {
|
||||
return if (amqpConfig.useOpenSsl && amqpConfig.enableSNI && amqpConfig.keyStore.aliases().size > 1) {
|
||||
val keyManagerFactoriesMap = splitKeystore(amqpConfig)
|
||||
// SNI matching needed only when multiple nodes exist behind the server.
|
||||
Pair(createServerSNIOpenSslHandler(keyManagerFactoriesMap, trustManagerFactory), keyManagerFactoriesMap)
|
||||
} else {
|
||||
val keyManagerFactory = CertHoldingKeyManagerFactoryWrapper(keyManagerFactory, amqpConfig)
|
||||
val handler = if (amqpConfig.useOpenSsl) {
|
||||
createServerOpenSslHandler(keyManagerFactory, trustManagerFactory, ch.alloc())
|
||||
} else {
|
||||
// For javaSSL, SNI matching is handled at key manager level.
|
||||
createServerSslHandler(amqpConfig.keyStore, keyManagerFactory, trustManagerFactory)
|
||||
}
|
||||
Pair(handler, mapOf(DEFAULT to keyManagerFactory))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun start() {
|
||||
@ -189,10 +179,7 @@ class AMQPServer(val hostName: String,
|
||||
}
|
||||
|
||||
fun dropConnection(connectionRemoteHost: InetSocketAddress) {
|
||||
val channel = clientChannels[connectionRemoteHost]
|
||||
if (channel != null) {
|
||||
channel.close()
|
||||
}
|
||||
clientChannels[connectionRemoteHost]?.close()
|
||||
}
|
||||
|
||||
fun complete(delivery: Delivery, target: InetSocketAddress) {
|
||||
|
||||
@ -26,15 +26,11 @@ class CertHoldingKeyManagerFactorySpiWrapper(private val factorySpi: KeyManagerF
|
||||
return if (factorySpi is CertHoldingKeyManagerFactorySpiWrapper) keyManagers else keyManagers.map {
|
||||
val aliasProvidingKeyManager = getDefaultKeyManager(it)
|
||||
// Use the SNIKeyManager if keystore has several entries and only for clients and non-openSSL servers.
|
||||
if (amqpConfig.keyStore.aliases().size > 1) {
|
||||
// Clients
|
||||
if (amqpConfig.sourceX500Name != null) {
|
||||
SNIKeyManager(aliasProvidingKeyManager as X509ExtendedKeyManager, amqpConfig)
|
||||
} else if (!amqpConfig.useOpenSsl) { // JDK SSL servers
|
||||
SNIKeyManager(aliasProvidingKeyManager as X509ExtendedKeyManager, amqpConfig)
|
||||
} else {
|
||||
aliasProvidingKeyManager
|
||||
}
|
||||
// Condition of using SNIKeyManager: if its client, or JDKSsl server.
|
||||
val isClient = amqpConfig.sourceX500Name != null
|
||||
val enableSNI = amqpConfig.enableSNI && amqpConfig.keyStore.aliases().size > 1
|
||||
if (enableSNI && (isClient || !amqpConfig.useOpenSsl)) {
|
||||
SNIKeyManager(aliasProvidingKeyManager as X509ExtendedKeyManager, amqpConfig)
|
||||
} else {
|
||||
aliasProvidingKeyManager
|
||||
}
|
||||
@ -78,5 +74,4 @@ class CertHoldingKeyManagerFactoryWrapper(factory: KeyManagerFactory, amqpConfig
|
||||
keyManager.getCertificateChain(alias)
|
||||
} else null
|
||||
}
|
||||
|
||||
}
|
||||
@ -149,9 +149,9 @@ internal fun createClientOpenSslHandler(target: NetworkHostAndPort,
|
||||
return SslHandler(sslEngine)
|
||||
}
|
||||
|
||||
internal fun createServerSslHelper(keyStore: CertificateStore,
|
||||
keyManagerFactory: KeyManagerFactory,
|
||||
trustManagerFactory: TrustManagerFactory): SslHandler {
|
||||
internal fun createServerSslHandler(keyStore: CertificateStore,
|
||||
keyManagerFactory: KeyManagerFactory,
|
||||
trustManagerFactory: TrustManagerFactory): SslHandler {
|
||||
val sslContext = SSLContext.getInstance("TLS")
|
||||
val keyManagers = keyManagerFactory.keyManagers
|
||||
val trustManagers = trustManagerFactory.trustManagers.filterIsInstance(X509ExtendedTrustManager::class.java).map { LoggingTrustManagerWrapper(it) }.toTypedArray()
|
||||
@ -189,12 +189,10 @@ internal fun initialiseTrustStoreAndEnableCrlChecking(trustStore: CertificateSto
|
||||
internal fun createServerOpenSslHandler(keyManagerFactory: KeyManagerFactory,
|
||||
trustManagerFactory: TrustManagerFactory,
|
||||
alloc: ByteBufAllocator): SslHandler {
|
||||
val sslContext = SslContextBuilder.forServer(keyManagerFactory).sslProvider(SslProvider.OPENSSL).trustManager(LoggingTrustManagerFactoryWrapper(trustManagerFactory)).build()
|
||||
|
||||
val sslContext = getServerSslContextBuilder(keyManagerFactory, trustManagerFactory).build()
|
||||
val sslEngine = sslContext.newEngine(alloc)
|
||||
sslEngine.useClientMode = false
|
||||
sslEngine.needClientAuth = true
|
||||
sslEngine.enabledProtocols = ArtemisTcpTransport.TLS_VERSIONS.toTypedArray()
|
||||
sslEngine.enabledCipherSuites = ArtemisTcpTransport.CIPHER_SUITES.toTypedArray()
|
||||
return SslHandler(sslEngine)
|
||||
}
|
||||
|
||||
@ -205,20 +203,21 @@ internal fun createServerSNIOpenSslHandler(keyManagerFactoriesMap: Map<String, K
|
||||
trustManagerFactory: TrustManagerFactory): SniHandler {
|
||||
|
||||
// Default value can be any in the map.
|
||||
val sslCtxBuilder = SslContextBuilder.forServer(keyManagerFactoriesMap.values.first())
|
||||
val sslCtxBuilder = getServerSslContextBuilder(keyManagerFactoriesMap.values.first(), trustManagerFactory)
|
||||
val mapping = DomainNameMappingBuilder(sslCtxBuilder.build())
|
||||
keyManagerFactoriesMap.forEach {
|
||||
mapping.add(it.key, sslCtxBuilder.keyManager(it.value).build())
|
||||
}
|
||||
return SniHandler(mapping.build())
|
||||
}
|
||||
|
||||
private fun getServerSslContextBuilder(keyManagerFactory: KeyManagerFactory, trustManagerFactory: TrustManagerFactory): SslContextBuilder {
|
||||
return SslContextBuilder.forServer(keyManagerFactory)
|
||||
.sslProvider(SslProvider.OPENSSL)
|
||||
.trustManager(LoggingTrustManagerFactoryWrapper(trustManagerFactory))
|
||||
.clientAuth(ClientAuth.REQUIRE)
|
||||
.ciphers(ArtemisTcpTransport.CIPHER_SUITES)
|
||||
.protocols(*ArtemisTcpTransport.TLS_VERSIONS.toTypedArray())
|
||||
|
||||
val mapping = DomainNameMappingBuilder(sslCtxBuilder.build())
|
||||
|
||||
keyManagerFactoriesMap.forEach {
|
||||
mapping.add(it.key, sslCtxBuilder.keyManager(it.value).build())
|
||||
}
|
||||
|
||||
return SniHandler(mapping.build())
|
||||
}
|
||||
|
||||
internal fun splitKeystore(config: AMQPConfiguration): Map<String, CertHoldingKeyManagerFactoryWrapper> {
|
||||
|
||||
@ -94,5 +94,5 @@ class TestKeyManagerFactoryWrapper {
|
||||
assertNull(otherWrappedKeyManagerFactory.getCurrentCertChain())
|
||||
}
|
||||
|
||||
private class AMQPConfigurationImpl(override val keyStore: CertificateStore, override val trustStore: CertificateStore, override val maxMessageSize: Int) : AMQPConfiguration
|
||||
private class AMQPConfigurationImpl(override val keyStore: CertificateStore, override val trustStore: CertificateStore, override val maxMessageSize: Int, override val enableSNI: Boolean = true) : AMQPConfiguration
|
||||
}
|
||||
Reference in New Issue
Block a user