CORDA-1010: Send a request signature in addition to a transaction to the notary (#2527)

CORDA-1010: Notary flow - clients now send a signature over a notarisation
request in addition to the transaction. This will be logged by the notary
to be able to prove that a particular party has requested the consumption
of a particular state.

.ci/api-current.txt

@@ -1128,7 +1128,8 @@ public final class net.corda.core.flows.ContractUpgradeFlow extends java.lang.Ob
   public <init>(net.corda.core.contracts.StateAndRef, Class)
   @co.paralleluniverse.fibers.Suspendable @org.jetbrains.annotations.NotNull protected net.corda.core.flows.AbstractStateReplacementFlow$UpgradeTx assembleTx()
 ##
-public abstract class net.corda.core.flows.DataVendingFlow extends net.corda.core.flows.FlowLogic
+public class net.corda.core.flows.DataVendingFlow extends net.corda.core.flows.FlowLogic
+  public <init>(net.corda.core.flows.FlowSession, Object)
   @co.paralleluniverse.fibers.Suspendable @org.jetbrains.annotations.Nullable public Void call()
   @org.jetbrains.annotations.NotNull public final net.corda.core.flows.FlowSession getOtherSideSession()
   @org.jetbrains.annotations.NotNull public final Object getPayload()
@@ -1322,11 +1323,11 @@ public @interface net.corda.core.flows.InitiatingFlow
   @org.jetbrains.annotations.NotNull public String toString()
 ##
 @net.corda.core.serialization.CordaSerializable public static final class net.corda.core.flows.NotaryError$General extends net.corda.core.flows.NotaryError
-  public <init>(String)
+  public <init>(Throwable)
-  @org.jetbrains.annotations.NotNull public final String component1()
+  @org.jetbrains.annotations.NotNull public final Throwable component1()
-  @org.jetbrains.annotations.NotNull public final net.corda.core.flows.NotaryError$General copy(String)
+  @org.jetbrains.annotations.NotNull public final net.corda.core.flows.NotaryError$General copy(Throwable)
   public boolean equals(Object)
-  @org.jetbrains.annotations.NotNull public final String getCause()
+  @org.jetbrains.annotations.NotNull public final Throwable getCause()
   public int hashCode()
   @org.jetbrains.annotations.NotNull public String toString()
 ##
@@ -1370,8 +1371,6 @@ public final class net.corda.core.flows.NotaryFlow extends java.lang.Object
   @co.paralleluniverse.fibers.Suspendable @org.jetbrains.annotations.NotNull public List call()
   @org.jetbrains.annotations.NotNull public net.corda.core.utilities.ProgressTracker getProgressTracker()
   @co.paralleluniverse.fibers.Suspendable @org.jetbrains.annotations.NotNull protected final net.corda.core.utilities.UntrustworthyData notarise(net.corda.core.identity.Party)
-  @co.paralleluniverse.fibers.Suspendable @org.jetbrains.annotations.NotNull protected net.corda.core.utilities.UntrustworthyData sendAndReceiveNonValidating(net.corda.core.identity.Party, net.corda.core.flows.FlowSession)
-  @co.paralleluniverse.fibers.Suspendable @org.jetbrains.annotations.NotNull protected net.corda.core.utilities.UntrustworthyData sendAndReceiveValidating(net.corda.core.flows.FlowSession)
   @org.jetbrains.annotations.NotNull protected final List validateResponse(net.corda.core.utilities.UntrustworthyData, net.corda.core.identity.Party)
   public static final net.corda.core.flows.NotaryFlow$Client$Companion Companion
 ##
@@ -2923,7 +2922,7 @@ public static final class net.corda.core.serialization.SingletonSerializationTok
   @org.jetbrains.annotations.NotNull public final net.corda.core.crypto.SecureHash getId()
   @org.jetbrains.annotations.NotNull public final String getReason()
 ##
-@net.corda.core.DoNotImplement public abstract class net.corda.core.transactions.CoreTransaction extends net.corda.core.transactions.BaseTransaction
+@net.corda.core.serialization.CordaSerializable @net.corda.core.DoNotImplement public abstract class net.corda.core.transactions.CoreTransaction extends net.corda.core.transactions.BaseTransaction
   public <init>()
   @org.jetbrains.annotations.NotNull public abstract List getInputs()
 ##
@@ -3165,7 +3164,7 @@ public class net.corda.core.transactions.TransactionBuilder extends java.lang.Ob
   public abstract void verifyRequiredSignatures()
   public abstract void verifySignaturesExcept(Collection)
 ##
-@net.corda.core.DoNotImplement public abstract class net.corda.core.transactions.TraversableTransaction extends net.corda.core.transactions.CoreTransaction
+@net.corda.core.serialization.CordaSerializable @net.corda.core.DoNotImplement public abstract class net.corda.core.transactions.TraversableTransaction extends net.corda.core.transactions.CoreTransaction
   public <init>(List)
   @org.jetbrains.annotations.NotNull public final List getAttachments()
   @org.jetbrains.annotations.NotNull public final List getAvailableComponentGroups()

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

@@ -0,0 +1,101 @@
+package net.corda.core.flows
+
+import net.corda.core.contracts.StateRef
+import net.corda.core.crypto.DigitalSignature
+import net.corda.core.crypto.SecureHash
+import net.corda.core.identity.Party
+import net.corda.core.serialization.CordaSerializable
+import net.corda.core.serialization.serialize
+import net.corda.core.transactions.CoreTransaction
+import net.corda.core.transactions.SignedTransaction
+import java.security.InvalidKeyException
+import java.security.SignatureException
+
+/**
+ * A notarisation request specifies a list of states to consume and the id of the consuming transaction. Its primary
+ * purpose is for notarisation traceability – a signature over the notarisation request, [NotarisationRequestSignature],
+ * allows a notary to prove that a certain party requested the consumption of a particular state.
+ *
+ * While the signature must be retained, the notarisation request does not need to be transferred or stored anywhere - it
+ * can be built from a [SignedTransaction] or a [CoreTransaction]. The notary can recompute it from the committed states index.
+ *
+ * In case there is a need to prove that a party spent a particular state, the notary will:
+ * 1) Locate the consuming transaction id in the index, along with all other states consumed in the same transaction.
+ * 2) Build a [NotarisationRequest].
+ * 3) Locate the [NotarisationRequestSignature] for the transaction id. The signature will contain the signing public key.
+ * 4) Demonstrate the signature verifies against the serialized request. The provided states are always sorted internally,
+ *    to ensure the serialization does not get affected by the order.
+ */
+@CordaSerializable
+class NotarisationRequest(statesToConsume: List<StateRef>, val transactionId: SecureHash) {
+    companion object {
+        /** Sorts in ascending order first by transaction hash, then by output index. */
+        private val stateRefComparator = compareBy<StateRef>({ it.txhash }, { it.index })
+    }
+
+    private val _statesToConsumeSorted = statesToConsume.sortedWith(stateRefComparator)
+
+    /** States this request specifies to be consumed. Sorted to ensure the serialized form does not get affected by the state order. */
+    val statesToConsume: List<StateRef> get() = _statesToConsumeSorted // Getter required for AMQP serialization
+
+    /** Verifies the signature against this notarisation request. Checks that the signature is issued by the right party. */
+    fun verifySignature(requestSignature: NotarisationRequestSignature, intendedSigner: Party) {
+        val signature = requestSignature.digitalSignature
+        if (intendedSigner.owningKey != signature.by) {
+            val errorMessage = "Expected a signature by ${intendedSigner.owningKey}, but received by ${signature.by}}"
+            throw NotaryException(NotaryError.RequestSignatureInvalid(IllegalArgumentException(errorMessage)))
+        }
+        // TODO: if requestSignature was generated over an old version of NotarisationRequest, we need to be able to
+        // reserialize it in that version to get the exact same bytes. Modify the serialization logic once that's
+        // available.
+        val expectedSignedBytes = this.serialize().bytes
+        verifyCorrectBytesSigned(signature, expectedSignedBytes)
+    }
+
+    private fun verifyCorrectBytesSigned(signature: DigitalSignature.WithKey, bytes: ByteArray) {
+        try {
+            signature.verify(bytes)
+        } catch (e: Exception) {
+            when (e) {
+                is InvalidKeyException, is SignatureException -> {
+                    val error = NotaryError.RequestSignatureInvalid(e)
+                    throw NotaryException(error)
+                }
+                else -> throw e
+            }
+        }
+    }
+}
+
+/**
+ * A wrapper around a digital signature used for notarisation requests.
+ *
+ * The [platformVersion] is required so the notary can verify the signature against the right version of serialized
+ * bytes of the [NotarisationRequest]. Otherwise, the request may be rejected.
+ */
+@CordaSerializable
+data class NotarisationRequestSignature(val digitalSignature: DigitalSignature.WithKey, val platformVersion: Int)
+
+/**
+ * Container for the transaction and notarisation request signature.
+ * This is the payload that gets sent by a client to a notary service for committing the input states of the [transaction].
+ */
+@CordaSerializable
+data class NotarisationPayload(val transaction: Any, val requestSignature: NotarisationRequestSignature) {
+    init {
+        require(transaction is SignedTransaction || transaction is CoreTransaction) {
+            "Unsupported transaction type in the notarisation payload: ${transaction.javaClass.simpleName}"
+        }
+    }
+
+    /**
+     * A helper for automatically casting the underlying [transaction] payload to a [SignedTransaction].
+     * Should only be used by validating notaries.
+     */
+    val signedTransaction get() = transaction as SignedTransaction
+    /**
+     * A helper for automatically casting the underlying [transaction] payload to a [CoreTransaction].
+     * Should only be used by non-validating notaries.
+     */
+    val coreTransaction get() = transaction as CoreTransaction
+}

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

@@ -9,10 +9,12 @@ import net.corda.core.crypto.TransactionSignature
 import net.corda.core.crypto.keys
 import net.corda.core.identity.Party
 import net.corda.core.internal.FetchDataFlow
+import net.corda.core.internal.generateSignature
 import net.corda.core.node.services.NotaryService
 import net.corda.core.node.services.TrustedAuthorityNotaryService
 import net.corda.core.node.services.UniquenessProvider
 import net.corda.core.serialization.CordaSerializable
+import net.corda.core.transactions.CoreTransaction
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.UntrustworthyData
@@ -73,15 +75,17 @@ class NotaryFlow {
             return notaryParty
         }
 
+        /** Notarises the transaction with the [notaryParty], obtains the notary's signature(s). */
         @Throws(NotaryException::class)
         @Suspendable
         protected fun notarise(notaryParty: Party): UntrustworthyData<List<TransactionSignature>> {
             return try {
                 val session = initiateFlow(notaryParty)
+                val requestSignature = NotarisationRequest(stx.inputs, stx.id).generateSignature(serviceHub)
                 if (serviceHub.networkMapCache.isValidatingNotary(notaryParty)) {
-                    sendAndReceiveValidating(session)
+                    sendAndReceiveValidating(session, requestSignature)
                 } else {
-                    sendAndReceiveNonValidating(notaryParty, session)
+                    sendAndReceiveNonValidating(notaryParty, session, requestSignature)
                 }
             } catch (e: NotaryException) {
                 if (e.error is NotaryError.Conflict) {
@@ -92,21 +96,23 @@ class NotaryFlow {
         }
 
         @Suspendable
-        protected open fun sendAndReceiveValidating(session: FlowSession): UntrustworthyData<List<TransactionSignature>> {
+        private fun sendAndReceiveValidating(session: FlowSession, signature: NotarisationRequestSignature): UntrustworthyData<List<TransactionSignature>> {
-            subFlow(SendTransactionWithRetry(session, stx))
+            val payload = NotarisationPayload(stx, signature)
+            subFlow(NotarySendTransactionFlow(session, payload))
             return session.receive()
         }
 
         @Suspendable
-        protected open fun sendAndReceiveNonValidating(notaryParty: Party, session: FlowSession): UntrustworthyData<List<TransactionSignature>> {
+        private fun sendAndReceiveNonValidating(notaryParty: Party, session: FlowSession, signature: NotarisationRequestSignature): UntrustworthyData<List<TransactionSignature>> {
-            val tx: Any = if (stx.isNotaryChangeTransaction()) {
+            val tx: CoreTransaction = if (stx.isNotaryChangeTransaction()) {
                 stx.notaryChangeTx // Notary change transactions do not support filtering
             } else {
                 stx.buildFilteredTransaction(Predicate { it is StateRef || it is TimeWindow || it == notaryParty })
             }
-            return session.sendAndReceiveWithRetry(tx)
+            return session.sendAndReceiveWithRetry(NotarisationPayload(tx, signature))
         }
 
+        /** Checks that the notary's signature(s) is/are valid. */
         protected fun validateResponse(response: UntrustworthyData<List<TransactionSignature>>, notaryParty: Party): List<TransactionSignature> {
             return response.unwrap { signatures ->
                 signatures.forEach { validateSignature(it, stx.id, notaryParty) }
@@ -118,16 +124,16 @@ class NotaryFlow {
             check(sig.by in notaryParty.owningKey.keys) { "Invalid signer for the notary result" }
             sig.verify(txId)
         }
-    }
 
-    /**
+        /**
-     * The [SendTransactionWithRetry] flow is equivalent to [SendTransactionFlow] but using [sendAndReceiveWithRetry]
+         * The [NotarySendTransactionFlow] flow is similar to [SendTransactionFlow], but uses [NotarisationPayload] as the
-     * instead of [sendAndReceive], [SendTransactionWithRetry] is intended to be use by the notary client only.
+         * initial message, and retries message delivery.
-     */
+         */
-    private class SendTransactionWithRetry(otherSideSession: FlowSession, stx: SignedTransaction) : SendTransactionFlow(otherSideSession, stx) {
+        private class NotarySendTransactionFlow(otherSide: FlowSession, payload: NotarisationPayload) : DataVendingFlow(otherSide, payload) {
-        @Suspendable
+            @Suspendable
-        override fun sendPayloadAndReceiveDataRequest(otherSideSession: FlowSession, payload: Any): UntrustworthyData<FetchDataFlow.Request> {
+            override fun sendPayloadAndReceiveDataRequest(otherSideSession: FlowSession, payload: Any): UntrustworthyData<FetchDataFlow.Request> {
-            return otherSideSession.sendAndReceiveWithRetry(payload)
+                return otherSideSession.sendAndReceiveWithRetry(payload)
+            }
         }
     }
 
@@ -186,10 +192,16 @@ class NotaryFlow {
  */
 data class TransactionParts(val id: SecureHash, val inputs: List<StateRef>, val timestamp: TimeWindow?, val notary: Party?)
 
+/**
+ * Exception thrown by the notary service if any issues are encountered while trying to commit a transaction. The
+ * underlying [error] specifies the cause of failure.
+ */
 class NotaryException(val error: NotaryError) : FlowException("Unable to notarise: $error")
 
+/** Specifies the cause for notarisation request failure. */
 @CordaSerializable
 sealed class NotaryError {
+    /** Occurs when one or more input states of transaction with [txId] have already been consumed by another transaction. */
     data class Conflict(val txId: SecureHash, val conflict: SignedData<UniquenessProvider.Conflict>) : NotaryError() {
         override fun toString() = "One or more input states for transaction $txId have been used in another transaction"
     }
@@ -199,18 +211,27 @@ sealed class NotaryError {
         override fun toString() = "Current time $currentTime is outside the time bounds specified by the transaction: $txTimeWindow"
 
         companion object {
-            @JvmField @Deprecated("Here only for binary compatibility purposes, do not use.")
+            @JvmField
+            @Deprecated("Here only for binary compatibility purposes, do not use.")
             val INSTANCE = TimeWindowInvalid(Instant.EPOCH, TimeWindow.fromOnly(Instant.EPOCH))
         }
     }
 
+    /** Occurs when the provided transaction fails to verify. */
     data class TransactionInvalid(val cause: Throwable) : NotaryError() {
         override fun toString() = cause.toString()
     }
 
+    /** Occurs when the transaction sent for notarisation is assigned to a different notary identity. */
     object WrongNotary : NotaryError()
 
-    data class General(val cause: String): NotaryError() {
+    /** Occurs when the notarisation request signature does not verify for the provided transaction. */
-        override fun toString() = cause
+    data class RequestSignatureInvalid(val cause: Throwable) : NotaryError() {
+        override fun toString() = "Request signature invalid: $cause"
+    }
+
+    /** Occurs when the notary service encounters an unexpected issue or becomes temporarily unavailable. */
+    data class General(val cause: Throwable) : NotaryError() {
+        override fun toString() = cause.toString()
     }
 }

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

@@ -28,7 +28,7 @@ open class SendTransactionFlow(otherSide: FlowSession, stx: SignedTransaction) :
  */
 open class SendStateAndRefFlow(otherSideSession: FlowSession, stateAndRefs: List<StateAndRef<*>>) : DataVendingFlow(otherSideSession, stateAndRefs)
 
-sealed class DataVendingFlow(val otherSideSession: FlowSession, val payload: Any) : FlowLogic<Void?>() {
+open class DataVendingFlow(val otherSideSession: FlowSession, val payload: Any) : FlowLogic<Void?>() {
     @Suspendable
     protected open fun sendPayloadAndReceiveDataRequest(otherSideSession: FlowSession, payload: Any) = otherSideSession.sendAndReceive<FetchDataFlow.Request>(payload)
 

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

@@ -7,7 +7,11 @@ import net.corda.core.cordapp.CordappConfig
 import net.corda.core.cordapp.CordappContext
 import net.corda.core.cordapp.CordappProvider
 import net.corda.core.crypto.*
+import net.corda.core.flows.NotarisationRequest
+import net.corda.core.flows.NotarisationRequestSignature
+import net.corda.core.flows.NotaryFlow
 import net.corda.core.identity.CordaX500Name
+import net.corda.core.node.ServiceHub
 import net.corda.core.node.ServicesForResolution
 import net.corda.core.serialization.SerializationContext
 import net.corda.core.serialization.SerializedBytes
@@ -381,4 +385,20 @@ fun ByteBuffer.copyBytes() = ByteArray(remaining()).also { get(it) }
 
 fun createCordappContext(cordapp: Cordapp, attachmentId: SecureHash?, classLoader: ClassLoader, config: CordappConfig): CordappContext {
     return CordappContext(cordapp, attachmentId, classLoader, config)
+}
+/** Verifies that the correct notarisation request was signed by the counterparty. */
+fun NotaryFlow.Service.validateRequest(request: NotarisationRequest, signature: NotarisationRequestSignature) {
+    val requestingParty = otherSideSession.counterparty
+    request.verifySignature(signature, requestingParty)
+    // TODO: persist the signature for traceability. Do we need to persist the request as well?
+}
+
+/** Creates a signature over the notarisation request using the legal identity key. */
+fun NotarisationRequest.generateSignature(serviceHub: ServiceHub): NotarisationRequestSignature {
+    val serializedRequest = this.serialize().bytes
+    val signature = with(serviceHub) {
+        val myLegalIdentity = myInfo.legalIdentitiesAndCerts.first().owningKey
+        keyManagementService.sign(serializedRequest, myLegalIdentity)
+    }
+    return NotarisationRequestSignature(signature, serviceHub.myInfo.platformVersion)
 }

core/src/main/kotlin/net/corda/core/node/services/NotaryService.kt

@@ -94,7 +94,7 @@ abstract class TrustedAuthorityNotaryService : NotaryService() {
             }
         } catch (e: Exception) {
             log.error("Internal error", e)
-            throw NotaryException(NotaryError.General("Service unavailable, please try again later"))
+            throw NotaryException(NotaryError.General(Exception("Service unavailable, please try again later")))
         }
     }
 

core/src/main/kotlin/net/corda/core/transactions/BaseTransactions.kt

@@ -3,12 +3,14 @@ package net.corda.core.transactions
 import net.corda.core.contracts.ContractState
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.StateRef
+import net.corda.core.serialization.CordaSerializable
 
 /**
  * A transaction with the minimal amount of information required to compute the unique transaction [id], and
  * resolve a [FullTransaction]. This type of transaction, wrapped in [SignedTransaction], gets transferred across the
  * wire and recorded to storage.
  */
+@CordaSerializable
 abstract class CoreTransaction : BaseTransaction() {
     /** The inputs of this transaction, containing state references only **/
     abstract override val inputs: List<StateRef>

node/src/main/kotlin/net/corda/node/services/transactions/BFTNonValidatingNotaryService.kt

@@ -12,13 +12,19 @@ import net.corda.core.flows.NotaryError
 import net.corda.core.flows.NotaryException
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.identity.Party
+import net.corda.core.flows.NotarisationPayload
+import net.corda.core.flows.NotarisationRequest
 import net.corda.core.node.services.NotaryService
 import net.corda.core.node.services.UniquenessProvider
 import net.corda.core.schemas.PersistentStateRef
 import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.serialize
+import net.corda.core.transactions.CoreTransaction
 import net.corda.core.transactions.FilteredTransaction
-import net.corda.core.utilities.*
+import net.corda.core.utilities.contextLogger
+import net.corda.core.utilities.debug
+import net.corda.core.utilities.getOrThrow
+import net.corda.core.utilities.unwrap
 import net.corda.node.services.api.ServiceHubInternal
 import net.corda.node.services.config.BFTSMaRtConfiguration
 import net.corda.node.utilities.AppendOnlyPersistentMap
@@ -67,25 +73,25 @@ class BFTNonValidatingNotaryService(
         replicaHolder.getOrThrow() // It's enough to wait for the ServiceReplica constructor to return.
     }
 
-    fun commitTransaction(tx: Any, otherSide: Party) = client.commitTransaction(tx, otherSide)
+    fun commitTransaction(payload: NotarisationPayload, otherSide: Party) = client.commitTransaction(payload, otherSide)
 
     override fun createServiceFlow(otherPartySession: FlowSession): FlowLogic<Void?> = ServiceFlow(otherPartySession, this)
 
     private class ServiceFlow(val otherSideSession: FlowSession, val service: BFTNonValidatingNotaryService) : FlowLogic<Void?>() {
         @Suspendable
         override fun call(): Void? {
-            val stx = otherSideSession.receive<FilteredTransaction>().unwrap { it }
+            val payload = otherSideSession.receive<NotarisationPayload>().unwrap { it }
-            val signatures = commit(stx)
+            val signatures = commit(payload)
             otherSideSession.send(signatures)
             return null
         }
 
-        private fun commit(stx: FilteredTransaction): List<DigitalSignature> {
+        private fun commit(payload: NotarisationPayload): List<DigitalSignature> {
-            val response = service.commitTransaction(stx, otherSideSession.counterparty)
+            val response = service.commitTransaction(payload, otherSideSession.counterparty)
             when (response) {
                 is BFTSMaRt.ClusterResponse.Error -> throw NotaryException(response.error)
                 is BFTSMaRt.ClusterResponse.Signatures -> {
-                    log.debug("All input states of transaction ${stx.id} have been committed")
+                    log.debug("All input states of transaction ${payload.coreTransaction.id} have been committed")
                     return response.txSignatures
                 }
             }
@@ -132,28 +138,34 @@ class BFTNonValidatingNotaryService(
                           notaryIdentityKey: PublicKey) : BFTSMaRt.Replica(config, replicaId, createMap, services, notaryIdentityKey) {
 
         override fun executeCommand(command: ByteArray): ByteArray {
-            val request = command.deserialize<BFTSMaRt.CommitRequest>()
+            val commitRequest = command.deserialize<BFTSMaRt.CommitRequest>()
-            val ftx = request.tx as FilteredTransaction
+            verifyRequest(commitRequest)
-            val response = verifyAndCommitTx(ftx, request.callerIdentity)
+            val response = verifyAndCommitTx(commitRequest.payload.coreTransaction, commitRequest.callerIdentity)
             return response.serialize().bytes
         }
 
-        fun verifyAndCommitTx(ftx: FilteredTransaction, callerIdentity: Party): BFTSMaRt.ReplicaResponse {
+        private fun verifyAndCommitTx(transaction: CoreTransaction, callerIdentity: Party): BFTSMaRt.ReplicaResponse {
             return try {
-                val id = ftx.id
+                val id = transaction.id
-                val inputs = ftx.inputs
+                val inputs = transaction.inputs
-                val notary = ftx.notary
+                val notary = transaction.notary
-                NotaryService.validateTimeWindow(services.clock, ftx.timeWindow)
+                if (transaction is FilteredTransaction) NotaryService.validateTimeWindow(services.clock, transaction.timeWindow)
                 if (notary !in services.myInfo.legalIdentities) throw NotaryException(NotaryError.WrongNotary)
                 commitInputStates(inputs, id, callerIdentity)
                 log.debug { "Inputs committed successfully, signing $id" }
-                BFTSMaRt.ReplicaResponse.Signature(sign(ftx))
+                BFTSMaRt.ReplicaResponse.Signature(sign(id))
             } catch (e: NotaryException) {
                 log.debug { "Error processing transaction: ${e.error}" }
                 BFTSMaRt.ReplicaResponse.Error(e.error)
             }
         }
 
+        private fun verifyRequest(commitRequest: BFTSMaRt.CommitRequest) {
+            val transaction = commitRequest.payload.coreTransaction
+            val notarisationRequest = NotarisationRequest(transaction.inputs, transaction.id)
+            notarisationRequest.verifySignature(commitRequest.payload.requestSignature, commitRequest.callerIdentity)
+            // TODO: persist the signature for traceability.
+        }
     }
 
     override fun start() {

node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRt.kt

@@ -17,6 +17,7 @@ import net.corda.core.crypto.*
 import net.corda.core.flows.NotaryError
 import net.corda.core.flows.NotaryException
 import net.corda.core.identity.Party
+import net.corda.core.flows.NotarisationPayload
 import net.corda.core.internal.declaredField
 import net.corda.core.internal.toTypedArray
 import net.corda.core.node.services.UniquenessProvider
@@ -25,8 +26,6 @@ import net.corda.core.serialization.CordaSerializable
 import net.corda.core.serialization.SingletonSerializeAsToken
 import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.serialize
-import net.corda.core.transactions.FilteredTransaction
-import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.contextLogger
 import net.corda.core.utilities.debug
 import net.corda.node.services.api.ServiceHubInternal
@@ -52,7 +51,7 @@ import java.util.*
 object BFTSMaRt {
     /** Sent from [Client] to [Replica]. */
     @CordaSerializable
-    data class CommitRequest(val tx: Any, val callerIdentity: Party)
+    data class CommitRequest(val payload: NotarisationPayload, val callerIdentity: Party)
 
     /** Sent from [Replica] to [Client]. */
     @CordaSerializable
@@ -101,13 +100,12 @@ object BFTSMaRt {
          * Sends a transaction commit request to the BFT cluster. The [proxy] will deliver the request to every
          * replica, and block until a sufficient number of replies are received.
          */
-        fun commitTransaction(transaction: Any, otherSide: Party): ClusterResponse {
+        fun commitTransaction(payload: NotarisationPayload, otherSide: Party): ClusterResponse {
-            require(transaction is FilteredTransaction || transaction is SignedTransaction) { "Unsupported transaction type: ${transaction.javaClass.name}" }
             awaitClientConnectionToCluster()
             cluster.waitUntilAllReplicasHaveInitialized()
-            val requestBytes = CommitRequest(transaction, otherSide).serialize().bytes
+            val requestBytes = CommitRequest(payload, otherSide).serialize().bytes
             val responseBytes = proxy.invokeOrdered(requestBytes)
-            return responseBytes.deserialize<ClusterResponse>()
+            return responseBytes.deserialize()
         }
 
         /** A comparator to check if replies from two replicas are the same. */
@@ -242,12 +240,15 @@ object BFTSMaRt {
             }
         }
 
+        /** Generates a signature over an arbitrary array of bytes. */
         protected fun sign(bytes: ByteArray): DigitalSignature.WithKey {
             return services.database.transaction { services.keyManagementService.sign(bytes, notaryIdentityKey) }
         }
 
-        protected fun sign(filteredTransaction: FilteredTransaction): TransactionSignature {
+        /** Generates a transaction signature over the specified transaction [txId]. */
-            return services.database.transaction { services.createSignature(filteredTransaction, notaryIdentityKey) }
+        protected fun sign(txId: SecureHash): TransactionSignature {
+            val signableData = SignableData(txId, SignatureMetadata(services.myInfo.platformVersion, Crypto.findSignatureScheme(notaryIdentityKey).schemeNumberID))
+            return services.keyManagementService.sign(signableData, notaryIdentityKey)
         }
 
         // TODO:

node/src/main/kotlin/net/corda/node/services/transactions/NonValidatingNotaryFlow.kt

@@ -1,11 +1,15 @@
 package net.corda.node.services.transactions
 
 import co.paralleluniverse.fibers.Suspendable
-import net.corda.core.flows.FlowSession
 import net.corda.core.contracts.ComponentGroupEnum
+import net.corda.core.flows.FlowSession
 import net.corda.core.flows.NotaryFlow
 import net.corda.core.flows.TransactionParts
+import net.corda.core.flows.NotarisationPayload
+import net.corda.core.flows.NotarisationRequest
+import net.corda.core.internal.validateRequest
 import net.corda.core.node.services.TrustedAuthorityNotaryService
+import net.corda.core.transactions.CoreTransaction
 import net.corda.core.transactions.FilteredTransaction
 import net.corda.core.transactions.NotaryChangeWireTransaction
 import net.corda.core.utilities.unwrap
@@ -21,22 +25,30 @@ class NonValidatingNotaryFlow(otherSideSession: FlowSession, service: TrustedAut
      */
     @Suspendable
     override fun receiveAndVerifyTx(): TransactionParts {
-        val parts = otherSideSession.receive<Any>().unwrap {
+        return otherSideSession.receive<NotarisationPayload>().unwrap { payload ->
-            when (it) {
+            val transaction = payload.coreTransaction
-                is FilteredTransaction -> {
+            val request = NotarisationRequest(transaction.inputs, transaction.id)
-                    it.verify()
+            validateRequest(request, payload.requestSignature)
-                    it.checkAllComponentsVisible(ComponentGroupEnum.INPUTS_GROUP)
+            extractParts(transaction)
-                    it.checkAllComponentsVisible(ComponentGroupEnum.TIMEWINDOW_GROUP)
+        }
-                    val notary = it.notary
+    }
-                    TransactionParts(it.id, it.inputs, it.timeWindow, notary)
+
-                }
+    private fun extractParts(tx: CoreTransaction): TransactionParts {
-                is NotaryChangeWireTransaction -> TransactionParts(it.id, it.inputs, null, it.notary)
+        return when (tx) {
-                else -> {
+            is FilteredTransaction -> {
-                    throw IllegalArgumentException("Received unexpected transaction type: ${it::class.java.simpleName}," +
+                tx.apply {
-                            "expected either ${FilteredTransaction::class.java.simpleName} or ${NotaryChangeWireTransaction::class.java.simpleName}")
+                    verify()
+                    checkAllComponentsVisible(ComponentGroupEnum.INPUTS_GROUP)
+                    checkAllComponentsVisible(ComponentGroupEnum.TIMEWINDOW_GROUP)
                 }
+                val notary = tx.notary
+                TransactionParts(tx.id, tx.inputs, tx.timeWindow, notary)
+            }
+            is NotaryChangeWireTransaction -> TransactionParts(tx.id, tx.inputs, null, tx.notary)
+            else -> {
+                throw IllegalArgumentException("Received unexpected transaction type: ${tx::class.java.simpleName}," +
+                        "expected either ${FilteredTransaction::class.java.simpleName} or ${NotaryChangeWireTransaction::class.java.simpleName}")
             }
         }
-        return parts
     }
 }

node/src/main/kotlin/net/corda/node/services/transactions/ValidatingNotaryFlow.kt

@@ -4,8 +4,14 @@ import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.contracts.TimeWindow
 import net.corda.core.contracts.TransactionVerificationException
 import net.corda.core.flows.*
+import net.corda.core.flows.NotarisationPayload
+import net.corda.core.flows.NotarisationRequest
+import net.corda.core.internal.ResolveTransactionsFlow
+import net.corda.core.internal.validateRequest
 import net.corda.core.node.services.TrustedAuthorityNotaryService
+import net.corda.core.transactions.SignedTransaction
 import net.corda.core.transactions.TransactionWithSignatures
+import net.corda.core.utilities.unwrap
 import java.security.SignatureException
 
 /**
@@ -22,15 +28,15 @@ class ValidatingNotaryFlow(otherSideSession: FlowSession, service: TrustedAuthor
     @Suspendable
     override fun receiveAndVerifyTx(): TransactionParts {
         try {
-            val stx = subFlow(ReceiveTransactionFlow(otherSideSession, checkSufficientSignatures = false))
+            val stx = receiveTransaction()
             val notary = stx.notary
             checkNotary(notary)
             val timeWindow: TimeWindow? = if (stx.isNotaryChangeTransaction())
                 null
             else
                 stx.tx.timeWindow
-            val transactionWithSignatures = stx.resolveTransactionWithSignatures(serviceHub)
+            resolveAndContractVerify(stx)
-            checkSignatures(transactionWithSignatures)
+            verifySignatures(stx)
             return TransactionParts(stx.id, stx.inputs, timeWindow, notary!!)
         } catch (e: Exception) {
             throw when (e) {
@@ -41,6 +47,26 @@ class ValidatingNotaryFlow(otherSideSession: FlowSession, service: TrustedAuthor
         }
     }
 
+    @Suspendable
+    private fun receiveTransaction(): SignedTransaction {
+        return otherSideSession.receive<NotarisationPayload>().unwrap {
+            val stx = it.signedTransaction
+            validateRequest(NotarisationRequest(stx.inputs, stx.id), it.requestSignature)
+            stx
+        }
+    }
+
+    @Suspendable
+    private fun resolveAndContractVerify(stx: SignedTransaction) {
+        subFlow(ResolveTransactionsFlow(stx, otherSideSession))
+        stx.verify(serviceHub, false)
+    }
+
+    private fun verifySignatures(stx: SignedTransaction) {
+        val transactionWithSignatures = stx.resolveTransactionWithSignatures(serviceHub)
+        checkSignatures(transactionWithSignatures)
+    }
+
     private fun checkSignatures(tx: TransactionWithSignatures) {
         try {
             tx.verifySignaturesExcept(service.notaryIdentityKey)

node/src/test/kotlin/net/corda/node/services/transactions/NotaryServiceTests.kt

@@ -3,24 +3,35 @@ package net.corda.node.services.transactions
 import net.corda.core.concurrent.CordaFuture
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.StateRef
+import net.corda.core.crypto.Crypto
+import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.TransactionSignature
+import net.corda.core.crypto.sign
 import net.corda.core.flows.NotaryError
 import net.corda.core.flows.NotaryException
 import net.corda.core.flows.NotaryFlow
 import net.corda.core.identity.Party
+import net.corda.core.flows.NotarisationPayload
+import net.corda.core.flows.NotarisationRequest
+import net.corda.core.flows.NotarisationRequestSignature
+import net.corda.core.internal.generateSignature
+import net.corda.core.messaging.MessageRecipients
 import net.corda.core.node.ServiceHub
+import net.corda.core.serialization.deserialize
+import net.corda.core.serialization.serialize
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.transactions.TransactionBuilder
+import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.getOrThrow
 import net.corda.core.utilities.seconds
 import net.corda.node.services.api.StartedNodeServices
-import net.corda.testing.core.ALICE_NAME
+import net.corda.node.services.messaging.Message
+import net.corda.node.services.statemachine.InitialSessionMessage
 import net.corda.testing.contracts.DummyContract
+import net.corda.testing.core.ALICE_NAME
 import net.corda.testing.core.dummyCommand
-import net.corda.testing.node.MockNetwork
-import net.corda.testing.node.MockNodeParameters
 import net.corda.testing.core.singleIdentity
-import net.corda.testing.node.startFlow
+import net.corda.testing.node.*
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.After
 import org.junit.Before
@@ -34,6 +45,7 @@ import kotlin.test.assertTrue
 class NotaryServiceTests {
     private lateinit var mockNet: MockNetwork
     private lateinit var notaryServices: StartedNodeServices
+    private lateinit var aliceNode: StartedMockNode
     private lateinit var aliceServices: StartedNodeServices
     private lateinit var notary: Party
     private lateinit var alice: Party
@@ -41,7 +53,8 @@ class NotaryServiceTests {
     @Before
     fun setup() {
         mockNet = MockNetwork(cordappPackages = listOf("net.corda.testing.contracts"))
-        aliceServices = mockNet.createNode(MockNodeParameters(legalName = ALICE_NAME)).services
+        aliceNode = mockNet.createNode(MockNodeParameters(legalName = ALICE_NAME))
+        aliceServices = aliceNode.services
         notaryServices = mockNet.defaultNotaryNode.services //TODO get rid of that
         notary = mockNet.defaultNotaryIdentity
         alice = aliceServices.myInfo.singleIdentity()
@@ -159,6 +172,70 @@ class NotaryServiceTests {
         notaryError.conflict.verified()
     }
 
+    @Test
+    fun `should reject when notarisation request not signed by the requesting party`() {
+        runNotarisationAndInterceptClientPayload { originalPayload ->
+            val transaction = originalPayload.signedTransaction
+            val randomKeyPair = Crypto.generateKeyPair()
+            val bytesToSign = NotarisationRequest(transaction.inputs, transaction.id).serialize().bytes
+            val modifiedSignature = NotarisationRequestSignature(randomKeyPair.sign(bytesToSign), aliceServices.myInfo.platformVersion)
+            originalPayload.copy(requestSignature = modifiedSignature)
+        }
+    }
+
+    @Test
+    fun `should reject when incorrect notarisation request signed - inputs don't match`() {
+        runNotarisationAndInterceptClientPayload { originalPayload ->
+            val transaction = originalPayload.signedTransaction
+            val wrongInputs = listOf(StateRef(SecureHash.randomSHA256(), 0))
+            val request = NotarisationRequest(wrongInputs, transaction.id)
+            val modifiedSignature = request.generateSignature(aliceServices)
+            originalPayload.copy(requestSignature = modifiedSignature)
+        }
+    }
+
+    @Test
+    fun `should reject when incorrect notarisation request signed - transaction id doesn't match`() {
+        runNotarisationAndInterceptClientPayload { originalPayload ->
+            val transaction = originalPayload.signedTransaction
+            val wrongTransactionId = SecureHash.randomSHA256()
+            val request = NotarisationRequest(transaction.inputs, wrongTransactionId)
+            val modifiedSignature = request.generateSignature(aliceServices)
+            originalPayload.copy(requestSignature = modifiedSignature)
+        }
+    }
+
+    private fun runNotarisationAndInterceptClientPayload(payloadModifier: (NotarisationPayload) -> NotarisationPayload) {
+        aliceNode.setMessagingServiceSpy(object : MessagingServiceSpy(aliceNode.network) {
+            override fun send(message: Message, target: MessageRecipients, retryId: Long?, sequenceKey: Any, additionalHeaders: Map<String, String>) {
+                val messageData = message.data.deserialize<Any>() as? InitialSessionMessage
+                val payload = messageData?.firstPayload!!.deserialize()
+
+                if (payload is NotarisationPayload) {
+                    val alteredPayload = payloadModifier(payload)
+                    val alteredMessageData = messageData.copy(firstPayload = alteredPayload.serialize())
+                    val alteredMessage = InMemoryMessagingNetwork.InMemoryMessage(message.topic, OpaqueBytes(alteredMessageData.serialize().bytes), message.uniqueMessageId)
+                    messagingService.send(alteredMessage, target, retryId)
+
+                } else {
+                    messagingService.send(message, target, retryId)
+                }
+            }
+        })
+
+        val stx = run {
+            val inputState = issueState(aliceServices, alice)
+            val tx = TransactionBuilder(notary)
+                    .addInputState(inputState)
+                    .addCommand(dummyCommand(alice.owningKey))
+            aliceServices.signInitialTransaction(tx)
+        }
+
+        val future = runNotaryClient(stx)
+        val ex = assertFailsWith(NotaryException::class) { future.getOrThrow() }
+        assertThat(ex.error).isInstanceOf(NotaryError.RequestSignatureInvalid::class.java)
+    }
+
     private fun runNotaryClient(stx: SignedTransaction): CordaFuture<List<TransactionSignature>> {
         val flow = NotaryFlow.Client(stx)
         val future = aliceServices.startFlow(flow)

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

@@ -4,12 +4,16 @@ import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.contracts.TimeWindow
 import net.corda.core.contracts.TransactionVerificationException
 import net.corda.core.flows.*
+import net.corda.core.flows.NotarisationPayload
+import net.corda.core.flows.NotarisationRequest
+import net.corda.core.internal.ResolveTransactionsFlow
+import net.corda.core.internal.validateRequest
 import net.corda.core.node.AppServiceHub
 import net.corda.core.node.services.CordaService
-import net.corda.core.node.services.TimeWindowChecker
 import net.corda.core.node.services.TrustedAuthorityNotaryService
-import net.corda.core.transactions.LedgerTransaction
+import net.corda.core.transactions.SignedTransaction
 import net.corda.core.transactions.TransactionWithSignatures
+import net.corda.core.utilities.unwrap
 import net.corda.node.services.transactions.PersistentUniquenessProvider
 import java.security.PublicKey
 import java.security.SignatureException
@@ -17,7 +21,8 @@ import java.security.SignatureException
 /**
  * A custom notary service should provide a constructor that accepts two parameters of types [AppServiceHub] and [PublicKey].
  *
- * Note that at present only a single-node notary service can be customised.
+ * Note that the support for custom notaries is still experimental – at present only a single-node notary service can be customised.
+ * The notary-related APIs might change in the future.
  */
 // START 1
 @CordaService
@@ -41,19 +46,15 @@ class MyValidatingNotaryFlow(otherSide: FlowSession, service: MyCustomValidating
     @Suspendable
     override fun receiveAndVerifyTx(): TransactionParts {
         try {
-            val stx = subFlow(ReceiveTransactionFlow(otherSideSession, checkSufficientSignatures = false))
+            val stx = receiveTransaction()
             val notary = stx.notary
             checkNotary(notary)
-            var timeWindow: TimeWindow? = null
+            val timeWindow: TimeWindow? = if (stx.isNotaryChangeTransaction())
-            val transactionWithSignatures = if (stx.isNotaryChangeTransaction()) {
+                null
-                stx.resolveNotaryChangeTransaction(serviceHub)
+            else
-            } else {
+                stx.tx.timeWindow
-                val wtx = stx.tx
+            resolveAndContractVerify(stx)
-                customVerify(wtx.toLedgerTransaction(serviceHub))
+            verifySignatures(stx)
-                timeWindow = wtx.timeWindow
-                stx
-            }
-            checkSignatures(transactionWithSignatures)
             return TransactionParts(stx.id, stx.inputs, timeWindow, notary!!)
         } catch (e: Exception) {
             throw when (e) {
@@ -64,8 +65,25 @@ class MyValidatingNotaryFlow(otherSide: FlowSession, service: MyCustomValidating
         }
     }
 
-    private fun customVerify(transaction: LedgerTransaction) {
+    @Suspendable
-        // Add custom verification logic
+    private fun receiveTransaction(): SignedTransaction {
+        return otherSideSession.receive<NotarisationPayload>().unwrap {
+            val stx = it.signedTransaction
+            validateRequest(NotarisationRequest(stx.inputs, stx.id), it.requestSignature)
+            stx
+        }
+    }
+
+    @Suspendable
+    private fun resolveAndContractVerify(stx: SignedTransaction) {
+        subFlow(ResolveTransactionsFlow(stx, otherSideSession))
+        stx.verify(serviceHub, false)
+        customVerify(stx)
+    }
+
+    private fun verifySignatures(stx: SignedTransaction) {
+        val transactionWithSignatures = stx.resolveTransactionWithSignatures(serviceHub)
+        checkSignatures(transactionWithSignatures)
     }
 
     private fun checkSignatures(tx: TransactionWithSignatures) {
@@ -75,5 +93,9 @@ class MyValidatingNotaryFlow(otherSide: FlowSession, service: MyCustomValidating
             throw NotaryException(NotaryError.TransactionInvalid(e))
         }
     }
+
+    private fun customVerify(stx: SignedTransaction) {
+        // Add custom verification logic
+    }
 }
 // END 2