Register maker's swapcoins with a watchtower

Also edits the "test_standard_coin_swap" integration test to start up
and test a watchtower too.

README.md

@@ -70,7 +70,9 @@ total balance = 0.15000000 BTC
 
 * On one terminal run a maker server with `cargo run -- --wallet-file-name=maker1.teleport run-maker 6102`. You should see the message `listening on port 6102`.
 
-* On another terminal run a maker server with `cargo run -- --wallet-file-name=maker2.teleport run-maker 16102`. You should see the message `listening on port 16102`.
+* On another terminal run another maker server with `cargo run -- --wallet-file-name=maker2.teleport run-maker 16102`. You should see the message `listening on port 16102`.
+
+* On another terminal run a watchtower with `cargo run -- run-watchtower`. You should see the message `Starting teleport watchtower`. In the teleport project contracts are enforced with one or more watchtowers, which are required for the coinswap protocol to be secure against the maker's coins being stolen.
 
 * On another terminal start a coinswap with `cargo run -- --wallet-file-name=taker.teleport coinswap-send`. When you see the terminal messages `waiting for funding transaction to confirm` and `waiting for maker's funding transaction to confirm` then tell regtest to generate another block (or just wait if you're using testnet).
 

src/main.rs

@@ -28,10 +28,10 @@ mod maker_protocol;
 mod messages;
 mod offerbook_sync;
 mod taker_protocol;
-mod watchtower_client;
-
 mod watchtower_protocol;
-use watchtower_protocol::ContractInfo;
+
+mod watchtower_client;
+use watchtower_client::ContractInfo;
 
 fn generate_wallet(wallet_file_name: &PathBuf) -> std::io::Result<()> {
     let rpc = match get_bitcoin_rpc() {
@@ -594,7 +594,7 @@ mod test {
             .unwrap();
     }
 
-    async fn kill_maker(addr: &str) {
+    async fn send_test_kill_signal(addr: &str) {
         {
             let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
             let (_, mut writer) = stream.split();
@@ -686,7 +686,11 @@ mod test {
         assert_eq!(maker1.lock_all_nonwallet_unspents(&rpc).unwrap(), ());
         assert_eq!(maker2.lock_all_nonwallet_unspents(&rpc).unwrap(), ());
 
-        // Start threads and execute swaps
+        // Start watchtower, makers and taker to execute a coinswap
+        let watchtower_thread = thread::spawn(|| {
+            run_watchtower(); //which listens on port 6103
+        });
+
         let maker1_thread = thread::spawn(|| {
             run_maker(&PathBuf::from_str(MAKER1).unwrap(), 6102);
         });
@@ -703,13 +707,13 @@ mod test {
 
         taker_thread.join().unwrap();
 
-        kill_maker("127.0.0.1:6102").await;
-
-        kill_maker("127.0.0.1:16102").await;
+        send_test_kill_signal("127.0.0.1:6102").await;
+        send_test_kill_signal("127.0.0.1:16102").await;
+        send_test_kill_signal("127.0.0.1:6103").await;
 
         maker1_thread.join().unwrap();
-
         maker2_thread.join().unwrap();
+        watchtower_thread.join().unwrap();
 
         // Recreate the wallet
         let taker = Wallet::load_wallet_from_file(&TAKER).unwrap();

src/maker_protocol.rs

@@ -28,6 +28,7 @@ use crate::messages::{
     SignSendersContractTx, SwapCoinPrivateKey, TakerToMakerMessage,
 };
 use crate::wallet_sync::{CoreAddressLabelType, IncomingSwapCoin, OutgoingSwapCoin, Wallet};
+use crate::watchtower_client::{register_coinswap_with_watchtowers, ContractInfo};
 
 // A structure denoting expectation of type of taker message.
 // Used in the [ConnectionState] structure.
@@ -157,7 +158,8 @@ async fn run(rpc: Arc<Client>, wallet: Arc<RwLock<Wallet>>, port: u16) -> Result
                     Arc::clone(&client_rpc),
                     Arc::clone(&client_wallet),
                     addr,
-                );
+                )
+                .await;
                 match message_result {
                     Ok(reply) => {
                         if let Some(message) = reply {
@@ -198,7 +200,7 @@ async fn send_message(
     Ok(())
 }
 
-fn handle_message(
+async fn handle_message(
     line: String,
     connection_state: &mut ConnectionState,
     rpc: Arc<Client>,
@@ -318,7 +320,8 @@ fn handle_message(
                 // Nothing to send. Maker now creates and broadcasts his funding Txs
                 log::debug!("{:#?}", message);
                 connection_state.allowed_message = ExpectedMessage::SignReceiversContractTx;
-                handle_senders_and_receivers_contract_sigs(connection_state, rpc, wallet, message)?
+                handle_senders_and_receivers_contract_sigs(connection_state, rpc, wallet, message)
+                    .await?
             } else {
                 return Err(Error::Protocol(
                     "Expected sender's and reciever's contract signatures",
@@ -571,7 +574,7 @@ fn handle_proof_of_funding(
     ))
 }
 
-fn handle_senders_and_receivers_contract_sigs(
+async fn handle_senders_and_receivers_contract_sigs(
     connection_state: &mut ConnectionState,
     rpc: Arc<Client>,
     wallet: Arc<RwLock<Wallet>>,
@@ -615,6 +618,19 @@ fn handle_senders_and_receivers_contract_sigs(
             outgoing_swapcoin.others_contract_sig = Some(senders_sig)
         });
 
+    register_coinswap_with_watchtowers(
+        incoming_swapcoins
+            .iter()
+            .map(|isc| ContractInfo {
+                contract_tx: isc.contract_tx.clone(),
+            })
+            .chain(outgoing_swapcoins.iter().map(|osc| ContractInfo {
+                contract_tx: osc.contract_tx.clone(),
+            }))
+            .collect::<Vec<ContractInfo>>(),
+    )
+    .await?;
+
     let mut w = wallet.write().unwrap();
     incoming_swapcoins
         .iter()
@@ -624,8 +640,6 @@ fn handle_senders_and_receivers_contract_sigs(
         .for_each(|outgoing_swapcoin| w.add_outgoing_swapcoin(outgoing_swapcoin.clone()));
     w.update_swap_coins_list()?;
 
-    //TODO add coin to watchtowers
-
     for my_funding_tx in connection_state.pending_funding_txes.as_ref().unwrap() {
         log::debug!("Broadcasting My Funding Tx : {:#?}", my_funding_tx);
         let txid = rpc.send_raw_transaction(my_funding_tx)?;

src/watchtower_client.rs

@@ -2,14 +2,23 @@ use tokio::io::BufReader;
 use tokio::net::TcpStream;
 use tokio::prelude::*;
 
+use serde::{Deserialize, Serialize};
+
+use bitcoin::Transaction;
+
 use crate::error::Error;
 use crate::watchtower_protocol::{
-    ContractInfo, MakerToWatchtowerMessage, WatchContractTxes, WatchtowerToMakerMessage,
+    MakerToWatchtowerMessage, WatchContractTxes, WatchtowerToMakerMessage,
 };
 
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ContractInfo {
+    pub contract_tx: Transaction,
+}
+
 #[tokio::main]
 pub async fn test_watchtower_client(contracts_to_watch: Vec<ContractInfo>) {
-    register_coinswap_with_watchtower(contracts_to_watch)
+    register_coinswap_with_watchtowers(contracts_to_watch)
         .await
         .unwrap();
 }
@@ -19,7 +28,7 @@ fn parse_message(line: &str) -> Result<WatchtowerToMakerMessage, Error> {
         .map_err(|_| Error::Protocol("watchtower sent invalid message"))
 }
 
-pub async fn register_coinswap_with_watchtower(
+pub async fn register_coinswap_with_watchtowers(
     contracts_to_watch: Vec<ContractInfo>,
 ) -> Result<(), Error> {
     //TODO add support for registering with multiple watchtowers concurrently

src/watchtower_protocol.rs

@@ -14,11 +14,12 @@ use tokio::time::sleep;
 
 use serde::{Deserialize, Serialize};
 
-use bitcoin::{Transaction, Txid};
+use bitcoin::Txid;
 use bitcoincore_rpc;
 use bitcoincore_rpc::{json::GetBlockResult, Client, RpcApi};
 
 use crate::error::Error;
+use crate::watchtower_client::ContractInfo;
 
 //needed things for each contract: contract_redeemscript, fully signed contract_tx
 // tx which spends from the hashlock branch minus the preimage
@@ -27,11 +28,6 @@ use crate::error::Error;
 // of all contract_tx
 //later will add part about
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct ContractInfo {
-    pub contract_tx: Transaction,
-}
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct WatchContractTxes {
     pub protocol_version_min: u32,
@@ -164,6 +160,15 @@ async fn run(rpc: Arc<Client>) -> Result<(), Error> {
                         break;
                     }
                 };
+                #[cfg(test)]
+                if line == "kill".to_string() {
+                    server_loop_err_comms_tx
+                        .send(Error::Protocol("kill signal"))
+                        .await
+                        .unwrap();
+                    log::info!("Kill signal received, stopping watchtower....");
+                    break;
+                }
 
                 line = line.trim_end().to_string();
                 let message_result = handle_message(line, &watched_txes_comms_tx).await;