Raft notaries can share a single key pair for the service identity (i… (#2269)

* Raft notaries can share a single key pair for the service identity (in contrast to a shared composite public key, and individual signing key pairs). This allows adjusting the cluster size on the fly.

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

@@ -12,6 +12,9 @@ import net.corda.nodeapi.internal.config.NodeSSLConfiguration
 import net.corda.nodeapi.internal.crypto.*
 import org.slf4j.LoggerFactory
 import java.nio.file.Path
+import java.security.KeyPair
+import java.security.PublicKey
+import java.security.cert.X509Certificate
 
 /**
  * Contains utility methods for generating identities for a node.
@@ -47,37 +50,56 @@ object DevIdentityGenerator {
         return identity.party
     }
 
-    fun generateDistributedNotaryIdentity(dirs: List<Path>, notaryName: CordaX500Name, threshold: Int = 1): Party {
+    fun generateDistributedNotaryCompositeIdentity(dirs: List<Path>, notaryName: CordaX500Name, threshold: Int = 1): Party {
         require(dirs.isNotEmpty())
 
-        log.trace { "Generating identity \"$notaryName\" for nodes: ${dirs.joinToString()}" }
+        log.trace { "Generating composite identity \"$notaryName\" for nodes: ${dirs.joinToString()}" }
-        val keyPairs = (1..dirs.size).map { generateKeyPair() }
-        val compositeKey = CompositeKey.Builder().addKeys(keyPairs.map { it.public }).build(threshold)
-
         val caKeyStore = loadKeyStore(javaClass.classLoader.getResourceAsStream("certificates/cordadevcakeys.jks"), "cordacadevpass")
         val intermediateCa = caKeyStore.getCertificateAndKeyPair(X509Utilities.CORDA_INTERMEDIATE_CA, "cordacadevkeypass")
-        val rootCert = caKeyStore.getCertificate(X509Utilities.CORDA_ROOT_CA)
+        val rootCert = caKeyStore.getX509Certificate(X509Utilities.CORDA_ROOT_CA)
 
+        val keyPairs = (1..dirs.size).map { generateKeyPair() }
+        val notaryKey = CompositeKey.Builder().addKeys(keyPairs.map { it.public }).build(threshold)
         keyPairs.zip(dirs) { keyPair, nodeDir ->
-            val (serviceKeyCert, compositeKeyCert) = listOf(keyPair.public, compositeKey).map { publicKey ->
+            generateCertificates(keyPair, notaryKey, intermediateCa, notaryName, nodeDir, rootCert)
-                X509Utilities.createCertificate(
-                        CertificateType.SERVICE_IDENTITY,
-                        intermediateCa.certificate,
-                        intermediateCa.keyPair,
-                        notaryName.x500Principal,
-                        publicKey)
-            }
-            val distServKeyStoreFile = (nodeDir / "certificates").createDirectories() / "distributedService.jks"
-            val keystore = loadOrCreateKeyStore(distServKeyStoreFile, "cordacadevpass")
-            keystore.setCertificateEntry("$DISTRIBUTED_NOTARY_ALIAS_PREFIX-composite-key", compositeKeyCert)
-            keystore.setKeyEntry(
-                    "$DISTRIBUTED_NOTARY_ALIAS_PREFIX-private-key",
-                    keyPair.private,
-                    "cordacadevkeypass".toCharArray(),
-                    arrayOf(serviceKeyCert, intermediateCa.certificate, rootCert))
-            keystore.save(distServKeyStoreFile, "cordacadevpass")
         }
 
-        return Party(notaryName, compositeKey)
+        return Party(notaryName, notaryKey)
+    }
+
+    fun generateDistributedNotarySingularIdentity(dirs: List<Path>, notaryName: CordaX500Name): Party {
+        require(dirs.isNotEmpty())
+
+        log.trace { "Generating singular identity \"$notaryName\" for nodes: ${dirs.joinToString()}" }
+        val caKeyStore = loadKeyStore(javaClass.classLoader.getResourceAsStream("certificates/cordadevcakeys.jks"), "cordacadevpass")
+        val intermediateCa = caKeyStore.getCertificateAndKeyPair(X509Utilities.CORDA_INTERMEDIATE_CA, "cordacadevkeypass")
+        val rootCert = caKeyStore.getX509Certificate(X509Utilities.CORDA_ROOT_CA)
+
+        val keyPair = generateKeyPair()
+        val notaryKey = keyPair.public
+        dirs.forEach { dir ->
+            generateCertificates(keyPair, notaryKey, intermediateCa, notaryName, dir, rootCert)
+        }
+        return Party(notaryName, notaryKey)
+    }
+
+    private fun generateCertificates(keyPair: KeyPair, notaryKey: PublicKey, intermediateCa: CertificateAndKeyPair, notaryName: CordaX500Name, nodeDir: Path, rootCert: X509Certificate) {
+        val (serviceKeyCert, compositeKeyCert) = listOf(keyPair.public, notaryKey).map { publicKey ->
+            X509Utilities.createCertificate(
+                    CertificateType.SERVICE_IDENTITY,
+                    intermediateCa.certificate,
+                    intermediateCa.keyPair,
+                    notaryName.x500Principal,
+                    publicKey)
+        }
+        val distServKeyStoreFile = (nodeDir / "certificates").createDirectories() / "distributedService.jks"
+        val keystore = loadOrCreateKeyStore(distServKeyStoreFile, "cordacadevpass")
+        keystore.setCertificateEntry("$DISTRIBUTED_NOTARY_ALIAS_PREFIX-composite-key", compositeKeyCert)
+        keystore.setKeyEntry(
+                "$DISTRIBUTED_NOTARY_ALIAS_PREFIX-private-key",
+                keyPair.private,
+                "cordacadevkeypass".toCharArray(),
+                arrayOf(serviceKeyCert, intermediateCa.certificate, rootCert))
+        keystore.save(distServKeyStoreFile, "cordacadevpass")
     }
 }

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

@@ -60,7 +60,7 @@ class BFTNotaryServiceTests {
         (Paths.get("config") / "currentView").deleteIfExists() // XXX: Make config object warn if this exists?
         val replicaIds = (0 until clusterSize)
 
-        notary = DevIdentityGenerator.generateDistributedNotaryIdentity(
+        notary = DevIdentityGenerator.generateDistributedNotaryCompositeIdentity(
                 replicaIds.map { mockNet.baseDirectory(mockNet.nextNodeId + it) },
                 CordaX500Name("BFT", "Zurich", "CH"))
 

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

@@ -16,11 +16,11 @@ import net.corda.node.services.Permissions.Companion.startFlow
 import net.corda.nodeapi.internal.config.User
 import net.corda.testing.*
 import net.corda.testing.driver.NodeHandle
+import net.corda.testing.driver.PortAllocation
 import net.corda.testing.driver.driver
-import net.corda.testing.node.ClusterSpec
 import net.corda.testing.node.NotarySpec
+import net.corda.testing.node.internal.DummyClusterSpec
 import org.assertj.core.api.Assertions.assertThat
-import org.junit.Ignore
 import org.junit.Test
 import rx.Observable
 import java.util.*
@@ -32,18 +32,23 @@ class DistributedServiceTests {
     private lateinit var raftNotaryIdentity: Party
     private lateinit var notaryStateMachines: Observable<Pair<Party, StateMachineUpdate>>
 
-    private fun setup(testBlock: () -> Unit) {
+    private fun setup(compositeIdentity: Boolean = false, testBlock: () -> Unit) {
         val testUser = User("test", "test", permissions = setOf(
                 startFlow<CashIssueFlow>(),
                 startFlow<CashPaymentFlow>(),
                 invokeRpc(CordaRPCOps::nodeInfo),
                 invokeRpc(CordaRPCOps::stateMachinesFeed))
         )
-
         driver(
                 extraCordappPackagesToScan = listOf("net.corda.finance.contracts"),
-                notarySpecs = listOf(NotarySpec(DUMMY_NOTARY_NAME, rpcUsers = listOf(testUser), cluster = ClusterSpec.Raft(clusterSize = 3))))
+                notarySpecs = listOf(
-        {
+                        NotarySpec(
+                                DUMMY_NOTARY_NAME,
+                                rpcUsers = listOf(testUser),
+                                cluster = DummyClusterSpec(clusterSize = 3, compositeServiceIdentity = compositeIdentity))
+                ),
+                portAllocation = PortAllocation.RandomFree
+        ) {
             alice = startNode(providedName = ALICE_NAME, rpcUsers = listOf(testUser)).getOrThrow()
             raftNotaryIdentity = defaultNotaryIdentity
             notaryNodes = defaultNotaryHandle.nodeHandles.getOrThrow().map { it as NodeHandle.OutOfProcess }
@@ -72,11 +77,60 @@ class DistributedServiceTests {
         }
     }
 
+    // TODO This should be in RaftNotaryServiceTests
+    @Test
+    fun `cluster survives if a notary is killed`() {
+        setup {
+            // Issue 100 pounds, then pay ourselves 10x5 pounds
+            issueCash(100.POUNDS)
+
+            for (i in 1..10) {
+                paySelf(5.POUNDS)
+            }
+
+            // Now kill a notary node
+            with(notaryNodes[0].process) {
+                destroy()
+                waitFor()
+            }
+
+            // Pay ourselves another 20x5 pounds
+            for (i in 1..20) {
+                paySelf(5.POUNDS)
+            }
+
+            val notarisationsPerNotary = HashMap<Party, Int>()
+            notaryStateMachines.expectEvents(isStrict = false) {
+                replicate<Pair<Party, StateMachineUpdate>>(30) {
+                    expect(match = { it.second is StateMachineUpdate.Added }) { (notary, update) ->
+                        update as StateMachineUpdate.Added
+                        notarisationsPerNotary.compute(notary) { _, number -> number?.plus(1) ?: 1 }
+                    }
+                }
+            }
+
+            println("Notarisation distribution: $notarisationsPerNotary")
+            require(notarisationsPerNotary.size == 3)
+        }
+    }
+
     // TODO Use a dummy distributed service rather than a Raft Notary Service as this test is only about Artemis' ability
     // to handle distributed services
-    @Ignore("Test has undeterministic capacity to hang, ignore till fixed")
     @Test
-    fun `requests are distributed evenly amongst the nodes`() = setup {
+    fun `requests are distributed evenly amongst the nodes`() {
+        setup {
+            checkRequestsDistributedEvenly()
+        }
+    }
+
+    @Test
+    fun `requests are distributed evenly amongst the nodes with a composite public key`() {
+        setup(true) {
+            checkRequestsDistributedEvenly()
+        }
+    }
+
+    private fun checkRequestsDistributedEvenly() {
         // Issue 100 pounds, then pay ourselves 50x2 pounds
         issueCash(100.POUNDS)
 
@@ -102,42 +156,6 @@ class DistributedServiceTests {
         require(notarisationsPerNotary.values.all { it > 10 })
     }
 
-    // TODO This should be in RaftNotaryServiceTests
-    @Ignore("Test has undeterministic capacity to hang, ignore till fixed")
-    @Test
-    fun `cluster survives if a notary is killed`() = setup {
-        // Issue 100 pounds, then pay ourselves 10x5 pounds
-        issueCash(100.POUNDS)
-
-        for (i in 1..10) {
-            paySelf(5.POUNDS)
-        }
-
-        // Now kill a notary node
-        with(notaryNodes[0].process) {
-            destroy()
-            waitFor()
-        }
-
-        // Pay ourselves another 20x5 pounds
-        for (i in 1..20) {
-            paySelf(5.POUNDS)
-        }
-
-        val notarisationsPerNotary = HashMap<Party, Int>()
-        notaryStateMachines.expectEvents(isStrict = false) {
-            replicate<Pair<Party, StateMachineUpdate>>(30) {
-                expect(match = { it.second is StateMachineUpdate.Added }) { (notary, update) ->
-                    update as StateMachineUpdate.Added
-                    notarisationsPerNotary.compute(notary) { _, number -> number?.plus(1) ?: 1 }
-                }
-            }
-        }
-
-        println("Notarisation distribution: $notarisationsPerNotary")
-        require(notarisationsPerNotary.size == 3)
-    }
-
     private fun issueCash(amount: Amount<Currency>) {
         aliceProxy.startFlow(::CashIssueFlow, amount, OpaqueBytes.of(0), raftNotaryIdentity).returnValue.getOrThrow()
     }

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

@@ -173,11 +173,11 @@ abstract class AbstractNode(val configuration: NodeConfiguration,
     }
 
     private inline fun signNodeInfo(nodeInfo: NodeInfo, sign: (PublicKey, SerializedBytes<NodeInfo>) -> DigitalSignature): SignedNodeInfo {
-        // For now we assume the node has only one identity (excluding any composite ones)
+        // For now we exclude any composite identities, see [SignedNodeInfo]
-        val owningKey = nodeInfo.legalIdentities.single { it.owningKey !is CompositeKey }.owningKey
+        val owningKeys = nodeInfo.legalIdentities.map { it.owningKey }.filter { it !is CompositeKey }
         val serialised = nodeInfo.serialize()
-        val signature = sign(owningKey, serialised)
+        val signatures = owningKeys.map { sign(it, serialised) }
-        return SignedNodeInfo(serialised, listOf(signature))
+        return SignedNodeInfo(serialised, signatures)
     }
 
     open fun generateAndSaveNodeInfo(): NodeInfo {

samples/notary-demo/src/main/kotlin/net/corda/notarydemo/BFTNotaryCordform.kt

@@ -62,7 +62,7 @@ class BFTNotaryCordform : CordformDefinition() {
     }
 
     override fun setup(context: CordformContext) {
-        DevIdentityGenerator.generateDistributedNotaryIdentity(
+        DevIdentityGenerator.generateDistributedNotaryCompositeIdentity(
                 notaryNames.map { context.baseDirectory(it.toString()) },
                 clusterName,
                 minCorrectReplicas(clusterSize)

samples/notary-demo/src/main/kotlin/net/corda/notarydemo/Notarise.kt

@@ -1,6 +1,7 @@
 package net.corda.notarydemo
 
 import net.corda.client.rpc.CordaRPCClient
+import net.corda.core.crypto.CompositeKey
 import net.corda.core.crypto.toStringShort
 import net.corda.core.identity.PartyAndCertificate
 import net.corda.core.messaging.CordaRPCOps
@@ -38,7 +39,8 @@ private class NotaryDemoClientApi(val rpc: CordaRPCOps) {
 
     /** Makes calls to the node rpc to start transaction notarisation. */
     fun notarise(count: Int) {
-        println("Notary: \"${notary.name}\", with composite key: ${notary.owningKey.toStringShort()}")
+        val keyType = if (notary.owningKey is CompositeKey) "composite" else "public"
+        println("Notary: \"${notary.name}\", with $keyType key: ${notary.owningKey.toStringShort()}")
         val transactions = buildTransactions(count)
         println("Notarised ${transactions.size} transactions:")
         transactions.zip(notariseTransactions(transactions)).forEach { (tx, signersFuture) ->

samples/notary-demo/src/main/kotlin/net/corda/notarydemo/RaftNotaryCordform.kt

@@ -58,7 +58,7 @@ class RaftNotaryCordform : CordformDefinition() {
     }
 
     override fun setup(context: CordformContext) {
-        DevIdentityGenerator.generateDistributedNotaryIdentity(
+        DevIdentityGenerator.generateDistributedNotarySingularIdentity(
                 notaryNames.map { context.baseDirectory(it.toString()) },
                 clusterName
         )

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

@@ -14,10 +14,12 @@ data class NotarySpec(
 )
 
 @DoNotImplement
-sealed class ClusterSpec {
+abstract class ClusterSpec {
     abstract val clusterSize: Int
 
-    data class Raft(override val clusterSize: Int) : ClusterSpec() {
+    data class Raft(
+            override val clusterSize: Int
+    ) : ClusterSpec() {
         init {
             require(clusterSize > 0)
         }

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

@@ -267,7 +267,7 @@ class DriverDSLImpl(
             if (cordform.notary == null) continue
             val name = CordaX500Name.parse(cordform.name)
             val notaryConfig = ConfigFactory.parseMap(cordform.notary).parseAs<NotaryConfig>()
-            // We need to first group the nodes that form part of a cluser. We assume for simplicity that nodes of the
+            // We need to first group the nodes that form part of a cluster. We assume for simplicity that nodes of the
             // same cluster type and validating flag are part of the same cluster.
             if (notaryConfig.raft != null) {
                 val key = if (notaryConfig.validating) VALIDATING_RAFT else NON_VALIDATING_RAFT
@@ -282,10 +282,17 @@ class DriverDSLImpl(
         }
 
         clusterNodes.asMap().forEach { type, nodeNames ->
-            val identity = DevIdentityGenerator.generateDistributedNotaryIdentity(
+            val identity = if (type == ClusterType.NON_VALIDATING_RAFT || type == ClusterType.VALIDATING_RAFT) {
-                    dirs = nodeNames.map { baseDirectory(it) },
+                DevIdentityGenerator.generateDistributedNotarySingularIdentity(
-                    notaryName = type.clusterName
+                        dirs = nodeNames.map { baseDirectory(it) },
-            )
+                        notaryName = type.clusterName
+                )
+            } else {
+                DevIdentityGenerator.generateDistributedNotaryCompositeIdentity(
+                        dirs = nodeNames.map { baseDirectory(it) },
+                        notaryName = type.clusterName
+                )
+            }
             notaryInfos += NotaryInfo(identity, type.validating)
         }
 
@@ -382,13 +389,30 @@ class DriverDSLImpl(
     private fun startNotaryIdentityGeneration(): CordaFuture<List<NotaryInfo>> {
         return executorService.fork {
             notarySpecs.map { spec ->
-                val identity = if (spec.cluster == null) {
+                val identity = when (spec.cluster) {
-                    DevIdentityGenerator.installKeyStoreWithNodeIdentity(baseDirectory(spec.name), spec.name)
+                    null -> {
-                } else {
+                        DevIdentityGenerator.installKeyStoreWithNodeIdentity(baseDirectory(spec.name), spec.name)
-                    DevIdentityGenerator.generateDistributedNotaryIdentity(
+                    }
-                            dirs = generateNodeNames(spec).map { baseDirectory(it) },
+                    is ClusterSpec.Raft -> {
-                            notaryName = spec.name
+                        DevIdentityGenerator.generateDistributedNotarySingularIdentity(
-                    )
+                                dirs = generateNodeNames(spec).map { baseDirectory(it) },
+                                notaryName = spec.name
+                        )
+                    }
+                    is DummyClusterSpec -> {
+                        if (spec.cluster.compositeServiceIdentity) {
+                            DevIdentityGenerator.generateDistributedNotarySingularIdentity(
+                                    dirs = generateNodeNames(spec).map { baseDirectory(it) },
+                                    notaryName = spec.name
+                            )
+                        } else {
+                            DevIdentityGenerator.generateDistributedNotaryCompositeIdentity(
+                                    dirs = generateNodeNames(spec).map { baseDirectory(it) },
+                                    notaryName = spec.name
+                            )
+                        }
+                    }
+                    else -> throw UnsupportedOperationException("Cluster spec ${spec.cluster} not supported by Driver")
                 }
                 NotaryInfo(identity, spec.validating)
             }
@@ -433,9 +457,12 @@ class DriverDSLImpl(
 
     private fun startNotaries(localNetworkMap: LocalNetworkMap?): List<CordaFuture<List<NodeHandle>>> {
         return notarySpecs.map {
-            when {
+            when (it.cluster) {
-                it.cluster == null -> startSingleNotary(it, localNetworkMap)
+                null -> startSingleNotary(it, localNetworkMap)
-                it.cluster is ClusterSpec.Raft -> startRaftNotaryCluster(it, localNetworkMap)
+                is ClusterSpec.Raft,
+                // DummyCluster is used for testing the notary communication path, and it does not matter
+                // which underlying consensus algorithm is used, so we just stick to Raft
+                is DummyClusterSpec -> startRaftNotaryCluster(it, localNetworkMap)
                 else -> throw IllegalArgumentException("BFT-SMaRt not supported")
             }
         }

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

@@ -0,0 +1,20 @@
+package net.corda.testing.node.internal
+
+import net.corda.testing.node.ClusterSpec
+
+/**
+ * Only used for testing the notary communication path. Can be configured to act as a Raft (singular identity),
+ * or a BFT (composite key identity) notary service.
+ */
+data class DummyClusterSpec(
+        override val clusterSize: Int,
+        /**
+         * If *true*, the cluster will use a shared composite public key for the service identity, with individual
+         * private keys. If *false*, the same "singular" key pair will be shared by all replicas.
+         */
+        val compositeServiceIdentity: Boolean = false
+) : ClusterSpec() {
+    init {
+        require(clusterSize > 0)
+    }
+}