ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-06-05 00:50:52 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Refactor the 2-party trading protocol

Browse Source 
- Fix a security bug/TODO by having seller send back the signatures rather than a full blown transaction (which would allow a malicious seller to try and confuse the buyer by sending back a completely different TX to the one he proposed)
- Introduce an UntrustworthyData<T> wrapper as an (inefficient) form of taint tracking, to make it harder to forget that data has come from an untrustworthy source and may be malicious.
- Split the giant {Buyer, Seller}.call() methods into a set of smaller methods that make it easier to unit test various kinds of failure/skip bits in tests that aren't needed.
This commit is contained in:
Mike Hearn 2016-02-09 15:11:17 +01:00
parent 28daae5bd4
commit d98a3871da
6 changed files with 337 additions and 213 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

248
docs/source/protocol-state-machines.rst
View File

@ -125,6 +125,16 @@ each side.
return buyer.resultFuture return buyer.resultFuture
} }
// This object is serialised to the network and is the first protocol message the seller sends to the buyer.
class SellerTradeInfo(
val assetForSale: StateAndRef<OwnableState>,
val price: Amount,
val sellerOwnerKey: PublicKey,
val sessionID: Long
)
class SignaturesFromSeller(val timestampAuthoritySig: DigitalSignature.WithKey, val sellerSig: DigitalSignature.WithKey)
class Seller(val otherSide: SingleMessageRecipient, class Seller(val otherSide: SingleMessageRecipient,
val timestampingAuthority: LegallyIdentifiableNode, val timestampingAuthority: LegallyIdentifiableNode,
val assetToSell: StateAndRef<OwnableState>, val assetToSell: StateAndRef<OwnableState>,
@ -137,21 +147,11 @@ each side.
} }
} }
// This object is serialised to the network and is the first protocol message the seller sends to the buyer.
private class SellerTradeInfo(
val assetForSale: StateAndRef<OwnableState>,
val price: Amount,
val sellerOwnerKey: PublicKey,
val sessionID: Long
)
class UnacceptablePriceException(val givenPrice: Amount) : Exception() class UnacceptablePriceException(val givenPrice: Amount) : Exception()
class AssetMismatchException(val expectedTypeName: String, val typeName: String) : Exception() { class AssetMismatchException(val expectedTypeName: String, val typeName: String) : Exception() {
override fun toString() = "The submitted asset didn't match the expected type: $expectedTypeName vs $typeName" override fun toString() = "The submitted asset didn't match the expected type: $expectedTypeName vs $typeName"
} }
// The buyer's side of the protocol. See note above Seller to learn about the caveats here.
class Buyer(val otherSide: SingleMessageRecipient, class Buyer(val otherSide: SingleMessageRecipient,
val timestampingAuthority: Party, val timestampingAuthority: Party,
val acceptablePrice: Amount, val acceptablePrice: Amount,
@ -167,7 +167,7 @@ each side.
Let's unpack what this code does: Let's unpack what this code does:
- It defines a several classes nested inside the main ``TwoPartyTradeProtocol`` singleton, and a couple of methods, one - It defines a several classes nested inside the main ``TwoPartyTradeProtocol`` singleton, and a couple of methods, one
to run the buyer side of the protocol and one to run the seller side. to run the buyer side of the protocol and one to run the seller side. Some of the classes are simply protocol messages.
- It defines the "trade topic", which is just a string that namespaces this protocol. The prefix "platform." is reserved - It defines the "trade topic", which is just a string that namespaces this protocol. The prefix "platform." is reserved
by the DLG, but you can define your own protocols using standard Java-style reverse DNS notation. by the DLG, but you can define your own protocols using standard Java-style reverse DNS notation.
- The ``runBuyer`` and ``runSeller`` methods take a number of parameters that specialise the protocol for this run, - The ``runBuyer`` and ``runSeller`` methods take a number of parameters that specialise the protocol for this run,
@ -205,7 +205,8 @@ to either runBuyer or runSeller, depending on who we are, and then call ``.get()
block the calling thread until the protocol has finished. Or we could register a callback on the returned future that block the calling thread until the protocol has finished. Or we could register a callback on the returned future that
will be invoked when it's done, where we could e.g. update a user interface. will be invoked when it's done, where we could e.g. update a user interface.
Finally, we define a couple of exceptions, and a class that will be used as a protocol message called ``SellerTradeInfo``. Finally, we define a couple of exceptions, and two classes that will be used as a protocol message called
``SellerTradeInfo`` and ``SignaturesFromSeller``.
Suspendable methods Suspendable methods
------------------- -------------------
@ -244,13 +245,57 @@ Let's implement the ``Seller.call`` method. This will be invoked by the platform
.. sourcecode:: kotlin .. sourcecode:: kotlin
val sessionID = random63BitValue() val partialTX: SignedWireTransaction = receiveAndCheckProposedTransaction()
// Make the first message we'll send to kick off the protocol. // These two steps could be done in parallel, in theory. Our framework doesn't support that yet though.
val hello = SellerTradeInfo(assetToSell, price, myKeyPair.public, sessionID) val ourSignature = signWithOurKey(partialTX)
val tsaSig = timestamp(partialTX)
val partialTX = sendAndReceive(TRADE_TOPIC, buyerSessionID, sessionID, hello, SignedWireTransaction::class.java) val ledgerTX = sendSignatures(partialTX, ourSignature, tsaSig)
logger().trace { "Received partially signed transaction" }
return Pair(partialTX.tx, ledgerTX)
Here we see the outline of the procedure. We receive a proposed trade transaction from the buyer and check that it's
valid. Then we sign with our own key, request a timestamping authority to assert with another signature that the
timestamp in the transaction (if any) is valid, and finally we send back both our signature and the TSA's signature.
Finally, we hand back to the code that invoked the protocol the finished transaction in a couple of different forms.
Let's fill out the ``receiveAndCheckProposedTransaction()`` method.
.. container:: codeset
.. sourcecode:: kotlin
@Suspendable
open fun receiveAndCheckProposedTransaction(): SignedWireTransaction {
val sessionID = random63BitValue()
// Make the first message we'll send to kick off the protocol.
val hello = SellerTradeInfo(assetToSell, price, myKeyPair.public, sessionID)
val maybePartialTX = sendAndReceive(TRADE_TOPIC, buyerSessionID, sessionID, hello, SignedWireTransaction::class.java)
val partialTX = maybePartialTX.validate {
it.verifySignatures()
logger.trace { "Received partially signed transaction" }
val wtx: WireTransaction = it.tx
requireThat {
"transaction sends us the right amount of cash" by (wtx.outputs.sumCashBy(myKeyPair.public) == price)
// There are all sorts of funny games a malicious secondary might play here, we should fix them:
//
// - This tx may attempt to send some assets we aren't intending to sell to the secondary, if
// we're reusing keys! So don't reuse keys!
// - This tx may not be valid according to the contracts of the input states, so we must resolve
// and fully audit the transaction chains to convince ourselves that it is actually valid.
// - This tx may include output states that impose odd conditions on the movement of the cash,
// once we implement state pairing.
//
// but the goal of this code is not to be fully secure, but rather, just to find good ways to
// express protocol state machines on top of the messaging layer.
}
}
return partialTX
}
That's pretty straightforward. We generate a session ID to identify what's happening on the seller side, fill out That's pretty straightforward. We generate a session ID to identify what's happening on the seller side, fill out
the initial protocol message, and then call ``sendAndReceive``. This function takes a few arguments: the initial protocol message, and then call ``sendAndReceive``. This function takes a few arguments:
@ -260,6 +305,11 @@ the initial protocol message, and then call ``sendAndReceive``. This function ta
- The thing to send. It'll be serialised and sent automatically. - The thing to send. It'll be serialised and sent automatically.
- Finally a type argument, which is the kind of object we're expecting to receive from the other side. - Finally a type argument, which is the kind of object we're expecting to receive from the other side.
It returns a simple wrapper class, ``UntrustworthyData<SignedWireTransaction>``, which is just a marker class that reminds
us that the data came from a potentially malicious external source and may have been tampered with or be unexpected in
other ways. It doesn't add any functionality, but acts as a reminder to "scrub" the data before use. Here, our scrubbing
simply involves checking the signatures on it. Then we could go ahead and do some more involved checks.
Once sendAndReceive is called, the call method will be suspended into a continuation. When it gets back we'll do a log Once sendAndReceive is called, the call method will be suspended into a continuation. When it gets back we'll do a log
message. The buyer is supposed to send us a transaction with all the right inputs/outputs/commands in return, with their message. The buyer is supposed to send us a transaction with all the right inputs/outputs/commands in return, with their
cash put into the transaction and their signature on it authorising the movement of the cash. cash put into the transaction and their signature on it authorising the movement of the cash.
@ -273,51 +323,45 @@ cash put into the transaction and their signature on it authorising the movement
doing things like creating threads from inside these calls would be a bad idea. They should only contain business doing things like creating threads from inside these calls would be a bad idea. They should only contain business
logic. logic.
OK, let's keep going: Here's the rest of the code:
.. container:: codeset .. container:: codeset
.. sourcecode:: kotlin .. sourcecode:: kotlin
partialTX.verifySignatures() open fun signWithOurKey(partialTX: SignedWireTransaction) = myKeyPair.signWithECDSA(partialTX.txBits)
val wtx = partialTX.txBits.deserialize<WireTransaction>()
requireThat { @Suspendable
"transaction sends us the right amount of cash" by (wtx.outputStates.sumCashBy(args.myKeyPair.public) == args.price) open fun timestamp(partialTX: SignedWireTransaction): DigitalSignature.LegallyIdentifiable {
// There are all sorts of funny games a malicious secondary might play here, we should fix them: return TimestamperClient(this, timestampingAuthority).timestamp(partialTX.txBits)
//
// - This tx may attempt to send some assets we aren't intending to sell to the secondary, if
// we're reusing keys! So don't reuse keys!
// - This tx may not be valid according to the contracts of the input states, so we must resolve
// and fully audit the transaction chains to convince ourselves that it is actually valid.
// - This tx may include output states that impose odd conditions on the movement of the cash,
// once we implement state pairing.
} }
val ourSignature = args.myKeyPair.signWithECDSA(partialTX.txBits.bits) @Suspendable
val fullySigned: SignedWireTransaction = partialTX.copy(sigs = partialTX.sigs + ourSignature) open fun sendSignatures(partialTX: SignedWireTransaction, ourSignature: DigitalSignature.WithKey,
fullySigned.verify() tsaSig: DigitalSignature.LegallyIdentifiable): LedgerTransaction {
val timestamped: TimestampedWireTransaction = fullySigned.toTimestampedTransaction(serviceHub.timestampingService) val fullySigned = partialTX + tsaSig + ourSignature
logger().trace { "Built finished transaction, sending back to secondary!" } val ltx = fullySigned.verifyToLedgerTransaction(serviceHub.identityService)
send(TRADE_TOPIC, sessionID, timestamped) // TODO: We should run it through our full TransactionGroup of all transactions here.
return Pair(timestamped, timestamped.verifyToLedgerTransaction(serviceHub.timestampingService, serviceHub.identityService)) logger.trace { "Built finished transaction, sending back to secondary!" }
Here, we see some assertions and signature checking to satisfy ourselves that we're not about to sign something send(TRADE_TOPIC, otherSide, buyerSessionID, SignaturesFromSeller(tsaSig, ourSignature))
incorrect. Once we're happy, we calculate a signature over the transaction to authorise the movement of the asset return ltx
we are selling, and then we verify things to make sure it's all OK. Finally, we request timestamping of the }
transaction, in case the contracts governing the asset we're selling require it, and send the now finalised and
validated transaction back to the buyer. It's should be all pretty straightforward: here, ``txBits`` is the raw byte array representing the transaction.
In ``sendSignatures``, we take the two signatures we calculated, then add them to the partial transaction we were sent
and verify that the signatures all make sense. This should never fail: it's just a sanity check. Finally, we wrap the
two signatures in a simple wrapper message class and send it back. The send won't block waiting for an acknowledgement,
but the underlying message queue software will retry delivery if the other side has gone away temporarily.
.. warning:: This code is **not secure**. Other than not checking for all possible invalid constructions, if the .. warning:: This code is **not secure**. Other than not checking for all possible invalid constructions, if the
seller stops before sending the finalised transaction to the buyer, the seller is left with a valid transaction seller stops before sending the finalised transaction to the buyer, the seller is left with a valid transaction
but the buyer isn't, so they can't spend the asset they just purchased! This sort of thing will be fixed in a but the buyer isn't, so they can't spend the asset they just purchased! This sort of thing will be fixed in a
future version of the code. future version of the code.
Finally, the call function returns with the result of the protocol: in our case, the final transaction in two different
forms.
Implementing the buyer Implementing the buyer
---------------------- ----------------------
@ -328,41 +372,54 @@ OK, let's do the same for the buyer side:
.. sourcecode:: kotlin .. sourcecode:: kotlin
@Suspendable @Suspendable
override fun call(): Pair<TimestampedWireTransaction, LedgerTransaction> { override fun call(): Pair<WireTransaction, LedgerTransaction> {
val tradeRequest = receiveAndValidateTradeRequest()
val (ptx, cashSigningPubKeys) = assembleSharedTX(tradeRequest)
val stx = signWithOurKeys(cashSigningPubKeys, ptx)
val signatures = swapSignaturesWithSeller(stx, tradeRequest.sessionID)
logger.trace { "Got signatures from seller, verifying ... "}
val fullySigned = stx + signatures.timestampAuthoritySig + signatures.sellerSig
val ltx = fullySigned.verifyToLedgerTransaction(serviceHub.identityService)
logger.trace { "Fully signed transaction was valid. Trade complete! :-)" }
return Pair(fullySigned.tx, ltx)
}
@Suspendable
open fun receiveAndValidateTradeRequest(): SellerTradeInfo {
// Wait for a trade request to come in on our pre-provided session ID. // Wait for a trade request to come in on our pre-provided session ID.
val tradeRequest = receive(TRADE_TOPIC, args.sessionID, SellerTradeInfo::class.java) val maybeTradeRequest = receive(TRADE_TOPIC, sessionID, SellerTradeInfo::class.java)
// What is the seller trying to sell us? val tradeRequest = maybeTradeRequest.validate {
val assetTypeName = tradeRequest.assetForSale.state.javaClass.name // What is the seller trying to sell us?
logger().trace { "Got trade request for a $assetTypeName" } val assetTypeName = it.assetForSale.state.javaClass.name
logger.trace { "Got trade request for a $assetTypeName" }
// Check the start message for acceptability. // Check the start message for acceptability.
check(tradeRequest.sessionID > 0) check(it.sessionID > 0)
if (tradeRequest.price > acceptablePrice) if (it.price > acceptablePrice)
throw UnacceptablePriceException(tradeRequest.price) throw UnacceptablePriceException(it.price)
if (!typeToBuy.isInstance(tradeRequest.assetForSale.state)) if (!typeToBuy.isInstance(it.assetForSale.state))
throw AssetMismatchException(typeToBuy.name, assetTypeName) throw AssetMismatchException(typeToBuy.name, assetTypeName)
}
// TODO: Either look up the stateref here in our local db, or accept a long chain // TODO: Either look up the stateref here in our local db, or accept a long chain of states and
// of states and validate them to audit the other side and ensure it actually owns // validate them to audit the other side and ensure it actually owns the state we are being offered!
// the state we are being offered! For now, just assume validity! // For now, just assume validity!
return tradeRequest
}
// Generate the shared transaction that both sides will sign, using the data we have. @Suspendable
val ptx = TransactionBuilder() open fun swapSignaturesWithSeller(stx: SignedWireTransaction, theirSessionID: Long): SignaturesFromSeller {
// Add input and output states for the movement of cash, by using the Cash contract logger.trace { "Sending partially signed transaction to seller" }
// to generate the states.
val wallet = serviceHub.walletService.currentWallet
val cashStates = wallet.statesOfType<Cash.State>()
val cashSigningPubKeys = Cash().craftSpend(ptx, tradeRequest.price,
tradeRequest.sellerOwnerKey, cashStates)
// Add inputs/outputs/a command for the movement of the asset.
ptx.addInputState(tradeRequest.assetForSale.ref)
// Just pick some new public key for now.
val freshKey = serviceHub.keyManagementService.freshKey()
val (command, state) = tradeRequest.assetForSale.state.withNewOwner(freshKey.public)
ptx.addOutputState(state)
ptx.addArg(WireCommand(command, tradeRequest.assetForSale.state.owner))
// TODO: Protect against the seller terminating here and leaving us in the lurch without the final tx.
return sendAndReceive(TRADE_TOPIC, otherSide, theirSessionID, sessionID, stx, SignaturesFromSeller::class.java).validate {}
}
open fun signWithOurKeys(cashSigningPubKeys: List<PublicKey>, ptx: TransactionBuilder): SignedWireTransaction {
// Now sign the transaction with whatever keys we need to move the cash. // Now sign the transaction with whatever keys we need to move the cash.
for (k in cashSigningPubKeys) { for (k in cashSigningPubKeys) {
val priv = serviceHub.keyManagementService.toPrivate(k) val priv = serviceHub.keyManagementService.toPrivate(k)
@ -374,36 +431,43 @@ OK, let's do the same for the buyer side:
// TODO: Could run verify() here to make sure the only signature missing is the sellers. // TODO: Could run verify() here to make sure the only signature missing is the sellers.
logger().trace { "Sending partially signed transaction to seller" } return stx
}
// TODO: Protect against the buyer terminating here and leaving us in the lurch without open fun assembleSharedTX(tradeRequest: SellerTradeInfo): Pair<TransactionBuilder, List<PublicKey>> {
// the final tx. val ptx = TransactionBuilder()
// TODO: Protect against a malicious buyer sending us back a different transaction to // Add input and output states for the movement of cash, by using the Cash contract to generate the states.
// the one we built. val wallet = serviceHub.walletService.currentWallet
val fullySigned = sendAndReceive(TRADE_TOPIC, tradeRequest.sessionID, sessionID, stx, val cashStates = wallet.statesOfType<Cash.State>()
TimestampedWireTransaction::class.java) val cashSigningPubKeys = Cash().generateSpend(ptx, tradeRequest.price, tradeRequest.sellerOwnerKey, cashStates)
// Add inputs/outputs/a command for the movement of the asset.
ptx.addInputState(tradeRequest.assetForSale.ref)
// Just pick some new public key for now. This won't be linked with our identity in any way, which is what
// we want for privacy reasons: the key is here ONLY to manage and control ownership, it is not intended to
// reveal who the owner actually is. The key management service is expected to derive a unique key from some
// initial seed in order to provide privacy protection.
val freshKey = serviceHub.keyManagementService.freshKey()
val (command, state) = tradeRequest.assetForSale.state.withNewOwner(freshKey.public)
ptx.addOutputState(state)
ptx.addCommand(command, tradeRequest.assetForSale.state.owner)
logger().trace { "Got fully signed transaction, verifying ... "} // And add a request for timestamping: it may be that none of the contracts need this! But it can't hurt
// to have one.
val ltx = fullySigned.verifyToLedgerTransaction(serviceHub.timestampingService, ptx.setTime(Instant.now(), timestampingAuthority, 30.seconds)
serviceHub.identityService) return Pair(ptx, cashSigningPubKeys)
logger().trace { "Fully signed transaction was valid. Trade complete! :-)" }
return Pair(fullySigned, ltx)
} }
This code is longer but still fairly straightforward. Here are some things to pay attention to: This code is longer but still fairly straightforward. Here are some things to pay attention to:
1. We do some sanity checking on the received message to ensure we're being offered what we expected to be offered. 1. We do some sanity checking on the received message to ensure we're being offered what we expected to be offered.
2. We create a cash spend in the normal way, by using ``Cash().craftSpend``. See the contracts tutorial if this isn't 2. We create a cash spend in the normal way, by using ``Cash().generateSpend``. See the contracts tutorial if this isn't
clear. clear.
3. We access the *service hub* when we need it to access things that are transient and may change or be recreated 3. We access the *service hub* when we need it to access things that are transient and may change or be recreated
whilst a protocol is suspended, things like the wallet or the timestamping service. Remember that a protocol may whilst a protocol is suspended, things like the wallet or the timestamping service. Remember that a protocol may
be suspended when it waits to receive a message across node or computer restarts, so objects representing a service be suspended when it waits to receive a message across node or computer restarts, so objects representing a service
or data which may frequently change should be accessed 'just in time'. or data which may frequently change should be accessed 'just in time'.
4. Finally, we send the unfinsished, invalid transaction to the seller so they can sign it. They are expected to send 4. Finally, we send the unfinished, invalid transaction to the seller so they can sign it. They are expected to send
back to us a ``TimestampedWireTransaction``, which once we verify it, should be the final outcome of the trade. back to us a ``SignaturesFromSeller``, which once we verify it, should be the final outcome of the trade.
As you can see, the protocol logic is straightforward and does not contain any callbacks or network glue code, despite As you can see, the protocol logic is straightforward and does not contain any callbacks or network glue code, despite
the fact that it takes minimal resources and can survive node restarts. the fact that it takes minimal resources and can survive node restarts.

232
src/main/kotlin/contracts/protocols/TwoPartyTradeProtocol.kt
View File

@ -13,13 +13,13 @@ import com.google.common.util.concurrent.ListenableFuture
import contracts.Cash import contracts.Cash
import contracts.sumCashBy import contracts.sumCashBy
import core.* import core.*
import core.crypto.DigitalSignature
import core.crypto.signWithECDSA import core.crypto.signWithECDSA
import core.messaging.LegallyIdentifiableNode import core.messaging.LegallyIdentifiableNode
import core.messaging.ProtocolStateMachine import core.messaging.ProtocolStateMachine
import core.messaging.SingleMessageRecipient import core.messaging.SingleMessageRecipient
import core.messaging.StateMachineManager import core.messaging.StateMachineManager
import core.node.TimestamperClient import core.node.TimestamperClient
import core.serialization.deserialize
import core.utilities.trace import core.utilities.trace
import java.security.KeyPair import java.security.KeyPair
import java.security.PublicKey import java.security.PublicKey
@ -67,98 +67,162 @@ object TwoPartyTradeProtocol {
return buyer.resultFuture return buyer.resultFuture
} }
class Seller(val otherSide: SingleMessageRecipient,
val timestampingAuthority: LegallyIdentifiableNode,
val assetToSell: StateAndRef<OwnableState>,
val price: Amount,
val myKeyPair: KeyPair,
val buyerSessionID: Long) : ProtocolStateMachine<Pair<WireTransaction, LedgerTransaction>>() {
@Suspendable
override fun call(): Pair<WireTransaction, LedgerTransaction> {
val sessionID = random63BitValue()
// Make the first message we'll send to kick off the protocol.
val hello = SellerTradeInfo(assetToSell, price, myKeyPair.public, sessionID)
val partialTX = sendAndReceive(TRADE_TOPIC, otherSide, buyerSessionID, sessionID, hello, SignedWireTransaction::class.java)
logger.trace { "Received partially signed transaction" }
partialTX.verifySignatures()
val wtx: WireTransaction = partialTX.txBits.deserialize()
requireThat {
"transaction sends us the right amount of cash" by (wtx.outputs.sumCashBy(myKeyPair.public) == price)
// There are all sorts of funny games a malicious secondary might play here, we should fix them:
//
// - This tx may attempt to send some assets we aren't intending to sell to the secondary, if
// we're reusing keys! So don't reuse keys!
// - This tx may not be valid according to the contracts of the input states, so we must resolve
// and fully audit the transaction chains to convince ourselves that it is actually valid.
// - This tx may include output states that impose odd conditions on the movement of the cash,
// once we implement state pairing.
//
// but the goal of this code is not to be fully secure, but rather, just to find good ways to
// express protocol state machines on top of the messaging layer.
}
// Sign with our key and get the timestamping authorities key as well.
// These two steps could be done in parallel, in theory.
val ourSignature = myKeyPair.signWithECDSA(partialTX.txBits)
val tsaSig = TimestamperClient(this, timestampingAuthority).timestamp(partialTX.txBits)
val fullySigned = partialTX.withAdditionalSignature(tsaSig).withAdditionalSignature(ourSignature)
val ltx = fullySigned.verifyToLedgerTransaction(serviceHub.identityService)
// We should run it through our full TransactionGroup of all transactions here.
logger.trace { "Built finished transaction, sending back to secondary!" }
send(TRADE_TOPIC, otherSide, buyerSessionID, fullySigned)
return Pair(wtx, ltx)
}
}
// This object is serialised to the network and is the first protocol message the seller sends to the buyer. // This object is serialised to the network and is the first protocol message the seller sends to the buyer.
private class SellerTradeInfo( class SellerTradeInfo(
val assetForSale: StateAndRef<OwnableState>, val assetForSale: StateAndRef<OwnableState>,
val price: Amount, val price: Amount,
val sellerOwnerKey: PublicKey, val sellerOwnerKey: PublicKey,
val sessionID: Long val sessionID: Long
) )
class SignaturesFromSeller(val timestampAuthoritySig: DigitalSignature.WithKey, val sellerSig: DigitalSignature.WithKey)
open class Seller(val otherSide: SingleMessageRecipient,
val timestampingAuthority: LegallyIdentifiableNode,
val assetToSell: StateAndRef<OwnableState>,
val price: Amount,
val myKeyPair: KeyPair,
val buyerSessionID: Long) : ProtocolStateMachine<Pair<WireTransaction, LedgerTransaction>>() {
@Suspendable
override fun call(): Pair<WireTransaction, LedgerTransaction> {
val partialTX: SignedWireTransaction = receiveAndCheckProposedTransaction()
// These two steps could be done in parallel, in theory. Our framework doesn't support that yet though.
val ourSignature = signWithOurKey(partialTX)
val tsaSig = timestamp(partialTX)
val ledgerTX = sendSignatures(partialTX, ourSignature, tsaSig)
return Pair(partialTX.tx, ledgerTX)
}
@Suspendable
open fun receiveAndCheckProposedTransaction(): SignedWireTransaction {
val sessionID = random63BitValue()
// Make the first message we'll send to kick off the protocol.
val hello = SellerTradeInfo(assetToSell, price, myKeyPair.public, sessionID)
val maybePartialTX = sendAndReceive(TRADE_TOPIC, otherSide, buyerSessionID, sessionID, hello, SignedWireTransaction::class.java)
val partialTX = maybePartialTX.validate {
it.verifySignatures()
logger.trace { "Received partially signed transaction" }
val wtx: WireTransaction = it.tx
requireThat {
"transaction sends us the right amount of cash" by (wtx.outputs.sumCashBy(myKeyPair.public) == price)
// There are all sorts of funny games a malicious secondary might play here, we should fix them:
//
// - This tx may attempt to send some assets we aren't intending to sell to the secondary, if
// we're reusing keys! So don't reuse keys!
// - This tx may not be valid according to the contracts of the input states, so we must resolve
// and fully audit the transaction chains to convince ourselves that it is actually valid.
// - This tx may include output states that impose odd conditions on the movement of the cash,
// once we implement state pairing.
//
// but the goal of this code is not to be fully secure, but rather, just to find good ways to
// express protocol state machines on top of the messaging layer.
}
}
return partialTX
}
open fun signWithOurKey(partialTX: SignedWireTransaction) = myKeyPair.signWithECDSA(partialTX.txBits)
@Suspendable
open fun timestamp(partialTX: SignedWireTransaction): DigitalSignature.LegallyIdentifiable {
return TimestamperClient(this, timestampingAuthority).timestamp(partialTX.txBits)
}
@Suspendable
open fun sendSignatures(partialTX: SignedWireTransaction, ourSignature: DigitalSignature.WithKey,
tsaSig: DigitalSignature.LegallyIdentifiable): LedgerTransaction {
val fullySigned = partialTX + tsaSig + ourSignature
val ltx = fullySigned.verifyToLedgerTransaction(serviceHub.identityService)
// TODO: We should run it through our full TransactionGroup of all transactions here.
logger.trace { "Built finished transaction, sending back to secondary!" }
send(TRADE_TOPIC, otherSide, buyerSessionID, SignaturesFromSeller(tsaSig, ourSignature))
return ltx
}
}
class UnacceptablePriceException(val givenPrice: Amount) : Exception() class UnacceptablePriceException(val givenPrice: Amount) : Exception()
class AssetMismatchException(val expectedTypeName: String, val typeName: String) : Exception() { class AssetMismatchException(val expectedTypeName: String, val typeName: String) : Exception() {
override fun toString() = "The submitted asset didn't match the expected type: $expectedTypeName vs $typeName" override fun toString() = "The submitted asset didn't match the expected type: $expectedTypeName vs $typeName"
} }
// The buyer's side of the protocol. See note above Seller to learn about the caveats here. open class Buyer(val otherSide: SingleMessageRecipient,
class Buyer(val otherSide: SingleMessageRecipient, val timestampingAuthority: Party,
val timestampingAuthority: Party, val acceptablePrice: Amount,
val acceptablePrice: Amount, val typeToBuy: Class<out OwnableState>,
val typeToBuy: Class<out OwnableState>, val sessionID: Long) : ProtocolStateMachine<Pair<WireTransaction, LedgerTransaction>>() {
val sessionID: Long) : ProtocolStateMachine<Pair<WireTransaction, LedgerTransaction>>() {
@Suspendable @Suspendable
override fun call(): Pair<WireTransaction, LedgerTransaction> { override fun call(): Pair<WireTransaction, LedgerTransaction> {
val tradeRequest = receiveAndValidateTradeRequest()
val (ptx, cashSigningPubKeys) = assembleSharedTX(tradeRequest)
val stx = signWithOurKeys(cashSigningPubKeys, ptx)
val signatures = swapSignaturesWithSeller(stx, tradeRequest.sessionID)
logger.trace { "Got signatures from seller, verifying ... "}
val fullySigned = stx + signatures.timestampAuthoritySig + signatures.sellerSig
val ltx = fullySigned.verifyToLedgerTransaction(serviceHub.identityService)
logger.trace { "Fully signed transaction was valid. Trade complete! :-)" }
return Pair(fullySigned.tx, ltx)
}
@Suspendable
open fun receiveAndValidateTradeRequest(): SellerTradeInfo {
// Wait for a trade request to come in on our pre-provided session ID. // Wait for a trade request to come in on our pre-provided session ID.
val tradeRequest = receive(TRADE_TOPIC, sessionID, SellerTradeInfo::class.java) val maybeTradeRequest = receive(TRADE_TOPIC, sessionID, SellerTradeInfo::class.java)
// What is the seller trying to sell us? val tradeRequest = maybeTradeRequest.validate {
val assetTypeName = tradeRequest.assetForSale.state.javaClass.name // What is the seller trying to sell us?
logger.trace { "Got trade request for a $assetTypeName" } val assetTypeName = it.assetForSale.state.javaClass.name
logger.trace { "Got trade request for a $assetTypeName" }
// Check the start message for acceptability. // Check the start message for acceptability.
check(tradeRequest.sessionID > 0) check(it.sessionID > 0)
if (tradeRequest.price > acceptablePrice) if (it.price > acceptablePrice)
throw UnacceptablePriceException(tradeRequest.price) throw UnacceptablePriceException(it.price)
if (!typeToBuy.isInstance(tradeRequest.assetForSale.state)) if (!typeToBuy.isInstance(it.assetForSale.state))
throw AssetMismatchException(typeToBuy.name, assetTypeName) throw AssetMismatchException(typeToBuy.name, assetTypeName)
}
// TODO: Either look up the stateref here in our local db, or accept a long chain of states and // TODO: Either look up the stateref here in our local db, or accept a long chain of states and
// validate them to audit the other side and ensure it actually owns the state we are being offered! // validate them to audit the other side and ensure it actually owns the state we are being offered!
// For now, just assume validity! // For now, just assume validity!
return tradeRequest
}
// Generate the shared transaction that both sides will sign, using the data we have. @Suspendable
open fun swapSignaturesWithSeller(stx: SignedWireTransaction, theirSessionID: Long): SignaturesFromSeller {
logger.trace { "Sending partially signed transaction to seller" }
// TODO: Protect against the seller terminating here and leaving us in the lurch without the final tx.
return sendAndReceive(TRADE_TOPIC, otherSide, theirSessionID, sessionID, stx, SignaturesFromSeller::class.java).validate {}
}
open fun signWithOurKeys(cashSigningPubKeys: List<PublicKey>, ptx: TransactionBuilder): SignedWireTransaction {
// Now sign the transaction with whatever keys we need to move the cash.
for (k in cashSigningPubKeys) {
val priv = serviceHub.keyManagementService.toPrivate(k)
ptx.signWith(KeyPair(k, priv))
}
val stx = ptx.toSignedTransaction(checkSufficientSignatures = false)
stx.verifySignatures() // Verifies that we generated a signed transaction correctly.
// TODO: Could run verify() here to make sure the only signature missing is the sellers.
return stx
}
open fun assembleSharedTX(tradeRequest: SellerTradeInfo): Pair<TransactionBuilder, List<PublicKey>> {
val ptx = TransactionBuilder() val ptx = TransactionBuilder()
// Add input and output states for the movement of cash, by using the Cash contract to generate the states. // Add input and output states for the movement of cash, by using the Cash contract to generate the states.
val wallet = serviceHub.walletService.currentWallet val wallet = serviceHub.walletService.currentWallet
@ -178,33 +242,7 @@ object TwoPartyTradeProtocol {
// And add a request for timestamping: it may be that none of the contracts need this! But it can't hurt // And add a request for timestamping: it may be that none of the contracts need this! But it can't hurt
// to have one. // to have one.
ptx.setTime(Instant.now(), timestampingAuthority, 30.seconds) ptx.setTime(Instant.now(), timestampingAuthority, 30.seconds)
return Pair(ptx, cashSigningPubKeys)
// Now sign the transaction with whatever keys we need to move the cash.
for (k in cashSigningPubKeys) {
val priv = serviceHub.keyManagementService.toPrivate(k)
ptx.signWith(KeyPair(k, priv))
}
val stx = ptx.toSignedTransaction(checkSufficientSignatures = false)
stx.verifySignatures() // Verifies that we generated a signed transaction correctly.
// TODO: Could run verify() here to make sure the only signature missing is the sellers.
logger.trace { "Sending partially signed transaction to seller" }
// TODO: Protect against the buyer terminating here and leaving us in the lurch without the final tx.
// TODO: Protect against a malicious buyer sending us back a different transaction to the one we built.
val fullySigned = sendAndReceive(TRADE_TOPIC, otherSide, tradeRequest.sessionID,
sessionID, stx, SignedWireTransaction::class.java)
logger.trace { "Got fully signed transaction, verifying ... "}
val ltx = fullySigned.verifyToLedgerTransaction(serviceHub.identityService)
logger.trace { "Fully signed transaction was valid. Trade complete! :-)" }
return Pair(fullySigned.tx, ltx)
} }
} }
} }

