NOTICK Merge OS 4.6 -> 4.7

core/src/main/kotlin/net/corda/core/node/NetworkParameters.kt

@@ -21,7 +21,7 @@ import java.time.Instant
  *
  * @property minimumPlatformVersion Minimum version of Corda platform that is required for nodes in the network.
  * @property notaries List of well known and trusted notary identities with information on validation type.
- * @property maxMessageSize This is currently ignored. However, it will be wired up in a future release.
+ * @property maxMessageSize Maximum allowed size in bytes of an individual message sent over the wire.
  * @property maxTransactionSize Maximum permitted transaction size in bytes.
  * @property modifiedTime ([AutoAcceptable]) Last modification time of network parameters set.
  * @property epoch ([AutoAcceptable]) Version number of the network parameters. Starting from 1, this will always increment on each new set

node-api-tests/src/test/kotlin/net/corda/nodeapitests/internal/crypto/X509UtilitiesTest.kt

@@ -43,6 +43,7 @@ import net.corda.coretesting.internal.NettyTestServer
 import net.corda.testing.internal.createDevIntermediateCaCertPath
 import net.corda.coretesting.internal.stubs.CertificateStoreStubs
 import net.corda.nodeapi.internal.crypto.CertificateType
+import net.corda.nodeapi.internal.crypto.X509CertificateFactory
 import net.corda.nodeapi.internal.crypto.X509Utilities
 import net.corda.nodeapi.internal.crypto.checkValidity
 import net.corda.nodeapi.internal.crypto.getSupportedKey
@@ -50,6 +51,7 @@ import net.corda.nodeapi.internal.crypto.loadOrCreateKeyStore
 import net.corda.nodeapi.internal.crypto.save
 import net.corda.nodeapi.internal.crypto.toBc
 import net.corda.nodeapi.internal.crypto.x509
+import net.corda.nodeapi.internal.crypto.x509Certificates
 import net.corda.testing.internal.IS_OPENJ9
 import net.i2p.crypto.eddsa.EdDSAPrivateKey
 import org.assertj.core.api.Assertions.assertThat
@@ -570,4 +572,16 @@ class X509UtilitiesTest {
             cert.checkValidity({ "Error text" }, { }, Date.from(today.toInstant() + 51.days))
         }
     }
+
+    @Test(timeout = 300_000)
+    fun `check certificate serial number`() {
+        val keyPair = generateKeyPair()
+        val subject = X500Principal("CN=Test,O=R3 Ltd,L=London,C=GB")
+        val cert = X509Utilities.createSelfSignedCACertificate(subject, keyPair)
+        assertTrue(cert.serialNumber.signum() > 0)
+        assertEquals(127, cert.serialNumber.bitLength())
+        val serialized = X509Utilities.buildCertPath(cert).encoded
+        val deserialized = X509CertificateFactory().delegate.generateCertPath(serialized.inputStream()).x509Certificates.first()
+        assertEquals(cert.serialNumber, deserialized.serialNumber)
+    }
 }

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

@@ -2,7 +2,7 @@ package net.corda.nodeapi.internal.crypto
 
 import net.corda.core.CordaOID
 import net.corda.core.crypto.Crypto
-import net.corda.core.crypto.random63BitValue
+import net.corda.core.crypto.newSecureRandom
 import net.corda.core.internal.*
 import net.corda.core.utilities.days
 import net.corda.core.utilities.millis
@@ -35,6 +35,8 @@ import java.time.Instant
 import java.time.temporal.ChronoUnit
 import java.util.*
 import javax.security.auth.x500.X500Principal
