CORDA-1724 Add tx access right validity check during resolution (#3732)

* CORDA-1724 Add tx access right validity check during resolution

* CORDA-1724 Fix missing payload type

* CORDA-1724 Fix test

* CORDA-1724 Add test

* CORDA-1865 Improve comments

* CORDA-1724 Address code review comments

core/src/main/kotlin/net/corda/core/flows/SendTransactionFlow.kt

@@ -2,8 +2,10 @@ package net.corda.core.flows
 
 import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.contracts.StateAndRef
+import net.corda.core.crypto.SecureHash
 import net.corda.core.internal.FetchDataFlow
 import net.corda.core.internal.readFully
+import net.corda.core.serialization.CordaSerializable
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.unwrap
 
@@ -42,6 +44,25 @@ open class DataVendingFlow(val otherSideSession: FlowSession, val payload: Any)
     override fun call(): Void? {
         // The first payload will be the transaction data, subsequent payload will be the transaction/attachment data.
         var payload = payload
+
+        // Depending on who called this flow, the type of the initial payload is different.
+        // The authorisation logic is to maintain a dynamic list of transactions that the caller is authorised to make based on the transactions that were made already.
+        // Each time an authorised transaction is requested, the input transactions are added to the list.
+        // Once a transaction has been requested, it will be removed from the authorised list. This means that it is a protocol violation to request a transaction twice.
+        val authorisedTransactions = when (payload) {
+            is NotarisationPayload -> TransactionAuthorisationFilter().addAuthorised(getInputTransactions(payload.signedTransaction))
+            is SignedTransaction -> TransactionAuthorisationFilter().addAuthorised(getInputTransactions(payload))
+            is RetrieveAnyTransactionPayload -> TransactionAuthorisationFilter(acceptAll = true)
+            is List<*> -> TransactionAuthorisationFilter().addAuthorised(payload.flatMap { stateAndRef ->
+                if (stateAndRef is StateAndRef<*>) {
+                    getInputTransactions(serviceHub.validatedTransactions.getTransaction(stateAndRef.ref.txhash)!!) + stateAndRef.ref.txhash
+                } else {
+                    throw Exception("Unknown payload type: ${stateAndRef!!::class.java} ?")
+                }
+            }.toSet())
+            else -> throw Exception("Unknown payload type: ${payload::class.java} ?")
+        }
+
         // This loop will receive [FetchDataFlow.Request] continuously until the `otherSideSession` has all the data they need
         // to resolve the transaction, a [FetchDataFlow.EndRequest] will be sent from the `otherSideSession` to indicate end of
         // data request.
@@ -56,14 +77,47 @@ open class DataVendingFlow(val otherSideSession: FlowSession, val payload: Any)
                     FetchDataFlow.Request.End -> return null
                 }
             }
+
             payload = when (dataRequest.dataType) {
-                FetchDataFlow.DataType.TRANSACTION -> dataRequest.hashes.map {
+                FetchDataFlow.DataType.TRANSACTION -> dataRequest.hashes.map { txId ->
-                    serviceHub.validatedTransactions.getTransaction(it) ?: throw FetchDataFlow.HashNotFound(it)
+                    if (!authorisedTransactions.isAuthorised(txId)) {
+                        throw FetchDataFlow.IllegalTransactionRequest(txId)
+                    }
+                    val tx = serviceHub.validatedTransactions.getTransaction(txId)
+                            ?: throw FetchDataFlow.HashNotFound(txId)
+                    authorisedTransactions.removeAuthorised(tx.id)
+                    authorisedTransactions.addAuthorised(getInputTransactions(tx))
+                    tx
                 }
                 FetchDataFlow.DataType.ATTACHMENT -> dataRequest.hashes.map {
-                    serviceHub.attachments.openAttachment(it)?.open()?.readFully() ?: throw FetchDataFlow.HashNotFound(it)
+                    serviceHub.attachments.openAttachment(it)?.open()?.readFully()
+                            ?: throw FetchDataFlow.HashNotFound(it)
                 }
             }
         }
     }
