Test justice tx formation from persistence

Here we implement `WatchtowerPersister`, which provides a test-only
sample implementation of `Persist` similar to how we might imagine a
user to build watchtower-like functionality in the persistence pipeline.

We test that the `WatchtowerPersister` is able to successfully build and
sign a valid justice transaction that sweeps a counterparty's funds if
they broadcast an old commitment.

lightning/src/ln/functional_test_utils.rs

@@ -10,7 +10,7 @@
 //! A bunch of useful utilities for building networks of nodes and exchanging messages between
 //! nodes for functional tests.
 
-use crate::chain::{BestBlock, ChannelMonitorUpdateStatus, Confirm, Listen, Watch};
+use crate::chain::{BestBlock, ChannelMonitorUpdateStatus, Confirm, Listen, Watch, chainmonitor::Persist};
 use crate::sign::EntropySource;
 use crate::chain::channelmonitor::ChannelMonitor;
 use crate::chain::transaction::OutPoint;
@@ -2643,10 +2643,14 @@ pub fn create_chanmon_cfgs(node_count: usize) -> Vec<TestChanMonCfg> {
 }
 
 pub fn create_node_cfgs<'a>(node_count: usize, chanmon_cfgs: &'a Vec<TestChanMonCfg>) -> Vec<NodeCfg<'a>> {
