Persistent Identity

Fixup tests after rebase

Add unit tests of Persistent Identity. Fix bugs in PersistentMap.

Wrap identity and network map RPC calls in database transaction

Address PR comments

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

@@ -34,7 +34,7 @@ import net.corda.node.services.config.NodeConfiguration
 import net.corda.node.services.config.configureWithDevSSLCertificate
 import net.corda.node.services.events.NodeSchedulerService
 import net.corda.node.services.events.ScheduledActivityObserver
-import net.corda.node.services.identity.InMemoryIdentityService
+import net.corda.node.services.identity.PersistentIdentityService
 import net.corda.node.services.keys.PersistentKeyManagementService
 import net.corda.node.services.messaging.MessagingService
 import net.corda.node.services.messaging.sendRequest
@@ -658,7 +658,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
         val caCertificates: Array<X509Certificate> = listOf(legalIdentity.certificate.cert, clientCa?.certificate?.cert)
                 .filterNotNull()
                 .toTypedArray()
-        val service = InMemoryIdentityService(setOf(info.legalIdentityAndCert), trustRoot = trustRoot, caCertificates = *caCertificates)
+        val service = PersistentIdentityService(setOf(info.legalIdentityAndCert), trustRoot = trustRoot, caCertificates = *caCertificates)
         services.networkMapCache.partyNodes.forEach { service.verifyAndRegisterIdentity(it.legalIdentityAndCert) }
         services.networkMapCache.changed.subscribe { mapChange ->
             // TODO how should we handle network map removal

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

@@ -1,6 +1,7 @@
 package net.corda.node.internal
 
 import net.corda.client.rpc.notUsed
+import net.corda.core.concurrent.CordaFuture
 import net.corda.core.contracts.ContractState
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.UpgradedContract
@@ -18,10 +19,10 @@ import net.corda.core.node.services.vault.PageSpecification
 import net.corda.core.node.services.vault.QueryCriteria
 import net.corda.core.node.services.vault.Sort
 import net.corda.core.transactions.SignedTransaction
+import net.corda.node.services.FlowPermissions.Companion.startFlowPermission
 import net.corda.node.services.api.ServiceHubInternal
 import net.corda.node.services.messaging.getRpcContext
 import net.corda.node.services.messaging.requirePermission
-import net.corda.node.services.FlowPermissions.Companion.startFlowPermission
 import net.corda.node.services.statemachine.FlowStateMachineImpl
 import net.corda.node.services.statemachine.StateMachineManager
 import net.corda.node.utilities.CordaPersistence
@@ -173,17 +174,49 @@ class CordaRPCOpsImpl(
     override fun authoriseContractUpgrade(state: StateAndRef<*>, upgradedContractClass: Class<out UpgradedContract<*, *>>) = services.contractUpgradeService.authoriseContractUpgrade(state, upgradedContractClass)
     override fun deauthoriseContractUpgrade(state: StateAndRef<*>) = services.contractUpgradeService.deauthoriseContractUpgrade(state)
     override fun currentNodeTime(): Instant = Instant.now(services.clock)
-    override fun waitUntilNetworkReady() = services.networkMapCache.nodeReady
-    override fun partyFromAnonymous(party: AbstractParty): Party? = services.identityService.partyFromAnonymous(party)
-    override fun partyFromKey(key: PublicKey) = services.identityService.partyFromKey(key)
-    override fun partyFromX500Name(x500Name: X500Name) = services.identityService.partyFromX500Name(x500Name)
-    override fun partiesFromName(query: String, exactMatch: Boolean): Set<Party> = services.identityService.partiesFromName(query, exactMatch)
-    override fun nodeIdentityFromParty(party: AbstractParty): NodeInfo? = services.networkMapCache.getNodeByLegalIdentity(party)
+
+    override fun waitUntilNetworkReady(): CordaFuture<Void?> {
+        return database.transaction {
+            services.networkMapCache.nodeReady
+        }
+    }
+
+    override fun partyFromAnonymous(party: AbstractParty): Party? {
+        return database.transaction {
+            services.identityService.partyFromAnonymous(party)
+        }
+    }
+
+    override fun partyFromKey(key: PublicKey): Party? {
+        return database.transaction {
+            services.identityService.partyFromKey(key)
+        }
+    }
+
+    override fun partyFromX500Name(x500Name: X500Name): Party? {
+        return database.transaction {
+            services.identityService.partyFromX500Name(x500Name)
+        }
+    }
+
+    override fun partiesFromName(query: String, exactMatch: Boolean): Set<Party> {
+        return database.transaction {
+            services.identityService.partiesFromName(query, exactMatch)
+        }
+    }
+
+    override fun nodeIdentityFromParty(party: AbstractParty): NodeInfo? {
+        return database.transaction {
+            services.networkMapCache.getNodeByLegalIdentity(party)
+        }
+    }
 
     override fun registeredFlows(): List<String> = services.rpcFlows.map { it.name }.sorted()
 
     override fun clearNetworkMapCache() {
-        services.networkMapCache.clearNetworkMapCache()
+        database.transaction {
+            services.networkMapCache.clearNetworkMapCache()
+        }
     }
 
     companion object {

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

@@ -0,0 +1,199 @@
+package net.corda.node.services.identity
+
+import net.corda.core.contracts.PartyAndReference
+import net.corda.core.crypto.SecureHash
+import net.corda.core.crypto.toStringShort
+import net.corda.core.identity.AbstractParty
+import net.corda.core.identity.AnonymousParty
+import net.corda.core.identity.Party
+import net.corda.core.identity.PartyAndCertificate
+import net.corda.core.internal.toX509CertHolder
+import net.corda.core.node.services.IdentityService
+import net.corda.core.node.services.UnknownAnonymousPartyException
+import net.corda.core.serialization.SingletonSerializeAsToken
+import net.corda.core.utilities.cert
+import net.corda.core.utilities.loggerFor
+import net.corda.node.utilities.NODE_DATABASE_PREFIX
+import net.corda.node.utilities.PersistentMap
+import org.bouncycastle.asn1.x500.X500Name
+import org.bouncycastle.cert.X509CertificateHolder
+import java.io.ByteArrayInputStream
+import java.security.InvalidAlgorithmParameterException
+import java.security.PublicKey
+import java.security.cert.*
+import javax.annotation.concurrent.ThreadSafe
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.Id
+import javax.persistence.Lob
+
+@ThreadSafe
+class PersistentIdentityService(identities: Iterable<PartyAndCertificate> = emptySet(),
+                                confidentialIdentities: Iterable<PartyAndCertificate> = emptySet(),
+                                override val trustRoot: X509Certificate,
+                                vararg caCertificates: X509Certificate) : SingletonSerializeAsToken(), IdentityService {
+    constructor(wellKnownIdentities: Iterable<PartyAndCertificate> = emptySet(),
+                confidentialIdentities: Iterable<PartyAndCertificate> = emptySet(),
+                trustRoot: X509CertificateHolder) : this(wellKnownIdentities, confidentialIdentities, trustRoot.cert)
+
+    companion object {
+        private val log = loggerFor<PersistentIdentityService>()
+        private val certFactory: CertificateFactory = CertificateFactory.getInstance("X.509")
+
+        fun createPKMap(): PersistentMap<SecureHash, PartyAndCertificate, PersistentIdentity, String> {
+            return PersistentMap(
+                    toPersistentEntityKey = { it.toString() },
+                    fromPersistentEntity = {
+                        Pair(SecureHash.parse(it.publicKeyHash),
+                                PartyAndCertificate(ByteArrayInputStream(it.identity).use {
+                                    certFactory.generateCertPath(it)
+                                }))
+                    },
+                    toPersistentEntity = { key: SecureHash, value: PartyAndCertificate ->
+                        PersistentIdentity(key.toString(), value.certPath.encoded)
+                    },
+                    persistentEntityClass = PersistentIdentity::class.java
+            )
+        }
+
+        fun createX500Map(): PersistentMap<X500Name, SecureHash, PersistentIdentityNames, String> {
+            return PersistentMap(
+                    toPersistentEntityKey = { it.toString() },
+                    fromPersistentEntity = { Pair(X500Name(it.name), SecureHash.parse(it.publicKeyHash)) },
+                    toPersistentEntity = { key: X500Name, value: SecureHash ->
+                        PersistentIdentityNames(key.toString(), value.toString())
+                    },
+                    persistentEntityClass = PersistentIdentityNames::class.java
+            )
+        }
+
+        private fun mapToKey(owningKey: PublicKey) = SecureHash.sha256(owningKey.encoded)
+        private fun mapToKey(party: PartyAndCertificate) = mapToKey(party.owningKey)
+    }
+
+    @Entity
+    @javax.persistence.Table(name = "${NODE_DATABASE_PREFIX}identities")
+    class PersistentIdentity(
+            @Id
+            @Column(name = "pk_hash", length = 64)
+            var publicKeyHash: String = "",
+
+            @Lob
+            @Column
+            var identity: ByteArray = ByteArray(0)
+    )
+
+    @Entity
+    @javax.persistence.Table(name = "${NODE_DATABASE_PREFIX}named_identities")
+    class PersistentIdentityNames(
+            @Id
+            @Column(name = "name", length = 128)
+            var name: String = "",
+
+            @Column(name = "pk_hash", length = 64)
+            var publicKeyHash: String = ""
+    )
+
+    override val caCertStore: CertStore
+    override val trustRootHolder = trustRoot.toX509CertHolder()
+    override val trustAnchor: TrustAnchor = TrustAnchor(trustRoot, null)
+
+    private val keyToParties = createPKMap()
+    private val principalToParties = createX500Map()
+
+    init {
+        val caCertificatesWithRoot: Set<X509Certificate> = caCertificates.toSet() + trustRoot
+        caCertStore = CertStore.getInstance("Collection", CollectionCertStoreParameters(caCertificatesWithRoot))
+        keyToParties.putAll(identities.associateBy { mapToKey(it) })
+        principalToParties.putAll(identities.associateBy({ it.name }, { mapToKey(it) }))
+        confidentialIdentities.forEach { identity ->
+            principalToParties.computeIfAbsent(identity.name) { mapToKey(identity) }
+        }
+    }
+
+    override fun registerIdentity(party: PartyAndCertificate) {
+        verifyAndRegisterIdentity(party)
+    }
+
+    // TODO: Check the certificate validation logic
+    @Throws(CertificateExpiredException::class, CertificateNotYetValidException::class, InvalidAlgorithmParameterException::class)
+    override fun verifyAndRegisterIdentity(identity: PartyAndCertificate): PartyAndCertificate? {
+        // Validate the chain first, before we do anything clever with it
+        identity.verify(trustAnchor)
+
+        log.info("Registering identity $identity")
+        keyToParties[mapToKey(identity)] = identity
+        // Always keep the first party we registered, as that's the well known identity
+        principalToParties.computeIfAbsent(identity.name) { mapToKey(identity) }
+        val parentId = mapToKey(identity.certPath.certificates[1].publicKey)
+        return keyToParties[parentId]
+    }
+
+    override fun certificateFromKey(owningKey: PublicKey): PartyAndCertificate? = keyToParties[mapToKey(owningKey)]
+    private fun certificateFromX500Name(name: X500Name): PartyAndCertificate? {
+        val partyId = principalToParties[name]
+        return if (partyId != null) {
+            keyToParties[partyId]
+        } else null
+    }
+
+    override fun certificateFromParty(party: Party): PartyAndCertificate = certificateFromX500Name(party.name) ?: throw IllegalArgumentException("Unknown identity ${party.name}")
+
+    // We give the caller a copy of the data set to avoid any locking problems
+    override fun getAllIdentities(): Iterable<PartyAndCertificate> = ArrayList(keyToParties.values)
+
+    override fun partyFromKey(key: PublicKey): Party? = certificateFromKey(key)?.party
+    override fun partyFromX500Name(principal: X500Name): Party? = certificateFromX500Name(principal)?.party
+    override fun partyFromAnonymous(party: AbstractParty): Party? {
+        // Expand the anonymous party to a full party (i.e. has a name) if possible
+        val candidate = party as? Party ?: partyFromKey(party.owningKey)
+        // TODO: This should be done via the network map cache, which is the authoritative source of well known identities
+        // Look up the well known identity for that name
+        return if (candidate != null) {
+            // If we have a well known identity by that name, use it in preference to the candidate. Otherwise default
+            // back to the candidate.
+            val res = partyFromX500Name(candidate.name) ?: candidate
+            res
+        } else {
+            null
+        }
+    }
+
+    override fun partyFromAnonymous(partyRef: PartyAndReference) = partyFromAnonymous(partyRef.party)
+    override fun requirePartyFromAnonymous(party: AbstractParty): Party {
+        return partyFromAnonymous(party) ?: throw IllegalStateException("Could not deanonymise party ${party.owningKey.toStringShort()}")
+    }
+
+    override fun partiesFromName(query: String, exactMatch: Boolean): Set<Party> {
+        val results = LinkedHashSet<Party>()
+        for ((x500name, partyId) in principalToParties) {
+            val party = keyToParties[partyId]!!.party
+            for (rdn in x500name.rdNs) {
+                val component = rdn.first.value.toString()
+                if (exactMatch && component == query) {
+                    results += party
+                } else if (!exactMatch) {
+                    // We can imagine this being a query over a lucene index in future.
+                    //
+                    // Kostas says: We can easily use the Jaro-Winkler distance metric as it is best suited for short
+                    // strings such as entity/company names, and to detect small typos. We can also apply it for city
+                    // or any keyword related search in lists of records (not raw text - for raw text we need indexing)
+                    // and we can return results in hierarchical order (based on normalised String similarity 0.0-1.0).
+                    if (component.contains(query, ignoreCase = true))
+                        results += party
+                }
+            }
+        }
+        return results
+    }
+
+    @Throws(UnknownAnonymousPartyException::class)
+    override fun assertOwnership(party: Party, anonymousParty: AnonymousParty) {
+        val anonymousIdentity = certificateFromKey(anonymousParty.owningKey) ?:
+                throw UnknownAnonymousPartyException("Unknown $anonymousParty")
+        val issuingCert = anonymousIdentity.certPath.certificates[1]
+        require(issuingCert.publicKey == party.owningKey) {
+            "Issuing certificate's public key must match the party key ${party.owningKey.toStringShort()}."
+        }
+    }
+}

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

@@ -11,6 +11,7 @@ import net.corda.core.schemas.QueryableState
 import net.corda.core.serialization.SingletonSerializeAsToken
 import net.corda.node.services.api.SchemaService
 import net.corda.node.services.events.NodeSchedulerService
+import net.corda.node.services.identity.PersistentIdentityService
 import net.corda.node.services.keys.PersistentKeyManagementService
 import net.corda.node.services.messaging.NodeMessagingClient
 import net.corda.node.services.network.PersistentNetworkMapService
@@ -51,7 +52,9 @@ class NodeSchemaService(customSchemas: Set<MappedSchema> = emptySet()) : SchemaS
                     NodeMessagingClient.RetryMessage::class.java,
                     NodeAttachmentService.DBAttachment::class.java,
                     RaftUniquenessProvider.RaftState::class.java,
-                    BFTNonValidatingNotaryService.PersistedCommittedState::class.java
+                    BFTNonValidatingNotaryService.PersistedCommittedState::class.java,
+                    PersistentIdentityService.PersistentIdentity::class.java,
+                    PersistentIdentityService.PersistentIdentityNames::class.java
                     ))
 
     // Required schemas are those used by internal Corda services

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

@@ -57,7 +57,7 @@ class PersistentMap<K, V, E, out EK> (
     }
 
     fun all(): Sequence<Pair<K, V>> {
-        return cache.asMap().asSequence().map { Pair(it.key, it.value.get()) }
+        return cache.asMap().asSequence().filter { it.value.isPresent }.map { Pair(it.key, it.value.get()) }
     }
 
     override val size get() = cache.size().toInt()

node/src/test/kotlin/net/corda/node/messaging/TwoPartyTradeFlowTests.kt

@@ -11,7 +11,6 @@ import net.corda.core.flows.StateMachineRunId
 import net.corda.core.identity.AbstractParty
 import net.corda.core.identity.AnonymousParty
 import net.corda.core.identity.Party
-import net.corda.core.identity.PartyAndCertificate
 import net.corda.core.internal.FlowStateMachine
 import net.corda.core.internal.concurrent.map
 import net.corda.core.internal.rootCause
@@ -205,11 +204,17 @@ class TwoPartyTradeFlowTests {
             // Let the nodes know about each other - normally the network map would handle this
             val allNodes = listOf(notaryNode, aliceNode, bobNode, bankNode)
             allNodes.forEach { node ->
-                allNodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity -> node.services.identityService.registerIdentity(identity) }
+                node.database.transaction {
+                    allNodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity -> node.services.identityService.registerIdentity(identity) }
+                }
             }
 
