Clean up transaction key flow

* Identities returned from TxKeyFlow were backwards, meaning keys were incorrectly assigned to the remote and local identities. Added unit test covering this case and corrected the flow logic.
* Rename TxKeyFlow to TransactionKeyFlow
* Correct registration of transaction key flows
* Move TransactionKeyFlow.Provider into CoreFlowHandlers
* Move TransactionKeyFlow.Request up to the top level class instead of being a class within an object.
* Remove AbstractIdentityFlow and move the validation logic into individual flows to make it clearer that it's registering the received identities.
* Cash flows now return the recipient identity instead of full identity lookup, as this is what
the caller actually needs and simplifies a lot of cases.

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

@@ -212,8 +212,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
                 findRPCFlows(scanResult)
             }
 
-            // TODO: Investigate having class path scanning find this flow
-            registerInitiatedFlow(TxKeyFlow.Provider::class.java)
             // TODO Remove this once the cash stuff is in its own CorDapp
             registerInitiatedFlow(IssuerFlow.Issuer::class.java)
 
@@ -413,6 +411,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
         installCoreFlow(BroadcastTransactionFlow::class) { otherParty, _ -> NotifyTransactionHandler(otherParty) }
         installCoreFlow(NotaryChangeFlow::class) { otherParty, _ -> NotaryChangeHandler(otherParty) }
         installCoreFlow(ContractUpgradeFlow::class) { otherParty, _ -> ContractUpgradeHandler(otherParty) }
+        installCoreFlow(TransactionKeyFlow::class) { otherParty, _ -> TransactionKeyHandler(otherParty) }
     }
 
     /**

node/src/main/kotlin/net/corda/node/services/CoreFlowHandlers.kt

@@ -10,6 +10,7 @@ import net.corda.core.flows.FlowException
 import net.corda.core.flows.FlowLogic
 import net.corda.core.identity.Party
 import net.corda.core.transactions.SignedTransaction
+import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.unwrap
 import net.corda.flows.*
 
@@ -122,3 +123,24 @@ class ContractUpgradeHandler(otherSide: Party) : AbstractStateReplacementFlow.Ac
         ContractUpgradeFlow.verify(oldStateAndRef.state.data, expectedTx.outRef<ContractState>(0).state.data, expectedTx.commands.single())
     }
 }
+
+class TransactionKeyHandler(val otherSide: Party, val revocationEnabled: Boolean) : FlowLogic<Unit>() {
+    constructor(otherSide: Party) : this(otherSide, false)
+    companion object {
+        object SENDING_KEY : ProgressTracker.Step("Sending key")
+    }
+
+    override val progressTracker: ProgressTracker = ProgressTracker(SENDING_KEY)
+
+    @Suspendable
+    override fun call(): Unit {
+        val revocationEnabled = false
+        progressTracker.currentStep = SENDING_KEY
+        val legalIdentityAnonymous = serviceHub.keyManagementService.freshKeyAndCert(serviceHub.myInfo.legalIdentityAndCert, revocationEnabled)
+        val otherSideAnonymous = sendAndReceive<AnonymisedIdentity>(otherSide, legalIdentityAnonymous).unwrap { TransactionKeyFlow.validateIdentity(otherSide, it) }
+        val (certPath, theirCert, txIdentity) = otherSideAnonymous
+        // Validate then store their identity so that we can prove the key in the transaction is owned by the
+        // counterparty.
+        serviceHub.identityService.registerAnonymousIdentity(txIdentity, otherSide, certPath)
+    }
+}

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

@@ -7,7 +7,6 @@ import net.corda.core.identity.PartyAndCertificate
 import net.corda.core.node.services.IdentityService
 import net.corda.core.utilities.*
 import net.corda.flows.AnonymisedIdentity
-import net.corda.flows.TxKeyFlow
 import net.corda.node.services.identity.InMemoryIdentityService
 import net.corda.testing.ALICE_PUBKEY
 import net.corda.testing.BOB_PUBKEY