+import kotlin.experimental.and
+import kotlin.experimental.or
 
 object X509Utilities {
     // Note that this default value only applies to BCCryptoService. Other implementations of CryptoService may have to use different
@@ -58,6 +60,7 @@ object X509Utilities {
     const val TLS_CERTIFICATE_DAYS_TO_EXPIRY_WARNING_THRESHOLD = 30
     private const val KEY_ALIAS_REGEX = "[a-z0-9-]+"
     private const val KEY_ALIAS_MAX_LENGTH = 100
+    private const val CERTIFICATE_SERIAL_NUMBER_LENGTH = 16
 
     /**
      * Checks if the provided key alias does not exceed maximum length and
@@ -184,7 +187,7 @@ object X509Utilities {
                                  nameConstraints: NameConstraints? = null,
                                  crlDistPoint: String? = null,
                                  crlIssuer: X500Name? = null): X509v3CertificateBuilder {
-        val serial = BigInteger.valueOf(random63BitValue())
+        val serial = generateCertificateSerialNumber()
         val keyPurposes = DERSequence(ASN1EncodableVector().apply { certificateType.purposes.forEach { add(it) } })
         val subjectPublicKeyInfo = SubjectPublicKeyInfo.getInstance(ASN1Sequence.getInstance(subjectPublicKey.encoded))
         val role = certificateType.role
@@ -364,6 +367,15 @@ object X509Utilities {
             builder.addExtension(Extension.cRLDistributionPoints, false, CRLDistPoint(arrayOf(distPoint)))
         }
     }
+
+    @Suppress("MagicNumber")
+    private fun generateCertificateSerialNumber(): BigInteger {
+        val bytes = ByteArray(CERTIFICATE_SERIAL_NUMBER_LENGTH)
+        newSecureRandom().nextBytes(bytes)
+        // Set highest byte to 01xxxxxx to ensure positive sign and constant bit length.
+        bytes[0] = bytes[0].and(0x3F).or(0x40)
+        return BigInteger(bytes)
+    }
 }
 
 // Assuming cert type to role is 1:1

node/src/integration-test/kotlin/net/corda/node/AuthDBTests.kt

@@ -23,9 +23,9 @@ import org.junit.Before
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
+import java.sql.Connection
 import java.sql.Statement
 import java.util.*
-import javax.sql.DataSource
 import kotlin.test.assertFailsWith
 
 /*
@@ -34,7 +34,7 @@ import kotlin.test.assertFailsWith
  */
 @RunWith(Parameterized::class)
 class AuthDBTests : NodeBasedTest(cordappPackages = CORDAPPS) {
-    private lateinit var node: NodeWithInfo
+    private var node: NodeWithInfo? = null
     private lateinit var client: CordaRPCClient
     private lateinit var db: UsersDB
 
@@ -97,8 +97,9 @@ class AuthDBTests : NodeBasedTest(cordappPackages = CORDAPPS) {
                 )
         )
 
-        node = startNode(ALICE_NAME, rpcUsers = emptyList(), configOverrides = securityConfig)
+        node = startNode(ALICE_NAME, rpcUsers = emptyList(), configOverrides = securityConfig).also { node ->
-        client = CordaRPCClient(node.node.configuration.rpcOptions.address)
+            client = CordaRPCClient(node.node.configuration.rpcOptions.address)
+        }
     }
 
     @Test(timeout=300_000)
@@ -219,6 +220,7 @@ class AuthDBTests : NodeBasedTest(cordappPackages = CORDAPPS) {
 
     @After
     fun tearDown() {
+        node?.node?.stop()
         db.close()
     }
 
@@ -232,7 +234,7 @@ private data class RoleAndPermissions(val role: String, val permissions: List<St
  * Manage in-memory DB mocking a users database with the schema expected by Node's security manager
  */
 private class UsersDB(name: String, users: List<UserAndRoles> = emptyList(), roleAndPermissions: List<RoleAndPermissions> = emptyList()) : AutoCloseable {
-    val jdbcUrl = "jdbc:h2:mem:$name;DB_CLOSE_DELAY=-1"
+    val jdbcUrl = "jdbc:h2:mem:$name"
 
     companion object {
         const val DB_CREATE_SCHEMA = """
@@ -273,36 +275,34 @@ private class UsersDB(name: String, users: List<UserAndRoles> = emptyList(), rol
         }
     }
 
-    private val dataSource: DataSource
+    private val connection: Connection
     private inline fun session(statement: (Statement) -> Unit) {
-        dataSource.connection.use {
+        connection.createStatement().use(statement)
-            it.autoCommit = false
+        connection.commit()
-            it.createStatement().use(statement)
-            it.commit()
-        }
     }
 
     init {
-        dataSource = DataSourceFactory.createDataSource(Properties().apply {
+        require(users.map { it.username }.toSet().size == users.size) {
+            "Duplicate username in input"
+        }
+        connection = DataSourceFactory.createDataSource(Properties().apply {
             put("dataSourceClassName", "org.h2.jdbcx.JdbcDataSource")
             put("dataSource.url", jdbcUrl)
         }, false)
+                .connection
+                .apply {
+                    autoCommit = false
+                }
         session {
             it.execute(DB_CREATE_SCHEMA)
         }
-        require(users.map { it.username }.toSet().size == users.size) {
-            "Duplicate username in input"
-        }
         users.forEach { insert(it) }
         roleAndPermissions.forEach { insert(it) }
     }
 
     override fun close() {
-        dataSource.connection.use {
+        // Close the connection, at which point the database will shut down
-            it.createStatement().use {
+        connection.close()
-                it.execute("DROP ALL OBJECTS")
-            }
-        }
     }
 }
 

node/src/integration-test/kotlin/net/corda/node/services/events/ScheduledFlowIntegrationTests.kt

@@ -89,18 +89,15 @@ class ScheduledFlowIntegrationTests {
 
     private fun MutableList<CordaFuture<*>>.getOrThrowAll() {
         forEach {
-            try {
+            it.getOrThrow()
-                it.getOrThrow()
-            } catch (ex: Exception) {
-            }
         }
     }
 
     @Test(timeout=300_000)
 	fun `test that when states are being spent at the same time that schedules trigger everything is processed`() {
         driver(DriverParameters(
-                startNodesInProcess = true,
+                startNodesInProcess = false,
-                cordappsForAllNodes = listOf(DUMMY_CONTRACTS_CORDAPP, cordappWithPackages("net.corda.testMessage"), enclosedCordapp())
+                cordappsForAllNodes = listOf(DUMMY_CONTRACTS_CORDAPP, cordappWithPackages("net.corda.testMessage", "net.corda.testing.core"), enclosedCordapp())
         )) {
             val N = 23
             val rpcUser = User("admin", "admin", setOf("ALL"))
@@ -127,6 +124,7 @@ class ScheduledFlowIntegrationTests {
                         scheduledFor
                 ).returnValue)
             }
+
             initialiseFutures.getOrThrowAll()
 
             val spendAttemptFutures = mutableListOf<CordaFuture<*>>()
@@ -134,6 +132,7 @@ class ScheduledFlowIntegrationTests {
                 spendAttemptFutures.add(aliceClient.proxy.startFlow(::AnotherFlow, (i).toString()).returnValue)
                 spendAttemptFutures.add(bobClient.proxy.startFlow(::AnotherFlow, (i + 100).toString()).returnValue)
             }
+
             spendAttemptFutures.getOrThrowAll()
 
             // TODO: the queries below are not atomic so we need to allow enough time for the scheduler to finish.  Would be better to query scheduler.
@@ -144,7 +143,7 @@ class ScheduledFlowIntegrationTests {
 
             val bobStates = bobClient.proxy.vaultQuery(ScheduledState::class.java).states.filter { it.state.data.processed }
             val bobSpentStates = bobClient.proxy.vaultQuery(SpentState::class.java).states
-
+            
             assertEquals(aliceStates.count() + aliceSpentStates.count(), N * 2)
             assertEquals(bobStates.count() + bobSpentStates.count(), N * 2)
             assertEquals(aliceSpentStates.count(), bobSpentStates.count())

node/src/integration-test/kotlin/net/corda/node/services/vault/VaultObserverExceptionTest.kt

@@ -52,6 +52,7 @@ import kotlin.test.assertTrue
 class VaultObserverExceptionTest {
     companion object {
 
+        val waitForFlowDuration = 45.seconds
         val log = contextLogger()
 
         private fun testCordapps() = listOf(
@@ -99,7 +100,7 @@ class VaultObserverExceptionTest {
                 "Syntax Error in Custom SQL",
                 CreateStateFlow.errorTargetsToNum(CreateStateFlow.ErrorTarget.ServiceSqlSyntaxError)
             ).returnValue.then { testControlFuture.complete(false) }
-            val foundExpectedException = testControlFuture.getOrThrow(30.seconds)
+            val foundExpectedException = testControlFuture.getOrThrow(waitForFlowDuration)
 
             Assert.assertTrue(foundExpectedException)
         }
@@ -133,7 +134,7 @@ class VaultObserverExceptionTest {
                 "Syntax Error in Custom SQL",
                 CreateStateFlow.errorTargetsToNum(CreateStateFlow.ErrorTarget.ServiceSqlSyntaxError)
             ).returnValue.then { testControlFuture.complete(false) }
-            val foundExpectedException = testControlFuture.getOrThrow(30.seconds)
+            val foundExpectedException = testControlFuture.getOrThrow(waitForFlowDuration)
 
             Assert.assertTrue(foundExpectedException)
         }
@@ -224,7 +225,7 @@ class VaultObserverExceptionTest {
             assertFailsWith<TimeoutException>("PersistenceException") {
                 aliceNode.rpc.startFlow(CreateStateFlow::Initiator, "EntityManager", errorTargetsToNum(
                         CreateStateFlow.ErrorTarget.TxInvalidState))
-                        .returnValue.getOrThrow(30.seconds)
+                        .returnValue.getOrThrow(waitForFlowDuration)
             }
         }
         Assert.assertTrue("Flow has not been to hospital", counter > 0)
@@ -260,7 +261,7 @@ class VaultObserverExceptionTest {
                             CreateStateFlow.ErrorTarget.TxInvalidState,
                             CreateStateFlow.ErrorTarget.FlowSwallowErrors))
             val flowResult = flowHandle.returnValue
-            assertFailsWith<TimeoutException>("PersistenceException") { flowResult.getOrThrow(30.seconds) }
+            assertFailsWith<TimeoutException>("PersistenceException") { flowResult.getOrThrow(waitForFlowDuration) }
             Assert.assertTrue("Flow has not been to hospital", counter > 0)
         }
     }
@@ -291,7 +292,7 @@ class VaultObserverExceptionTest {
                 log.info("Flow has finished")
                 testControlFuture.set(false)
             }
-            Assert.assertTrue("Flow has not been kept in hospital", testControlFuture.getOrThrow(30.seconds))
+            Assert.assertTrue("Flow has not been kept in hospital", testControlFuture.getOrThrow(waitForFlowDuration))
         }
     }
 
@@ -310,7 +311,7 @@ class VaultObserverExceptionTest {
                     CreateStateFlow.ErrorTarget.ServiceSqlSyntaxError,
                     CreateStateFlow.ErrorTarget.ServiceSwallowErrors))
             val flowResult = flowHandle.returnValue
-            flowResult.getOrThrow(30.seconds)
+            flowResult.getOrThrow(waitForFlowDuration)
         }
     }
 
@@ -411,7 +412,7 @@ class VaultObserverExceptionTest {
                     testControlFuture.complete(true)
                 }
                 startNode(providedName = ALICE_NAME, rpcUsers = listOf(aliceUser), startInSameProcess = true).getOrThrow()
-                assert(testControlFuture.getOrThrow(30.seconds))
+                assert(testControlFuture.getOrThrow(waitForFlowDuration))
             } else {
                 throw IllegalStateException("Out of process node is still up and running!")
             }
@@ -464,7 +465,7 @@ class VaultObserverExceptionTest {
                     CreateStateFlow::Initiator,
                     "AllGood",
                     errorTargetsToNum(CreateStateFlow.ErrorTarget.ServiceSqlSyntaxErrorOnConsumed)
-                ).returnValue.getOrThrow(30.seconds)
+                ).returnValue.getOrThrow(waitForFlowDuration)
 
                 println("Created new state")
 
@@ -558,7 +559,7 @@ class VaultObserverExceptionTest {
                     "AllGood",
                     // should be a hospital exception
                     errorTargetsToNum(CreateStateFlow.ErrorTarget.ServiceSqlSyntaxErrorOnConsumed)
-                ).returnValue.getOrThrow(30.seconds)
+                ).returnValue.getOrThrow(waitForFlowDuration)
 
                 val flowHandle = aliceNode.rpc.startFlow(
                     SendStateFlow::PassErroneousOwnableState,
@@ -642,7 +643,7 @@ class VaultObserverExceptionTest {
                     CreateStateFlow::Initiator,
                     "AllGood",
                     errorTargetsToNum(CreateStateFlow.ErrorTarget.NoError)
-                ).returnValue.getOrThrow(30.seconds)
+                ).returnValue.getOrThrow(waitForFlowDuration)
 
                 aliceNode.rpc.startFlow(
                     SendStateFlow::PassErroneousOwnableState,
@@ -722,7 +723,7 @@ class VaultObserverExceptionTest {
                     CreateStateFlow::Initiator,
                     "AllGood",
                     errorTargetsToNum(CreateStateFlow.ErrorTarget.ServiceSqlSyntaxErrorOnConsumed)
-                ).returnValue.getOrThrow(30.seconds)
+                ).returnValue.getOrThrow(waitForFlowDuration)
 
                 val flowHandle = aliceNode.rpc.startFlow(
                     SendStateFlow::PassErroneousOwnableState,
@@ -777,7 +778,7 @@ class VaultObserverExceptionTest {
                 "Flow ${SubscribingRawUpdatesFlow::class.java.name} tried to access VaultService.rawUpdates " +
                         "- Rx.Observables should only be accessed outside the context of a flow "
             ) {
-                flowHandle.returnValue.getOrThrow(30.seconds)
+                flowHandle.returnValue.getOrThrow(waitForFlowDuration)
             }
         }
     }

node/src/test/kotlin/net/corda/node/services/statemachine/RetryFlowMockTest.kt

@@ -126,7 +126,9 @@ class RetryFlowMockTest {
         ReceiveFlow3.lock.release()
         assertTrue(expectedMessagesSent.await(20, TimeUnit.SECONDS))
         assertEquals(3, messagesSent.size)
-        assertNull(messagesSent.last().senderUUID)
+        // CORDA-4045: We can't be sure that the last message sent will be the last we record, so
+        // instead check we have exactly one message (the first) with sender UUID
+        assertNotNull(messagesSent.singleOrNull { it.senderUUID != null })
     }
 
     @Test(timeout=300_000)

testing/node-driver/src/main/kotlin/net/corda/testing/node/DatabaseSnapshot.kt

@@ -13,11 +13,12 @@ object DatabaseSnapshot {
         return resourceUri.openStream()
     }
 
+    fun databaseFilename(baseDirectory: Path) = baseDirectory.resolve(databaseName)
+
     fun copyDatabaseSnapshot(baseDirectory: Path) {
         getDatabaseSnapshotStream().use { stream ->
             Files.createDirectories(baseDirectory)
-            val path = baseDirectory.resolve(databaseName)
+            Files.copy(stream, databaseFilename(baseDirectory))
-            Files.copy(stream, path)
         }
     }
 }

testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/DriverDSLImpl.kt

@@ -36,6 +36,7 @@ import net.corda.core.internal.cordapp.CordappImpl.Companion.MIN_PLATFORM_VERSIO
 import net.corda.core.internal.cordapp.CordappImpl.Companion.TARGET_PLATFORM_VERSION
 import net.corda.core.internal.cordapp.get
 import net.corda.core.internal.createDirectories
+import net.corda.core.internal.deleteIfExists
 import net.corda.core.internal.div
 import net.corda.core.internal.isRegularFile
 import net.corda.core.internal.list
@@ -57,6 +58,7 @@ import net.corda.core.utilities.toHexString
 import net.corda.coretesting.internal.stubs.CertificateStoreStubs
 import net.corda.node.NodeRegistrationOption
 import net.corda.node.VersionInfo
+import net.corda.node.internal.DataSourceFactory
 import net.corda.node.internal.Node
 import net.corda.node.internal.NodeWithInfo
 import net.corda.node.internal.clientSslOptionsCompatibleWith
@@ -268,14 +270,17 @@ class DriverDSLImpl(
         val name = parameters.providedName ?: CordaX500Name("${oneOf(names).organisation}-${p2pAddress.port}", "London", "GB")
 
         val config = createConfig(name, parameters, p2pAddress)
-        if (premigrateH2Database && isH2Database(config)) {
+        if (isH2Database(config) && !inMemoryDB) {
-            if (!inMemoryDB) {
+            if (premigrateH2Database) {
                 try {
                     DatabaseSnapshot.copyDatabaseSnapshot(config.corda.baseDirectory)
                 } catch (ex: java.nio.file.FileAlreadyExistsException) {
                     log.warn("Database already exists on disk, not attempting to pre-migrate database.")
                 }
             }
+            shutdownManager.registerShutdown {
+                shutdownAndDeleteDatabase(config.corda)
+            }
         }
         val registrationFuture = if (compatibilityZone?.rootCert != null) {
             // We don't need the network map to be available to be able to register the node
@@ -1141,6 +1146,17 @@ class DriverDSLImpl(
         private fun Map<String, Any>.removeResolvedClasspath(): Map<String, Any> {
             return filterNot { it.key == "java.class.path" }
         }
+
+        private fun shutdownAndDeleteDatabase(config: NodeConfiguration) {
+            DataSourceFactory.createDataSource(config.dataSourceProperties).also { dataSource ->
+                dataSource.connection.use { connection ->
+                    connection.createStatement().use { statement ->
+                        statement.execute("SHUTDOWN")
+                    }
+                }
+            }
+            DatabaseSnapshot.databaseFilename(config.baseDirectory).deleteIfExists()
+        }
     }
 }
 

testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/InternalMockNetwork.kt

@@ -16,6 +16,7 @@ import net.corda.core.internal.NetworkParametersStorage
 import net.corda.core.internal.PLATFORM_VERSION
 import net.corda.core.internal.VisibleForTesting
 import net.corda.core.internal.createDirectories
+import net.corda.core.internal.deleteIfExists
 import net.corda.core.internal.div
 import net.corda.core.internal.notary.NotaryService
 import net.corda.core.internal.uncheckedCast
@@ -62,6 +63,7 @@ import net.corda.nodeapi.internal.network.NetworkParametersCopier
 import net.corda.nodeapi.internal.persistence.CordaPersistence
 import net.corda.nodeapi.internal.persistence.DatabaseConfig
 import net.corda.testing.common.internal.testNetworkParameters
+import net.corda.testing.node.DatabaseSnapshot
 import net.corda.testing.node.InMemoryMessagingNetwork
 import net.corda.testing.node.MockNetworkNotarySpec
 import net.corda.testing.node.MockNetworkParameters
@@ -599,7 +601,10 @@ open class InternalMockNetwork(cordappPackages: List<String> = emptyList(),
         cordappClassLoader.use { _ ->
             // Serialization env must be unset even if other parts of this method fail.
             serializationEnv.use {
-                nodes.forEach { it.started?.dispose() }
+                nodes.forEach { node ->
+                    node.started?.dispose()
+                    DatabaseSnapshot.databaseFilename(node.configuration.baseDirectory).deleteIfExists()
+                }
             }
             messagingNetwork.stop()
         }