Skip to content

Commit d056812

Browse files
author
Matthew Nesbit
authored
Some code tidy up ahead of post-GA development. (corda#3500)
Handle SSL Handshake timeouts properly
1 parent a5f4311 commit d056812

File tree

10 files changed

+200
-135
lines changed

10 files changed

+200
-135
lines changed

‎node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt

+31-23
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,13 @@ import net.corda.nodeapi.internal.ArtemisMessagingClient
1111
import net.corda.nodeapi.internal.ArtemisMessagingComponent
1212
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_P2P_USER
1313
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2PMessagingHeaders
14-
import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEER_USER
1514
import net.corda.nodeapi.internal.ArtemisMessagingComponent.RemoteInboxAddress.Companion.translateLocalQueueToInboxAddress
1615
import net.corda.nodeapi.internal.ArtemisSessionProvider
1716
import net.corda.nodeapi.internal.bridging.AMQPBridgeManager.AMQPBridge.Companion.getBridgeName
1817
import net.corda.nodeapi.internal.config.NodeSSLConfiguration
1918
import net.corda.nodeapi.internal.protonwrapper.messages.MessageStatus
2019
import net.corda.nodeapi.internal.protonwrapper.netty.AMQPClient
20+
import net.corda.nodeapi.internal.protonwrapper.netty.AMQPConfiguration
2121
import org.apache.activemq.artemis.api.core.SimpleString
2222
import org.apache.activemq.artemis.api.core.client.ActiveMQClient.DEFAULT_ACK_BATCH_SIZE
2323
import org.apache.activemq.artemis.api.core.client.ClientConsumer
@@ -37,16 +37,24 @@ import kotlin.concurrent.withLock
3737
* The Netty thread pool used by the AMQPBridges is also shared and managed by the AMQPBridgeManager.
3838
*/
3939
@VisibleForTesting
40-
class AMQPBridgeManager(config: NodeSSLConfiguration, private val maxMessageSize: Int, private val artemisMessageClientFactory: () -> ArtemisSessionProvider) : BridgeManager {
40+
class AMQPBridgeManager(config: NodeSSLConfiguration, maxMessageSize: Int, private val artemisMessageClientFactory: () -> ArtemisSessionProvider) : BridgeManager {
4141

4242
private val lock = ReentrantLock()
4343
private val bridgeNameToBridgeMap = mutableMapOf<String, AMQPBridge>()
44+
45+
private class AMQPConfigurationImpl private constructor(override val keyStore: KeyStore,
46+
override val keyStorePrivateKeyPassword: CharArray,
47+
override val trustStore: KeyStore,
48+
override val maxMessageSize: Int) : AMQPConfiguration {
49+
constructor(config: NodeSSLConfiguration, maxMessageSize: Int) : this(config.loadSslKeyStore().internal,
50+
config.keyStorePassword.toCharArray(),
51+
config.loadTrustStore().internal,
52+
maxMessageSize)
53+
}
54+
55+
private val amqpConfig: AMQPConfiguration = AMQPConfigurationImpl(config, maxMessageSize)
4456
private var sharedEventLoopGroup: EventLoopGroup? = null
45-
private val keyStore = config.loadSslKeyStore().internal
46-
private val keyStorePrivateKeyPassword: String = config.keyStorePassword
47-
private val trustStore = config.loadTrustStore().internal
4857
private var artemis: ArtemisSessionProvider? = null
49-
private val crlCheckSoftFail: Boolean = config.crlCheckSoftFail
5058

5159
constructor(config: NodeSSLConfiguration, p2pAddress: NetworkHostAndPort, maxMessageSize: Int) : this(config, maxMessageSize, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
5260

@@ -65,26 +73,26 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, private val maxMessageSize
6573
private class AMQPBridge(private val queueName: String,
6674
private val target: NetworkHostAndPort,
6775
private val legalNames: Set<CordaX500Name>,
68-
keyStore: KeyStore,
69-
keyStorePrivateKeyPassword: String,
70-
trustStore: KeyStore,
71-
crlCheckSoftFail: Boolean,
76+
private val amqpConfig: AMQPConfiguration,
7277
sharedEventGroup: EventLoopGroup,
73-
private val artemis: ArtemisSessionProvider,
74-
private val maxMessageSize: Int) {
78+
private val artemis: ArtemisSessionProvider) {
7579
companion object {
7680
fun getBridgeName(queueName: String, hostAndPort: NetworkHostAndPort): String = "$queueName -> $hostAndPort"
7781
private val log = contextLogger()
7882
}
7983

8084
private fun withMDC(block: () -> Unit) {
81-
MDC.put("queueName", queueName)
82-
MDC.put("target", target.toString())
83-
MDC.put("bridgeName", bridgeName)
84-
MDC.put("legalNames", legalNames.joinToString(separator = ";") { it.toString() })
85-
MDC.put("maxMessageSize", maxMessageSize.toString())
86-
block()
87-
MDC.clear()
85+
val oldMDC = MDC.getCopyOfContextMap()
86+
try {
87+
MDC.put("queueName", queueName)
88+
MDC.put("target", target.toString())
89+
MDC.put("bridgeName", bridgeName)
90+
MDC.put("legalNames", legalNames.joinToString(separator = ";") { it.toString() })
91+
MDC.put("maxMessageSize", amqpConfig.maxMessageSize.toString())
92+
block()
93+
} finally {
94+
MDC.setContextMap(oldMDC)
95+
}
8896
}
8997

9098
private fun logDebugWithMDC(msg: () -> String) {
@@ -97,7 +105,7 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, private val maxMessageSize
97105

98106
private fun logWarnWithMDC(msg: String) = withMDC { log.warn(msg) }
99107

100-
val amqpClient = AMQPClient(listOf(target), legalNames, PEER_USER, PEER_USER, keyStore, keyStorePrivateKeyPassword, trustStore, crlCheckSoftFail, sharedThreadPool = sharedEventGroup, maxMessageSize = maxMessageSize)
108+
val amqpClient = AMQPClient(listOf(target), legalNames, amqpConfig, sharedThreadPool = sharedEventGroup)
101109
val bridgeName: String get() = getBridgeName(queueName, target)
102110
private val lock = ReentrantLock() // lock to serialise session level access
103111
private var session: ClientSession? = null
@@ -149,8 +157,8 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, private val maxMessageSize
149157
}
150158

151159
private fun clientArtemisMessageHandler(artemisMessage: ClientMessage) {
152-
if (artemisMessage.bodySize > maxMessageSize) {
153-
logWarnWithMDC("Message exceeds maxMessageSize network parameter, maxMessageSize: [$maxMessageSize], message size: [${artemisMessage.bodySize}], " +
160+
if (artemisMessage.bodySize > amqpConfig.maxMessageSize) {
161+
logWarnWithMDC("Message exceeds maxMessageSize network parameter, maxMessageSize: [${amqpConfig.maxMessageSize}], message size: [${artemisMessage.bodySize}], " +
154162
"dropping message, uuid: ${artemisMessage.getObjectProperty("_AMQ_DUPL_ID")}")
155163
// Ack the message to prevent same message being sent to us again.
156164
artemisMessage.acknowledge()
@@ -198,7 +206,7 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, private val maxMessageSize
198206
if (bridgeExists(getBridgeName(queueName, target))) {
199207
return
200208
}
201-
val newBridge = AMQPBridge(queueName, target, legalNames, keyStore, keyStorePrivateKeyPassword, trustStore, crlCheckSoftFail, sharedEventLoopGroup!!, artemis!!, maxMessageSize)
209+
val newBridge = AMQPBridge(queueName, target, legalNames, amqpConfig, sharedEventLoopGroup!!, artemis!!)
202210
lock.withLock {
203211
bridgeNameToBridgeMap[newBridge.bridgeName] = newBridge
204212
}

‎node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/engine/ConnectionStateMachine.kt

+9-5
Original file line numberDiff line numberDiff line change
@@ -47,11 +47,15 @@ internal class ConnectionStateMachine(private val serverMode: Boolean,
4747
}
4848

4949
private fun withMDC(block: () -> Unit) {
50-
MDC.put("serverMode", serverMode.toString())
51-
MDC.put("localLegalName", localLegalName)
52-
MDC.put("remoteLegalName", remoteLegalName)
53-
block()
54-
MDC.clear()
50+
val oldMDC = MDC.getCopyOfContextMap()
51+
try {
52+
MDC.put("serverMode", serverMode.toString())
53+
MDC.put("localLegalName", localLegalName)
54+
MDC.put("remoteLegalName", remoteLegalName)
55+
block()
56+
} finally {
57+
MDC.setContextMap(oldMDC)
58+
}
5559
}
5660

5761
private fun logDebugWithMDC(msg: () -> String) {

‎node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/engine/EventProcessor.kt

+9-5
Original file line numberDiff line numberDiff line change
@@ -41,11 +41,15 @@ internal class EventProcessor(channel: Channel,
4141
}
4242

4343
private fun withMDC(block: () -> Unit) {
44-
MDC.put("serverMode", serverMode.toString())
45-
MDC.put("localLegalName", localLegalName)
46-
MDC.put("remoteLegalName", remoteLegalName)
47-
block()
48-
MDC.clear()
44+
val oldMDC = MDC.getCopyOfContextMap()
45+
try {
46+
MDC.put("serverMode", serverMode.toString())
47+
MDC.put("localLegalName", localLegalName)
48+
MDC.put("remoteLegalName", remoteLegalName)
49+
block()
50+
} finally {
51+
MDC.setContextMap(oldMDC)
52+
}
4953
}
5054

5155
private fun logDebugWithMDC(msg: () -> String) {

‎node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/AMQPChannelHandler.kt

+16-8
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import org.slf4j.MDC
2323
import java.net.InetSocketAddress
2424
import java.nio.channels.ClosedChannelException
2525
import java.security.cert.X509Certificate
26+
import javax.net.ssl.SSLException
2627

2728
/**
2829
* An instance of AMQPChannelHandler sits inside the netty pipeline and controls the socket level lifecycle.
@@ -48,13 +49,17 @@ internal class AMQPChannelHandler(private val serverMode: Boolean,
4849
private var badCert: Boolean = false
4950

5051
private fun withMDC(block: () -> Unit) {
51-
MDC.put("serverMode", serverMode.toString())
52-
MDC.put("remoteAddress", remoteAddress.toString())
53-
MDC.put("localCert", localCert?.subjectDN?.toString())
54-
MDC.put("remoteCert", remoteCert?.subjectDN?.toString())
55-
MDC.put("allowedRemoteLegalNames", allowedRemoteLegalNames?.joinToString(separator = ";") { it.toString() })
56-
block()
57-
MDC.clear()
52+
val oldMDC = MDC.getCopyOfContextMap()
53+
try {
54+
MDC.put("serverMode", serverMode.toString())
55+
MDC.put("remoteAddress", remoteAddress.toString())
56+
MDC.put("localCert", localCert?.subjectDN?.toString())
57+
MDC.put("remoteCert", remoteCert?.subjectDN?.toString())
58+
MDC.put("allowedRemoteLegalNames", allowedRemoteLegalNames?.joinToString(separator = ";") { it.toString() })
59+
block()
60+
} finally {
61+
MDC.setContextMap(oldMDC)
62+
}
5863
}
5964

6065
private fun logDebugWithMDC(msg: () -> String) {
@@ -129,9 +134,12 @@ internal class AMQPChannelHandler(private val serverMode: Boolean,
129134
createAMQPEngine(ctx)
130135
onOpen(Pair(ctx.channel() as SocketChannel, ConnectionChange(remoteAddress, remoteCert, true, false)))
131136
} else {
137+
val cause = evt.cause()
132138
// This happens when the peer node is closed during SSL establishment.
133-
if (evt.cause() is ClosedChannelException) {
139+
if (cause is ClosedChannelException) {
134140
logWarnWithMDC("SSL Handshake closed early.")
141+
} else if (cause is SSLException && cause.message == "handshake timed out") { // Sadly the exception thrown by Netty wrapper requires that we check the message.
142+
logWarnWithMDC("SSL Handshake timed out")
135143
} else {
136144
badCert = true
137145
}

‎node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/AMQPClient.kt

+10-17
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ import net.corda.nodeapi.internal.requireMessageSize
1919
import rx.Observable
2020
import rx.subjects.PublishSubject
2121
import java.lang.Long.min
22-
import java.security.KeyStore
2322
import java.util.concurrent.TimeUnit
2423
import java.util.concurrent.locks.ReentrantLock
2524
import javax.net.ssl.KeyManagerFactory
@@ -35,15 +34,8 @@ import kotlin.concurrent.withLock
3534
*/
3635
class AMQPClient(val targets: List<NetworkHostAndPort>,
3736
val allowedRemoteLegalNames: Set<CordaX500Name>,
38-
private val userName: String?,
39-
private val password: String?,
40-
private val keyStore: KeyStore,
41-
private val keyStorePrivateKeyPassword: String,
42-
private val trustStore: KeyStore,
43-
private val crlCheckSoftFail: Boolean,
44-
private val trace: Boolean = false,
45-
private val sharedThreadPool: EventLoopGroup? = null,
46-
private val maxMessageSize: Int) : AutoCloseable {
37+
private val configuration: AMQPConfiguration,
38+
private val sharedThreadPool: EventLoopGroup? = null) : AutoCloseable {
4739
companion object {
4840
init {
4941
InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE)
@@ -121,23 +113,24 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
121113
private class ClientChannelInitializer(val parent: AMQPClient) : ChannelInitializer<SocketChannel>() {
122114
private val keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm())
123115
private val trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm())
116+
private val conf = parent.configuration
124117

125118
init {
126-
keyManagerFactory.init(parent.keyStore, parent.keyStorePrivateKeyPassword.toCharArray())
127-
trustManagerFactory.init(initialiseTrustStoreAndEnableCrlChecking(parent.trustStore, parent.crlCheckSoftFail))
119+
keyManagerFactory.init(conf.keyStore, conf.keyStorePrivateKeyPassword)
120+
trustManagerFactory.init(initialiseTrustStoreAndEnableCrlChecking(conf.trustStore, conf.crlCheckSoftFail))
128121
}
129122

130123
override fun initChannel(ch: SocketChannel) {
131124
val pipeline = ch.pipeline()
132125
val target = parent.currentTarget
133126
val handler = createClientSslHelper(target, keyManagerFactory, trustManagerFactory)
134127
pipeline.addLast("sslHandler", handler)
135-
if (parent.trace) pipeline.addLast("logger", LoggingHandler(LogLevel.INFO))
128+
if (conf.trace) pipeline.addLast("logger", LoggingHandler(LogLevel.INFO))
136129
pipeline.addLast(AMQPChannelHandler(false,
137130
parent.allowedRemoteLegalNames,
138-
parent.userName,
139-
parent.password,
140-
parent.trace,
131+
conf.userName,
132+
conf.password,
133+
conf.trace,
141134
{
142135
parent.retryInterval = MIN_RETRY_INTERVAL // reset to fast reconnect if we connect properly
143136
parent._onConnection.onNext(it.second)
@@ -205,7 +198,7 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
205198
topic: String,
206199
destinationLegalName: String,
207200
properties: Map<String, Any?>): SendableMessage {
208-
requireMessageSize(payload.size, maxMessageSize)
201+
requireMessageSize(payload.size, configuration.maxMessageSize)
209202
return SendableMessageImpl(payload, topic, destinationLegalName, currentTarget, properties)
210203
}
211204

‎node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/AMQPConfiguration.kt

+59
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package net.corda.nodeapi.internal.protonwrapper.netty
2+
3+
import net.corda.nodeapi.internal.ArtemisMessagingComponent
4+
import java.security.KeyStore
5+
6+
interface AMQPConfiguration {
7+
/**
8+
* SASL User name presented during protocol handshake. No SASL login if NULL.
9+
* For legacy interoperability with Artemis authorisation we typically require this to be "PEER_USER"
10+
*/
11+
@JvmDefault
12+
val userName: String?
13+
get() = ArtemisMessagingComponent.PEER_USER
14+
15+
/**
16+
* SASL plain text password presented during protocol handshake. No SASL login if NULL.
17+
* For legacy interoperability with Artemis authorisation we typically require this to be "PEER_USER"
18+
*/
19+
@JvmDefault
20+
val password: String?
21+
get() = ArtemisMessagingComponent.PEER_USER
22+
23+
/**
24+
* The keystore used for TLS connections
25+
*/
26+
val keyStore: KeyStore
27+
28+
/**
29+
* Password used to unlock TLS private keys in the KeyStore.
30+
*/
31+
val keyStorePrivateKeyPassword: CharArray
32+
33+
/**
34+
* The trust root KeyStore to validate the peer certificates against
35+
*/
36+
val trustStore: KeyStore
37+
38+
/**
39+
* Setting crlCheckSoftFail to true allows certificate paths where some leaf certificates do not contain cRLDistributionPoints
40+
* and also allows validation to continue if the CRL distribution server is not contactable.
41+
*/
42+
@JvmDefault
43+
val crlCheckSoftFail: Boolean
44+
get() = true
45+
46+
/**
47+
* Enables full debug tracing of all netty and AMQP level packets. This logs aat very high volume and is only for developers.
48+
*/
49+
@JvmDefault
50+
val trace: Boolean
51+
get() = false
52+
53+
/**
54+
* The maximum allowed size for packets, which will be dropped ahead of send. In future may also be enforced on receive,
55+
* but currently that is deferred to Artemis and the bridge code.
56+
*/
57+
val maxMessageSize: Int
58+
}
59+

0 commit comments

Comments
 (0)