CORDA-3972: Support for node identity rotation in IdentityService (#6752)

core/src/main/kotlin/net/corda/core/identity/Party.kt

@@ -4,6 +4,7 @@ import net.corda.core.KeepForDJVM
 import net.corda.core.contracts.PartyAndReference
 import net.corda.core.crypto.CompositeKey
 import net.corda.core.crypto.Crypto
+import net.corda.core.crypto.toStringShort
 import net.corda.core.flows.Destination
 import net.corda.core.flows.FlowLogic
 import net.corda.core.utilities.OpaqueBytes
@@ -43,4 +44,5 @@ class Party(val name: CordaX500Name, owningKey: PublicKey) : Destination, Abstra
     fun anonymise(): AnonymousParty = AnonymousParty(owningKey)
     override fun ref(bytes: OpaqueBytes): PartyAndReference = PartyAndReference(this, bytes)
     override fun toString() = name.toString()
+    fun description() = "$name (owningKey = ${owningKey.toStringShort()})"
 }

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

@@ -0,0 +1,197 @@
+package net.corda.node.services.identity
+
+import net.corda.core.internal.div
+import net.corda.core.utilities.OpaqueBytes
+import net.corda.coretesting.internal.stubs.CertificateStoreStubs
+import net.corda.finance.DOLLARS
+import net.corda.finance.GBP
+import net.corda.finance.POUNDS
+import net.corda.finance.USD
+import net.corda.finance.flows.CashIssueAndPaymentFlow
+import net.corda.finance.flows.CashPaymentFlow
+import net.corda.finance.workflows.getCashBalance
+import net.corda.node.services.keys.KeyManagementServiceInternal
+import net.corda.nodeapi.internal.DEV_CA_KEY_STORE_PASS
+import net.corda.nodeapi.internal.crypto.X509Utilities.NODE_IDENTITY_KEY_ALIAS
+import net.corda.nodeapi.internal.storeLegalIdentity
+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.node.internal.FINANCE_CORDAPPS
+import net.corda.testing.node.internal.InternalMockNetwork
+import net.corda.testing.node.internal.TestStartedNode
+import net.corda.testing.node.internal.startFlow
+import org.junit.After
+import org.junit.Test
+import java.nio.file.Path
+import java.security.PublicKey
+import kotlin.test.assertEquals
+import kotlin.test.assertNotEquals
+import kotlin.test.assertNull
+
+class CertificateRotationTest {
+    private val ref = OpaqueBytes.of(0x01)
+
+    private val TestStartedNode.party get() = info.legalIdentities.first()
+
+    private lateinit var mockNet: InternalMockNetwork
+
+    @After
+    fun tearDown() {
+        mockNet.stopNodes()
+    }
+
+    @Test(timeout = 300_000)
+    fun `restart with the same identities`() {
+        mockNet = InternalMockNetwork(cordappsForAllNodes = FINANCE_CORDAPPS)
+        val alice = mockNet.createPartyNode(ALICE_NAME)
+        val bob = mockNet.createPartyNode(BOB_NAME)
+
+        alice.services.startFlow(CashIssueAndPaymentFlow(300.DOLLARS, ref, alice.party, false, mockNet.defaultNotaryIdentity))
+        alice.services.startFlow(CashIssueAndPaymentFlow(1000.DOLLARS, ref, bob.party, false, mockNet.defaultNotaryIdentity))
+        bob.services.startFlow(CashIssueAndPaymentFlow(300.POUNDS, ref, bob.party, false, mockNet.defaultNotaryIdentity))
+        bob.services.startFlow(CashIssueAndPaymentFlow(1000.POUNDS, ref, alice.party, false, mockNet.defaultNotaryIdentity))
+        mockNet.runNetwork()
+
+        val alice2 = mockNet.restartNode(alice)
+        val bob2 = mockNet.restartNode(bob)
+
+        assertEquals(alice.party, alice2.party)
+        assertEquals(bob.party, bob2.party)
+        assertEquals(alice2.party, alice2.services.identityService.wellKnownPartyFromX500Name(ALICE_NAME))
+        assertEquals(bob2.party, alice2.services.identityService.wellKnownPartyFromX500Name(BOB_NAME))
+        assertEquals(alice2.party, bob2.services.identityService.wellKnownPartyFromX500Name(ALICE_NAME))
+        assertEquals(bob2.party, bob2.services.identityService.wellKnownPartyFromX500Name(BOB_NAME))
+
+        alice2.services.startFlow(CashPaymentFlow(300.DOLLARS, bob2.party, false))
+        bob2.services.startFlow(CashPaymentFlow(300.POUNDS, alice2.party, false))
+        mockNet.runNetwork()
+        bob2.services.startFlow(CashPaymentFlow(1300.DOLLARS, alice2.party, false))
+        alice2.services.startFlow(CashPaymentFlow(1300.POUNDS, bob2.party, false))
+        mockNet.runNetwork()
+
+        assertEquals(1300.DOLLARS, alice2.services.getCashBalance(USD))
+        assertEquals(0.POUNDS, alice2.services.getCashBalance(GBP))
+        assertEquals(0.DOLLARS, bob2.services.getCashBalance(USD))
+        assertEquals(1300.POUNDS, bob2.services.getCashBalance(GBP))
+    }
+
+    @Test(timeout = 300_000)
+    fun `restart with rotated key for one node`() {
+        mockNet = InternalMockNetwork(cordappsForAllNodes = FINANCE_CORDAPPS)
+        val alice = mockNet.createPartyNode(ALICE_NAME)
+        val bob = mockNet.createPartyNode(BOB_NAME)
+
+        alice.services.startFlow(CashIssueAndPaymentFlow(300.DOLLARS, ref, alice.party, false, mockNet.defaultNotaryIdentity))
+        alice.services.startFlow(CashIssueAndPaymentFlow(1000.DOLLARS, ref, bob.party, false, mockNet.defaultNotaryIdentity))
+        bob.services.startFlow(CashIssueAndPaymentFlow(300.POUNDS, ref, bob.party, false, mockNet.defaultNotaryIdentity))
+        bob.services.startFlow(CashIssueAndPaymentFlow(1000.POUNDS, ref, alice.party, false, mockNet.defaultNotaryIdentity))
+        mockNet.runNetwork()
+
+        val alice2 = mockNet.restartNodeWithRotateIdentityKey(alice)
+        val bob2 = mockNet.restartNode(bob)
+
+        assertNotEquals(alice.party, alice2.party)
+        assertEquals(bob.party, bob2.party)
+        assertEquals(alice2.party, alice2.services.identityService.wellKnownPartyFromX500Name(ALICE_NAME))
+        assertEquals(bob2.party, alice2.services.identityService.wellKnownPartyFromX500Name(BOB_NAME))
+        assertEquals(alice2.party, bob2.services.identityService.wellKnownPartyFromX500Name(ALICE_NAME))
+        assertEquals(bob2.party, bob2.services.identityService.wellKnownPartyFromX500Name(BOB_NAME))
+
+        alice2.services.startFlow(CashPaymentFlow(300.DOLLARS, bob2.party, false))
+        bob2.services.startFlow(CashPaymentFlow(300.POUNDS, alice2.party, false))
+        mockNet.runNetwork()
+        bob2.services.startFlow(CashPaymentFlow(1300.DOLLARS, alice2.party, false))
+        alice2.services.startFlow(CashPaymentFlow(1300.POUNDS, bob2.party, false))
+        mockNet.runNetwork()
+
+        assertEquals(1300.DOLLARS, alice2.services.getCashBalance(USD))
+        assertEquals(0.POUNDS, alice2.services.getCashBalance(GBP))
+        assertEquals(0.DOLLARS, bob2.services.getCashBalance(USD))
+        assertEquals(1300.POUNDS, bob2.services.getCashBalance(GBP))
+    }
+
+    @Test(timeout = 300_000)
+    fun `backchain resolution with rotated issuer key`() {
+        mockNet = InternalMockNetwork(cordappsForAllNodes = FINANCE_CORDAPPS)
+        val alice = mockNet.createPartyNode(ALICE_NAME)
+        val bob = mockNet.createPartyNode(BOB_NAME)
+
+        alice.services.startFlow(CashIssueAndPaymentFlow(1000.DOLLARS, ref, alice.party, false, mockNet.defaultNotaryIdentity))
+        mockNet.runNetwork()
+        alice.services.startFlow(CashPaymentFlow(1000.DOLLARS, bob.party, false))
+        mockNet.runNetwork()
+
+        val alice2 = mockNet.restartNodeWithRotateIdentityKey(alice)
+        val bob2 = mockNet.restartNode(bob)
+        val charlie = mockNet.createPartyNode(CHARLIE_NAME)
+
+        assertNotEquals(alice.party, alice2.party)
+        assertEquals(alice2.party, charlie.services.identityService.wellKnownPartyFromX500Name(ALICE_NAME))
+        assertEquals(bob2.party, charlie.services.identityService.wellKnownPartyFromX500Name(BOB_NAME))
+        assertEquals(charlie.party, charlie.services.identityService.wellKnownPartyFromX500Name(CHARLIE_NAME))
+
+        bob2.services.startFlow(CashPaymentFlow(1000.DOLLARS, charlie.party, false))
+        mockNet.runNetwork()
+
+        assertEquals(0.DOLLARS, alice2.services.getCashBalance(USD))
+        assertEquals(0.DOLLARS, bob2.services.getCashBalance(USD))
+        assertEquals(1000.DOLLARS, charlie.services.getCashBalance(USD))
+    }
+
+    @Test(timeout = 300_000)
+    fun `backchain resolution with issuer removed from network map`() {
+        mockNet = InternalMockNetwork(cordappsForAllNodes = FINANCE_CORDAPPS, autoVisibleNodes = false)
+        val alice = mockNet.createPartyNode(ALICE_NAME)
+        val bob = mockNet.createPartyNode(BOB_NAME)
+
+        advertiseNodesToNetwork(mockNet.defaultNotaryNode, alice, bob)
+
+        alice.services.startFlow(CashIssueAndPaymentFlow(1000.DOLLARS, ref, alice.party, false, mockNet.defaultNotaryIdentity))
+        mockNet.runNetwork()
+        alice.services.startFlow(CashPaymentFlow(1000.DOLLARS, bob.party, false))
+        mockNet.runNetwork()
+
+        bob.services.networkMapCache.clearNetworkMapCache()
+
+        val bob2 = mockNet.restartNode(bob)
+        val charlie = mockNet.createPartyNode(CHARLIE_NAME)
+
+        advertiseNodesToNetwork(mockNet.defaultNotaryNode, bob2, charlie)
+
+        assertNull(bob2.services.identityService.wellKnownPartyFromX500Name(ALICE_NAME))
+        assertNull(charlie.services.identityService.wellKnownPartyFromX500Name(ALICE_NAME))
+
+        bob2.services.startFlow(CashPaymentFlow(1000.DOLLARS, charlie.party, false))
+        mockNet.runNetwork()
+        charlie.services.startFlow(CashPaymentFlow(300.DOLLARS, bob2.party, false))
+        mockNet.runNetwork()
+
+        assertEquals(300.DOLLARS, bob2.services.getCashBalance(USD))
+        assertEquals(700.DOLLARS, charlie.services.getCashBalance(USD))
+    }
+
+    private fun InternalMockNetwork.restartNodeWithRotateIdentityKey(node: TestStartedNode): TestStartedNode {
+        val oldIdentity = rotateIdentityKey(baseDirectory(node) / "certificates")
+        val restartedNode = restartNode(node)
+        (restartedNode.services.keyManagementService as KeyManagementServiceInternal).start(listOf(oldIdentity))
+        return restartedNode
+    }
+
+    private fun rotateIdentityKey(certificatesDirectory: Path): Pair<PublicKey, String> {
+        val oldIdentityAlias = "old-identity"
+        val certStore = CertificateStoreStubs.Signing.withCertificatesDirectory(certificatesDirectory).get()
+        certStore.update {
+            val oldKey = getPrivateKey(NODE_IDENTITY_KEY_ALIAS, DEV_CA_KEY_STORE_PASS)
+            setPrivateKey(oldIdentityAlias, oldKey, getCertificateChain(NODE_IDENTITY_KEY_ALIAS), DEV_CA_KEY_STORE_PASS)
+        }
+        certStore.storeLegalIdentity(NODE_IDENTITY_KEY_ALIAS)
+        return certStore[oldIdentityAlias].publicKey to oldIdentityAlias
+    }
+
+    private fun advertiseNodesToNetwork(vararg nodes: TestStartedNode) {
+        nodes.forEach { node ->
+            nodes.forEach { node.services.networkMapCache.addOrUpdateNode(it.info) }
+        }
+    }
+}

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

