Augment the chanmon fuzzer with async ops

Originally from Matt. This augments the channel monitor consistency tests with
ops that disable and enable the signer to simulate asynchronous remote
signing.

Author: waterson

fuzz/src/chanmon_consistency.rs

@@ -46,7 +46,7 @@ use lightning::ln::script::ShutdownScript;
 use lightning::ln::functional_test_utils::*;
 use lightning::offers::invoice::UnsignedBolt12Invoice;
 use lightning::offers::invoice_request::UnsignedInvoiceRequest;
-use lightning::util::test_channel_signer::{TestChannelSigner, EnforcementState};
+use lightning::util::test_channel_signer::{TestChannelSigner, EnforcementState, ops};
 use lightning::util::errors::APIError;
 use lightning::util::logger::Logger;
 use lightning::util::config::UserConfig;
@@ -69,6 +69,7 @@ use std::sync::atomic;
 use std::io::Cursor;
 use bitcoin::bech32::u5;
 
+const ASYNC_OPS: u32 = ops::GET_PER_COMMITMENT_POINT | ops::RELEASE_COMMITMENT_SECRET | ops::SIGN_COUNTERPARTY_COMMITMENT;
 const MAX_FEE: u32 = 10_000;
 struct FuzzEstimator {
 	ret_val: atomic::AtomicU32,
@@ -168,10 +169,15 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 	}
 }
 
+struct SignerState {
+	enforcement: Arc<Mutex<EnforcementState>>,
+	unavailable: Arc<Mutex<u32>>,
+}
+
 struct KeyProvider {
 	node_secret: SecretKey,
 	rand_bytes_id: atomic::AtomicU32,
-	enforcement_states: Mutex<HashMap<[u8;32], Arc<Mutex<EnforcementState>>>>,
+	enforcement_states: Mutex<HashMap<[u8;32], SignerState>>,
 }
 
 impl EntropySource for KeyProvider {
@@ -253,22 +259,19 @@ impl SignerProvider for KeyProvider {
 			channel_keys_id,
 			channel_keys_id,
 		);
-		let revoked_commitment = self.make_enforcement_state_cell(keys.commitment_seed);
-		TestChannelSigner::new_with_revoked(keys, revoked_commitment, false)
+		let mut revoked_commitments = self.enforcement_states.lock().unwrap();
+		let new_state = revoked_commitments.entry(keys.commitment_seed)
+			.or_insert(SignerState {
+				enforcement: Arc::new(Mutex::new(EnforcementState::new())),
+				unavailable: Arc::new(Mutex::new(0)),
+			});
+		let mut ret = TestChannelSigner::new_with_revoked(keys, Arc::clone(&new_state.enforcement), false);
+		ret.unavailable = Arc::clone(&new_state.unavailable);
+		ret
 	}
 
-	fn read_chan_signer(&self, buffer: &[u8]) -> Result<Self::Signer, DecodeError> {
-		let mut reader = std::io::Cursor::new(buffer);
-
-		let inner: InMemorySigner = ReadableArgs::read(&mut reader, self)?;
-		let state = self.make_enforcement_state_cell(inner.commitment_seed);
-
-		Ok(TestChannelSigner {
-			inner,
-			state,
-			disable_revocation_policy_check: false,
-			unavailable: Arc::new(Mutex::new(0)),
-		})
+	fn read_chan_signer(&self, _buffer: &[u8]) -> Result<Self::Signer, DecodeError> {
+		unreachable!();
 	}
 
 	fn get_destination_script(&self) -> Result<Script, ()> {
@@ -286,17 +289,6 @@ impl SignerProvider for KeyProvider {
 	}
 }
 
-impl KeyProvider {
-	fn make_enforcement_state_cell(&self, commitment_seed: [u8; 32]) -> Arc<Mutex<EnforcementState>> {
-		let mut revoked_commitments = self.enforcement_states.lock().unwrap();
-		if !revoked_commitments.contains_key(&commitment_seed) {
-			revoked_commitments.insert(commitment_seed, Arc::new(Mutex::new(EnforcementState::new())));
-		}
-		let cell = revoked_commitments.get(&commitment_seed).unwrap();
-		Arc::clone(cell)
-	}
-}
-
 #[inline]
 fn check_api_err(api_err: APIError, sendable_bounds_violated: bool) {
 	match api_err {
@@ -799,7 +791,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == node_id {
 									for update_add in update_add_htlcs.iter() {
-										out.locked_write(format!("Delivering update_add_htlc to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_add_htlc to node {} from node {}.\n", idx, $node).as_bytes());
 										if !$corrupt_forward {
 											dest.handle_update_add_htlc(&nodes[$node].get_our_node_id(), update_add);
 										} else {
@@ -814,19 +806,19 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 										}
 									}
 									for update_fulfill in update_fulfill_htlcs.iter() {
-										out.locked_write(format!("Delivering update_fulfill_htlc to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_fulfill_htlc to node {} from node {}.\n", idx, $node).as_bytes());
 										dest.handle_update_fulfill_htlc(&nodes[$node].get_our_node_id(), update_fulfill);
 									}
 									for update_fail in update_fail_htlcs.iter() {
-										out.locked_write(format!("Delivering update_fail_htlc to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_fail_htlc to node {} from node {}.\n", idx, $node).as_bytes());
 										dest.handle_update_fail_htlc(&nodes[$node].get_our_node_id(), update_fail);
 									}
 									for update_fail_malformed in update_fail_malformed_htlcs.iter() {
-										out.locked_write(format!("Delivering update_fail_malformed_htlc to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_fail_malformed_htlc to node {} from node {}.\n", idx, $node).as_bytes());
 										dest.handle_update_fail_malformed_htlc(&nodes[$node].get_our_node_id(), update_fail_malformed);
 									}
 									if let Some(msg) = update_fee {
-										out.locked_write(format!("Delivering update_fee to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_fee to node {} from node {}.\n", idx, $node).as_bytes());
 										dest.handle_update_fee(&nodes[$node].get_our_node_id(), &msg);
 									}
 									let processed_change = !update_add_htlcs.is_empty() || !update_fulfill_htlcs.is_empty() ||
@@ -843,7 +835,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 										} });
 										break;
 									}