5
src/main/kotlin/core/Transactions.kt
View File

@ -79,7 +79,7 @@ data class WireTransaction(val inputs: List<StateRef>,
data class SignedWireTransaction(val txBits: SerializedBytes<WireTransaction>, val sigs: List<DigitalSignature.WithKey>) { data class SignedWireTransaction(val txBits: SerializedBytes<WireTransaction>, val sigs: List<DigitalSignature.WithKey>) {
init { check(sigs.isNotEmpty()) } init { check(sigs.isNotEmpty()) }
// Lazily calculated access to the deserialised/hashed transaction data. /** Lazily calculated access to the deserialised/hashed transaction data. */
val tx: WireTransaction by lazy { txBits.deserialize() } val tx: WireTransaction by lazy { txBits.deserialize() }
/** A transaction ID is the hash of the [WireTransaction]. Thus adding or removing a signature does not change it. */ /** A transaction ID is the hash of the [WireTransaction]. Thus adding or removing a signature does not change it. */
@ -124,6 +124,9 @@ data class SignedWireTransaction(val txBits: SerializedBytes<WireTransaction>, v
/** Returns the same transaction but with an additional (unchecked) signature */ /** Returns the same transaction but with an additional (unchecked) signature */
fun withAdditionalSignature(sig: DigitalSignature.WithKey) = copy(sigs = sigs + sig) fun withAdditionalSignature(sig: DigitalSignature.WithKey) = copy(sigs = sigs + sig)
/** Alias for [withAdditionalSignature] to let you use Kotlin operator overloading. */
operator fun plus(sig: DigitalSignature.WithKey) = withAdditionalSignature(sig)
} }
/** A mutable transaction that's in the process of being built, before all signatures are present. */ /** A mutable transaction that's in the process of being built, before all signatures are present. */

34
src/main/kotlin/core/messaging/StateMachines.kt
View File

@ -118,8 +118,7 @@ class StateMachineManager(val serviceHub: ServiceHub, val runInThread: Executor)
} }
/** /**
* Kicks off a brand new state machine of the given class. It will log with the named logger, and the * Kicks off a brand new state machine of the given class. It will log with the named logger.
* [initialArgs] object will be passed to the call method of the [ProtocolStateMachine] object.
* The state machine will be persisted when it suspends, with automated restart if the StateMachineManager is * The state machine will be persisted when it suspends, with automated restart if the StateMachineManager is
* restarted with checkpointed state machines in the storage service. * restarted with checkpointed state machines in the storage service.
*/ */
@ -262,7 +261,7 @@ abstract class ProtocolStateMachine<R> : Fiber<R>("protocol", SameThreadFiberSch
} }
@Suspendable @Suppress("UNCHECKED_CAST") @Suspendable @Suppress("UNCHECKED_CAST")
private fun <T : Any> suspendAndExpectReceive(with: FiberRequest): T { private fun <T : Any> suspendAndExpectReceive(with: FiberRequest): UntrustworthyData<T> {
Fiber.parkAndSerialize { fiber, serializer -> Fiber.parkAndSerialize { fiber, serializer ->
// We don't use the passed-in serializer here, because we need to use our own augmented Kryo. // We don't use the passed-in serializer here, because we need to use our own augmented Kryo.
val deserializer = Fiber.getFiberSerializer() as KryoSerializer val deserializer = Fiber.getFiberSerializer() as KryoSerializer
@ -275,18 +274,18 @@ abstract class ProtocolStateMachine<R> : Fiber<R>("protocol", SameThreadFiberSch
} }
val tmp = resumeWithObject ?: throw IllegalStateException("Expected to receive something") val tmp = resumeWithObject ?: throw IllegalStateException("Expected to receive something")
resumeWithObject = null resumeWithObject = null
return tmp as T return UntrustworthyData(tmp as T)
} }
@Suspendable @Suppress("UNCHECKED_CAST") @Suspendable @Suppress("UNCHECKED_CAST")
fun <T : Any> sendAndReceive(topic: String, destination: MessageRecipients, sessionIDForSend: Long, sessionIDForReceive: Long, fun <T : Any> sendAndReceive(topic: String, destination: MessageRecipients, sessionIDForSend: Long, sessionIDForReceive: Long,
obj: Any, recvType: Class<T>): T { obj: Any, recvType: Class<T>): UntrustworthyData<T> {
val result = FiberRequest.ExpectingResponse(topic, destination, sessionIDForSend, sessionIDForReceive, obj, recvType) val result = FiberRequest.ExpectingResponse(topic, destination, sessionIDForSend, sessionIDForReceive, obj, recvType)
return suspendAndExpectReceive(result) return suspendAndExpectReceive(result)
} }
@Suspendable @Suspendable
fun <T : Any> receive(topic: String, sessionIDForReceive: Long, recvType: Class<T>): T { fun <T : Any> receive(topic: String, sessionIDForReceive: Long, recvType: Class<T>): UntrustworthyData<T> {
val result = FiberRequest.ExpectingResponse(topic, null, -1, sessionIDForReceive, null, recvType) val result = FiberRequest.ExpectingResponse(topic, null, -1, sessionIDForReceive, null, recvType)
return suspendAndExpectReceive(result) return suspendAndExpectReceive(result)
} }
@ -298,6 +297,29 @@ abstract class ProtocolStateMachine<R> : Fiber<R>("protocol", SameThreadFiberSch
} }
} }
/**
* A small utility to approximate taint tracking: if a method gives you back one of these, it means the data came from
* a remote source that may be incentivised to pass us junk that violates basic assumptions and thus must be checked
* first. The wrapper helps you to avoid forgetting this vital step. Things you might want to check are:
*
* - Is this object the one you actually expected? Did the other side hand you back something technically valid but
* not what you asked for?
* - Is the object disobeying its own invariants?
* - Are any objects *reachable* from this object mismatched or not what you expected?
* - Is it suspiciously large or small?
*/
class UntrustworthyData<T>(private val fromUntrustedWorld: T) {
val data: T
@Deprecated("Accessing the untrustworthy data directly without validating it first is a bad idea")
get() = fromUntrustedWorld
@Suppress("DEPRECATION")
inline fun validate(validator: (T) -> Unit): T {
validator(data)
return data
}
}
// TODO: Clean this up // TODO: Clean this up
open class FiberRequest(val topic: String, val destination: MessageRecipients?, open class FiberRequest(val topic: String, val destination: MessageRecipients?,
val sessionIDForSend: Long, val sessionIDForReceive: Long, val obj: Any?) { val sessionIDForSend: Long, val sessionIDForReceive: Long, val obj: Any?) {

7
src/main/kotlin/core/node/TimestamperNodeService.kt
View File

@ -117,10 +117,13 @@ class TimestamperClient(private val psm: ProtocolStateMachine<*>, private val no
val sessionID = random63BitValue() val sessionID = random63BitValue()
val replyTopic = "${TimestamperNodeService.TIMESTAMPING_PROTOCOL_TOPIC}.$sessionID" val replyTopic = "${TimestamperNodeService.TIMESTAMPING_PROTOCOL_TOPIC}.$sessionID"
val req = TimestampingMessages.Request(wtxBytes, psm.serviceHub.networkService.myAddress, replyTopic) val req = TimestampingMessages.Request(wtxBytes, psm.serviceHub.networkService.myAddress, replyTopic)
val signature = psm.sendAndReceive(TimestamperNodeService.TIMESTAMPING_PROTOCOL_TOPIC, node.address, 0,
val maybeSignature = psm.sendAndReceive(TimestamperNodeService.TIMESTAMPING_PROTOCOL_TOPIC, node.address, 0,
sessionID, req, DigitalSignature.LegallyIdentifiable::class.java) sessionID, req, DigitalSignature.LegallyIdentifiable::class.java)
// Check that the timestamping authority gave us back a valid signature and didn't break somehow // Check that the timestamping authority gave us back a valid signature and didn't break somehow
signature.verifyWithECDSA(wtxBytes) val signature = maybeSignature.validate { it.verifyWithECDSA(wtxBytes) }
return signature return signature
} }
} }