@@ -572,7 +572,6 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
         }
 
         val (nodeInfo, signedNodeInfo) = nodeInfoAndSigned
-        identityService.ourNames = nodeInfo.legalIdentities.map { it.name }.toSet()
         services.start(nodeInfo, netParams)
 
         val networkParametersHotloader = if (networkMapClient == null) {

node/src/main/kotlin/net/corda/node/migration/CordaMigration.kt

@@ -6,7 +6,6 @@ import liquibase.database.Database
 import liquibase.database.jvm.JdbcConnection
 import liquibase.exception.ValidationErrors
 import liquibase.resource.ResourceAccessor
-import net.corda.core.identity.CordaX500Name
 import net.corda.core.schemas.MappedSchema
 import net.corda.node.SimpleClock
 import net.corda.node.services.identity.PersistentIdentityService
@@ -15,7 +14,6 @@ import net.corda.node.services.persistence.DBTransactionStorage
 import net.corda.node.services.persistence.NodeAttachmentService
 import net.corda.node.services.persistence.PublicKeyToTextConverter
 import net.corda.nodeapi.internal.persistence.CordaPersistence
-import net.corda.nodeapi.internal.persistence.SchemaMigration.Companion.NODE_X500_NAME
 import java.io.PrintWriter
 import java.sql.Connection
 import java.sql.SQLFeatureNotSupportedException
@@ -62,10 +60,8 @@ abstract class CordaMigration : CustomTaskChange {
         _cordaDB = createDatabase(url, cacheFactory, identityService, schema)
         cordaDB.start(dataSource)
         identityService.database = cordaDB
-        val ourName = CordaX500Name.parse(System.getProperty(NODE_X500_NAME))
 
         cordaDB.transaction {
-            identityService.ourNames = setOf(ourName)
              val dbTransactions = DBTransactionStorage(cordaDB, cacheFactory, SimpleClock(Clock.systemUTC()))
              val attachmentsService = NodeAttachmentService(metricRegistry, cacheFactory, cordaDB)
             _servicesForResolution = MigrationServicesForResolution(identityService, attachmentsService, dbTransactions, cordaDB, cacheFactory)

node/src/main/kotlin/net/corda/node/migration/MigrationNamedCacheFactory.kt

@@ -28,9 +28,10 @@ class MigrationNamedCacheFactory(private val metricRegistry: MetricRegistry?,
                     nodeConfiguration?.transactionCacheSizeBytes ?: NodeConfiguration.defaultTransactionCacheSize
             )
             "PersistentIdentityService_keyToPartyAndCert" -> caffeine.maximumSize(defaultCacheSize)
-            "PersistentIdentityService_nameToKey" -> caffeine.maximumSize(defaultCacheSize)
-            "PersistentIdentityService_keyToName" -> caffeine.maximumSize(defaultCacheSize)
-            "PersistentIdentityService_hashToKey" -> caffeine.maximumSize(defaultCacheSize)
+            "PersistentIdentityService_nameToParty" -> caffeine.maximumSize(defaultCacheSize)
+            "PersistentIdentityService_keyToParty" -> caffeine.maximumSize(defaultCacheSize)
+            "PersistentNetworkMap_nodesByKey" -> caffeine.maximumSize(defaultCacheSize)
+            "PersistentNetworkMap_idByLegalName" -> caffeine.maximumSize(defaultCacheSize)
             "BasicHSMKeyManagementService_keys" -> caffeine.maximumSize(defaultCacheSize)
             "NodeAttachmentService_attachmentContent" -> caffeine.maximumWeight(defaultCacheSize)
             "NodeAttachmentService_attachmentPresence" -> caffeine.maximumSize(defaultCacheSize)

node/src/main/kotlin/net/corda/node/migration/NodeIdentitiesNoCertMigration.kt

@@ -0,0 +1,102 @@
+package net.corda.node.migration
+
+import liquibase.change.custom.CustomTaskChange
+import liquibase.database.Database
+import liquibase.database.jvm.JdbcConnection
+import liquibase.exception.ValidationErrors
+import liquibase.resource.ResourceAccessor
+import net.corda.core.identity.PartyAndCertificate
+import net.corda.core.utilities.contextLogger
+import net.corda.nodeapi.internal.crypto.X509CertificateFactory
+import java.sql.ResultSet
+
+class NodeIdentitiesNoCertMigration : CustomTaskChange {
+    companion object {
+        private val logger = contextLogger()
+
+        const val UNRESOLVED = "Unresolved"
+    }
+
+    @Suppress("MagicNumber")
+    override fun execute(database: Database) {
+        val connection = database.connection as JdbcConnection
+
+        logger.info("Preparing to migrate node_identities_no_cert.")
+
+        val nodeKeysByHash = mutableMapOf<String, ByteArray>()
+        connection.queryAll("SELECT pk_hash, identity_value FROM node_identities") { resultSet ->
+            val hash = resultSet.getString(1)
+            val certificateBytes = resultSet.getBytes(2)
+            nodeKeysByHash[hash] = certificateBytes.toKeyBytes()
+        }
+
+        val nodeKeyHashesByName = mutableMapOf<String, String>()
+        connection.queryAll("SELECT name, pk_hash FROM node_named_identities") { resultSet ->
+            val name = resultSet.getString(1)
+            val hash = resultSet.getString(2)
+            nodeKeyHashesByName[name] = hash
+        }
+
+        logger.info("Starting to migrate node_identities_no_cert.")
+
+        var count = 0
+        connection.queryAll("SELECT pk_hash, name FROM node_identities_no_cert") { resultSet ->
+            val hash = resultSet.getString(1)
+            val name = resultSet.getString(2)
+
+            val partyKeyHash = nodeKeysByHash[hash]?.let { hash }
+                    ?: nodeKeyHashesByName[name]
+                    ?: UNRESOLVED.also { logger.warn("Unable to find party key hash for [$name] [$hash]") }
+
+            val key = nodeKeysByHash[hash]
+                    ?: connection.query("SELECT public_key FROM v_our_key_pairs WHERE public_key_hash = ?", hash)
+                    ?: connection.query("SELECT public_key FROM node_hash_to_key WHERE pk_hash = ?", hash)
+                    ?: UNRESOLVED.toByteArray().also { logger.warn("Unable to find key for [$name] [$hash]") }
+
+            connection.prepareStatement("UPDATE node_identities_no_cert SET party_pk_hash = ?, public_key = ? WHERE pk_hash = ?").use {
+                it.setString(1, partyKeyHash)
+                it.setBytes(2, key)
+                it.setString(3, hash)
+                it.executeUpdate()
+            }
+            count++
+        }
+
+        logger.info("Migrated $count node_identities_no_cert entries.")
+    }
+
+    private fun JdbcConnection.queryAll(statement: String, action: (ResultSet) -> Unit) = createStatement().use {
+        it.executeQuery(statement).use { resultSet ->
+            while (resultSet.next()) {
+                action.invoke(resultSet)
+            }
+        }
+    }
+
+    private fun JdbcConnection.query(statement: String, key: String): ByteArray? = prepareStatement(statement).use {
+        it.setString(1, key)
+        it.executeQuery().use { resultSet ->
+            if (resultSet.next()) resultSet.getBytes(1) else null
+        }
+    }
+
+    private fun ByteArray.toKeyBytes(): ByteArray {
+        val certPath = X509CertificateFactory().delegate.generateCertPath(inputStream())
+        val partyAndCertificate = PartyAndCertificate(certPath)
+        return partyAndCertificate.party.owningKey.encoded
+    }
+
+    override fun setUp() {
+    }
+
+    override fun setFileOpener(resourceAccessor: ResourceAccessor?) {
+    }
+
+    override fun getConfirmationMessage(): String? {
+        return null
+    }
+
+    override fun validate(database: Database?): ValidationErrors? {
+        return null
+    }
+}

node/src/main/kotlin/net/corda/node/migration/VaultStateMigration.kt

@@ -3,6 +3,7 @@ package net.corda.node.migration
 import liquibase.database.Database
 import net.corda.core.contracts.*
 import net.corda.core.crypto.SecureHash
+import net.corda.core.identity.CordaX500Name
 import net.corda.core.node.services.Vault
 import net.corda.core.schemas.MappedSchema
 import net.corda.core.schemas.PersistentStateRef
@@ -10,6 +11,7 @@ import net.corda.core.serialization.SerializationContext
 import net.corda.core.serialization.internal.*
 import net.corda.core.utilities.contextLogger
 import net.corda.node.internal.DBNetworkParametersStorage
+import net.corda.node.internal.schemas.NodeInfoSchemaV1
 import net.corda.node.services.identity.PersistentIdentityService
 import net.corda.node.services.keys.BasicHSMKeyManagementService
 import net.corda.node.services.persistence.DBTransactionStorage
@@ -18,6 +20,7 @@ import net.corda.node.services.vault.NodeVaultService
 import net.corda.node.services.vault.VaultSchemaV1
 import net.corda.nodeapi.internal.persistence.CordaPersistence
 import net.corda.nodeapi.internal.persistence.DatabaseTransaction
+import net.corda.nodeapi.internal.persistence.SchemaMigration
 import net.corda.nodeapi.internal.persistence.currentDBSession
 import net.corda.serialization.internal.AMQP_P2P_CONTEXT
 import net.corda.serialization.internal.AMQP_STORAGE_CONTEXT
@@ -74,13 +77,14 @@ class VaultStateMigration : CordaMigration() {
             logger.error("Cannot migrate vault states: Liquibase failed to provide a suitable database connection")
             throw VaultStateMigrationException("Cannot migrate vault states as liquibase failed to provide a suitable database connection")
         }
-        initialiseNodeServices(database, setOf(VaultMigrationSchemaV1, VaultSchemaV1))
+        initialiseNodeServices(database, setOf(VaultMigrationSchemaV1, VaultSchemaV1, NodeInfoSchemaV1))
         var statesSkipped = 0
         val persistentStates = VaultStateIterator(cordaDB)
         if (persistentStates.numStates > 0) {
             logger.warn("Found ${persistentStates.numStates} states to update from a previous version. This may take a while for large "
             + "volumes of data.")
         }
+        val ourName = CordaX500Name.parse(System.getProperty(SchemaMigration.NODE_X500_NAME))
         VaultStateIterator.withSerializationEnv {
             persistentStates.forEach {
                 val session = currentDBSession()
@@ -91,9 +95,8 @@ class VaultStateMigration : CordaMigration() {
 
                     // Can get away without checking for AbstractMethodErrors here as these will have already occurred when trying to add
                     // state parties.
-                    val myKeys = identityService.stripNotOurKeys(stateAndRef.state.data.participants.map { participant ->
-                        participant.owningKey
-                    }).toSet()
+                    val myKeys = stateAndRef.state.data.participants.map { participant -> participant.owningKey}
+                            .filter { key -> identityService.certificateFromKey(key)?.name == ourName }.toSet()
                     if (!NodeVaultService.isRelevant(stateAndRef.state.data, myKeys)) {
                         it.relevancyStatus = Vault.RelevancyStatus.NOT_RELEVANT
                     }
@@ -127,7 +130,6 @@ object VaultMigrationSchemaV1 : MappedSchema(schemaFamily = VaultMigrationSchema
         mappedTypes = listOf(
                 DBTransactionStorage.DBTransaction::class.java,
                 PersistentIdentityService.PersistentPublicKeyHashToCertificate::class.java,
-                PersistentIdentityService.PersistentPartyToPublicKeyHash::class.java,
                 PersistentIdentityService.PersistentPublicKeyHashToParty::class.java,
                 BasicHSMKeyManagementService.PersistentKey::class.java,
                 NodeAttachmentService.DBAttachment::class.java,

node/src/main/kotlin/net/corda/node/services/api/IdentityServiceInternal.kt

@@ -1,22 +1,15 @@
 package net.corda.node.services.api
 
+import net.corda.core.identity.CordaX500Name
 import net.corda.core.identity.PartyAndCertificate
 import net.corda.core.node.services.IdentityService
-import net.corda.core.utilities.contextLogger
 import java.security.InvalidAlgorithmParameterException
 import java.security.cert.CertificateExpiredException
 import java.security.cert.CertificateNotYetValidException
 
 interface IdentityServiceInternal : IdentityService {
-    private companion object {
-        val log = contextLogger()
-    }
-
-    /** This method exists so it can be mocked with doNothing, rather than having to make up a possibly invalid return value. */
-    fun justVerifyAndRegisterIdentity(identity: PartyAndCertificate, isNewRandomIdentity: Boolean = false) {
-        verifyAndRegisterIdentity(identity, isNewRandomIdentity)
-    }
-
     @Throws(CertificateExpiredException::class, CertificateNotYetValidException::class, InvalidAlgorithmParameterException::class)
-    fun verifyAndRegisterIdentity(identity: PartyAndCertificate, isNewRandomIdentity: Boolean): PartyAndCertificate?
+    fun verifyAndRegisterNewRandomIdentity(identity: PartyAndCertificate)
+
+    fun invalidateCaches(name: CordaX500Name) {}
 }

node/src/main/kotlin/net/corda/node/services/identity/InMemoryIdentityService.kt

@@ -59,8 +59,8 @@ class InMemoryIdentityService(
     }
 
     @Throws(CertificateExpiredException::class, CertificateNotYetValidException::class, InvalidAlgorithmParameterException::class)
-    override fun verifyAndRegisterIdentity(identity: PartyAndCertificate, isNewRandomIdentity: Boolean): PartyAndCertificate? {
-        return verifyAndRegisterIdentity(trustAnchor, identity)
+    override fun verifyAndRegisterNewRandomIdentity(identity: PartyAndCertificate) {
+        verifyAndRegisterIdentity(trustAnchor, identity)
     }
 
     @Throws(CertificateExpiredException::class, CertificateNotYetValidException::class, InvalidAlgorithmParameterException::class)
@@ -134,9 +134,6 @@ class InMemoryIdentityService(
             if (externalId != null) {
                 registerKeyToExternalId(publicKey, externalId)
             }
-        } else {
-            if (party.name != existingEntry) {
-            }
         }
     }
 

node/src/main/kotlin/net/corda/node/services/identity/PersistentIdentityService.kt

@@ -11,26 +11,27 @@ import net.corda.core.identity.x500Matches
 import net.corda.core.internal.CertRole
 import net.corda.core.internal.NamedCacheFactory
 import net.corda.core.internal.hash
-import net.corda.core.internal.toSet
 import net.corda.core.node.NotaryInfo
 import net.corda.core.node.services.UnknownAnonymousPartyException
 import net.corda.core.serialization.SingletonSerializeAsToken
 import net.corda.core.utilities.MAX_HASH_HEX_SIZE
 import net.corda.core.utilities.contextLogger
 import net.corda.core.utilities.debug
+import net.corda.node.internal.schemas.NodeInfoSchemaV1
 import net.corda.node.services.api.IdentityServiceInternal
-import net.corda.node.services.keys.BasicHSMKeyManagementService
 import net.corda.node.services.network.NotaryUpdateListener
 import net.corda.node.services.persistence.PublicKeyHashToExternalId
 import net.corda.node.services.persistence.WritablePublicKeyToOwningIdentityCache
 import net.corda.node.utilities.AppendOnlyPersistentMap
+import net.corda.node.utilities.NonInvalidatingCache
 import net.corda.nodeapi.internal.KeyOwningIdentity
 import net.corda.nodeapi.internal.crypto.X509CertificateFactory
 import net.corda.nodeapi.internal.crypto.X509Utilities
 import net.corda.nodeapi.internal.crypto.x509Certificates
 import net.corda.nodeapi.internal.persistence.CordaPersistence
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
-import org.apache.commons.lang3.ArrayUtils
+import net.corda.nodeapi.internal.persistence.currentDBSession
+import org.hibernate.Session
 import org.hibernate.annotations.Type
 import org.hibernate.internal.util.collections.ArrayHelper.EMPTY_BYTE_ARRAY
 import java.security.InvalidAlgorithmParameterException
@@ -47,7 +48,6 @@ import javax.annotation.concurrent.ThreadSafe
 import javax.persistence.Column
 import javax.persistence.Entity
 import javax.persistence.Id
-import kotlin.collections.HashSet
 import kotlin.streams.toList
 
 /**
@@ -61,15 +61,7 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
     companion object {
         private val log = contextLogger()
 
-        const val HASH_TO_IDENTITY_TABLE_NAME = "${NODE_DATABASE_PREFIX}identities"
-        const val NAME_TO_HASH_TABLE_NAME = "${NODE_DATABASE_PREFIX}named_identities"
-        const val KEY_TO_NAME_TABLE_NAME = "${NODE_DATABASE_PREFIX}identities_no_cert"
-        const val HASH_TO_KEY_TABLE_NAME = "${NODE_DATABASE_PREFIX}hash_to_key"
-        const val PK_HASH_COLUMN_NAME = "pk_hash"
-        const val IDENTITY_COLUMN_NAME = "identity_value"
-        const val NAME_COLUMN_NAME = "name"
-
-        fun createKeyToPartyAndCertMap(cacheFactory: NamedCacheFactory): AppendOnlyPersistentMap<String, PartyAndCertificate,
+        private fun createKeyToPartyAndCertMap(cacheFactory: NamedCacheFactory): AppendOnlyPersistentMap<String, PartyAndCertificate,
                 PersistentPublicKeyHashToCertificate, String> {
             return AppendOnlyPersistentMap(
                     cacheFactory = cacheFactory,
@@ -88,106 +80,76 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
             )
         }
 
-        fun createX500ToKeyMap(cacheFactory: NamedCacheFactory): AppendOnlyPersistentMap<CordaX500Name, String,
-                PersistentPartyToPublicKeyHash, String> {
-            return AppendOnlyPersistentMap(
-                    cacheFactory = cacheFactory,
-                    name = "PersistentIdentityService_nameToKey",
-                    toPersistentEntityKey = { it.toString() },
-                    fromPersistentEntity = {
-                        Pair(CordaX500Name.parse(it.name), it.publicKeyHash)
-                    },
-                    toPersistentEntity = { key: CordaX500Name, value: String ->
-                        PersistentPartyToPublicKeyHash(key.toString(), value)
-                    },
-                    persistentEntityClass = PersistentPartyToPublicKeyHash::class.java
-            )
+        /**
+         * Link anonymous public key to well known party (substituting well-known party public key with its hash).
+         * Public key for well-known party is linked to itself.
+         */
+        private data class KeyToParty(val publicKey: PublicKey, val name: CordaX500Name, val partyPublicKeyHash: String) {
+            constructor(party: Party, publicKey: PublicKey = party.owningKey) : this(publicKey, party.name, party.owningKey.toStringShort())
+            val party get() = Party(name, publicKey)
         }
 
-        fun createKeyToX500Map(cacheFactory: NamedCacheFactory): AppendOnlyPersistentMap<String, CordaX500Name,
+        private fun createKeyToPartyMap(cacheFactory: NamedCacheFactory): AppendOnlyPersistentMap<String, KeyToParty,
                 PersistentPublicKeyHashToParty, String> {
             return AppendOnlyPersistentMap(
                     cacheFactory = cacheFactory,
-                    name = "PersistentIdentityService_keyToName",
+                    name = "PersistentIdentityService_keyToParty",
                     toPersistentEntityKey = { it },
                     fromPersistentEntity = {
                         Pair(
                                 it.publicKeyHash,
-                                CordaX500Name.parse(it.name)
+                                KeyToParty(Crypto.decodePublicKey(it.publicKey), CordaX500Name.parse(it.name), it.partyPublicKeyHash)
                         )
                     },
-                    toPersistentEntity = { key: String, value: CordaX500Name ->
-                        PersistentPublicKeyHashToParty(key, value.toString())
+                    toPersistentEntity = { key: String, value: KeyToParty ->
+                        PersistentPublicKeyHashToParty(key, value.name.toString(), value.partyPublicKeyHash, value.publicKey.encoded)
                     },
                     persistentEntityClass = PersistentPublicKeyHashToParty::class.java)
         }
 
-        fun createHashToKeyMap(cacheFactory: NamedCacheFactory): AppendOnlyPersistentMap<String, PublicKey, PersistentHashToPublicKey,
-                String> {
-            return AppendOnlyPersistentMap(
+        private fun createNameToPartyMap(cacheFactory: NamedCacheFactory): NonInvalidatingCache<CordaX500Name, Optional<Party>> {
+            return NonInvalidatingCache(
                     cacheFactory = cacheFactory,
-                    name = "PersistentIdentityService_hashToKey",
-                    toPersistentEntityKey = { it },
-                    fromPersistentEntity = {
-                        Pair(
-                                it.publicKeyHash,
-                                Crypto.decodePublicKey(it.publicKey)
-                        )
-                    },
-                    toPersistentEntity = { key: String, value: PublicKey ->
-                        PersistentHashToPublicKey(key, value.encoded)
-                    },
-                    persistentEntityClass = PersistentHashToPublicKey::class.java)
+                    name = "PersistentIdentityService_nameToParty",
+                    loadFunction = {
+                        val result = currentDBSession().find(NodeInfoSchemaV1.DBPartyAndCertificate::class.java, it.toString())
+                        Optional.ofNullable(result?.toLegalIdentityAndCert()?.party)
+                    }
+            )
         }
 
         private fun mapToKey(party: PartyAndCertificate) = party.owningKey.toStringShort()
     }
 
     @Entity
-    @javax.persistence.Table(name = HASH_TO_IDENTITY_TABLE_NAME)
+    @javax.persistence.Table(name = "${NODE_DATABASE_PREFIX}identities")
     class PersistentPublicKeyHashToCertificate(
             @Id
-            @Column(name = PK_HASH_COLUMN_NAME, length = MAX_HASH_HEX_SIZE, nullable = false)
+            @Column(name = "pk_hash", length = MAX_HASH_HEX_SIZE, nullable = false)
             var publicKeyHash: String = "",
 
             @Type(type = "corda-blob")
-            @Column(name = IDENTITY_COLUMN_NAME, nullable = false)
+            @Column(name = "identity_value", nullable = false)
             var identity: ByteArray = EMPTY_BYTE_ARRAY
     )
 
     @Entity
-    @javax.persistence.Table(name = NAME_TO_HASH_TABLE_NAME)
-    class PersistentPartyToPublicKeyHash(
+    @javax.persistence.Table(name = "${NODE_DATABASE_PREFIX}identities_no_cert")
+    class PersistentPublicKeyHashToParty(
+            @Suppress("Unused")
             @Id
-            @Suppress("MagicNumber") // database column width
-            @Column(name = NAME_COLUMN_NAME, length = 128, nullable = false)
+            @Column(name = "pk_hash", length = MAX_HASH_HEX_SIZE, nullable = false)
+            var publicKeyHash: String = "",
+
+            @Column(name = "name", length = 128, nullable = false)
             var name: String = "",
 
-            @Column(name = PK_HASH_COLUMN_NAME, length = MAX_HASH_HEX_SIZE, nullable = false)
-            var publicKeyHash: String = ""
-    )
-
-    @Entity
-    @javax.persistence.Table(name = KEY_TO_NAME_TABLE_NAME)
-    class PersistentPublicKeyHashToParty(
-            @Id
-            @Column(name = PK_HASH_COLUMN_NAME, length = MAX_HASH_HEX_SIZE, nullable = false)
-            var publicKeyHash: String = "",
-
-            @Column(name = NAME_COLUMN_NAME, length = 128, nullable = false)
-            var name: String = ""
-    )
-
-    @Entity
-    @javax.persistence.Table(name = HASH_TO_KEY_TABLE_NAME)
-    class PersistentHashToPublicKey(
-            @Id
-            @Column(name = PK_HASH_COLUMN_NAME, length = MAX_HASH_HEX_SIZE, nullable = false)
-            var publicKeyHash: String = "",
+            @Column(name = "party_pk_hash", length = MAX_HASH_HEX_SIZE, nullable = false)
+            var partyPublicKeyHash: String = "",
 
             @Type(type = "corda-blob")
             @Column(name = "public_key", nullable = false)
-            var publicKey: ByteArray = ArrayUtils.EMPTY_BYTE_ARRAY
+            var publicKey: ByteArray = EMPTY_BYTE_ARRAY
     )
 
     private lateinit var _caCertStore: CertStore
@@ -199,6 +161,8 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
     private lateinit var _trustAnchor: TrustAnchor
     override val trustAnchor: TrustAnchor get() = _trustAnchor
 
+    private lateinit var ourParty: Party
+
     /** Stores notary identities obtained from the network parameters, for which we don't need to perform a database lookup. */
     @Volatile
     private var notaryIdentityCache = HashSet<Party>()
@@ -209,9 +173,8 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
     private lateinit var _pkToIdCache: WritablePublicKeyToOwningIdentityCache
 
     private val keyToPartyAndCert = createKeyToPartyAndCertMap(cacheFactory)
-    private val nameToKey = createX500ToKeyMap(cacheFactory)
-    private val keyToName = createKeyToX500Map(cacheFactory)
-    private val hashToKey = createHashToKeyMap(cacheFactory)
+    private val keyToParty = createKeyToPartyMap(cacheFactory)
+    private val nameToParty = createNameToPartyMap(cacheFactory)
 
     fun start(
             trustRoot: X509Certificate,
@@ -226,50 +189,38 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
         _caCertStore = CertStore.getInstance("Collection", CollectionCertStoreParameters(certificates))
         _pkToIdCache = pkToIdCache
         notaryIdentityCache.addAll(notaryIdentities)
+        ourParty = ourIdentity.party
     }
 
-    fun loadIdentities(identities: Collection<PartyAndCertificate> = emptySet(), confidentialIdentities: Collection<PartyAndCertificate> =
-            emptySet()) {
+    fun loadIdentities(identities: Collection<PartyAndCertificate>) {
         identities.forEach {
             val key = mapToKey(it)
             keyToPartyAndCert.addWithDuplicatesAllowed(key, it, false)
-            nameToKey.addWithDuplicatesAllowed(it.name, key, false)
-            keyToName.addWithDuplicatesAllowed(mapToKey(it), it.name, false)
-        }
-        confidentialIdentities.forEach {
-            keyToName.addWithDuplicatesAllowed(mapToKey(it), it.name, false)
+            keyToParty.addWithDuplicatesAllowed(it.owningKey.toStringShort(), KeyToParty(it.party), false)
+            nameToParty.asMap()[it.name] = Optional.of(it.party)
         }
         log.debug("Identities loaded")
     }
 
     @Throws(CertificateExpiredException::class, CertificateNotYetValidException::class, InvalidAlgorithmParameterException::class)
     override fun verifyAndRegisterIdentity(identity: PartyAndCertificate): PartyAndCertificate? {
-        return verifyAndRegisterIdentity(identity, false)
+        return verifyAndRegisterIdentity(trustAnchor, identity)
     }
 
     @Throws(CertificateExpiredException::class, CertificateNotYetValidException::class, InvalidAlgorithmParameterException::class)
-    override fun verifyAndRegisterIdentity(identity: PartyAndCertificate, isNewRandomIdentity: Boolean): PartyAndCertificate? {
-        return database.transaction {
-            verifyAndRegisterIdentity(trustAnchor, identity, isNewRandomIdentity)
-        }
+    override fun verifyAndRegisterNewRandomIdentity(identity: PartyAndCertificate) {
+        verifyAndRegisterIdentity(trustAnchor, identity, true)
     }
 
-    /**
-     * Verifies that an identity is valid. If it is valid, it gets registered in the database and the [PartyAndCertificate] is returned.
-     *
-     * @param trustAnchor The trust anchor that will verify the identity's validity
-     * @param identity The identity to verify
-     * @param isNewRandomIdentity true if identity will not have been registered before (e.g. because it is randomly generated by us)
-     */
     @Throws(CertificateExpiredException::class, CertificateNotYetValidException::class, InvalidAlgorithmParameterException::class)
     private fun verifyAndRegisterIdentity(trustAnchor: TrustAnchor, identity: PartyAndCertificate, isNewRandomIdentity: Boolean = false):
             PartyAndCertificate? {
-            // Validate the chain first, before we do anything clever with it
+        // Validate the chain first, before we do anything clever with it
         val identityCertChain = identity.certPath.x509Certificates
         try {
             identity.verify(trustAnchor)
         } catch (e: CertPathValidatorException) {
-        log.warn("Certificate validation failed for ${identity.name} against trusted root ${trustAnchor.trustedCert.subjectX500Principal}.")
+            log.warn("Certificate validation failed for ${identity.name} against trusted root ${trustAnchor.trustedCert.subjectX500Principal}.")
             log.warn("Certificate path :")
             identityCertChain.reversed().forEachIndexed { index, certificate ->
                 val space = (0 until index).joinToString("") { "   " }
@@ -292,38 +243,37 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
         val identityCertChain = identity.certPath.x509Certificates
         val key = mapToKey(identity)
 
-        if (isNewRandomIdentity) {
-            // Because this is supposed to be new and random, there's no way we have it in the database already, so skip the this check
-            keyToPartyAndCert[key] = identity
-            val parentId = identityCertChain[1].publicKey.toStringShort()
-            return keyToPartyAndCert[parentId]
-        } else {
-            return database.transaction {
+        return database.transaction {
+            if (isNewRandomIdentity) {
+                // Because this is supposed to be new and random, there's no way we have it in the database already, so skip the this check
+                keyToPartyAndCert[key] = identity
+                // keyToParty is already registered via KMS freshKeyInternal()
+            } else {
                 keyToPartyAndCert.addWithDuplicatesAllowed(key, identity, false)
-                nameToKey.addWithDuplicatesAllowed(identity.name, key, false)
-                keyToName.addWithDuplicatesAllowed(key, identity.name, false)
-                val parentId = identityCertChain[1].publicKey.toStringShort()
-                keyToPartyAndCert[parentId]
+                keyToParty.addWithDuplicatesAllowed(identity.owningKey.toStringShort(), KeyToParty(identity.party), false)
             }
+            val parentId = identityCertChain[1].publicKey.toStringShort()
+            keyToPartyAndCert[parentId]
         }
     }
 
+    override fun invalidateCaches(name: CordaX500Name) {
+        nameToParty.invalidate(name)
+    }
+
     override fun certificateFromKey(owningKey: PublicKey): PartyAndCertificate? = database.transaction {
         keyToPartyAndCert[owningKey.toStringShort()]
     }
 
-    override fun partyFromKey(key: PublicKey): Party? {
-        return certificateFromKey(key)?.party ?: database.transaction {
-            keyToName[key.toStringShort()]
-        }?.let { wellKnownPartyFromX500Name(it) }
-    }
-
-    private fun certificateFromCordaX500Name(name: CordaX500Name): PartyAndCertificate? {
-        return database.transaction {
-            val partyId = nameToKey[name]
-            if (partyId != null) {
-                keyToPartyAndCert[partyId]
-            } else null
+    override fun partyFromKey(key: PublicKey): Party? = database.transaction {
+        keyToParty[key.toStringShort()]?.let {
+            if (it.partyPublicKeyHash == key.toStringShort()) {
+                // Well-known party is linked to itself.
+                it.party
+            } else {
+                // Anonymous party is linked to well-known party.
+                keyToParty[it.partyPublicKeyHash]?.party
+            }
         }
     }
 
@@ -333,45 +283,47 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
             keyToPartyAndCert.allPersisted.use { it.map { it.second }.toList() }
         }
     }
+
     override fun wellKnownPartyFromX500Name(name: CordaX500Name): Party? = database.transaction {
-        certificateFromCordaX500Name(name)?.party
+        nameToParty[name]?.orElse(null)
     }
 
     override fun wellKnownPartyFromAnonymous(party: AbstractParty): Party? {
-        // Skip database lookup if the party is a notary identity.
-        // This also prevents an issue where the notary identity can't be resolved if it's not in the network map cache. The node obtains
-        // a trusted list of notary identities from the network parameters automatically.
-        return if (party is Party && party in notaryIdentityCache) {
-            party
-        } else {
-            database.transaction {
-                // Try and resolve the party from the table to public keys to party and certificates
-                // If we cannot find it then we perform a lookup on the public key to X500 name table
-                val legalIdentity = super.wellKnownPartyFromAnonymous(party)
-                if (legalIdentity == null) {
-                    // If there is no entry in the legal keyToPartyAndCert table then the party must be a confidential identity so we
-                    // perform a lookup in the keyToName table. If an entry for that public key exists, then we attempt look up the
-                    // associated node's PartyAndCertificate.
-                    val name = keyToName[party.owningKey.toStringShort()]
-                    if (name != null) {
-                        // This should never return null as this node would not be able to communicate with the node providing a
-                        // confidential identity unless its NodeInfo/PartyAndCertificate were available.
-                        wellKnownPartyFromX500Name(name)
-                    } else {
-                        null
-                    }
+        return database.transaction {
+            log.debug("Attempting to find wellKnownParty for: ${party.owningKey.toStringShort()}")
+            if (party is Party) {
+                val candidate = wellKnownPartyFromX500Name(party.name)
+                if (candidate != null && candidate != party) {
+                    // Party doesn't match existing well-known party: check that the key is registered, otherwise return null.
+                    require(party.name == candidate.name) { "Candidate party $candidate does not match expected $party" }
+                    keyToParty[party.owningKey.toStringShort()]?.let { candidate }
                 } else {
-                    legalIdentity
+                    // Party is a well-known party or well-known party doesn't exist: skip checks.
+                    // If the notary is not in the network map cache, try getting it from the network parameters
+                    // to prevent database conversion issues with vault updates (CORDA-2745).
+                    candidate ?: party.takeIf { it in notaryIdentityCache }
+                }
+            } else {
+                keyToParty[party.owningKey.toStringShort()]?.let {
+                    // Resolved party can be stale due to key rotation: always convert it to the actual well-known party.
+                    wellKnownPartyFromX500Name(it.name)
                 }
             }
         }
     }
 
+    private fun getAllCertificates(session: Session): List<NodeInfoSchemaV1.DBPartyAndCertificate> {
+        val criteria = session.criteriaBuilder.createQuery(NodeInfoSchemaV1.DBPartyAndCertificate::class.java)
+        criteria.select(criteria.from(NodeInfoSchemaV1.DBPartyAndCertificate::class.java))
+        return session.createQuery(criteria).resultList
+    }
+
     override fun partiesFromName(query: String, exactMatch: Boolean): Set<Party> {
         return database.transaction {
-            nameToKey.allPersisted.use {
-                it.filter { x500Matches(query, exactMatch, it.first) }.map { keyToPartyAndCert[it.second]!!.party }.toSet()
-            }
+            getAllCertificates(session)
+                    .map { it.toLegalIdentityAndCert() }
+                    .filter { x500Matches(query, exactMatch, it.name) }
+                    .map { it.party }.toSet()
         }
     }
 
@@ -379,34 +331,24 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
     override fun assertOwnership(party: Party, anonymousParty: AnonymousParty) = database.transaction { super.assertOwnership(party,
             anonymousParty) }
 
-    lateinit var ourNames: Set<CordaX500Name>
-
-    // Allows us to eliminate keys we know belong to others by using the cache contents that might have been seen during other identity
-    // activity. Concentrating activity on the identity cache works better than spreading checking across identity and key management,
-    // because we cache misses too.
-    fun stripNotOurKeys(keys: Iterable<PublicKey>): Iterable<PublicKey> {
-        return keys.filter { (@Suppress("DEPRECATION") certificateFromKey(it))?.name in ourNames }
-    }
-
     override fun registerKey(publicKey: PublicKey, party: Party, externalId: UUID?) {
         return database.transaction {
-            val publicKeyHash = publicKey.toStringShort()
             // EVERY key should be mapped to a Party in the "keyToName" table. Therefore if there is already a record in that table for the
             // specified key then it's either our key which has been stored prior or another node's key which we have previously mapped.
-            val existingEntryForKey = keyToName[publicKeyHash]
+            val existingEntryForKey = keyToParty[publicKey.toStringShort()]
             if (existingEntryForKey == null) {
                 // Update the three tables as necessary. We definitely store the public key and map it to a party and we optionally update
                 // the public key to external ID mapping table. This block will only ever be reached when registering keys generated on
                 // other because when a node generates its own keys "registerKeyToParty" is automatically called by
                 // KeyManagementService.freshKey.
                 registerKeyToParty(publicKey, party)
-                hashToKey[publicKeyHash] = publicKey
                 if (externalId != null) {
                     registerKeyToExternalId(publicKey, externalId)
                 }
             } else {
+                val publicKeyHash = publicKey.toStringShort()
                 log.info("An existing entry for $publicKeyHash already exists.")
-                if (party.name != existingEntryForKey) {
+                if (party.name != existingEntryForKey.name) {
                     throw IllegalStateException("The public publicKey $publicKeyHash is already assigned to a different party than the " +
                             "supplied party.")
                 }
@@ -415,11 +357,11 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
     }
 
     // Internal function used by the KMS to register a public key to a Corda Party.
-    fun registerKeyToParty(publicKey: PublicKey, party: Party) {
+    fun registerKeyToParty(publicKey: PublicKey, party: Party = ourParty) {
         return database.transaction {
             log.info("Linking: ${publicKey.hash} to ${party.name}")
-            keyToName[publicKey.toStringShort()] = party.name
-            if (party == wellKnownPartyFromX500Name(ourNames.first())) {
+            keyToParty[publicKey.toStringShort()] = KeyToParty(party, publicKey)
+            if (party == ourParty) {
                 _pkToIdCache[publicKey] = KeyOwningIdentity.UnmappedIdentity
             }
         }
@@ -434,12 +376,12 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
         return _pkToIdCache[publicKey].uuid
     }
 
-    private fun publicKeysForExternalId(externalId: UUID, table: Class<*>): List<PublicKey> {
+    override fun publicKeysForExternalId(externalId: UUID): Iterable<PublicKey> {
         return database.transaction {
             val query = session.createQuery(
                     """
                         select a.publicKey
-                        from ${table.name} a, ${PublicKeyHashToExternalId::class.java.name} b
+                        from ${PersistentPublicKeyHashToParty::class.java.name} a, ${PublicKeyHashToExternalId::class.java.name} b
                         where b.externalId = :uuid
                         and b.publicKeyHash = a.publicKeyHash
                     """,
@@ -450,16 +392,6 @@ class PersistentIdentityService(cacheFactory: NamedCacheFactory) : SingletonSeri
         }
     }
 
-    override fun publicKeysForExternalId(externalId: UUID): Iterable<PublicKey> {
-        // If the externalId was created by this node then we'll find the keys in the KMS, otherwise they'll be in the IdentityService.
-        val keys = publicKeysForExternalId(externalId, BasicHSMKeyManagementService.PersistentKey::class.java)
-        return if (keys.isEmpty()) {
-            publicKeysForExternalId(externalId, PersistentHashToPublicKey::class.java)
-        } else {
-            keys
-        }
-    }
-
     override fun onNewNotaryList(notaries: List<NotaryInfo>) {
         notaryIdentityCache = HashSet(notaries.map { it.identity })
     }

node/src/main/kotlin/net/corda/node/services/keys/BasicHSMKeyManagementService.kt

@@ -37,6 +37,7 @@ class BasicHSMKeyManagementService(
     @Entity
     @Table(name = "${NODE_DATABASE_PREFIX}our_key_pairs")
     class PersistentKey(
+            @Suppress("Unused")
             @Id
             @Column(name = "public_key_hash", length = MAX_HASH_HEX_SIZE, nullable = false)
             var publicKeyHash: String,
@@ -101,10 +102,8 @@ class BasicHSMKeyManagementService(
         database.transaction {
             keysMap[keyPair.public] = keyPair.private
             // Register the key to our identity.
-            val ourIdentity = identityService.wellKnownPartyFromX500Name(identityService.ourNames.first())
-                    ?: throw IllegalStateException("Could not lookup node Identity.")
             // No checks performed here as entries for the new key couldn't have existed before in the maps.
-            identityService.registerKeyToParty(keyPair.public, ourIdentity)
+            identityService.registerKeyToParty(keyPair.public)
             if (externalId != null) {
                 identityService.registerKeyToExternalId(keyPair.public, externalId)
             }

node/src/main/kotlin/net/corda/node/services/keys/KMSUtils.kt

@@ -46,7 +46,7 @@ fun freshCertificate(identityService: IdentityService,
     val ourCertPath = X509Utilities.buildCertPath(ourCertificate, issuer.certPath.x509Certificates)
     val anonymisedIdentity = PartyAndCertificate(ourCertPath)
     if (identityService is IdentityServiceInternal) {
-        identityService.justVerifyAndRegisterIdentity(anonymisedIdentity, true)
+        identityService.verifyAndRegisterNewRandomIdentity(anonymisedIdentity)
     } else {
         identityService.verifyAndRegisterIdentity(anonymisedIdentity)
     }

node/src/main/kotlin/net/corda/node/services/network/PersistentNetworkMapCache.kt

@@ -13,7 +13,6 @@ import net.corda.core.internal.concurrent.openFuture
 import net.corda.core.messaging.DataFeed
 import net.corda.core.node.NodeInfo
 import net.corda.core.node.NotaryInfo
-import net.corda.core.node.services.IdentityService
 import net.corda.core.node.services.NetworkMapCache.MapChange
 import net.corda.core.node.services.PartyInfo
 import net.corda.core.serialization.SingletonSerializeAsToken
@@ -22,6 +21,7 @@ import net.corda.core.utilities.NetworkHostAndPort
 import net.corda.core.utilities.contextLogger
 import net.corda.core.utilities.debug
 import net.corda.node.internal.schemas.NodeInfoSchemaV1
+import net.corda.node.services.api.IdentityServiceInternal
 import net.corda.node.services.api.NetworkMapCacheInternal
 import net.corda.node.utilities.NonInvalidatingCache
 import net.corda.nodeapi.internal.persistence.CordaPersistence
@@ -41,7 +41,8 @@ import javax.persistence.PersistenceException
 @Suppress("TooManyFunctions")
 open class PersistentNetworkMapCache(cacheFactory: NamedCacheFactory,
                                      private val database: CordaPersistence,
-                                     private val identityService: IdentityService) : NetworkMapCacheInternal, SingletonSerializeAsToken(), NotaryUpdateListener {
+                                     private val identityService: IdentityServiceInternal
+) : NetworkMapCacheInternal, SingletonSerializeAsToken(), NotaryUpdateListener {
 
     companion object {
         private val logger = contextLogger()
@@ -182,8 +183,8 @@ open class PersistentNetworkMapCache(cacheFactory: NamedCacheFactory,
                             }
                             previousNode != node -> {
                                 logger.info("Previous node was found for ${node.legalIdentities.first().name} as: ${previousNode.printWithKey()}")
-                                // TODO We should be adding any new identities as well
-                                if (verifyIdentities(node)) {
+                                // Register new identities for rotated certificates
+                                if (verifyAndRegisterIdentities(node)) {
                                     updatedNodes.add(node to previousNode)
                                 }
                             }
@@ -246,6 +247,10 @@ open class PersistentNetworkMapCache(cacheFactory: NamedCacheFactory,
                 changePublisher.onNext(change)
             }
         }
+        // Invalidate caches outside database transaction to prevent reloading of uncommitted values.
+        nodeUpdates.forEach { (nodeInfo, _) ->
+            invalidateIdentityServiceCaches(nodeInfo)
+        }
     }
 
     override fun addOrUpdateNode(node: NodeInfo) {
@@ -277,13 +282,15 @@ open class PersistentNetworkMapCache(cacheFactory: NamedCacheFactory,
     }
 
     override fun removeNode(node: NodeInfo) {
-        logger.info("Removing node with info: $node")
+        logger.info("Removing node with info: ${node.printWithKey()}")
         synchronized(_changed) {
             database.transaction {
                 removeInfoDB(session, node)
                 changePublisher.onNext(MapChange.Removed(node))
             }
         }
+        // Invalidate caches outside database transaction to prevent reloading of uncommitted values.
+        invalidateIdentityServiceCaches(node)
         logger.debug { "Done removing node with info: $node" }
     }
 
@@ -398,6 +405,10 @@ open class PersistentNetworkMapCache(cacheFactory: NamedCacheFactory,
         identityByLegalNameCache.invalidateAll(nodeInfo.legalIdentities.map { it.name })
     }
 
+    private fun invalidateIdentityServiceCaches(nodeInfo: NodeInfo) {
+        nodeInfo.legalIdentities.forEach { identityService.invalidateCaches(it.name) }
+    }
+
     private fun invalidateCaches() {
         nodesByKeyCache.invalidateAll()
         identityByLegalNameCache.invalidateAll()

node/src/main/kotlin/net/corda/node/services/schema/NodeSchemaService.kt

@@ -45,9 +45,7 @@ class NodeSchemaService(private val extraSchemas: Set<MappedSchema> = emptySet()
                     NodeAttachmentService.DBAttachment::class.java,
                     P2PMessageDeduplicator.ProcessedMessage::class.java,
                     PersistentIdentityService.PersistentPublicKeyHashToCertificate::class.java,
-                    PersistentIdentityService.PersistentPartyToPublicKeyHash::class.java,
                     PersistentIdentityService.PersistentPublicKeyHashToParty::class.java,
-                    PersistentIdentityService.PersistentHashToPublicKey::class.java,
                     ContractUpgradeServiceImpl.DBContractUpgrade::class.java,
                     DBNetworkParametersStorage.PersistentNetworkParameters::class.java,
                     PublicKeyHashToExternalId::class.java

node/src/main/kotlin/net/corda/node/services/statemachine/FlowMessaging.kt

@@ -5,7 +5,7 @@ import com.esotericsoftware.kryo.KryoException
 import net.corda.core.context.InvocationOrigin
 import net.corda.core.flows.Destination
 import net.corda.core.flows.FlowException
-import net.corda.core.identity.AnonymousParty
+import net.corda.core.identity.AbstractParty
 import net.corda.core.identity.Party
 import net.corda.core.serialization.SerializedBytes
 import net.corda.core.serialization.serialize
@@ -68,19 +68,18 @@ class FlowMessagingImpl(val serviceHub: ServiceHubInternal): FlowMessaging {
     }
 
     private fun createMessage(destination: Destination, message: SessionMessage, deduplicationId: SenderDeduplicationId): MessagingService.AddressedMessage {
-        val party = if (destination is Party) {
-            log.trace { "Sending message $deduplicationId $message to $destination" }
-            destination
+        // We assume that the destination type has already been checked by initiateFlow.
+        // Destination may point to a stale well-known identity due to key rotation, so always resolve actual identity via IdentityService.
+        val party = requireNotNull(serviceHub.identityService.wellKnownPartyFromAnonymous(destination as AbstractParty)) {
+            "We do not know who $destination belongs to"
+        }
+        if (destination == party) {
+            log.trace { "Sending message $deduplicationId $message to $party" }
         } else {
-            // We assume that the destination type has already been checked by initiateFlow
-            val wellKnown = requireNotNull(serviceHub.identityService.wellKnownPartyFromAnonymous(destination as AnonymousParty)) {
-                "We do not know who $destination belongs to"
-            }
-            log.trace { "Sending message $deduplicationId $message to $wellKnown on behalf of $destination" }
-            wellKnown
+            log.trace { "Sending message $deduplicationId $message to $party on behalf of $destination" }
         }
         val networkMessage = serviceHub.networkService.createMessage(sessionTopic, serializeSessionMessage(message).bytes, deduplicationId, message.additionalHeaders(party))
-        val partyInfo = requireNotNull(serviceHub.networkMapCache.getPartyInfo(party)) { "Don't know about $party" }
+        val partyInfo = requireNotNull(serviceHub.networkMapCache.getPartyInfo(party)) { "Don't know about ${party.description()}" }
         val address = serviceHub.networkService.getAddressOfParty(partyInfo)
         val sequenceKey = when (message) {
             is InitialSessionMessage -> message.initiatorSessionId

node/src/main/kotlin/net/corda/node/utilities/NodeNamedCache.kt

@@ -46,9 +46,8 @@ open class DefaultNamedCacheFactory protected constructor(private val metricRegi
                 name == "NodeAttachmentService_attachmentPresence" -> caffeine.maximumSize(attachmentCacheBound)
                 name == "NodeAttachmentService_contractAttachmentVersions" -> caffeine.maximumSize(defaultCacheSize)
                 name == "PersistentIdentityService_keyToPartyAndCert" -> caffeine.maximumSize(defaultCacheSize)
-                name == "PersistentIdentityService_nameToKey" -> caffeine.maximumSize(defaultCacheSize)
-                name == "PersistentIdentityService_keyToName" -> caffeine.maximumSize(defaultCacheSize)
-                name == "PersistentIdentityService_hashToKey" -> caffeine.maximumSize(defaultCacheSize)
+                name == "PersistentIdentityService_nameToParty" -> caffeine.maximumSize(defaultCacheSize)
+                name == "PersistentIdentityService_keyToParty" -> caffeine.maximumSize(defaultCacheSize)
                 name == "PersistentNetworkMap_nodesByKey" -> caffeine.maximumSize(defaultCacheSize)
                 name == "PersistentNetworkMap_idByLegalName" -> caffeine.maximumSize(defaultCacheSize)
                 name == "PersistentKeyManagementService_keys" -> caffeine.maximumSize(defaultCacheSize)

node/src/main/resources/migration/node-core.changelog-master.xml

@@ -26,6 +26,7 @@
     <include file="migration/node-core.changelog-v15.xml"/>
     <include file="migration/node-core.changelog-v14-data.xml"/>
     <include file="migration/node-core.changelog-v16.xml"/>
+    <include file="migration/node-core.changelog-v20.xml"/>
 
     <!-- This must run after node-core.changelog-init.xml, to prevent database columns being created twice. -->
     <include file="migration/vault-schema.changelog-v9.xml"/>

node/src/main/resources/migration/node-core.changelog-v20.xml

@@ -0,0 +1,54 @@
+<?xml version="1.1" encoding="UTF-8" standalone="no"?>
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.5.xsd"
+                   logicalFilePath="migration/node-services.changelog-init.xml">
+
+    <changeSet author="R3.Corda" id="identity_service_key_rotation-start" dbms="!postgresql">
+        <addColumn tableName="node_identities_no_cert">
+            <column name="party_pk_hash" type="NVARCHAR(130)">
+                <constraints nullable="true"/>
+            </column>
+            <column name="public_key" type="blob">
+                <constraints nullable="true"/>
+            </column>
+        </addColumn>
+        <createView viewName="v_our_key_pairs">
+            select public_key_hash, public_key from node_our_key_pairs
+        </createView>
+    </changeSet>
+
+    <changeSet author="R3.Corda" id="identity_service_key_rotation-start-postgres" dbms="postgresql">
+        <addColumn tableName="node_identities_no_cert">
+            <column name="party_pk_hash" type="NVARCHAR(130)">
+                <constraints nullable="true"/>
+            </column>
+            <column name="public_key" type="varbinary(64000)">
+                <constraints nullable="true"/>
+            </column>
+        </addColumn>
+        <createView viewName="v_our_key_pairs">
+            select public_key_hash, lo_get(public_key) public_key from node_our_key_pairs
+        </createView>
+    </changeSet>
+
+    <changeSet author="R3.Corda" id="identity_service_key_rotation-custom">
+        <customChange class="net.corda.node.migration.NodeIdentitiesNoCertMigration"/>
+    </changeSet>
+
+    <changeSet author="R3.Corda" id="identity_service_key_rotation-end" dbms="!postgresql">
+        <addNotNullConstraint tableName="node_identities_no_cert" columnName="party_pk_hash" columnDataType="NVARCHAR(130)"/>
+        <addNotNullConstraint tableName="node_identities_no_cert" columnName="public_key" columnDataType="blob"/>
+        <dropTable tableName="node_hash_to_key"/>
+        <dropTable tableName="node_named_identities"/>
+        <dropView viewName="v_our_key_pairs"/>
+    </changeSet>
+
+    <changeSet author="R3.Corda" id="identity_service_key_rotation-end-postgres" dbms="postgresql">
+        <addNotNullConstraint tableName="node_identities_no_cert" columnName="party_pk_hash" columnDataType="NVARCHAR(130)"/>
+        <addNotNullConstraint tableName="node_identities_no_cert" columnName="public_key" columnDataType="varbinary(64000)"/>
+        <dropTable tableName="node_hash_to_key"/>
+        <dropTable tableName="node_named_identities"/>
+        <dropView viewName="v_our_key_pairs"/>
+    </changeSet>
+</databaseChangeLog>

node/src/test/kotlin/net/corda/node/migration/IdenityServiceKeyRotationMigrationTest.kt

@@ -0,0 +1,160 @@
+package net.corda.node.migration
+
+import com.nhaarman.mockito_kotlin.doReturn
+import com.nhaarman.mockito_kotlin.whenever
+import liquibase.Contexts
+import liquibase.Liquibase
+import liquibase.database.Database
+import liquibase.database.core.H2Database
+import liquibase.database.jvm.JdbcConnection
+import liquibase.resource.ClassLoaderResourceAccessor
+import net.corda.core.crypto.Crypto
+import net.corda.core.crypto.toStringShort
+import net.corda.core.identity.CordaX500Name
+import net.corda.core.identity.Party
+import net.corda.core.identity.PartyAndCertificate
+import net.corda.coretesting.internal.rigorousMock
+import net.corda.node.migration.NodeIdentitiesNoCertMigration.Companion.UNRESOLVED
+import net.corda.node.services.api.SchemaService
+import net.corda.nodeapi.internal.crypto.CertificateType
+import net.corda.nodeapi.internal.crypto.X509Utilities
+import net.corda.nodeapi.internal.crypto.x509Certificates
+import net.corda.nodeapi.internal.persistence.CordaPersistence
+import net.corda.nodeapi.internal.persistence.DatabaseConfig
+import net.corda.nodeapi.internal.persistence.contextTransactionOrNull
+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.TestIdentity
+import net.corda.testing.internal.configureDatabase
+import net.corda.testing.node.MockServices
+import org.junit.After
+import org.junit.Before
+import org.junit.Test
+import kotlin.test.assertEquals
+
+class IdenityServiceKeyRotationMigrationTest {
+    private lateinit var liquibaseDB: Database
+    private lateinit var cordaDB: CordaPersistence
+
+    @Before
+    fun setUp() {
+        val schemaService = rigorousMock<SchemaService>()
+        doReturn(setOf(IdentityTestSchemaV1, KMSTestSchemaV1)).whenever(schemaService).schemas
+
+        cordaDB = configureDatabase(
+                MockServices.makeTestDataSourceProperties(),
+                DatabaseConfig(),
+                { null },
+                { null },
+                schemaService = schemaService,
+                internalSchemas = setOf(),
+                ourName = ALICE_NAME)
+        liquibaseDB = H2Database()
+        liquibaseDB.connection = JdbcConnection(cordaDB.dataSource.connection)
+        liquibaseDB.isAutoCommit = true
+    }
+
+    @After
+    fun close() {
+        contextTransactionOrNull?.close()
+        cordaDB.close()
+        liquibaseDB.close()
+    }
+
+    private fun persist(vararg entries: Any) = cordaDB.transaction {
+        entries.forEach { session.persist(it) }
+    }
+
+    private fun PartyAndCertificate.dbCertificate() = IdentityTestSchemaV1.NodeIdentities(owningKey.toStringShort(), certPath.encoded)
+
+    private fun Party.dbParty() = IdentityTestSchemaV1.NodeIdentitiesNoCert(owningKey.toStringShort(), name.toString())
+
+    private fun TestIdentity.dbName() = IdentityTestSchemaV1.NodeNamedIdentities(name.toString(), publicKey.toStringShort())
+
+    private fun TestIdentity.dbKey() = IdentityTestSchemaV1.NodeHashToKey(publicKey.toStringShort(), publicKey.encoded)
+
+    private fun TestIdentity.dbKeyPair() =
+            KMSTestSchemaV1.PersistentKey(publicKey.toStringShort(), publicKey.encoded, keyPair.private.encoded)
+
+    private fun TestIdentity.generateConfidentialIdentityWithCert(): PartyAndCertificate {
+        val certificate = X509Utilities.createCertificate(
+                CertificateType.CONFIDENTIAL_LEGAL_IDENTITY,
+                identity.certificate,
+                keyPair,
+                name.x500Principal,
+                Crypto.generateKeyPair().public)
+        return PartyAndCertificate(X509Utilities.buildCertPath(certificate, identity.certPath.x509Certificates))
+    }
+
+    @Test(timeout = 300_000)
+    fun `test migration`() {
+        val alice = TestIdentity(ALICE_NAME, 70)
+        val bob = TestIdentity(BOB_NAME, 80)
+        val charlie = TestIdentity(CHARLIE_NAME, 90)
+
+        val alice2 = TestIdentity(ALICE_NAME, 71)
+        val alice3 = TestIdentity(ALICE_NAME, 72)
+        val aliceCiWithCert = alice.generateConfidentialIdentityWithCert()
+
+        val bob2 = TestIdentity(BOB_NAME, 81)
+        val bob3 = TestIdentity(BOB_NAME, 82)
+
+        val charlie2 = TestIdentity(CHARLIE_NAME, 91)
+        val charlie3 = TestIdentity(CHARLIE_NAME, 92)
+        val charlieCiWithCert = charlie.generateConfidentialIdentityWithCert()
+
+        persist(alice.identity.dbCertificate(), alice.party.dbParty(), alice.dbName())
+        persist(charlie.identity.dbCertificate(), charlie.party.dbParty())
+        persist(bob.identity.dbCertificate(), bob.party.dbParty(), bob.dbName())
+
+        persist(alice2.party.dbParty(), alice2.dbKeyPair())
+        persist(alice3.party.dbParty())
+        persist(aliceCiWithCert.dbCertificate(), aliceCiWithCert.party.dbParty())
+
+        persist(bob2.party.dbParty(), bob2.dbKey())
+        persist(bob3.party.dbParty())
+
+        persist(charlie2.party.dbParty(), charlie2.dbKey())
+        persist(charlie3.party.dbParty())
+        persist(charlieCiWithCert.dbCertificate(), charlieCiWithCert.party.dbParty())
+
+        Liquibase("migration/node-core.changelog-v20.xml", object : ClassLoaderResourceAccessor() {
+            override fun getResourcesAsStream(path: String) = super.getResourcesAsStream(path)?.firstOrNull()?.let { setOf(it) }
+        }, liquibaseDB).update(Contexts().toString())
+
+        val dummyKey = Crypto.generateKeyPair().public
+        val results = mutableMapOf<String, List<*>>()
+
+        cordaDB.transaction {
+            connection.prepareStatement("SELECT pk_hash, name, party_pk_hash, public_key FROM node_identities_no_cert").use { ps ->
+                ps.executeQuery().use { rs ->
+                    while (rs.next()) {
+                        val partyKeyHash = rs.getString(3).takeUnless { it == UNRESOLVED } ?: dummyKey.toStringShort()
+                        val key = if (UNRESOLVED.toByteArray().contentEquals(rs.getBytes(4))) {
+                            dummyKey
+                        } else {
+                            Crypto.decodePublicKey(rs.getBytes(4))
+                        }
+                        results[rs.getString(1)] = listOf(key, CordaX500Name.parse(rs.getString(2)), partyKeyHash)
+                    }
+                }
+            }
+        }
+
+        assertEquals(11, results.size)
+
+        listOf(alice.identity, bob.identity, charlie.identity, aliceCiWithCert, charlieCiWithCert).forEach {
+            assertEquals(results[it.owningKey.toStringShort()], listOf(it.owningKey, it.party.name, it.owningKey.toStringShort()))
+        }
+
+        assertEquals(results[alice2.publicKey.toStringShort()], listOf(alice2.publicKey, ALICE_NAME, alice.publicKey.toStringShort()))
+        assertEquals(results[alice3.publicKey.toStringShort()], listOf(dummyKey, ALICE_NAME, alice.publicKey.toStringShort()))
+
+        assertEquals(results[bob2.publicKey.toStringShort()], listOf(bob2.publicKey, BOB_NAME, bob.publicKey.toStringShort()))
+        assertEquals(results[bob3.publicKey.toStringShort()], listOf(dummyKey, BOB_NAME, bob.publicKey.toStringShort()))
+
+        assertEquals(results[charlie2.publicKey.toStringShort()], listOf(charlie2.publicKey, CHARLIE_NAME, dummyKey.toStringShort()))
+        assertEquals(results[charlie3.publicKey.toStringShort()], listOf(dummyKey, CHARLIE_NAME, dummyKey.toStringShort()))
+    }
+}

node/src/test/kotlin/net/corda/node/migration/MigrationTestSchema.kt

@@ -7,6 +7,7 @@ import org.hibernate.annotations.Type
 import javax.persistence.Column
 import javax.persistence.Entity
 import javax.persistence.Id
+import javax.persistence.Lob
 import javax.persistence.Table
 
 object MigrationTestSchema
@@ -75,4 +76,28 @@ object IdentityTestSchemaV1 : MappedSchema(
             @Column(name = "public_key", nullable = false)
             var publicKey: ByteArray = ArrayUtils.EMPTY_BYTE_ARRAY
     )
+}
+
+object KMSTestSchemaV1 : MappedSchema(
+        schemaFamily = MigrationTestSchema::class.java,
+        version = 1,
+        mappedTypes = listOf(
+                PersistentKey::class.java
+        )
+) {
+    @Entity
+    @Table(name = "node_our_key_pairs")
+    class PersistentKey(
+            @Id
+            @Column(name = "public_key_hash", length = MAX_HASH_HEX_SIZE, nullable = false)
+            var publicKeyHash: String,
+
+            @Lob
+            @Column(name = "public_key", nullable = false)
+            var publicKey: ByteArray = ArrayUtils.EMPTY_BYTE_ARRAY,
+
+            @Lob
+            @Column(name = "private_key", nullable = false)
+            var privateKey: ByteArray = ArrayUtils.EMPTY_BYTE_ARRAY
+    )
 }

node/src/test/kotlin/net/corda/node/migration/VaultStateMigrationTest.kt

@@ -26,6 +26,7 @@ import net.corda.finance.contracts.asset.Obligation
 import net.corda.finance.contracts.asset.OnLedgerAsset
 import net.corda.finance.schemas.CashSchemaV1
 import net.corda.node.internal.DBNetworkParametersStorage
+import net.corda.node.internal.schemas.NodeInfoSchemaV1
 import net.corda.node.services.identity.PersistentIdentityService
 import net.corda.node.services.keys.BasicHSMKeyManagementService
 import net.corda.node.services.persistence.DBTransactionStorage
@@ -194,11 +195,12 @@ class VaultStateMigrationTest {
 
     private fun saveAllIdentities(identities: List<PartyAndCertificate>) {
         cordaDB.transaction {
-            identities.groupBy { it.name }.forEach { name, certs ->
+            identities.groupBy { it.name }.forEach { (_, certs) ->
                 val persistentIDs = certs.map { PersistentIdentityService.PersistentPublicKeyHashToCertificate(it.owningKey.toStringShort(), it.certPath.encoded) }
-                val persistentName = PersistentIdentityService.PersistentPartyToPublicKeyHash(name.toString(), certs.first().owningKey.toStringShort())
                 persistentIDs.forEach { session.save(it) }
-                session.save(persistentName)
+                val networkIdentity = NodeInfoSchemaV1.DBPartyAndCertificate(certs.first(), true)
+                val persistentNodeInfo = NodeInfoSchemaV1.PersistentNodeInfo(0, "", listOf(), listOf(networkIdentity), 0, 0)
+                session.save(persistentNodeInfo)
             }
         }
     }

node/src/test/kotlin/net/corda/node/services/identity/PersistentIdentityServiceTests.kt

@@ -6,10 +6,13 @@ import net.corda.core.identity.AnonymousParty
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.identity.Party
 import net.corda.core.identity.PartyAndCertificate
+import net.corda.core.node.NodeInfo
 import net.corda.core.node.services.UnknownAnonymousPartyException
 import net.corda.core.serialization.serialize
+import net.corda.core.utilities.NetworkHostAndPort
 import net.corda.coretesting.internal.DEV_INTERMEDIATE_CA
 import net.corda.coretesting.internal.DEV_ROOT_CA
+import net.corda.node.services.network.PersistentNetworkMapCache
 import net.corda.node.services.persistence.PublicKeyToOwningIdentityCacheImpl
 import net.corda.nodeapi.internal.crypto.CertificateType
 import net.corda.nodeapi.internal.crypto.X509Utilities
@@ -18,13 +21,13 @@ import net.corda.nodeapi.internal.persistence.CordaPersistence
 import net.corda.nodeapi.internal.persistence.DatabaseConfig
 import net.corda.testing.core.ALICE_NAME
 import net.corda.testing.core.BOB_NAME
+import net.corda.testing.core.DUMMY_NOTARY_NAME
 import net.corda.testing.core.SerializationEnvironmentRule
 import net.corda.testing.core.TestIdentity
 import net.corda.testing.core.getTestPartyAndCertificate
 import net.corda.testing.internal.TestingNamedCacheFactory
 import net.corda.testing.internal.configureDatabase
 import net.corda.testing.node.MockServices.Companion.makeTestDataSourceProperties
-import net.corda.testing.node.makeTestIdentityService
 import org.junit.After
 import org.junit.Before
 import org.junit.Rule
@@ -38,6 +41,7 @@ class PersistentIdentityServiceTests {
     private companion object {
         val alice = TestIdentity(ALICE_NAME, 70)
         val bob = TestIdentity(BOB_NAME, 80)
+        val notary = TestIdentity(DUMMY_NOTARY_NAME, 90)
         val ALICE get() = alice.party
         val ALICE_IDENTITY get() = alice.identity
         val ALICE_PUBKEY get() = alice.publicKey
@@ -52,6 +56,7 @@ class PersistentIdentityServiceTests {
     private val cacheFactory = TestingNamedCacheFactory()
     private lateinit var database: CordaPersistence
     private lateinit var identityService: PersistentIdentityService
+    private lateinit var networkMapCache: PersistentNetworkMapCache
 
     @Before
     fun setup() {
@@ -64,11 +69,13 @@ class PersistentIdentityServiceTests {
                 identityService::wellKnownPartyFromAnonymous
         )
         identityService.database = database
-        identityService.ourNames = setOf(ALICE_NAME)
-        identityService.start(DEV_ROOT_CA.certificate, alice.identity, pkToIdCache = PublicKeyToOwningIdentityCacheImpl(
-                database,
-                cacheFactory
-        ))
+        identityService.start(
+                DEV_ROOT_CA.certificate,
+                alice.identity,
+                listOf(notary.party),
+                PublicKeyToOwningIdentityCacheImpl(database, cacheFactory)
+        )
+        networkMapCache = PersistentNetworkMapCache(cacheFactory, database, identityService)
     }
 
     @After
@@ -97,7 +104,7 @@ class PersistentIdentityServiceTests {
     @Test(timeout=300_000)
 	fun `get identity by key`() {
         assertNull(identityService.partyFromKey(ALICE_PUBKEY))
-        identityService.verifyAndRegisterIdentity(ALICE_IDENTITY)
+        networkMapCache.verifyAndRegisterIdentity(ALICE_IDENTITY)
         assertEquals(ALICE, identityService.partyFromKey(ALICE_PUBKEY))
         assertNull(identityService.partyFromKey(BOB_PUBKEY))
     }
@@ -107,31 +114,12 @@ class PersistentIdentityServiceTests {
         assertNull(identityService.wellKnownPartyFromX500Name(ALICE.name))
     }
 
-    @Test(timeout=300_000)
-	fun `stripping others when none registered strips`() {
-        assertEquals(identityService.stripNotOurKeys(listOf(BOB_PUBKEY)).firstOrNull(), null)
-    }
-
-    @Test(timeout=300_000)
-	fun `stripping others when only us registered strips`() {
-        identityService.verifyAndRegisterIdentity(ALICE_IDENTITY)
-        assertEquals(identityService.stripNotOurKeys(listOf(BOB_PUBKEY)).firstOrNull(), null)
-    }
-
-    @Test(timeout=300_000)
-	fun `stripping others when us and others registered does not strip us`() {
-        identityService.verifyAndRegisterIdentity(ALICE_IDENTITY)
-        identityService.verifyAndRegisterIdentity(BOB_IDENTITY)
-        val stripped = identityService.stripNotOurKeys(listOf(ALICE_PUBKEY, BOB_PUBKEY))
-        assertEquals(stripped.single(), ALICE_PUBKEY)
-    }
-
     @Test(timeout=300_000)
 	fun `get identity by substring match`() {
-        identityService.verifyAndRegisterIdentity(ALICE_IDENTITY)
-        identityService.verifyAndRegisterIdentity(BOB_IDENTITY)
+        networkMapCache.verifyAndRegisterIdentity(ALICE_IDENTITY)
+        networkMapCache.verifyAndRegisterIdentity(BOB_IDENTITY)
         val alicente = getTestPartyAndCertificate(CordaX500Name(organisation = "Alicente Worldwide", locality = "London", country = "GB"), generateKeyPair().public)
-        identityService.verifyAndRegisterIdentity(alicente)
+        networkMapCache.verifyAndRegisterIdentity(alicente)
         assertEquals(setOf(ALICE, alicente.party), identityService.partiesFromName("Alice", false))
         assertEquals(setOf(ALICE), identityService.partiesFromName("Alice Corp", true))
         assertEquals(setOf(BOB), identityService.partiesFromName("Bob Plc", true))
@@ -143,7 +131,7 @@ class PersistentIdentityServiceTests {
                 .map { getTestPartyAndCertificate(CordaX500Name(organisation = it, locality = "London", country = "GB"), generateKeyPair().public) }
         assertNull(identityService.wellKnownPartyFromX500Name(identities.first().name))
         identities.forEach {
-            identityService.verifyAndRegisterIdentity(it)
+            networkMapCache.verifyAndRegisterIdentity(it)
         }
         identities.forEach {
             assertEquals(it.party, identityService.wellKnownPartyFromX500Name(it.name))
@@ -224,8 +212,8 @@ class PersistentIdentityServiceTests {
         val (bob, anonymousBob) = createConfidentialIdentity(BOB.name)
 
         // Register well known identities
-        identityService.verifyAndRegisterIdentity(alice)
-        identityService.verifyAndRegisterIdentity(bob)
+        networkMapCache.verifyAndRegisterIdentity(alice)
+        networkMapCache.verifyAndRegisterIdentity(bob)
         // Register an anonymous identities
         identityService.verifyAndRegisterIdentity(anonymousAlice)
         identityService.verifyAndRegisterIdentity(anonymousBob)
@@ -270,7 +258,7 @@ class PersistentIdentityServiceTests {
     @Test(timeout=300_000)
 	fun `resolve key to party for key without certificate`() {
         // Register Alice's PartyAndCert as if it was done so via the network map cache.
-        identityService.verifyAndRegisterIdentity(alice.identity)
+        networkMapCache.verifyAndRegisterIdentity(alice.identity)
         // Use a key which is not tied to a cert.
         val publicKey = Crypto.generateKeyPair().public
         // Register the PublicKey to Alice's CordaX500Name.
@@ -280,7 +268,7 @@ class PersistentIdentityServiceTests {
 
     @Test(timeout=300_000)
 	fun `register incorrect party to public key `(){
-        database.transaction { identityService.verifyAndRegisterIdentity(ALICE_IDENTITY) }
+        networkMapCache.verifyAndRegisterIdentity(ALICE_IDENTITY)
         val (alice, anonymousAlice) = createConfidentialIdentity(ALICE.name)
         identityService.registerKey(anonymousAlice.owningKey, alice.party)
         // Should have no side effect but logs a warning that we tried to overwrite an existing mapping.
@@ -309,15 +297,18 @@ class PersistentIdentityServiceTests {
         return Pair(issuer, PartyAndCertificate(txCertPath))
     }
 
+    private fun PersistentNetworkMapCache.verifyAndRegisterIdentity(identity: PartyAndCertificate) {
+        addOrUpdateNode(NodeInfo(listOf(NetworkHostAndPort("localhost", 12345)), listOf(identity), 1, 0))
+    }
+
     /**
      * Ensure if we feed in a full identity, we get the same identity back.
      */
     @Test(timeout=300_000)
 	fun `deanonymising a well known identity should return the identity`() {
-        val service = makeTestIdentityService()
         val expected = ALICE
-        service.verifyAndRegisterIdentity(ALICE_IDENTITY)
-        val actual = service.wellKnownPartyFromAnonymous(expected)
+        networkMapCache.verifyAndRegisterIdentity(ALICE_IDENTITY)
+        val actual = identityService.wellKnownPartyFromAnonymous(expected)
         assertEquals(expected, actual)
     }
 
@@ -326,10 +317,37 @@ class PersistentIdentityServiceTests {
      */
     @Test(timeout=300_000)
 	fun `deanonymising a false well known identity should return null`() {
-        val service = makeTestIdentityService()
         val notAlice = Party(ALICE.name, generateKeyPair().public)
-        service.verifyAndRegisterIdentity(ALICE_IDENTITY)
-        val actual = service.wellKnownPartyFromAnonymous(notAlice)
+        networkMapCache.verifyAndRegisterIdentity(ALICE_IDENTITY)
+        val actual = identityService.wellKnownPartyFromAnonymous(notAlice)
         assertNull(actual)
     }
+
+    @Test(timeout = 300_000)
+    fun `rotate identity`() {
+        networkMapCache.verifyAndRegisterIdentity(ALICE_IDENTITY)
+        val anonymousParty = AnonymousParty(generateKeyPair().public)
+        identityService.registerKeyToParty(anonymousParty.owningKey)
+        assertEquals(ALICE, identityService.partyFromKey(anonymousParty.owningKey))
+
+        assertEquals(ALICE, identityService.wellKnownPartyFromAnonymous(anonymousParty))
+        assertEquals(ALICE, identityService.wellKnownPartyFromAnonymous(ALICE))
+        assertEquals(ALICE, identityService.wellKnownPartyFromX500Name(ALICE.name))
+
+        // Make sure that registration of CI with certificate doesn't disrupt well-known party resolution.
+        val (_, anonymousIdentityWithCert) = createConfidentialIdentity(ALICE.name)
+        identityService.verifyAndRegisterIdentity(anonymousIdentityWithCert)
+        assertEquals(ALICE, identityService.wellKnownPartyFromAnonymous(anonymousParty))
+        assertEquals(ALICE, identityService.wellKnownPartyFromAnonymous(ALICE))
+        assertEquals(ALICE, identityService.wellKnownPartyFromX500Name(ALICE.name))
+
+        val alice2 = getTestPartyAndCertificate(ALICE.name, generateKeyPair().public)
+        networkMapCache.verifyAndRegisterIdentity(alice2)
+        assertEquals(alice2.party, identityService.wellKnownPartyFromAnonymous(anonymousParty))
+        assertEquals(alice2.party, identityService.wellKnownPartyFromAnonymous(ALICE))
+        assertEquals(alice2.party, identityService.wellKnownPartyFromX500Name(ALICE.name))
+        assertEquals(alice2.party, identityService.wellKnownPartyFromAnonymous(alice2.party))
+
+        assertEquals(setOf(alice2.party), identityService.partiesFromName("Alice Corp", true))
+    }
 }

node/src/test/kotlin/net/corda/node/services/keys/FilterMyKeysTests.kt

@@ -3,12 +3,18 @@ package net.corda.node.services.keys
 import net.corda.core.crypto.Crypto
 import net.corda.core.identity.CordaX500Name
 import net.corda.testing.common.internal.testNetworkParameters
+import net.corda.testing.core.SerializationEnvironmentRule
 import net.corda.testing.core.TestIdentity
 import net.corda.testing.node.MockServices
+import org.junit.Rule
 import org.junit.Test
 import kotlin.test.assertEquals
 
 class FilterMyKeysTests {
+    @Rule
+    @JvmField
+    val testSerialization = SerializationEnvironmentRule()
+
     @Test(timeout=300_000)
 	fun test() {
         val name = CordaX500Name("Roger", "Office", "GB")

node/src/test/kotlin/net/corda/node/services/persistence/PublicKeyToOwningIdentityCacheImplTest.kt

@@ -9,11 +9,13 @@ import net.corda.nodeapi.internal.persistence.CordaPersistence
 import net.corda.nodeapi.internal.persistence.withoutDatabaseAccess
 import net.corda.testing.common.internal.testNetworkParameters
 import net.corda.testing.core.ALICE_NAME
+import net.corda.testing.core.SerializationEnvironmentRule
 import net.corda.testing.core.TestIdentity
 import net.corda.testing.internal.TestingNamedCacheFactory
 import net.corda.testing.node.MockServices
 import org.junit.After
 import org.junit.Before
+import org.junit.Rule
 import org.junit.Test
 import java.security.PublicKey
 import java.util.*
@@ -21,6 +23,9 @@ import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors
 
 class PublicKeyToOwningIdentityCacheImplTest {
+    @Rule
+    @JvmField
+    val testSerialization = SerializationEnvironmentRule()
 
     private lateinit var database: CordaPersistence
     private lateinit var testCache: PublicKeyToOwningIdentityCacheImpl

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

@@ -11,6 +11,7 @@ import net.corda.core.flows.StateMachineRunId
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.identity.Party
 import net.corda.core.identity.PartyAndCertificate
+import net.corda.core.internal.PLATFORM_VERSION
 import net.corda.core.messaging.DataFeed
 import net.corda.core.messaging.FlowHandle
 import net.corda.core.messaging.FlowProgressHandle
@@ -41,6 +42,7 @@ import net.corda.nodeapi.internal.persistence.contextTransaction
 import net.corda.testing.common.internal.testNetworkParameters
 import net.corda.testing.core.TestIdentity
 import net.corda.coretesting.internal.DEV_ROOT_CA
+import net.corda.node.services.network.PersistentNetworkMapCache
 import net.corda.testing.internal.MockCordappProvider
 import net.corda.testing.internal.TestingNamedCacheFactory
 import net.corda.testing.internal.configureDatabase
@@ -185,11 +187,14 @@ open class MockServices private constructor(
 
             // Create a persistent identity service and add all the supplied identities.
             identityService.apply {
-                ourNames = setOf(initialIdentity.name)
                 database = persistence
                 start(DEV_ROOT_CA.certificate, initialIdentity.identity, pkToIdCache = pkToIdCache)
                 persistence.transaction { identityService.loadIdentities(moreIdentities + initialIdentity.identity) }
             }
+            val networkMapCache = PersistentNetworkMapCache(cacheFactory, persistence, identityService)
+            (moreIdentities + initialIdentity.identity).forEach {
+                networkMapCache.addOrUpdateNode(NodeInfo(listOf(NetworkHostAndPort("localhost", 0)), listOf(it), PLATFORM_VERSION, 0))
+            }
 
             // Create a persistent key management service and add the key pair which was created for the TestIdentity.
             // We only add the keypair for the initial identity and any other keys which this node may control. Note: We don't add the keys