+	create_node_cfgs_with_persisters(node_count, chanmon_cfgs, chanmon_cfgs.iter().map(|c| &c.persister).collect())
+}
+
+pub fn create_node_cfgs_with_persisters<'a>(node_count: usize, chanmon_cfgs: &'a Vec<TestChanMonCfg>, persisters: Vec<&'a impl Persist<EnforcingSigner>>) -> Vec<NodeCfg<'a>> {
 	let mut nodes = Vec::new();
 
 	for i in 0..node_count {
-		let chain_monitor = test_utils::TestChainMonitor::new(Some(&chanmon_cfgs[i].chain_source), &chanmon_cfgs[i].tx_broadcaster, &chanmon_cfgs[i].logger, &chanmon_cfgs[i].fee_estimator, &chanmon_cfgs[i].persister, &chanmon_cfgs[i].keys_manager);
+		let chain_monitor = test_utils::TestChainMonitor::new(Some(&chanmon_cfgs[i].chain_source), &chanmon_cfgs[i].tx_broadcaster, &chanmon_cfgs[i].logger, &chanmon_cfgs[i].fee_estimator, persisters[i], &chanmon_cfgs[i].keys_manager);
 		let network_graph = Arc::new(NetworkGraph::new(Network::Testnet, &chanmon_cfgs[i].logger));
 		let seed = [i as u8; 32];
 		nodes.push(NodeCfg {

lightning/src/ln/functional_tests.rs

@@ -17,7 +17,7 @@ use crate::chain::chaininterface::LowerBoundedFeeEstimator;
 use crate::chain::channelmonitor;
 use crate::chain::channelmonitor::{CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY};
 use crate::chain::transaction::OutPoint;
-use crate::sign::{EcdsaChannelSigner, EntropySource};
+use crate::sign::{ChannelSigner, EcdsaChannelSigner, EntropySource, SignerProvider};
 use crate::events::{Event, MessageSendEvent, MessageSendEventsProvider, PathFailure, PaymentPurpose, ClosureReason, HTLCDestination, PaymentFailureReason};
 use crate::ln::{PaymentPreimage, PaymentSecret, PaymentHash};
 use crate::ln::channel::{commitment_tx_base_weight, COMMITMENT_TX_WEIGHT_PER_HTLC, CONCURRENT_INBOUND_HTLC_FEE_BUFFER, FEE_SPIKE_BUFFER_FEE_INCREASE_MULTIPLE, MIN_AFFORDABLE_HTLC_COUNT, get_holder_selected_channel_reserve_satoshis, OutboundV1Channel, InboundV1Channel};
@@ -31,7 +31,7 @@ use crate::ln::features::{ChannelFeatures, ChannelTypeFeatures, NodeFeatures};
 use crate::ln::msgs;
 use crate::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler, ErrorAction};
 use crate::util::enforcing_trait_impls::EnforcingSigner;
-use crate::util::test_utils;
+use crate::util::test_utils::{self, WatchtowerPersister};
 use crate::util::errors::APIError;
 use crate::util::ser::{Writeable, ReadableArgs};
 use crate::util::string::UntrustedString;
@@ -2554,6 +2554,72 @@ fn revoked_output_claim() {
 	check_closed_event!(nodes[0], 1, ClosureReason::CommitmentTxConfirmed, [nodes[1].node.get_our_node_id()], 100000);
 }
 
+#[test]
+fn test_forming_justice_tx_from_monitor_updates() {
+	do_test_forming_justice_tx_from_monitor_updates(true);
+	do_test_forming_justice_tx_from_monitor_updates(false);
+}
+
+fn do_test_forming_justice_tx_from_monitor_updates(broadcast_initial_commitment: bool) {
+	// Simple test to make sure that the justice tx formed in WatchtowerPersister
+	// is properly formed and can be broadcasted/confirmed successfully in the event
+	// that a revoked commitment transaction is broadcasted
+	// (Similar to `revoked_output_claim` test but we get the justice tx + broadcast manually)
+	let chanmon_cfgs = create_chanmon_cfgs(2);
+	let destination_script0 = chanmon_cfgs[0].keys_manager.get_destination_script().unwrap();
+	let destination_script1 = chanmon_cfgs[1].keys_manager.get_destination_script().unwrap();
+	let persisters = vec![WatchtowerPersister::new(destination_script0),
+		WatchtowerPersister::new(destination_script1)];
+	let node_cfgs = create_node_cfgs_with_persisters(2, &chanmon_cfgs, persisters.iter().collect());
+	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+	let (_, _, channel_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1);
+	let funding_txo = OutPoint { txid: funding_tx.txid(), index: 0 };
+
+	if !broadcast_initial_commitment {
+		// Send a payment to move the channel forward
+		send_payment(&nodes[0], &vec!(&nodes[1])[..], 5_000_000);
+	}
+
+	// node[0] is gonna to revoke an old state thus node[1] should be able to claim the revoked output.
+	// We'll keep this commitment transaction to broadcast once it's revoked.
+	let revoked_local_txn = get_local_commitment_txn!(nodes[0], channel_id);
+	assert_eq!(revoked_local_txn.len(), 1);
+	let revoked_commitment_tx = &revoked_local_txn[0];
+
+	// Send another payment, now revoking the previous commitment tx
+	send_payment(&nodes[0], &vec!(&nodes[1])[..], 5_000_000);
+
+	let justice_tx = persisters[1].justice_tx(funding_txo, &revoked_commitment_tx.txid()).unwrap();
+	check_spends!(justice_tx, revoked_commitment_tx);
+
+	mine_transactions(&nodes[1], &[revoked_commitment_tx, &justice_tx]);
+	mine_transactions(&nodes[0], &[revoked_commitment_tx, &justice_tx]);
+
+	check_added_monitors!(nodes[1], 1);
+	check_closed_event(&nodes[1], 1, ClosureReason::CommitmentTxConfirmed, false,
+		&[nodes[0].node.get_our_node_id()], 100_000);
+	get_announce_close_broadcast_events(&nodes, 1, 0);
+
+	check_added_monitors!(nodes[0], 1);
+	check_closed_event(&nodes[0], 1, ClosureReason::CommitmentTxConfirmed, false,
+		&[nodes[1].node.get_our_node_id()], 100_000);
+
+	// Check that the justice tx has sent the revoked output value to nodes[1]
+	let monitor = get_monitor!(nodes[1], channel_id);
+	let total_claimable_balance = monitor.get_claimable_balances().iter().fold(0, |sum, balance| {
+		match balance {
+			channelmonitor::Balance::ClaimableAwaitingConfirmations { amount_satoshis, .. } => sum + amount_satoshis,
+			_ => panic!("Unexpected balance type"),
+		}
+	});
+	// On the first commitment, node[1]'s balance was below dust so it didn't have an output
+	let node1_channel_balance = if broadcast_initial_commitment { 0 } else { revoked_commitment_tx.output[0].value };
+	let expected_claimable_balance = node1_channel_balance + justice_tx.output[0].value;
+	assert_eq!(total_claimable_balance, expected_claimable_balance);
+}
+
+
 #[test]
 fn claim_htlc_outputs_shared_tx() {
 	// Node revoked old state, htlcs haven't time out yet, claim them in shared justice tx

lightning/src/util/test_utils.rs

@@ -11,6 +11,7 @@ use crate::chain;
 use crate::chain::WatchedOutput;
 use crate::chain::chaininterface;
 use crate::chain::chaininterface::ConfirmationTarget;
+use crate::chain::chaininterface::FEERATE_FLOOR_SATS_PER_KW;
 use crate::chain::chainmonitor;
 use crate::chain::chainmonitor::MonitorUpdateId;
 use crate::chain::channelmonitor;
@@ -20,6 +21,7 @@ use crate::sign;
 use crate::events;
 use crate::events::bump_transaction::{WalletSource, Utxo};
 use crate::ln::channelmanager;
+use crate::ln::chan_utils::CommitmentTransaction;
 use crate::ln::features::{ChannelFeatures, InitFeatures, NodeFeatures};
 use crate::ln::{msgs, wire};
 use crate::ln::msgs::LightningError;
@@ -272,6 +274,108 @@ impl<'a> chain::Watch<EnforcingSigner> for TestChainMonitor<'a> {
 	}
 }
 
+struct JusticeTxData {
+	justice_tx: Transaction,
+	value: u64,
+	commitment_number: u64,
+}
+
+pub(crate) struct WatchtowerPersister {
+	persister: TestPersister,
+	/// Upon a new commitment_signed, we'll get a
+	/// ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTxInfo. We'll store the justice tx
+	/// amount, and commitment number so we can build the justice tx after our counterparty
+	/// revokes it.
+	unsigned_justice_tx_data: Mutex<HashMap<OutPoint, VecDeque<JusticeTxData>>>,
+	/// After receiving a revoke_and_ack for a commitment number, we'll form and store the justice
+	/// tx which would be used to provide a watchtower with the data it needs.
+	watchtower_state: Mutex<HashMap<OutPoint, HashMap<Txid, Transaction>>>,
+	destination_script: Script,
+}
+
+impl WatchtowerPersister {
+	pub(crate) fn new(destination_script: Script) -> Self {
+		WatchtowerPersister {
+			persister: TestPersister::new(),
+			unsigned_justice_tx_data: Mutex::new(HashMap::new()),
+			watchtower_state: Mutex::new(HashMap::new()),
+			destination_script,
+		}
+	}
+
+	pub(crate) fn justice_tx(&self, funding_txo: OutPoint, commitment_txid: &Txid)
+	-> Option<Transaction> {
+		self.watchtower_state.lock().unwrap().get(&funding_txo).unwrap().get(commitment_txid).cloned()
+	}
+
+	fn form_justice_data_from_commitment(&self, counterparty_commitment_tx: &CommitmentTransaction)
+	-> Option<JusticeTxData> {
+		let trusted_tx = counterparty_commitment_tx.trust();
+		let output_idx = trusted_tx.revokeable_output_index()?;
+		let built_tx = trusted_tx.built_transaction();
+		let value = built_tx.transaction.output[output_idx as usize].value;
+		let justice_tx = trusted_tx.build_to_local_justice_tx(
+			FEERATE_FLOOR_SATS_PER_KW as u64, self.destination_script.clone()).ok()?;
+		let commitment_number = counterparty_commitment_tx.commitment_number();
+		Some(JusticeTxData { justice_tx, value, commitment_number })
+	}
+}
+
+impl<Signer: sign::WriteableEcdsaChannelSigner> chainmonitor::Persist<Signer> for WatchtowerPersister {
+	fn persist_new_channel(&self, funding_txo: OutPoint,
+		data: &channelmonitor::ChannelMonitor<Signer>, id: MonitorUpdateId
+	) -> chain::ChannelMonitorUpdateStatus {
+		let res = self.persister.persist_new_channel(funding_txo, data, id);
+
+		assert!(self.unsigned_justice_tx_data.lock().unwrap()
+			.insert(funding_txo, VecDeque::new()).is_none());
+		assert!(self.watchtower_state.lock().unwrap()
+			.insert(funding_txo, HashMap::new()).is_none());
+
+		let initial_counterparty_commitment_tx = data.initial_counterparty_commitment_tx()
+			.expect("First and only call expects Some");
+		if let Some(justice_data)
+			= self.form_justice_data_from_commitment(&initial_counterparty_commitment_tx) {
+			self.unsigned_justice_tx_data.lock().unwrap()
+				.get_mut(&funding_txo).unwrap()
+				.push_back(justice_data);
+		}
+		res
+	}
+
+	fn update_persisted_channel(
+		&self, funding_txo: OutPoint, update: Option<&channelmonitor::ChannelMonitorUpdate>,
+		data: &channelmonitor::ChannelMonitor<Signer>, update_id: MonitorUpdateId
+	) -> chain::ChannelMonitorUpdateStatus {
+		let res = self.persister.update_persisted_channel(funding_txo, update, data, update_id);
+
+		if let Some(update) = update {
+			let commitment_txs = data.counterparty_commitment_txs_from_update(update);
+			let justice_datas = commitment_txs.into_iter()
+				.filter_map(|commitment_tx| self.form_justice_data_from_commitment(&commitment_tx));
+			let mut channels_justice_txs = self.unsigned_justice_tx_data.lock().unwrap();
+			let channel_state = channels_justice_txs.get_mut(&funding_txo).unwrap();
+			channel_state.extend(justice_datas);
+
+			while let Some(JusticeTxData { justice_tx, value, commitment_number }) = channel_state.front() {
+				let input_idx = 0;
+				let commitment_txid = justice_tx.input[input_idx].previous_output.txid;
+				match data.sign_to_local_justice_tx(justice_tx.clone(), input_idx, *value, *commitment_number) {
+					Ok(signed_justice_tx) => {
+						let dup = self.watchtower_state.lock().unwrap()
+							.get_mut(&funding_txo).unwrap()
+							.insert(commitment_txid, signed_justice_tx);
+						assert!(dup.is_none());
+						channel_state.pop_front();
+					},
+					Err(_) => break,
+				}
+			}
+		}
+		res
+	}
+}
+
 pub struct TestPersister {
 	/// The queue of update statuses we'll return. If none are queued, ::Completed will always be
 	/// returned.