24
src/test/kotlin/core/messaging/TwoPartyTradeProtocolTests.kt
View File

@ -14,14 +14,12 @@ import contracts.CommercialPaper
import contracts.protocols.TwoPartyTradeProtocol import contracts.protocols.TwoPartyTradeProtocol
import core.* import core.*
import core.testutils.* import core.testutils.*
import core.utilities.BriefLogFormatter
import org.junit.After import org.junit.After
import org.junit.Before import org.junit.Before
import org.junit.Test import org.junit.Test
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors import java.util.concurrent.Executors
import java.util.logging.Formatter
import java.util.logging.Level
import java.util.logging.LogRecord
import java.util.logging.Logger
import kotlin.test.assertEquals import kotlin.test.assertEquals
import kotlin.test.assertTrue import kotlin.test.assertTrue
@ -32,24 +30,21 @@ import kotlin.test.assertTrue
* We assume that Alice and Bob already found each other via some market, and have agreed the details already. * We assume that Alice and Bob already found each other via some market, and have agreed the details already.
*/ */
class TwoPartyTradeProtocolTests : TestWithInMemoryNetwork() { class TwoPartyTradeProtocolTests : TestWithInMemoryNetwork() {
lateinit var backgroundThread: ExecutorService
@Before @Before
fun initLogging() { fun before() {
Logger.getLogger("").handlers[0].level = Level.ALL backgroundThread = Executors.newSingleThreadExecutor()
Logger.getLogger("").handlers[0].formatter = object : Formatter() { BriefLogFormatter.initVerbose("platform.trade")
override fun format(record: LogRecord) = "${record.threadID} ${record.loggerName}: ${record.message}\n"
}
Logger.getLogger("com.r3cev.protocols.trade").level = Level.ALL
} }
@After @After
fun stopLogging() { fun after() {
Logger.getLogger("com.r3cev.protocols.trade").level = Level.INFO backgroundThread.shutdown()
} }
@Test @Test
fun cashForCP() { fun cashForCP() {
val backgroundThread = Executors.newSingleThreadExecutor()
transactionGroupFor<ContractState> { transactionGroupFor<ContractState> {
// Bob (Buyer) has some cash, Alice (Seller) has some commercial paper she wants to sell to Bob. // Bob (Buyer) has some cash, Alice (Seller) has some commercial paper she wants to sell to Bob.
roots { roots {
@ -96,7 +91,6 @@ class TwoPartyTradeProtocolTests : TestWithInMemoryNetwork() {
txns.add(aliceResult.get().second) txns.add(aliceResult.get().second)
verify() verify()
} }
backgroundThread.shutdown()
} }
@Test @Test