-            aliceNode.services.identityService.verifyAndRegisterIdentity(bobNode.info.legalIdentityAndCert)
-            bobNode.services.identityService.verifyAndRegisterIdentity(aliceNode.info.legalIdentityAndCert)
+            aliceNode.database.transaction {
+                aliceNode.services.identityService.verifyAndRegisterIdentity(bobNode.info.legalIdentityAndCert)
+            }
+            bobNode.database.transaction {
+                bobNode.services.identityService.verifyAndRegisterIdentity(aliceNode.info.legalIdentityAndCert)
+            }
             aliceNode.disableDBCloseOnStop()
             bobNode.disableDBCloseOnStop()
 
@@ -339,7 +344,9 @@ class TwoPartyTradeFlowTests {
 
         val allNodes = listOf(notaryNode, aliceNode, bobNode, bankNode)
         allNodes.forEach { node ->
-            allNodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity -> node.services.identityService.verifyAndRegisterIdentity(identity) }
+            node.database.transaction {
+                allNodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity -> node.services.identityService.verifyAndRegisterIdentity(identity) }
+            }
         }
 
         ledger(aliceNode.services, initialiseSerialization = false) {
@@ -448,7 +455,11 @@ class TwoPartyTradeFlowTests {
 
         val allNodes = listOf(notaryNode, aliceNode, bobNode, bankNode)
         allNodes.forEach { node ->
-            allNodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity -> node.services.identityService.verifyAndRegisterIdentity(identity) }
+            node.database.transaction {
+                allNodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity ->
+                    node.services.identityService.verifyAndRegisterIdentity(identity)
+                }
+            }
         }
 
         ledger(aliceNode.services, initialiseSerialization = false) {
@@ -607,7 +618,9 @@ class TwoPartyTradeFlowTests {
         // Let the nodes know about each other - normally the network map would handle this
         val allNodes = listOf(notaryNode, aliceNode, bobNode, bankNode)
         allNodes.forEach { node ->
-            allNodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity -> node.services.identityService.registerIdentity(identity) }
+            node.database.transaction {
+                allNodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity -> node.services.identityService.registerIdentity(identity) }
+            }
         }
 
         val bobsBadCash = bobNode.database.transaction {

node/src/test/kotlin/net/corda/node/services/network/NetworkMapCacheTest.kt

@@ -62,7 +62,7 @@ class NetworkMapCacheTest {
         val expected = n1.info
 
         mockNet.runNetwork()
-        val actual = node0Cache.getNodeByLegalIdentity(n1.info.legalIdentity)
+        val actual = n0.database.transaction { node0Cache.getNodeByLegalIdentity(n1.info.legalIdentity) }
         assertEquals(expected, actual)
 
         // TODO: Should have a test case with anonymous lookup

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

@@ -0,0 +1,279 @@
+package net.corda.node.services.network
+
+import net.corda.core.crypto.Crypto
+import net.corda.core.crypto.generateKeyPair
+import net.corda.core.identity.AnonymousParty
+import net.corda.core.identity.Party
+import net.corda.core.identity.PartyAndCertificate
+import net.corda.core.node.services.IdentityService
+import net.corda.core.node.services.UnknownAnonymousPartyException
+import net.corda.core.utilities.CertificateAndKeyPair
+import net.corda.core.utilities.cert
+import net.corda.node.services.identity.PersistentIdentityService
+import net.corda.node.utilities.CertificateType
+import net.corda.node.utilities.CordaPersistence
+import net.corda.node.utilities.X509Utilities
+import net.corda.testing.*
+import net.corda.testing.node.MockServices
+import org.bouncycastle.asn1.x500.X500Name
+import org.junit.After
+import org.junit.Before
+import org.junit.Test
+import java.security.cert.CertificateFactory
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertNull
+
+/**
+ * Tests for the in memory identity service.
+ */
+class PersistentIdentityServiceTests {
+
+    lateinit var database: CordaPersistence
+    lateinit var services: MockServices
+    lateinit var identityService: IdentityService
+
+    @Before
+    fun setup() {
+        val databaseAndServices = MockServices.makeTestDatabaseAndMockServices(keys = emptyList(), createIdentityService = { PersistentIdentityService(trustRoot = DUMMY_CA.certificate) })
+        database = databaseAndServices.first
+        services = databaseAndServices.second
+        identityService = services.identityService
+    }
+
+    @After
+    fun shutdown() {
+        database.close()
+    }
+
+    @Test
+    fun `get all identities`() {
+        // Nothing registered, so empty set
+        database.transaction {
+            assertNull(identityService.getAllIdentities().firstOrNull())
+        }
+
+        database.transaction {
+            identityService.verifyAndRegisterIdentity(ALICE_IDENTITY)
+        }
+        var expected = setOf(ALICE)
+        var actual = database.transaction {
+            identityService.getAllIdentities().map { it.party }.toHashSet()
+        }
+        assertEquals(expected, actual)
+
+        // Add a second party and check we get both back
+        database.transaction {
+            identityService.verifyAndRegisterIdentity(BOB_IDENTITY)
+        }
+        expected = setOf(ALICE, BOB)
+        actual = database.transaction {
+            identityService.getAllIdentities().map { it.party }.toHashSet()
+        }
+        assertEquals(expected, actual)
+    }
+
+    @Test
+    fun `get identity by key`() {
+        database.transaction {
+            assertNull(identityService.partyFromKey(ALICE_PUBKEY))
+            identityService.verifyAndRegisterIdentity(ALICE_IDENTITY)
+            assertEquals(ALICE, identityService.partyFromKey(ALICE_PUBKEY))
+            assertNull(identityService.partyFromKey(BOB_PUBKEY))
+        }
+    }
+
+    @Test
+    fun `get identity by name with no registered identities`() {
+        database.transaction {
+            assertNull(identityService.partyFromX500Name(ALICE.name))
+        }
+    }
+
+    @Test
+    fun `get identity by substring match`() {
+        database.transaction {
+            identityService.verifyAndRegisterIdentity(ALICE_IDENTITY)
+            identityService.verifyAndRegisterIdentity(BOB_IDENTITY)
+        }
+        val alicente = getTestPartyAndCertificate(X500Name("O=Alicente Worldwide,L=London,C=GB"), generateKeyPair().public)
+        database.transaction {
+            identityService.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))
+        }
+    }
+
+    @Test
+    fun `get identity by name`() {
+        val identities = listOf("Node A", "Node B", "Node C")
+                .map { getTestPartyAndCertificate(X500Name("CN=$it,O=R3,OU=corda,L=London,C=GB"), generateKeyPair().public) }
+        database.transaction {
+            assertNull(identityService.partyFromX500Name(identities.first().name))
+        }
+        identities.forEach {
+            database.transaction {
+                identityService.verifyAndRegisterIdentity(it)
+            }
+        }
+        identities.forEach {
+            database.transaction {
+                assertEquals(it.party, identityService.partyFromX500Name(it.name))
+            }
+        }
+    }
+
+    /**
+     * Generate a certificate path from a root CA, down to a transaction key, store and verify the association.
+     */
+    @Test
+    fun `assert unknown anonymous key is unrecognised`() {
+        withTestSerialization {
+            val rootKey = Crypto.generateKeyPair(X509Utilities.DEFAULT_TLS_SIGNATURE_SCHEME)
+            val rootCert = X509Utilities.createSelfSignedCACertificate(ALICE.name, rootKey)
+            val txKey = Crypto.generateKeyPair(X509Utilities.DEFAULT_IDENTITY_SIGNATURE_SCHEME)
+            val identity = Party(rootCert)
+            val txIdentity = AnonymousParty(txKey.public)
+
+            assertFailsWith<UnknownAnonymousPartyException> {
+                database.transaction {
+                    identityService.assertOwnership(identity, txIdentity)
+                }
+            }
+        }
+    }
+
+    /**
+     * Generate a pair of certificate paths from a root CA, down to a transaction key, store and verify the associations.
+     * Also checks that incorrect associations are rejected.
+     */
+    @Test
+    fun `get anonymous identity by key`() {
+        val trustRoot = DUMMY_CA
+        val (alice, aliceTxIdentity) = createParty(ALICE.name, trustRoot)
+        val (_, bobTxIdentity) = createParty(ALICE.name, trustRoot)
+
+        // Now we have identities, construct the service and let it know about both
+        database.transaction {
+            identityService.verifyAndRegisterIdentity(alice)
+            identityService.verifyAndRegisterIdentity(aliceTxIdentity)
+        }
+
+        var actual = database.transaction {
+            identityService.certificateFromKey(aliceTxIdentity.party.owningKey)
+        }
+        assertEquals(aliceTxIdentity, actual!!)
+
+        database.transaction {
+            assertNull(identityService.certificateFromKey(bobTxIdentity.party.owningKey))
+        }
+        database.transaction {
+            identityService.verifyAndRegisterIdentity(bobTxIdentity)
+        }
+        actual = database.transaction {
+            identityService.certificateFromKey(bobTxIdentity.party.owningKey)
+        }
+        assertEquals(bobTxIdentity, actual!!)
+    }
+
+    /**
+     * Generate a pair of certificate paths from a root CA, down to a transaction key, store and verify the associations.
+     * Also checks that incorrect associations are rejected.
+     */
+    @Test
+    fun `assert ownership`() {
+        withTestSerialization {
+            val trustRoot = DUMMY_CA
+            val (alice, anonymousAlice) = createParty(ALICE.name, trustRoot)
+            val (bob, anonymousBob) = createParty(BOB.name, trustRoot)
+
+            database.transaction {
+                // Now we have identities, construct the service and let it know about both
+                identityService.verifyAndRegisterIdentity(anonymousAlice)
+                identityService.verifyAndRegisterIdentity(anonymousBob)
+            }
+
+            // Verify that paths are verified
+            database.transaction {
+                identityService.assertOwnership(alice.party, anonymousAlice.party.anonymise())
+                identityService.assertOwnership(bob.party, anonymousBob.party.anonymise())
+            }
+            assertFailsWith<IllegalArgumentException> {
+                database.transaction {
+                    identityService.assertOwnership(alice.party, anonymousBob.party.anonymise())
+                }
+            }
+            assertFailsWith<IllegalArgumentException> {
+                database.transaction {
+                    identityService.assertOwnership(bob.party, anonymousAlice.party.anonymise())
+                }
+            }
+
+            assertFailsWith<IllegalArgumentException> {
+                val owningKey = Crypto.decodePublicKey(trustRoot.certificate.subjectPublicKeyInfo.encoded)
+                database.transaction {
+                    identityService.assertOwnership(Party(trustRoot.certificate.subject, owningKey), anonymousAlice.party.anonymise())
+                }
+            }
+        }
+    }
+
+    @Test
+    fun `Test Persistence`() {
+        val trustRoot = DUMMY_CA
+        val (alice, anonymousAlice) = createParty(ALICE.name, trustRoot)
+        val (bob, anonymousBob) = createParty(BOB.name, trustRoot)
+
+        database.transaction {
+            // Register well known identities
+            identityService.verifyAndRegisterIdentity(alice)
+            identityService.verifyAndRegisterIdentity(bob)
+            // Register an anonymous identities
+            identityService.verifyAndRegisterIdentity(anonymousAlice)
+            identityService.verifyAndRegisterIdentity(anonymousBob)
+        }
+
+        // Create new identity service mounted onto same DB
+        val newPersistentIdentityService = database.transaction {
+            PersistentIdentityService(trustRoot = DUMMY_CA.certificate)
+        }
+
+        database.transaction {
+            newPersistentIdentityService.assertOwnership(alice.party, anonymousAlice.party.anonymise())
+            newPersistentIdentityService.assertOwnership(bob.party, anonymousBob.party.anonymise())
+        }
+
+        val aliceParent = database.transaction {
+            newPersistentIdentityService.partyFromAnonymous(anonymousAlice.party.anonymise())
+        }
+        assertEquals(alice.party, aliceParent!!)
+
+        val bobReload = database.transaction {
+            newPersistentIdentityService.certificateFromKey(anonymousBob.party.owningKey)
+        }
+        assertEquals(anonymousBob, bobReload!!)
+    }
+
+    private fun createParty(x500Name: X500Name, ca: CertificateAndKeyPair): Pair<PartyAndCertificate, PartyAndCertificate> {
+        val certFactory = CertificateFactory.getInstance("X509")
+        val issuerKeyPair = generateKeyPair()
+        val issuer = getTestPartyAndCertificate(x500Name, issuerKeyPair.public, ca)
+        val txKey = Crypto.generateKeyPair()
+        val txCert = X509Utilities.createCertificate(CertificateType.IDENTITY, issuer.certificate, issuerKeyPair, x500Name, txKey.public)
+        val txCertPath = certFactory.generateCertPath(listOf(txCert.cert) + issuer.certPath.certificates)
+        return Pair(issuer, PartyAndCertificate(txCertPath))
+    }
+
+    /**
+     * Ensure if we feed in a full identity, we get the same identity back.
+     */
+    @Test
+    fun `deanonymising a well known identity`() {
+        val expected = ALICE
+        val actual = database.transaction {
+            identityService.partyFromAnonymous(expected)
+        }
+        assertEquals(expected, actual)
+    }
+}

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