-									out.locked_write(format!("Delivering commitment_signed to node {}.\n", idx).as_bytes());
+									out.locked_write(format!("Delivering commitment_signed to node {} from node {}.\n", idx, $node).as_bytes());
 									dest.handle_commitment_signed(&nodes[$node].get_our_node_id(), &commitment_signed);
 									break;
 								}
@@ -852,15 +844,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 						events::MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == *node_id {
-									out.locked_write(format!("Delivering revoke_and_ack to node {}.\n", idx).as_bytes());
+									out.locked_write(format!("Delivering revoke_and_ack to node {} from node {}.\n", idx, $node).as_bytes());
 									dest.handle_revoke_and_ack(&nodes[$node].get_our_node_id(), msg);
 								}
 							}
 						},
 						events::MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == *node_id {
-									out.locked_write(format!("Delivering channel_reestablish to node {}.\n", idx).as_bytes());
+									out.locked_write(format!("Delivering channel_reestablish to node {} from node {}.\n", idx, $node).as_bytes());
 									dest.handle_channel_reestablish(&nodes[$node].get_our_node_id(), msg);
 								}
 							}
@@ -1259,6 +1251,102 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			},
 			0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },
 
+			0xa0 => {
+				let signer_states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() = ASYNC_OPS;
+			}
+			0xa1 => {
+				let signer_states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::GET_PER_COMMITMENT_POINT;
+				nodes[0].signer_unblocked(None);
+			}
+			0xa2 => {
+				let signer_states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::RELEASE_COMMITMENT_SECRET;
+				nodes[0].signer_unblocked(None);
+			}
+			0xa3 => {
+				let signer_states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::SIGN_COUNTERPARTY_COMMITMENT;
+				nodes[0].signer_unblocked(None);
+			}
+
+			0xa4 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() = ASYNC_OPS;
+			}
+			0xa5 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::GET_PER_COMMITMENT_POINT;
+				nodes[1].signer_unblocked(None);
+			}
+			0xa6 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::RELEASE_COMMITMENT_SECRET;
+				nodes[1].signer_unblocked(None);
+			}
+			0xa7 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::SIGN_COUNTERPARTY_COMMITMENT;
+				nodes[1].signer_unblocked(None);
+			}
+
+			0xa8 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().last().unwrap().unavailable.lock().unwrap() = ASYNC_OPS;
+			}
+			0xa9 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().last().unwrap().unavailable.lock().unwrap() &= !ops::GET_PER_COMMITMENT_POINT;
+				nodes[1].signer_unblocked(None);
+			}
+			0xaa => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().last().unwrap().unavailable.lock().unwrap() &= !ops::RELEASE_COMMITMENT_SECRET;
+				nodes[1].signer_unblocked(None);
+			}
+			0xab => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().last().unwrap().unavailable.lock().unwrap() &= !ops::SIGN_COUNTERPARTY_COMMITMENT;
+				nodes[1].signer_unblocked(None);
+			}
+
+			0xac => {
+				let signer_states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() = ASYNC_OPS;
+			}
+			0xad => {
+				let signer_states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::GET_PER_COMMITMENT_POINT;
+				nodes[2].signer_unblocked(None);
+			}
+			0xae => {
+				let signer_states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::RELEASE_COMMITMENT_SECRET;
+				nodes[2].signer_unblocked(None);
+			}
+			0xaf => {
+				let signer_states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::SIGN_COUNTERPARTY_COMMITMENT;
+				nodes[2].signer_unblocked(None);
+			}
+
 			0xff => {
 				// Test that no channel is in a stuck state where neither party can send funds even
 				// after we resolve all pending events.
@@ -1268,6 +1356,19 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
 				*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
 
+				for signer_state in keys_manager_a.enforcement_states.lock().unwrap().values() {
+					*signer_state.unavailable.lock().unwrap() = 0;
+				}
+				for signer_state in keys_manager_b.enforcement_states.lock().unwrap().values() {
+					*signer_state.unavailable.lock().unwrap() = 0;
+				}
+				for signer_state in keys_manager_c.enforcement_states.lock().unwrap().values() {
+					*signer_state.unavailable.lock().unwrap() = 0;
+				}
+				nodes[0].signer_unblocked(None);
+				nodes[1].signer_unblocked(None);
+				nodes[2].signer_unblocked(None);
+
 				if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
 					monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
 					nodes[0].process_monitor_events();