diff --git a/node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingComponent.kt b/node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingComponent.kt index a5634e49a3..abf4727f6e 100644 --- a/node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingComponent.kt +++ b/node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingComponent.kt @@ -17,6 +17,8 @@ import net.corda.core.messaging.MessageRecipients import net.corda.core.messaging.SingleMessageRecipient import net.corda.core.serialization.CordaSerializable import net.corda.core.utilities.NetworkHostAndPort +import org.apache.activemq.artemis.api.core.Message +import org.apache.activemq.artemis.api.core.SimpleString import java.security.PublicKey /** @@ -38,6 +40,49 @@ class ArtemisMessagingComponent { const val BRIDGE_CONTROL = "${INTERNAL_PREFIX}bridge.control" const val BRIDGE_NOTIFY = "${INTERNAL_PREFIX}bridge.notify" const val NOTIFICATIONS_ADDRESS = "${INTERNAL_PREFIX}activemq.notifications" + + /** + * In the operation mode where we have an out of process bridge we cannot correctly populate the Artemis validated user header + * as the TLS does not terminate directly onto Artemis. We therefore use this internal only header to forward + * the equivalent information from the Float. + */ + val bridgedCertificateSubject = SimpleString("sender-subject-name") + + object P2PMessagingHeaders { + // This is a "property" attached to an Artemis MQ message object, which contains our own notion of "topic". + // We should probably try to unify our notion of "topic" (really, just a string that identifies an endpoint + // that will handle messages, like a URL) with the terminology used by underlying MQ libraries, to avoid + // confusion. + val topicProperty = SimpleString("platform-topic") + val cordaVendorProperty = SimpleString("corda-vendor") + val releaseVersionProperty = SimpleString("release-version") + val platformVersionProperty = SimpleString("platform-version") + val senderUUID = SimpleString("sender-uuid") + val senderSeqNo = SimpleString("send-seq-no") + /** + * In the operation mode where we have an out of process bridge we cannot correctly populate the Artemis validated user header + * as the TLS does not terminate directly onto Artemis. We therefore use this internal only header to forward + * the equivalent information from the Float. + */ + val bridgedCertificateSubject = SimpleString("sender-subject-name") + + + object Type { + const val KEY = "corda_p2p_message_type" + const val SESSION_INIT_VALUE = "session_init" + } + + val whitelistedHeaders: Set = setOf(topicProperty.toString(), + cordaVendorProperty.toString(), + releaseVersionProperty.toString(), + platformVersionProperty.toString(), + senderUUID.toString(), + senderSeqNo.toString(), + bridgedCertificateSubject.toString(), + Type.KEY, + Message.HDR_DUPLICATE_DETECTION_ID.toString(), + Message.HDR_VALIDATED_USER.toString()) + } } interface ArtemisAddress : MessageRecipients { diff --git a/node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt b/node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt index 69e577a1b1..5dc5840ceb 100644 --- a/node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt +++ b/node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt @@ -20,6 +20,7 @@ import net.corda.core.utilities.debug import net.corda.nodeapi.internal.ArtemisMessagingClient import net.corda.nodeapi.internal.ArtemisMessagingComponent import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_USER +import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2PMessagingHeaders import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEER_USER import net.corda.nodeapi.internal.ArtemisMessagingComponent.RemoteInboxAddress.Companion.translateLocalQueueToInboxAddress import net.corda.nodeapi.internal.ArtemisSessionProvider @@ -138,12 +139,14 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, val artemisMessageClientFa private fun clientArtemisMessageHandler(artemisMessage: ClientMessage) { val data = ByteArray(artemisMessage.bodySize).apply { artemisMessage.bodyBuffer.readBytes(this) } val properties = HashMap() - for (key in artemisMessage.propertyNames) { - var value = artemisMessage.getObjectProperty(key) - if (value is SimpleString) { - value = value.toString() + for (key in P2PMessagingHeaders.whitelistedHeaders) { + if (artemisMessage.containsProperty(key)) { + var value = artemisMessage.getObjectProperty(key) + if (value is SimpleString) { + value = value.toString() + } + properties[key] = value } - properties[key.toString()] = value } log.debug { "Bridged Send to ${legalNames.first()} uuid: ${artemisMessage.getObjectProperty("_AMQ_DUPL_ID")}" } val peerInbox = translateLocalQueueToInboxAddress(queueName) diff --git a/node/src/integration-test/kotlin/net/corda/node/amqp/AMQPBridgeTest.kt b/node/src/integration-test/kotlin/net/corda/node/amqp/AMQPBridgeTest.kt index e9d23fdffb..7953574dcb 100644 --- a/node/src/integration-test/kotlin/net/corda/node/amqp/AMQPBridgeTest.kt +++ b/node/src/integration-test/kotlin/net/corda/node/amqp/AMQPBridgeTest.kt @@ -23,6 +23,7 @@ import net.corda.node.services.config.configureWithDevSSLCertificate import net.corda.node.services.messaging.ArtemisMessagingServer import net.corda.nodeapi.internal.ArtemisMessagingClient import net.corda.nodeapi.internal.ArtemisMessagingComponent +import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2PMessagingHeaders import net.corda.nodeapi.internal.bridging.AMQPBridgeManager import net.corda.nodeapi.internal.bridging.BridgeManager import net.corda.nodeapi.internal.protonwrapper.netty.AMQPServer @@ -73,7 +74,7 @@ class AMQPBridgeTest { val artemis = artemisClient.started!! for (i in 0 until 3) { val artemisMessage = artemis.session.createMessage(true).apply { - putIntProperty("CountProp", i) + putIntProperty(P2PMessagingHeaders.senderUUID, i) writeBodyBufferBytes("Test$i".toByteArray()) // Use the magic deduplication property built into Artemis as our message identity too putStringProperty(HDR_DUPLICATE_DETECTION_ID, SimpleString(UUID.randomUUID().toString())) @@ -97,7 +98,7 @@ class AMQPBridgeTest { } val received1 = receive.next() - val messageID1 = received1.applicationProperties["CountProp"] as Int + val messageID1 = received1.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int assertArrayEquals("Test$messageID1".toByteArray(), received1.payload) assertEquals(0, messageID1) dedupeSet += received1.applicationProperties[HDR_DUPLICATE_DETECTION_ID.toString()] as String @@ -106,7 +107,7 @@ class AMQPBridgeTest { atNodeSequence += messageID1 val received2 = receive.next() - val messageID2 = received2.applicationProperties["CountProp"] as Int + val messageID2 = received2.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int assertArrayEquals("Test$messageID2".toByteArray(), received2.payload) assertEquals(1, messageID2, formatMessage("1", messageID2, receivedSequence)) received2.complete(false) // Reject message and don't add to dedupe @@ -115,7 +116,7 @@ class AMQPBridgeTest { // drop things until we get back to the replay while (true) { val received3 = receive.next() - val messageID3 = received3.applicationProperties["CountProp"] as Int + val messageID3 = received3.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int assertArrayEquals("Test$messageID3".toByteArray(), received3.payload) receivedSequence += messageID3 if (messageID3 != 1) { // keep rejecting any batched items following rejection @@ -134,7 +135,7 @@ class AMQPBridgeTest { // start receiving again, but discarding duplicates while (true) { val received4 = receive.next() - val messageID4 = received4.applicationProperties["CountProp"] as Int + val messageID4 = received4.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int assertArrayEquals("Test$messageID4".toByteArray(), received4.payload) receivedSequence += messageID4 val messageId = received4.applicationProperties[HDR_DUPLICATE_DETECTION_ID.toString()] as String @@ -150,7 +151,7 @@ class AMQPBridgeTest { // Send a fresh item and check receive val artemisMessage = artemis.session.createMessage(true).apply { - putIntProperty("CountProp", 3) + putIntProperty(P2PMessagingHeaders.senderUUID, 3) writeBodyBufferBytes("Test3".toByteArray()) // Use the magic deduplication property built into Artemis as our message identity too putStringProperty(HDR_DUPLICATE_DETECTION_ID, SimpleString(UUID.randomUUID().toString())) @@ -161,7 +162,7 @@ class AMQPBridgeTest { // start receiving again, discarding duplicates while (true) { val received5 = receive.next() - val messageID5 = received5.applicationProperties["CountProp"] as Int + val messageID5 = received5.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int assertArrayEquals("Test$messageID5".toByteArray(), received5.payload) receivedSequence += messageID5 val messageId = received5.applicationProperties[HDR_DUPLICATE_DETECTION_ID.toString()] as String diff --git a/node/src/main/kotlin/net/corda/node/services/messaging/Messaging.kt b/node/src/main/kotlin/net/corda/node/services/messaging/Messaging.kt index 0a75733280..46913bcdc4 100644 --- a/node/src/main/kotlin/net/corda/node/services/messaging/Messaging.kt +++ b/node/src/main/kotlin/net/corda/node/services/messaging/Messaging.kt @@ -176,10 +176,3 @@ interface AcknowledgeHandle { typealias MessageHandler = (ReceivedMessage, MessageHandlerRegistration, AcknowledgeHandle) -> Unit -object P2PMessagingHeaders { - - object Type { - const val KEY = "corda_p2p_message_type" - const val SESSION_INIT_VALUE = "session_init" - } -} diff --git a/node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt b/node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt index cb8176786b..1f7c18d442 100644 --- a/node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt +++ b/node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt @@ -46,6 +46,7 @@ import net.corda.nodeapi.internal.ArtemisMessagingComponent import net.corda.nodeapi.internal.ArtemisMessagingComponent.ArtemisAddress import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_CONTROL import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_NOTIFY +import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2PMessagingHeaders import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEERS_PREFIX import net.corda.nodeapi.internal.ArtemisMessagingComponent.NodeAddress import net.corda.nodeapi.internal.ArtemisMessagingComponent.RemoteInboxAddress @@ -124,17 +125,7 @@ class P2PMessagingClient(val config: NodeConfiguration, ) : SingletonSerializeAsToken(), MessagingService, AddressToArtemisQueueResolver, AutoCloseable { companion object { private val log = contextLogger() - // This is a "property" attached to an Artemis MQ message object, which contains our own notion of "topic". - // We should probably try to unify our notion of "topic" (really, just a string that identifies an endpoint - // that will handle messages, like a URL) with the terminology used by underlying MQ libraries, to avoid - // confusion. - val topicProperty = SimpleString("platform-topic") - val cordaVendorProperty = SimpleString("corda-vendor") - val releaseVersionProperty = SimpleString("release-version") - val platformVersionProperty = SimpleString("platform-version") - val senderUUID = SimpleString("sender-uuid") - val senderSeqNo = SimpleString("send-seq-no") - + private val amqDelayMillis = System.getProperty("amq.delivery.delay.ms", "0").toInt() private const val messageMaxRetryCount: Int = 3 fun createMessageToRedeliver(): PersistentMap, RetryMessage, Long> { @@ -403,9 +394,9 @@ class P2PMessagingClient(val config: NodeConfiguration, private fun artemisToCordaMessage(message: ClientMessage): ReceivedMessage? { try { - val topic = message.required(topicProperty) { getStringProperty(it) } + val topic = message.required(P2PMessagingHeaders.topicProperty) { getStringProperty(it) } val user = requireNotNull(message.getStringProperty(HDR_VALIDATED_USER)) { "Message is not authenticated" } - val platformVersion = message.required(platformVersionProperty) { getIntProperty(it) } + val platformVersion = message.required(P2PMessagingHeaders.platformVersionProperty) { getIntProperty(it) } // Use the magic deduplication property built into Artemis as our message identity too val uniqueMessageId = message.required(HDR_DUPLICATE_DETECTION_ID) { DeduplicationId(message.getStringProperty(it)) } val receivedSenderUUID = message.getStringProperty(senderUUID)