CORDA-864 Wire up max message size (#3057)

* add checks on message size
* added size check in AMQP bridge
* passing maxMessageSize to AMQPClient and server
* added Interceptor to enforce maxMessageSize on incoming messages

docs/source/network-map.rst

@@ -105,11 +105,9 @@ The current set of network parameters:
 :notaries: List of identity and validation type (either validating or non-validating) of the notaries which are permitted
         in the compatibility zone.
 
-:maxMessageSize: (This is currently ignored. However, it will be wired up in a future release.)
-
-.. TODO Replace the above with this once wired: Maximum allowed size in bytes of an individual message sent over the wire. Note that attachments are
-        a special case and may be fragmented for streaming transfer, however, an individual transaction or flow message
-        may not be larger than this value.
+:maxMessageSize: Maximum allowed size in bytes of an individual message sent over the wire. Note that attachments are
+            a special case and may be fragmented for streaming transfer, however, an individual transaction or flow message
+            may not be larger than this value.
 
 :maxTransactionSize: Maximum allowed size in bytes of a transaction. This is the size of the transaction object and its attachments.
 

node-api/build.gradle

@@ -48,6 +48,11 @@ dependencies {
     testCompile "junit:junit:$junit_version"
     testCompile "org.assertj:assertj-core:$assertj_version"
     testCompile project(':node-driver')
+
+    compile ("org.apache.activemq:artemis-amqp-protocol:${artemis_version}") {
+        // Gains our proton-j version from core module.
+        exclude group: 'org.apache.qpid', module: 'proton-j'
+    }
 }
 
 configurations {

node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingClient.kt

@@ -43,6 +43,7 @@ class ArtemisMessagingClient(private val config: SSLConfiguration,
             clientFailureCheckPeriod = -1
             minLargeMessageSize = maxMessageSize
             isUseGlobalPools = nodeSerializationEnv != null
+            addIncomingInterceptor(ArtemisMessageSizeChecksInterceptor(maxMessageSize))
         }
         val sessionFactory = locator.createSessionFactory()
         // Login using the node username. The broker will authenticate us as its node (as opposed to another peer)

node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingComponent.kt

@@ -30,7 +30,9 @@ 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"
-
+        // This is a rough guess on the extra space needed on top of maxMessageSize to store the journal.
+        // TODO: we might want to make this value configurable.
+        const val JOURNAL_HEADER_SIZE = 1024
         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

node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisUtils.kt

@@ -12,3 +12,7 @@ import java.nio.file.Path
 fun Path.requireOnDefaultFileSystem() {
     require(fileSystem == FileSystems.getDefault()) { "Artemis only uses the default file system" }
 }
+
+fun requireMessageSize(messageSize: Int, limit: Int) {
+    require(messageSize <= limit) { "Message exceeds maxMessageSize network parameter, maxMessageSize: [$limit], message size: [$messageSize]" }
+}

node-api/src/main/kotlin/net/corda/nodeapi/internal/MessageSizeChecksInterceptor.kt

@@ -0,0 +1,50 @@
+package net.corda.nodeapi.internal
+
+import net.corda.core.utilities.contextLogger
+import org.apache.activemq.artemis.api.core.BaseInterceptor
+import org.apache.activemq.artemis.api.core.Interceptor
+import org.apache.activemq.artemis.core.protocol.core.Packet
+import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.MessagePacket
+import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage
+import org.apache.activemq.artemis.protocol.amqp.broker.AmqpInterceptor
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection
+
+class ArtemisMessageSizeChecksInterceptor(maxMessageSize: Int) : MessageSizeChecksInterceptor<Packet>(maxMessageSize), Interceptor {
+    override fun getMessageSize(packet: Packet?): Int? {
+        return when (packet) {
+        // This is an estimate of how much memory a Message body takes up.
+        // Note, it is only an estimate
+            is MessagePacket -> (packet.message.persistentSize - packet.message.headersAndPropertiesEncodeSize - 4).toInt()
+        // Skip all artemis control messages.
+            else -> null
+        }
+    }
+}
+
+class AmqpMessageSizeChecksInterceptor(maxMessageSize: Int) : MessageSizeChecksInterceptor<AMQPMessage>(maxMessageSize), AmqpInterceptor {
+    override fun getMessageSize(packet: AMQPMessage?): Int? = packet?.length
+}
+
+/**
+ *  Artemis message interceptor to enforce maxMessageSize on incoming messages.
+ */
+sealed class MessageSizeChecksInterceptor<T : Any>(private val maxMessageSize: Int) : BaseInterceptor<T> {
+    companion object {
+        private val logger = contextLogger()
+    }
+
+    override fun intercept(packet: T, connection: RemotingConnection?): Boolean {
+        val messageSize = getMessageSize(packet) ?: return true
+        return if (messageSize > maxMessageSize) {
+            logger.warn("Message size exceeds maxMessageSize network parameter, maxMessageSize: [$maxMessageSize], message size: [$messageSize], " +
+                    "dropping message, client id :${connection?.clientID}")
+            false
+        } else {
+            true
+        }
+    }
+
+    // get size of the message in byte, returns null if the message is null or size don't need to be checked.
+    abstract fun getMessageSize(packet: T?): Int?
+}
+

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

@@ -37,7 +37,7 @@ import kotlin.concurrent.withLock
  *  The Netty thread pool used by the AMQPBridges is also shared and managed by the AMQPBridgeManager.
  */
 @VisibleForTesting
-class AMQPBridgeManager(config: NodeSSLConfiguration, private val artemisMessageClientFactory: () -> ArtemisSessionProvider) : BridgeManager {
+class AMQPBridgeManager(config: NodeSSLConfiguration, private val maxMessageSize: Int, private val artemisMessageClientFactory: () -> ArtemisSessionProvider) : BridgeManager {
 
     private val lock = ReentrantLock()
     private val bridgeNameToBridgeMap = mutableMapOf<String, AMQPBridge>()
@@ -48,7 +48,7 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, private val artemisMessage
     private var artemis: ArtemisSessionProvider? = null
     private val crlCheckSoftFail: Boolean = config.crlCheckSoftFail
 
-    constructor(config: NodeSSLConfiguration, p2pAddress: NetworkHostAndPort, maxMessageSize: Int) : this(config, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
+    constructor(config: NodeSSLConfiguration, p2pAddress: NetworkHostAndPort, maxMessageSize: Int) : this(config, maxMessageSize, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
 
     companion object {
         private const val NUM_BRIDGE_THREADS = 0 // Default sized pool
@@ -70,14 +70,15 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, private val artemisMessage
                              trustStore: KeyStore,
                              crlCheckSoftFail: Boolean,
                              sharedEventGroup: EventLoopGroup,
-                             private val artemis: ArtemisSessionProvider) {
+                             private val artemis: ArtemisSessionProvider,
+                             private val maxMessageSize: Int) {
         companion object {
             fun getBridgeName(queueName: String, hostAndPort: NetworkHostAndPort): String = "$queueName -> $hostAndPort"
         }
 
         private val log = LoggerFactory.getLogger("$bridgeName:${legalNames.first()}")
 
-        val amqpClient = AMQPClient(listOf(target), legalNames, PEER_USER, PEER_USER, keyStore, keyStorePrivateKeyPassword, trustStore, crlCheckSoftFail, sharedThreadPool = sharedEventGroup)
+        val amqpClient = AMQPClient(listOf(target), legalNames, PEER_USER, PEER_USER, keyStore, keyStorePrivateKeyPassword, trustStore, crlCheckSoftFail, sharedThreadPool = sharedEventGroup, maxMessageSize = maxMessageSize)
         val bridgeName: String get() = getBridgeName(queueName, target)
         private val lock = ReentrantLock() // lock to serialise session level access
         private var session: ClientSession? = null
@@ -129,6 +130,13 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, private val artemisMessage
         }
 
         private fun clientArtemisMessageHandler(artemisMessage: ClientMessage) {
+            if (artemisMessage.bodySize > maxMessageSize) {
+                log.warn("Message exceeds maxMessageSize network parameter, maxMessageSize: [$maxMessageSize], message size: [${artemisMessage.bodySize}], " +
+                        "dropping message, uuid: ${artemisMessage.getObjectProperty("_AMQ_DUPL_ID")}")
+                // Ack the message to prevent same message being sent to us again.
+                artemisMessage.acknowledge()
+                return
+            }
             val data = ByteArray(artemisMessage.bodySize).apply { artemisMessage.bodyBuffer.readBytes(this) }
             val properties = HashMap<String, Any?>()
             for (key in P2PMessagingHeaders.whitelistedHeaders) {
@@ -171,7 +179,7 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, private val artemisMessage
         if (bridgeExists(getBridgeName(queueName, target))) {
             return
         }
-        val newBridge = AMQPBridge(queueName, target, legalNames, keyStore, keyStorePrivateKeyPassword, trustStore, crlCheckSoftFail, sharedEventLoopGroup!!, artemis!!)
+        val newBridge = AMQPBridge(queueName, target, legalNames, keyStore, keyStorePrivateKeyPassword, trustStore, crlCheckSoftFail, sharedEventLoopGroup!!, artemis!!, maxMessageSize)
         lock.withLock {
             bridgeNameToBridgeMap[newBridge.bridgeName] = newBridge
         }

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

@@ -19,16 +19,17 @@ import org.apache.activemq.artemis.api.core.client.ClientMessage
 import java.util.*
 
 class BridgeControlListener(val config: NodeSSLConfiguration,
+                            maxMessageSize: Int,
                             val artemisMessageClientFactory: () -> ArtemisSessionProvider) : AutoCloseable {
     private val bridgeId: String = UUID.randomUUID().toString()
-    private val bridgeManager: BridgeManager = AMQPBridgeManager(config, artemisMessageClientFactory)
+    private val bridgeManager: BridgeManager = AMQPBridgeManager(config, maxMessageSize, artemisMessageClientFactory)
     private val validInboundQueues = mutableSetOf<String>()
     private var artemis: ArtemisSessionProvider? = null
     private var controlConsumer: ClientConsumer? = null
 
     constructor(config: NodeSSLConfiguration,
                 p2pAddress: NetworkHostAndPort,
-                maxMessageSize: Int) : this(config, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
+                maxMessageSize: Int) : this(config, maxMessageSize, { ArtemisMessagingClient(config, p2pAddress, maxMessageSize) })
 
     companion object {
         private val log = contextLogger()

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

@@ -15,6 +15,7 @@ import net.corda.core.utilities.contextLogger
 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 java.lang.Long.min
@@ -41,7 +42,8 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
                  private val trustStore: KeyStore,
                  private val crlCheckSoftFail: Boolean,
                  private val trace: Boolean = false,
-                 private val sharedThreadPool: EventLoopGroup? = null) : AutoCloseable {
+                 private val sharedThreadPool: EventLoopGroup? = null,
+                 private val maxMessageSize: Int) : AutoCloseable {
     companion object {
         init {
             InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE)
@@ -91,17 +93,15 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
         }
     }
 
-    private val closeListener = object : ChannelFutureListener {
-        override fun operationComplete(future: ChannelFuture) {
-            log.info("Disconnected from $currentTarget")
-            future.channel()?.disconnect()
-            clientChannel = null
-            if (!stopping) {
-                workerGroup?.schedule({
-                    nextTarget()
-                    restart()
-                }, retryInterval, TimeUnit.MILLISECONDS)
-            }
+    private val closeListener = ChannelFutureListener { future ->
+        log.info("Disconnected from $currentTarget")
+        future.channel()?.disconnect()
+        clientChannel = null
+        if (!stopping) {
+            workerGroup?.schedule({
+                nextTarget()
+                restart()
+            }, retryInterval, TimeUnit.MILLISECONDS)
         }
     }
 
@@ -182,6 +182,7 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
                       topic: String,
                       destinationLegalName: String,
                       properties: Map<String, Any?>): SendableMessage {
+        requireMessageSize(payload.size, maxMessageSize)
         return SendableMessageImpl(payload, topic, destinationLegalName, currentTarget, properties)
     }
 

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

@@ -17,6 +17,7 @@ import net.corda.core.utilities.contextLogger
 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 org.apache.qpid.proton.engine.Delivery
 import rx.Observable
 import rx.subjects.PublishSubject
@@ -41,7 +42,8 @@ class AMQPServer(val hostName: String,
                  private val keyStorePrivateKeyPassword: CharArray,
                  private val trustStore: KeyStore,
                  private val crlCheckSoftFail: Boolean,
-                 private val trace: Boolean = false) : AutoCloseable {
+                 private val trace: Boolean = false,
+                 private val maxMessageSize: Int) : AutoCloseable {
 
     companion object {
         init {
@@ -68,7 +70,8 @@ class AMQPServer(val hostName: String,
                 keyStorePrivateKeyPassword: String,
                 trustStore: KeyStore,
                 crlCheckSoftFail: Boolean,
-                trace: Boolean = false) : this(hostName, port, userName, password, keyStore, keyStorePrivateKeyPassword.toCharArray(), trustStore, crlCheckSoftFail, trace)
+                trace: Boolean = false,
+                maxMessageSize: Int) : this(hostName, port, userName, password, keyStore, keyStorePrivateKeyPassword.toCharArray(), trustStore, crlCheckSoftFail, trace, maxMessageSize)
 
     private class ServerChannelInitializer(val parent: AMQPServer) : ChannelInitializer<SocketChannel>() {
         private val keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm())
@@ -156,6 +159,7 @@ class AMQPServer(val hostName: String,
                       destinationLegalName: String,
                       destinationLink: NetworkHostAndPort,
                       properties: Map<String, Any?>): SendableMessage {
+        requireMessageSize(payload.size, maxMessageSize)
         val dest = InetSocketAddress(destinationLink.host, destinationLink.port)
         require(dest in clientChannels.keys) {
             "Destination not available"

node/src/integration-test/kotlin/net/corda/node/amqp/AMQPBridgeTest.kt

@@ -193,7 +193,7 @@ class AMQPBridgeTest {
         return Triple(artemisServer, artemisClient, bridgeManager)
     }
 
-    private fun createAMQPServer(): AMQPServer {
+    private fun createAMQPServer(maxMessageSize: Int = MAX_MESSAGE_SIZE): AMQPServer {
         val serverConfig = rigorousMock<AbstractNodeConfiguration>().also {
             doReturn(temporaryFolder.root.toPath() / "server").whenever(it).baseDirectory
             doReturn(BOB_NAME).whenever(it).myLegalName
@@ -210,7 +210,8 @@ class AMQPBridgeTest {
                 serverConfig.keyStorePassword,
                 serverConfig.loadTrustStore().internal,
                 crlCheckSoftFail = true,
-                trace = true
+                trace = true,
+                maxMessageSize = maxMessageSize
         )
     }
 }

node/src/integration-test/kotlin/net/corda/node/amqp/CertificateRevocationListNodeTests.kt

@@ -19,10 +19,7 @@ import net.corda.nodeapi.internal.crypto.*
 import net.corda.nodeapi.internal.protonwrapper.messages.MessageStatus
 import net.corda.nodeapi.internal.protonwrapper.netty.AMQPClient
 import net.corda.nodeapi.internal.protonwrapper.netty.AMQPServer
-import net.corda.testing.core.ALICE_NAME
-import net.corda.testing.core.BOB_NAME
-import net.corda.testing.core.CHARLIE_NAME
-import net.corda.testing.core.freePort
+import net.corda.testing.core.*
 import net.corda.testing.internal.DEV_INTERMEDIATE_CA
 import net.corda.testing.internal.DEV_ROOT_CA
 import net.corda.testing.internal.rigorousMock
@@ -272,7 +269,8 @@ class CertificateRevocationListNodeTests {
     private fun createClient(targetPort: Int,
                              crlCheckSoftFail: Boolean,
                              nodeCrlDistPoint: String = "http://${server.hostAndPort}/crl/node.crl",
-                             tlsCrlDistPoint: String? = "http://${server.hostAndPort}/crl/empty.crl"): Pair<AMQPClient, X509Certificate> {
+                             tlsCrlDistPoint: String? = "http://${server.hostAndPort}/crl/empty.crl",
+                             maxMessageSize: Int = MAX_MESSAGE_SIZE): Pair<AMQPClient, X509Certificate> {
         val clientConfig = rigorousMock<AbstractNodeConfiguration>().also {
             doReturn(temporaryFolder.root.toPath() / "client").whenever(it).baseDirectory
             doReturn(BOB_NAME).whenever(it).myLegalName
@@ -292,13 +290,15 @@ class CertificateRevocationListNodeTests {
                 clientKeystore,
                 clientConfig.keyStorePassword,
                 clientTruststore,
-                crlCheckSoftFail), nodeCert)
+                crlCheckSoftFail,
+                maxMessageSize = maxMessageSize), nodeCert)
     }
 
     private fun createServer(port: Int, name: CordaX500Name = ALICE_NAME,
                              crlCheckSoftFail: Boolean,
                              nodeCrlDistPoint: String = "http://${server.hostAndPort}/crl/node.crl",
-                             tlsCrlDistPoint: String? = "http://${server.hostAndPort}/crl/empty.crl"): Pair<AMQPServer, X509Certificate> {
+                             tlsCrlDistPoint: String? = "http://${server.hostAndPort}/crl/empty.crl",
+                             maxMessageSize: Int = MAX_MESSAGE_SIZE): Pair<AMQPServer, X509Certificate> {
         val serverConfig = rigorousMock<AbstractNodeConfiguration>().also {
             doReturn(temporaryFolder.root.toPath() / "server").whenever(it).baseDirectory
             doReturn(name).whenever(it).myLegalName
@@ -318,7 +318,8 @@ class CertificateRevocationListNodeTests {
                 serverKeystore,
                 serverConfig.keyStorePassword,
                 serverTruststore,
-                crlCheckSoftFail), nodeCert)
+                crlCheckSoftFail,
+                maxMessageSize = maxMessageSize), nodeCert)
     }
 
     private fun SSLConfiguration.recreateNodeCaAndTlsCertificates(nodeCaCrlDistPoint: String, tlsCrlDistPoint: String?): X509Certificate {

node/src/integration-test/kotlin/net/corda/node/amqp/ProtonWrapperTests.kt

@@ -26,6 +26,7 @@ import net.corda.testing.core.*
 import net.corda.testing.internal.createDevIntermediateCaCertPath
 import net.corda.testing.internal.rigorousMock
 import org.apache.activemq.artemis.api.core.RoutingType
+import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.Assert.assertArrayEquals
 import org.junit.Rule
 import org.junit.Test
@@ -260,6 +261,52 @@ class ProtonWrapperTests {
         server.stop()
     }
 
+    @Test
+    fun `Send a message larger then maxMessageSize from AMQP to Artemis inbox`() {
+        val maxMessageSize = 100_000
+        val (server, artemisClient) = createArtemisServerAndClient(maxMessageSize)
+        val amqpClient = createClient(maxMessageSize)
+        val clientConnected = amqpClient.onConnection.toFuture()
+        amqpClient.start()
+        assertEquals(true, clientConnected.get().connected)
+        assertEquals(CHARLIE_NAME, CordaX500Name.build(clientConnected.get().remoteCert!!.subjectX500Principal))
+        val artemis = artemisClient.started!!
+        val sendAddress = P2P_PREFIX + "Test"
+        artemis.session.createQueue(sendAddress, RoutingType.ANYCAST, "queue", true)
+        val consumer = artemis.session.createConsumer("queue")
+
+        val testProperty = mutableMapOf<String, Any?>()
+        testProperty["TestProp"] = "1"
+
+        // Send normal message.
+        val testData = ByteArray(maxMessageSize)
+        val message = amqpClient.createMessage(testData, sendAddress, CHARLIE_NAME.toString(), testProperty)
+        amqpClient.write(message)
+        assertEquals(MessageStatus.Acknowledged, message.onComplete.get())
+        val received = consumer.receive()
+        assertEquals("1", received.getStringProperty("TestProp"))
+        assertArrayEquals(testData, ByteArray(received.bodySize).apply { received.bodyBuffer.readBytes(this) })
+
+        // Send message larger then max message size.
+        val largeData = ByteArray(maxMessageSize + 1)
+        // Create message will fail.
+        assertThatThrownBy {
+            amqpClient.createMessage(largeData, sendAddress, CHARLIE_NAME.toString(), testProperty)
+        }.hasMessageContaining("Message exceeds maxMessageSize network parameter")
+
+        // Send normal message again to confirm the large message didn't reach the server and client is not killed by the message.
+        val message2 = amqpClient.createMessage(testData, sendAddress, CHARLIE_NAME.toString(), testProperty)
+        amqpClient.write(message2)
+        assertEquals(MessageStatus.Acknowledged, message2.onComplete.get())
+        val received2 = consumer.receive()
+        assertEquals("1", received2.getStringProperty("TestProp"))
+        assertArrayEquals(testData, ByteArray(received2.bodySize).apply { received2.bodyBuffer.readBytes(this) })
+
+        amqpClient.stop()
+        artemisClient.stop()
+        server.stop()
+    }
+
     @Test
     fun `shared AMQPClient threadpool tests`() {
         val amqpServer = createServer(serverPort)
@@ -310,7 +357,7 @@ class ProtonWrapperTests {
         }
     }
 
-    private fun createArtemisServerAndClient(): Pair<ArtemisMessagingServer, ArtemisMessagingClient> {
+    private fun createArtemisServerAndClient(maxMessageSize: Int = MAX_MESSAGE_SIZE): Pair<ArtemisMessagingServer, ArtemisMessagingClient> {
         val artemisConfig = rigorousMock<AbstractNodeConfiguration>().also {
             doReturn(temporaryFolder.root.toPath() / "artemis").whenever(it).baseDirectory
             doReturn(CHARLIE_NAME).whenever(it).myLegalName
@@ -323,14 +370,14 @@ class ProtonWrapperTests {
         }
         artemisConfig.configureWithDevSSLCertificate()
 
-        val server = ArtemisMessagingServer(artemisConfig, NetworkHostAndPort("0.0.0.0", artemisPort), MAX_MESSAGE_SIZE)
-        val client = ArtemisMessagingClient(artemisConfig, NetworkHostAndPort("localhost", artemisPort), MAX_MESSAGE_SIZE)
+        val server = ArtemisMessagingServer(artemisConfig, NetworkHostAndPort("0.0.0.0", artemisPort), maxMessageSize)
+        val client = ArtemisMessagingClient(artemisConfig, NetworkHostAndPort("localhost", artemisPort), maxMessageSize)
         server.start()
         client.start()
         return Pair(server, client)
     }
 
-    private fun createClient(): AMQPClient {
+    private fun createClient(maxMessageSize: Int = MAX_MESSAGE_SIZE): AMQPClient {
         val clientConfig = rigorousMock<AbstractNodeConfiguration>().also {
             doReturn(temporaryFolder.root.toPath() / "client").whenever(it).baseDirectory
             doReturn(BOB_NAME).whenever(it).myLegalName
@@ -352,10 +399,11 @@ class ProtonWrapperTests {
                 clientKeystore,
                 clientConfig.keyStorePassword,
                 clientTruststore,
-                true)
+                true,
+                maxMessageSize = maxMessageSize)
     }
 
-    private fun createSharedThreadsClient(sharedEventGroup: EventLoopGroup, id: Int): AMQPClient {
+    private fun createSharedThreadsClient(sharedEventGroup: EventLoopGroup, id: Int, maxMessageSize: Int = MAX_MESSAGE_SIZE): AMQPClient {
         val clientConfig = rigorousMock<AbstractNodeConfiguration>().also {
             doReturn(temporaryFolder.root.toPath() / "client_%$id").whenever(it).baseDirectory
             doReturn(CordaX500Name(null, "client $id", "Corda", "London", null, "GB")).whenever(it).myLegalName
@@ -376,10 +424,11 @@ class ProtonWrapperTests {
                 clientConfig.keyStorePassword,
                 clientTruststore,
                 true,
-                sharedThreadPool = sharedEventGroup)
+                sharedThreadPool = sharedEventGroup,
+                maxMessageSize = maxMessageSize)
     }
 
-    private fun createServer(port: Int, name: CordaX500Name = ALICE_NAME): AMQPServer {
+    private fun createServer(port: Int, name: CordaX500Name = ALICE_NAME, maxMessageSize: Int = MAX_MESSAGE_SIZE): AMQPServer {
         val serverConfig = rigorousMock<AbstractNodeConfiguration>().also {
             doReturn(temporaryFolder.root.toPath() / "server").whenever(it).baseDirectory
             doReturn(name).whenever(it).myLegalName
@@ -399,6 +448,7 @@ class ProtonWrapperTests {
                 serverKeystore,
                 serverConfig.keyStorePassword,
                 serverTruststore,
-                crlCheckSoftFail = true)
+                crlCheckSoftFail = true,
+                maxMessageSize = maxMessageSize)
     }
 }

node/src/main/kotlin/net/corda/node/internal/Node.kt

@@ -94,18 +94,15 @@ open class Node(configuration: NodeConfiguration,
         }
 
         private val sameVmNodeCounter = AtomicInteger()
-
         const val scanPackagesSystemProperty = "net.corda.node.cordapp.scan.packages"
         const val scanPackagesSeparator = ","
-
         private fun makeCordappLoader(configuration: NodeConfiguration): CordappLoader {
             return System.getProperty(scanPackagesSystemProperty)?.let { scanPackages ->
                 CordappLoader.createDefaultWithTestPackages(configuration, scanPackages.split(scanPackagesSeparator))
             } ?: CordappLoader.createDefault(configuration.baseDirectory)
         }
-
-        // TODO Wire up maxMessageSize
-        const val MAX_FILE_SIZE = 10485760
+        // TODO: make this configurable.
+        const val MAX_RPC_MESSAGE_SIZE = 10485760
     }
 
     override val log: Logger get() = staticLog
@@ -175,7 +172,7 @@ open class Node(configuration: NodeConfiguration,
 
         if (!configuration.messagingServerExternal) {
             val brokerBindAddress = configuration.messagingServerAddress ?: NetworkHostAndPort("0.0.0.0", configuration.p2pAddress.port)
-            messageBroker = ArtemisMessagingServer(configuration, brokerBindAddress, MAX_FILE_SIZE)
+            messageBroker = ArtemisMessagingServer(configuration, brokerBindAddress, networkParameters.maxMessageSize)
         }
 
         val serverAddress = configuration.messagingServerAddress ?: NetworkHostAndPort("localhost", configuration.p2pAddress.port)
@@ -185,11 +182,11 @@ open class Node(configuration: NodeConfiguration,
             startLocalRpcBroker()
         }
         val advertisedAddress = info.addresses[0]
-        bridgeControlListener = BridgeControlListener(configuration, serverAddress, /*networkParameters.maxMessageSize*/MAX_FILE_SIZE)
+        bridgeControlListener = BridgeControlListener(configuration, serverAddress, networkParameters.maxMessageSize)
 
         printBasicNodeInfo("Advertised P2P messaging addresses", info.addresses.joinToString())
         rpcServerAddresses?.let {
-            rpcMessagingClient = RPCMessagingClient(configuration.rpcOptions.sslConfig, it.admin, /*networkParameters.maxMessageSize*/MAX_FILE_SIZE)
+            rpcMessagingClient = RPCMessagingClient(configuration.rpcOptions.sslConfig, it.admin, MAX_RPC_MESSAGE_SIZE)
             printBasicNodeInfo("RPC connection address", it.primary.toString())
             printBasicNodeInfo("RPC admin connection address", it.admin.toString())
         }
@@ -209,7 +206,7 @@ open class Node(configuration: NodeConfiguration,
                 database,
                 services.networkMapCache,
                 advertisedAddress,
-                /*networkParameters.maxMessageSize*/MAX_FILE_SIZE,
+                networkParameters.maxMessageSize,
                 isDrainingModeOn = nodeProperties.flowsDrainingMode::isEnabled,
                 drainingModeWasChangedEvents = nodeProperties.flowsDrainingMode.values)
     }
@@ -221,9 +218,9 @@ open class Node(configuration: NodeConfiguration,
                 val rpcBrokerDirectory: Path = baseDirectory / "brokers" / "rpc"
                 with(rpcOptions) {
                     rpcBroker = if (useSsl) {
-                        ArtemisRpcBroker.withSsl(this.address!!, sslConfig, securityManager, certificateChainCheckPolicies, /*networkParameters.maxMessageSize*/MAX_FILE_SIZE, jmxMonitoringHttpPort != null, rpcBrokerDirectory)
+                        ArtemisRpcBroker.withSsl(this.address!!, sslConfig, securityManager, certificateChainCheckPolicies, MAX_RPC_MESSAGE_SIZE, jmxMonitoringHttpPort != null, rpcBrokerDirectory)
                     } else {
-                        ArtemisRpcBroker.withoutSsl(this.address!!, adminAddress!!, sslConfig, securityManager, certificateChainCheckPolicies, /*networkParameters.maxMessageSize*/MAX_FILE_SIZE, jmxMonitoringHttpPort != null, rpcBrokerDirectory)
+                        ArtemisRpcBroker.withoutSsl(this.address!!, adminAddress!!, sslConfig, securityManager, certificateChainCheckPolicies, MAX_RPC_MESSAGE_SIZE, jmxMonitoringHttpPort != null, rpcBrokerDirectory)
                     }
                 }
                 return rpcBroker!!.addresses

node/src/main/kotlin/net/corda/node/services/messaging/ArtemisMessagingServer.kt

@@ -19,7 +19,10 @@ import net.corda.node.services.messaging.NodeLoginModule.Companion.VERIFIER_ROLE
 import net.corda.nodeapi.ArtemisTcpTransport
 import net.corda.nodeapi.ConnectionDirection
 import net.corda.nodeapi.VerifierApi
+import net.corda.nodeapi.internal.AmqpMessageSizeChecksInterceptor
+import net.corda.nodeapi.internal.ArtemisMessageSizeChecksInterceptor
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.INTERNAL_PREFIX
+import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.JOURNAL_HEADER_SIZE
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_USER
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NOTIFICATIONS_ADDRESS
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2P_PREFIX
@@ -69,7 +72,7 @@ import javax.security.auth.spi.LoginModule
 @ThreadSafe
 class ArtemisMessagingServer(private val config: NodeConfiguration,
                              private val messagingServerAddress: NetworkHostAndPort,
-                             val maxMessageSize: Int) : ArtemisBroker, SingletonSerializeAsToken() {
+                             private val maxMessageSize: Int) : ArtemisBroker, SingletonSerializeAsToken() {
     companion object {
         private val log = contextLogger()
     }
@@ -117,8 +120,11 @@ class ArtemisMessagingServer(private val config: NodeConfiguration,
             registerPostQueueCreationCallback { log.debug { "Queue Created: $it" } }
             registerPostQueueDeletionCallback { address, qName -> log.debug { "Queue deleted: $qName for $address" } }
         }
-        // Config driven switch between legacy CORE bridges and the newer AMQP protocol bridges.
+
         activeMQServer.start()
+        activeMQServer.remotingService.addIncomingInterceptor(ArtemisMessageSizeChecksInterceptor(maxMessageSize))
+        activeMQServer.remotingService.addIncomingInterceptor(AmqpMessageSizeChecksInterceptor(maxMessageSize))
+        // Config driven switch between legacy CORE bridges and the newer AMQP protocol bridges.
         log.info("P2P messaging server listening on $messagingServerAddress")
     }
 
@@ -137,9 +143,9 @@ class ArtemisMessagingServer(private val config: NodeConfiguration,
         idCacheSize = 2000 // Artemis Default duplicate cache size i.e. a guess
         isPersistIDCache = true
         isPopulateValidatedUser = true
-        journalBufferSize_NIO = maxMessageSize // Artemis default is 490KiB - required to address IllegalArgumentException (when Artemis uses Java NIO): Record is too large to store.
-        journalBufferSize_AIO = maxMessageSize // Required to address IllegalArgumentException (when Artemis uses Linux Async IO): Record is too large to store.
-        journalFileSize = maxMessageSize // The size of each journal file in bytes. Artemis default is 10MiB.
+        journalBufferSize_NIO = maxMessageSize + JOURNAL_HEADER_SIZE // Artemis default is 490KiB - required to address IllegalArgumentException (when Artemis uses Java NIO): Record is too large to store.
+        journalBufferSize_AIO = maxMessageSize + JOURNAL_HEADER_SIZE // Required to address IllegalArgumentException (when Artemis uses Linux Async IO): Record is too large to store.
+        journalFileSize = maxMessageSize + JOURNAL_HEADER_SIZE// The size of each journal file in bytes. Artemis default is 10MiB.
         managementNotificationAddress = SimpleString(NOTIFICATIONS_ADDRESS)
 
         // JMX enablement

node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt

@@ -35,6 +35,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.JOURNAL_HEADER_SIZE
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2PMessagingHeaders
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEERS_PREFIX
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.NodeAddress
@@ -44,6 +45,7 @@ import net.corda.nodeapi.internal.bridging.BridgeControl
 import net.corda.nodeapi.internal.bridging.BridgeEntry
 import net.corda.nodeapi.internal.persistence.CordaPersistence
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
+import net.corda.nodeapi.internal.requireMessageSize
 import org.apache.activemq.artemis.api.core.ActiveMQObjectClosedException
 import org.apache.activemq.artemis.api.core.Message.HDR_DUPLICATE_DETECTION_ID
 import org.apache.activemq.artemis.api.core.Message.HDR_VALIDATED_USER
@@ -204,7 +206,7 @@ class P2PMessagingClient(val config: NodeConfiguration,
                 // would be the default and the two lines below can be deleted.
                 connectionTTL = -1
                 clientFailureCheckPeriod = -1
-                minLargeMessageSize = maxMessageSize
+                minLargeMessageSize = maxMessageSize + JOURNAL_HEADER_SIZE
                 isUseGlobalPools = nodeSerializationEnv != null
             }
             val sessionFactory = locator!!.createSessionFactory()
@@ -269,7 +271,7 @@ class P2PMessagingClient(val config: NodeConfiguration,
         networkChangeSubscription = networkMap.changed.subscribe { updateBridgesOnNetworkChange(it) }
     }
 
-     private fun sendBridgeControl(message: BridgeControl) {
+    private fun sendBridgeControl(message: BridgeControl) {
         state.locked {
             val controlPacket = message.serialize(context = SerializationDefaults.P2P_CONTEXT).bytes
             val artemisMessage = producerSession!!.createMessage(false)
@@ -377,6 +379,7 @@ class P2PMessagingClient(val config: NodeConfiguration,
 
     private fun artemisToCordaMessage(message: ClientMessage): ReceivedMessage? {
         try {
+            requireMessageSize(message.bodySize, maxMessageSize)
             val topic = message.required(P2PMessagingHeaders.topicProperty) { getStringProperty(it) }
             val user = requireNotNull(message.getStringProperty(HDR_VALIDATED_USER)) { "Message is not authenticated" }
             val platformVersion = message.required(P2PMessagingHeaders.platformVersionProperty) { getIntProperty(it) }
@@ -505,6 +508,7 @@ class P2PMessagingClient(val config: NodeConfiguration,
 
     @Suspendable
     override fun send(message: Message, target: MessageRecipients, retryId: Long?, sequenceKey: Any) {
+        requireMessageSize(message.data.size, maxMessageSize)
         messagingExecutor!!.send(message, target)
         retryId?.let {
             database.transaction {
@@ -539,7 +543,7 @@ class P2PMessagingClient(val config: NodeConfiguration,
 
         scheduledMessageRedeliveries[retryId] = nodeExecutor.schedule({
             sendWithRetry(retryCount + 1, message, target, retryId)
-        },messageRedeliveryDelaySeconds * Math.pow(backoffBase, retryCount.toDouble()).toLong(), TimeUnit.SECONDS)
+        }, messageRedeliveryDelaySeconds * Math.pow(backoffBase, retryCount.toDouble()).toLong(), TimeUnit.SECONDS)
     }
 
     override fun cancelRedelivery(retryId: Long) {
@@ -560,7 +564,8 @@ class P2PMessagingClient(val config: NodeConfiguration,
         } else {
             // Otherwise we send the message to an internal queue for the target residing on our broker. It's then the
             // broker's job to route the message to the target's P2P queue.
-            val internalTargetQueue = (address as? ArtemisAddress)?.queueName ?: throw IllegalArgumentException("Not an Artemis address")
+            val internalTargetQueue = (address as? ArtemisAddress)?.queueName
+                    ?: throw IllegalArgumentException("Not an Artemis address")
             state.locked {
                 createQueueIfAbsent(internalTargetQueue, producerSession!!)
             }

node/src/test/kotlin/net/corda/node/services/messaging/ArtemisMessagingTest.kt

@@ -21,6 +21,7 @@ import net.corda.testing.internal.LogHelper
 import net.corda.testing.internal.rigorousMock
 import net.corda.testing.node.MockServices.Companion.makeTestDataSourceProperties
 import net.corda.testing.node.internal.MOCK_VERSION_INFO
+import org.apache.activemq.artemis.api.core.ActiveMQConnectionTimedOutException
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.After
@@ -36,6 +37,7 @@ import java.util.concurrent.TimeUnit.MILLISECONDS
 import kotlin.concurrent.thread
 import kotlin.test.assertEquals
 import kotlin.test.assertNull
+import kotlin.test.assertTrue
 
 class ArtemisMessagingTest {
     companion object {
@@ -71,7 +73,7 @@ class ArtemisMessagingTest {
             doReturn(NetworkHostAndPort("0.0.0.0", serverPort)).whenever(it).p2pAddress
             doReturn(null).whenever(it).jmxMonitoringHttpPort
             doReturn(emptyList<CertChainPolicyConfig>()).whenever(it).certificateChainCheckPolicies
-            doReturn(P2PMessagingRetryConfiguration(5.seconds, 3, backoffBase=1.0)).whenever(it).p2pMessagingRetry
+            doReturn(P2PMessagingRetryConfiguration(5.seconds, 3, backoffBase = 1.0)).whenever(it).p2pMessagingRetry
         }
         LogHelper.setLevel(PersistentUniquenessProvider::class)
         database = configureDatabase(makeTestDataSourceProperties(), DatabaseConfig(), rigorousMock())
@@ -133,6 +135,41 @@ class ArtemisMessagingTest {
         assertNull(receivedMessages.poll(200, MILLISECONDS))
     }
 
+    @Test
+    fun `client should fail if message exceed maxMessageSize limit`() {
+        val (messagingClient, receivedMessages) = createAndStartClientAndServer()
+        val message = messagingClient.createMessage(TOPIC, data = ByteArray(MAX_MESSAGE_SIZE))
+        messagingClient.send(message, messagingClient.myAddress)
+
+        val actual: Message = receivedMessages.take()
+        assertTrue(ByteArray(MAX_MESSAGE_SIZE).contentEquals(actual.data.bytes))
+        assertNull(receivedMessages.poll(200, MILLISECONDS))
+
+        val tooLagerMessage = messagingClient.createMessage(TOPIC, data = ByteArray(MAX_MESSAGE_SIZE + 1))
+        assertThatThrownBy {
+            messagingClient.send(tooLagerMessage, messagingClient.myAddress)
+        }.isInstanceOf(IllegalArgumentException::class.java)
+                .hasMessageContaining("Message exceeds maxMessageSize network parameter")
+
+        assertNull(receivedMessages.poll(200, MILLISECONDS))
+    }
+    @Test
+    fun `server should not process if incoming message exceed maxMessageSize limit`() {
+        val (messagingClient, receivedMessages) = createAndStartClientAndServer(clientMaxMessageSize = 100_000, serverMaxMessageSize = 50_000)
+        val message = messagingClient.createMessage(TOPIC, data = ByteArray(50_000))
+        messagingClient.send(message, messagingClient.myAddress)
+
+        val actual: Message = receivedMessages.take()
+        assertTrue(ByteArray(50_000).contentEquals(actual.data.bytes))
+        assertNull(receivedMessages.poll(200, MILLISECONDS))
+
+        val tooLagerMessage = messagingClient.createMessage(TOPIC, data = ByteArray(100_000))
+        assertThatThrownBy {
+            messagingClient.send(tooLagerMessage, messagingClient.myAddress)
+        }.isInstanceOf(ActiveMQConnectionTimedOutException::class.java)
+        assertNull(receivedMessages.poll(200, MILLISECONDS))
+    }
+
     @Test
     fun `platform version is included in the message`() {
         val (messagingClient, receivedMessages) = createAndStartClientAndServer(platformVersion = 3)
@@ -147,12 +184,12 @@ class ArtemisMessagingTest {
         messagingClient!!.start()
     }
 
-    private fun createAndStartClientAndServer(platformVersion: Int = 1): Pair<P2PMessagingClient, BlockingQueue<ReceivedMessage>> {
+    private fun createAndStartClientAndServer(platformVersion: Int = 1, serverMaxMessageSize: Int = MAX_MESSAGE_SIZE, clientMaxMessageSize: Int = MAX_MESSAGE_SIZE): Pair<P2PMessagingClient, BlockingQueue<ReceivedMessage>> {
         val receivedMessages = LinkedBlockingQueue<ReceivedMessage>()
 
-        createMessagingServer().start()
+        createMessagingServer(maxMessageSize = serverMaxMessageSize).start()
 
-        val messagingClient = createMessagingClient(platformVersion = platformVersion)
+        val messagingClient = createMessagingClient(platformVersion = platformVersion, maxMessageSize = clientMaxMessageSize)
         messagingClient.addMessageHandler(TOPIC) { message, _, handle ->
             database.transaction { handle.insideDatabaseTransaction() }
             handle.afterDatabaseTransaction() // We ACK first so that if it fails we won't get a duplicate in [receivedMessages]