@@ -92,8 +92,10 @@ class FlowFrameworkTests {
         // We don't create a network map, so manually handle registrations
         val nodes = listOf(node1, node2, notary1, notary2)
         nodes.forEach { node ->
-            nodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity ->
-                node.services.identityService.verifyAndRegisterIdentity(identity)
+            node.database.transaction {
+                nodes.map { it.services.myInfo.legalIdentityAndCert }.forEach { identity ->
+                    node.services.identityService.verifyAndRegisterIdentity(identity)
+                }
             }
         }
     }
@@ -170,6 +172,7 @@ class FlowFrameworkTests {
         node3.services.startFlow(flow)
         assertEquals(false, flow.flowStarted) // Not started yet as no network activity has been allowed yet
         node3.disableDBCloseOnStop()
+        node3.services.networkMapCache.clearNetworkMapCache() // zap persisted NetworkMapCache to force use of network.
         node3.stop()
 
         node3 = mockNet.createNode(node1.network.myAddress, node3.id)
@@ -179,6 +182,7 @@ class FlowFrameworkTests {
         node3.smm.executor.flush()
         assertEquals(true, restoredFlow.flowStarted) // Now we should have run the flow and hopefully cleared the init checkpoint
         node3.disableDBCloseOnStop()
+        node3.services.networkMapCache.clearNetworkMapCache() // zap persisted NetworkMapCache to force use of network.
         node3.stop()
 
         // Now it is completed the flow should leave no Checkpoint.