+
+    @Suspendable
+    private fun getInputTransactions(tx: SignedTransaction): Set<SecureHash> = tx.inputs.map { it.txhash }.toSet()
+
+    private class TransactionAuthorisationFilter(private val authorisedTransactions: MutableSet<SecureHash> = mutableSetOf(), val acceptAll: Boolean = false) {
+        fun isAuthorised(txId: SecureHash) = acceptAll || authorisedTransactions.contains(txId)
+
+        fun addAuthorised(txs: Set<SecureHash>): TransactionAuthorisationFilter {
+            authorisedTransactions.addAll(txs)
+            return this
+        }
+
+        fun removeAuthorised(txId: SecureHash) {
+            authorisedTransactions.remove(txId)
+        }
+    }
 }
+
+/**
+ * This is a wildcard payload to be used by the invoker of the [DataVendingFlow] to allow unlimited access to its vault.
+ *
+ * Todo Fails with a serialization exception if it is not a list. Why?
+ */
+@CordaSerializable
+object RetrieveAnyTransactionPayload : ArrayList<Any>()

core/src/main/kotlin/net/corda/core/internal/FetchDataFlow.kt

@@ -50,6 +50,8 @@ sealed class FetchDataFlow<T : NamedByHash, in W : Any>(
 
     class HashNotFound(val requested: SecureHash) : FlowException()
 
+    class IllegalTransactionRequest(val requested: SecureHash) : FlowException("Illegal attempt to request a transaction (${requested}) that is not in the transitive dependency graph of the sent transaction.")
+
     @CordaSerializable
     data class Result<out T : NamedByHash>(val fromDisk: List<T>, val downloaded: List<T>)
 
@@ -179,9 +181,11 @@ class FetchAttachmentsFlow(requests: Set<SecureHash>,
  * Given a set of tx hashes (IDs), either loads them from local disk or asks the remote peer to provide them.
  *
  * A malicious response in which the data provided by the remote peer does not hash to the requested hash results in
- * [FetchDataFlow.DownloadedVsRequestedDataMismatch] being thrown. If the remote peer doesn't have an entry, it
+ * [FetchDataFlow.DownloadedVsRequestedDataMismatch] being thrown.
- * results in a [FetchDataFlow.HashNotFound] exception. Note that returned transactions are not inserted into
+ * If the remote peer doesn't have an entry, it results in a [FetchDataFlow.HashNotFound] exception.
- * the database, because it's up to the caller to actually verify the transactions are valid.
+ * If the remote peer is not authorized to request this transaction, it results in a [FetchDataFlow.IllegalTransactionRequest] exception.
+ * Authorisation is accorded only on valid ancestors of the root transation.
+ * Note that returned transactions are not inserted into the database, because it's up to the caller to actually verify the transactions are valid.
  */
 class FetchTransactionsFlow(requests: Set<SecureHash>, otherSide: FlowSession) :
         FetchDataFlow<SignedTransaction, SignedTransaction>(requests, otherSide, DataType.TRANSACTION) {

core/src/main/kotlin/net/corda/core/internal/ResolveTransactionsFlow.kt

@@ -72,7 +72,7 @@ class ResolveTransactionsFlow(txHashesArg: Set<SecureHash>,
         }
 
     @Suspendable
-    @Throws(FetchDataFlow.HashNotFound::class)
+    @Throws(FetchDataFlow.HashNotFound::class, FetchDataFlow.IllegalTransactionRequest::class)
     override fun call() {
         val newTxns = ArrayList<SignedTransaction>(txHashes.size)
         // Start fetching data.

core/src/test/kotlin/net/corda/core/flows/AttachmentTests.kt

@@ -116,7 +116,7 @@ class AttachmentTests : WithMockNet {
     @InitiatedBy(InitiatingFetchAttachmentsFlow::class)
     private class FetchAttachmentsResponse(val otherSideSession: FlowSession) : FlowLogic<Void?>() {
         @Suspendable
-        override fun call() = subFlow(TestDataVendingFlow(otherSideSession))
+        override fun call() = subFlow(TestNoSecurityDataVendingFlow(otherSideSession))
     }
 
     //region Generators

core/src/test/kotlin/net/corda/core/flows/TestNoSecurityDataVendingFlow.kt

@@ -5,7 +5,7 @@ import net.corda.core.internal.FetchDataFlow
 import net.corda.core.utilities.UntrustworthyData
 
 // Flow to start data vending without sending transaction. For testing only.
-class TestDataVendingFlow(otherSideSession: FlowSession) : SendStateAndRefFlow(otherSideSession, emptyList()) {
+class TestNoSecurityDataVendingFlow(otherSideSession: FlowSession) : DataVendingFlow(otherSideSession, RetrieveAnyTransactionPayload) {
     @Suspendable
     override fun sendPayloadAndReceiveDataRequest(otherSideSession: FlowSession, payload: Any): UntrustworthyData<FetchDataFlow.Request> {
         return if (payload is List<*> && payload.isEmpty()) {

core/src/test/kotlin/net/corda/core/internal/ResolveTransactionsFlowTest.kt

@@ -6,8 +6,10 @@ import net.corda.core.flows.*
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.identity.Party
 import net.corda.core.transactions.SignedTransaction
+import net.corda.core.utilities.NonEmptySet
 import net.corda.core.utilities.getOrThrow
 import net.corda.core.utilities.sequence
+import net.corda.core.utilities.unwrap
 import net.corda.testing.contracts.DummyContract
 import net.corda.testing.core.singleIdentity
 import net.corda.testing.node.MockNetwork
@@ -34,6 +36,8 @@ class ResolveTransactionsFlowTest {
     private lateinit var miniCorp: Party
     private lateinit var notary: Party
 
+    private lateinit var rootTx: SignedTransaction
+
     @Before
     fun setup() {
         mockNet = MockNetwork(cordappPackages = listOf("net.corda.testing.contracts", "net.corda.core.internal"))
@@ -160,6 +164,34 @@ class ResolveTransactionsFlowTest {
         }
     }
 
+    @Test
+    fun `Requesting a transaction while having the right to see it succeeds`() {
+        val (_, stx2) = makeTransactions()
+        val p = TestNoRightsVendingFlow(miniCorp, toVend = stx2, toRequest = stx2)
+        val future = megaCorpNode.startFlow(p)
+        mockNet.runNetwork()
+        future.getOrThrow()
+    }
+
+    @Test
+    fun `Requesting a transaction without having the right to see it results in exception`() {
+        val (_, stx2) = makeTransactions()
+        val (_, stx3) = makeTransactions()
+        val p = TestNoRightsVendingFlow(miniCorp, toVend = stx2, toRequest = stx3)
+        val future = megaCorpNode.startFlow(p)
+        mockNet.runNetwork()
+        assertFailsWith<FetchDataFlow.IllegalTransactionRequest> { future.getOrThrow() }
+    }
+
+    @Test
+    fun `Requesting a transaction twice results in exception`() {
+        val (_, stx2) = makeTransactions()
+        val p = TestResolveTwiceVendingFlow(miniCorp, stx2)
+        val future = megaCorpNode.startFlow(p)
+        mockNet.runNetwork()
+        assertFailsWith<FetchDataFlow.IllegalTransactionRequest> { future.getOrThrow() }
+    }
+
     // DOCSTART 2
     private fun makeTransactions(signFirstTX: Boolean = true, withAttachment: SecureHash? = null): Pair<SignedTransaction, SignedTransaction> {
         // Make a chain of custody of dummy states and insert into node A.
@@ -187,8 +219,9 @@ class ResolveTransactionsFlowTest {
     }
     // DOCEND 2
 
+
     @InitiatingFlow
-    private class TestFlow(val otherSide: Party, private val resolveTransactionsFlowFactory: (FlowSession) -> ResolveTransactionsFlow, private val txCountLimit: Int? = null) : FlowLogic<Unit>() {
+    private open class TestFlow(val otherSide: Party, private val resolveTransactionsFlowFactory: (FlowSession) -> ResolveTransactionsFlow, private val txCountLimit: Int? = null) : FlowLogic<Unit>() {
         constructor(txHashes: Set<SecureHash>, otherSide: Party, txCountLimit: Int? = null) : this(otherSide, { ResolveTransactionsFlow(txHashes, it) }, txCountLimit = txCountLimit)
         constructor(stx: SignedTransaction, otherSide: Party) : this(otherSide, { ResolveTransactionsFlow(stx, it) })
 
@@ -200,11 +233,54 @@ class ResolveTransactionsFlowTest {
             subFlow(resolveTransactionsFlow)
         }
     }
-
     @Suppress("unused")
     @InitiatedBy(TestFlow::class)
     private class TestResponseFlow(val otherSideSession: FlowSession) : FlowLogic<Void?>() {
         @Suspendable
-        override fun call() = subFlow(TestDataVendingFlow(otherSideSession))
+        override fun call() = subFlow(TestNoSecurityDataVendingFlow(otherSideSession))
+    }
+
+    // Used by the no-rights test
+    @InitiatingFlow
+    private class TestNoRightsVendingFlow(val otherSide: Party, val toVend: SignedTransaction, val toRequest: SignedTransaction) : FlowLogic<Unit>() {
+        @Suspendable
+        override fun call() {
+            val session = initiateFlow(otherSide)
+            session.send(toRequest)
+            subFlow(DataVendingFlow(session, toVend))
+        }
+    }
+    @Suppress("unused")
+    @InitiatedBy(TestNoRightsVendingFlow::class)
+    private open class TestResponseResolveNoRightsFlow(val otherSideSession: FlowSession) : FlowLogic<Unit>() {
+        @Suspendable
+        override fun call() {
+            val noRightsTx = otherSideSession.receive<SignedTransaction>().unwrap { it }
+            otherSideSession.receive<Any>().unwrap { it }
+            otherSideSession.sendAndReceive<Any>(FetchDataFlow.Request.Data(NonEmptySet.of(noRightsTx.inputs.first().txhash), FetchDataFlow.DataType.TRANSACTION)).unwrap { it }
+            otherSideSession.send(FetchDataFlow.Request.End)
+        }
+    }
+
+    //Used by the resolve twice test
+    @InitiatingFlow
+    private class TestResolveTwiceVendingFlow(val otherSide: Party, val tx: SignedTransaction) : FlowLogic<Unit>() {
+        @Suspendable
+        override fun call() {
+            val session = initiateFlow(otherSide)
+            subFlow(DataVendingFlow(session, tx))
+        }
+    }
+    @Suppress("unused")
+    @InitiatedBy(TestResolveTwiceVendingFlow::class)
+    private open class TestResponseResolveTwiceFlow(val otherSideSession: FlowSession) : FlowLogic<Unit>() {
+        @Suspendable
+        override fun call() {
+            val tx = otherSideSession.receive<SignedTransaction>().unwrap { it }
+            val parent1 = tx.inputs.first().txhash
+            otherSideSession.sendAndReceive<Any>(FetchDataFlow.Request.Data(NonEmptySet.of(parent1), FetchDataFlow.DataType.TRANSACTION)).unwrap { it }
+            otherSideSession.sendAndReceive<Any>(FetchDataFlow.Request.Data(NonEmptySet.of(parent1), FetchDataFlow.DataType.TRANSACTION)).unwrap { it }
+            otherSideSession.send(FetchDataFlow.Request.End)
+        }
     }
 }

core/src/test/kotlin/net/corda/core/serialization/AttachmentSerializationTest.kt

@@ -6,7 +6,7 @@ import net.corda.core.crypto.SecureHash
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.FlowSession
 import net.corda.core.flows.InitiatingFlow
-import net.corda.core.flows.TestDataVendingFlow
+import net.corda.core.flows.TestNoSecurityDataVendingFlow
 import net.corda.core.identity.Party
 import net.corda.core.internal.FetchAttachmentsFlow
 import net.corda.core.internal.FetchDataFlow
@@ -89,7 +89,7 @@ class AttachmentSerializationTest {
         @Suspendable
         override fun call() {
             if (sendData) {
-                subFlow(TestDataVendingFlow(clientSession))
+                subFlow(TestNoSecurityDataVendingFlow(clientSession))
             }
             clientSession.receive<String>().unwrap { assertEquals("ping one", it) }
             clientSession.sendAndReceive<String>("pong").unwrap { assertEquals("ping two", it) }