diff --git a/watchtower-plugin/README.md b/watchtower-plugin/README.md index 9afe8ab..e3ecdac 100644 --- a/watchtower-plugin/README.md +++ b/watchtower-plugin/README.md @@ -8,12 +8,15 @@ commitment transaction is generated. It also keeps a summary of the messages sen The plugin has the following methods: -- `registertower tower_id` : registers the user id (compressed public key) with a given tower. +- `registertower `: registers the user id (compressed public key) with a given tower. +- `gettowerinfo `: gets all the locally stored data about a given tower. +- `retrytower `: tries to send pending appointment to a (previously) unreachable tower. +- `abandontower `: deletes all data associated with a given tower. - `listtowers`: lists all registered towers. -- `gettowerinfo tower_id`: gets all the locally stored data about a given tower. -- `getsubscriptioninfo tower_id`: gets the subscription information by querying the tower. -- `retrytower tower_id`: tries to send pending appointment to a (previously) unreachable tower. -- `getappointment tower_id locator`: queries a given tower about an appointment. +- `getappointment `: queries a given tower about an appointment. +- `getsubscriptioninfo `: gets the subscription information by querying the tower. +- `getappointmentreceipt `: pulls a given appointment receipt from the local database. +- `getregistrationreceipt `: pulls the latest registration receipt from the local database. The plugin also has an implicit method to send appointments to the registered towers for every new commitment transaction. diff --git a/watchtower-plugin/src/dbm.rs b/watchtower-plugin/src/dbm.rs index 4a0c647..a59458c 100755 --- a/watchtower-plugin/src/dbm.rs +++ b/watchtower-plugin/src/dbm.rs @@ -244,6 +244,15 @@ impl DBM { Ok(receipt) } + /// Removes a tower record from the database. + /// + /// This triggers a cascade deletion of all related data, such as appointments, appointment receipts, etc. As long as there is a single + /// reference to them. + pub fn remove_tower_record(&self, tower_id: TowerId) -> Result<(), Error> { + let query = "DELETE FROM towers WHERE tower_id=?"; + self.remove_data(query, params![tower_id.to_vec()]) + } + /// Loads all tower records from the database. pub fn load_towers(&self) -> HashMap { let mut towers = HashMap::new(); @@ -316,7 +325,32 @@ impl DBM { tx.commit() } - /// Loads the appointment receipts associated to a given tower + /// Loads a given appointment receipt of a given tower from the database. + pub fn load_appointment_receipt( + &self, + tower_id: TowerId, + locator: Locator, + ) -> Result { + let mut stmt = self + .connection + .prepare("SELECT * FROM appointment_receipts WHERE tower_id = ?1 and locator = ?2") + .unwrap(); + + stmt.query_row(params![tower_id.to_vec(), locator.to_vec()], |row| { + let start_block = row.get::<_, u32>(2).unwrap(); + let user_sig = row.get::<_, String>(3).unwrap(); + let tower_sig = row.get::<_, String>(4).unwrap(); + + Ok(AppointmentReceipt::with_signature( + user_sig, + start_block, + tower_sig, + )) + }) + .map_err(|_| Error::NotFound) + } + + /// Loads the appointment receipts associated to a given tower. /// /// TODO: Currently this is only loading a summary of the receipt, if we need to really load all the information /// for any reason this method may need to be renamed. @@ -623,6 +657,19 @@ mod tests { .unwrap(); stmt.exists(params![locator.to_vec()]).unwrap() } + + pub(crate) fn appointment_receipt_exists( + &self, + locator: Locator, + tower_id: TowerId, + ) -> bool { + let mut stmt = self + .connection + .prepare("SELECT * FROM appointment_receipts WHERE locator=?1 AND tower_id=?2 ") + .unwrap(); + stmt.exists(params![locator.to_vec(), tower_id.to_vec()]) + .unwrap() + } } #[test] @@ -776,6 +823,29 @@ mod tests { assert_eq!(dbm.load_towers(), HashMap::new()); } + #[test] + fn test_remove_tower_record() { + let mut dbm = DBM::in_memory().unwrap(); + + let tower_id = get_random_user_id(); + let net_addr = "talaia.watch"; + let receipt = get_random_registration_receipt(); + dbm.store_tower_record(tower_id, net_addr, &receipt) + .unwrap(); + + assert!(matches!(dbm.remove_tower_record(tower_id), Ok(()))); + } + + #[test] + fn test_remove_tower_record_inexistent() { + let dbm = DBM::in_memory().unwrap(); + + assert!(matches!( + dbm.remove_tower_record(get_random_user_id()), + Err(Error::NotFound) + )); + } + #[test] fn test_store_load_appointment_receipts() { let mut dbm = DBM::in_memory().unwrap(); @@ -823,6 +893,57 @@ mod tests { assert_eq!(dbm.load_appointment_receipts(tower_id), receipts); } + #[test] + fn test_load_appointment_receipt() { + let mut dbm = DBM::in_memory().unwrap(); + let tower_id = get_random_user_id(); + let appointment = generate_random_appointment(None); + + // If there is no appointment receipt for the given (locator, tower_id) pair, Error::NotFound is returned + // Try first with both being unknown + assert!(matches!( + dbm.load_appointment_receipt(tower_id, appointment.locator), + Err(Error::NotFound) + )); + + // Add the tower but not the appointment and try again + let net_addr = "talaia.watch"; + let receipt = get_random_registration_receipt(); + dbm.store_tower_record(tower_id, net_addr, &receipt) + .unwrap(); + + assert!(matches!( + dbm.load_appointment_receipt(tower_id, appointment.locator), + Err(Error::NotFound) + )); + + // Add both + let tower_summary = TowerSummary::new( + net_addr.into(), + receipt.available_slots(), + receipt.subscription_start(), + receipt.subscription_expiry(), + ); + let appointment_receipt = AppointmentReceipt::with_signature( + "user_signature".into(), + 42, + "tower_signature".into(), + ); + dbm.store_appointment_receipt( + tower_id, + appointment.locator, + tower_summary.available_slots, + &appointment_receipt, + ) + .unwrap(); + + assert_eq!( + dbm.load_appointment_receipt(tower_id, appointment.locator) + .unwrap(), + appointment_receipt + ); + } + #[test] fn test_load_appointment_locators() { // `load_appointment_locators` is used to load locators from either `appointment_receipts`, `pending_appointments` or `invalid_appointments` diff --git a/watchtower-plugin/src/main.rs b/watchtower-plugin/src/main.rs index c5f3a16..c64982b 100755 --- a/watchtower-plugin/src/main.rs +++ b/watchtower-plugin/src/main.rs @@ -131,6 +131,68 @@ async fn register( Ok(json!(receipt)) } +/// Gets the latest registration receipt from the client to a given tower (if it exists). +/// +/// This is pulled from the database +async fn get_registration_receipt( + plugin: Plugin>>, + v: serde_json::Value, +) -> Result { + let tower_id = TowerId::try_from(v).map_err(|x| anyhow!(x))?; + let state = plugin.state().lock().unwrap(); + + let response = state.get_registration_receipt(tower_id).map_err(|_| { + anyhow!( + "Cannot find {} within the known towers. Have you registered?", + tower_id + ) + })?; + + Ok(json!(response)) +} + +/// Gets the subscription information directly form the tower. +async fn get_subscription_info( + plugin: Plugin>>, + v: serde_json::Value, +) -> Result { + let tower_id = TowerId::try_from(v).map_err(|x| anyhow!(x))?; + + let user_sk = plugin.state().lock().unwrap().user_sk; + let tower_net_addr = { + let state = plugin.state().lock().unwrap(); + if let Some(info) = state.towers.get(&tower_id) { + Ok(info.net_addr.clone()) + } else { + Err(anyhow!("Unknown tower id: {}", tower_id)) + } + }?; + + let get_subscription_info = format!("{}/get_subscription_info", tower_net_addr); + let signature = cryptography::sign("get subscription info".as_bytes(), &user_sk).unwrap(); + + let response: common_msgs::GetSubscriptionInfoResponse = process_post_response( + post_request( + &get_subscription_info, + &common_msgs::GetSubscriptionInfoRequest { signature }, + ) + .await, + ) + .await + .map_err(|e| { + if e.is_connection() { + plugin + .state() + .lock() + .unwrap() + .set_tower_status(tower_id, TowerStatus::TemporaryUnreachable); + } + to_cln_error(e) + })?; + + Ok(json!(response)) +} + /// Gets information about an appointment from the tower. async fn get_appointment( plugin: Plugin>>, @@ -180,44 +242,32 @@ async fn get_appointment( Ok(json!(response)) } -/// Gets the subscription information directly form the tower. -async fn get_subscription_info( +/// Gets an appointment receipt from the client given a tower_id and a locator (if it exists). +/// +/// This is pulled from the database +async fn get_appointment_receipt( plugin: Plugin>>, v: serde_json::Value, ) -> Result { - let tower_id = TowerId::try_from(v).map_err(|x| anyhow!(x))?; + let params = GetAppointmentParams::try_from(v).map_err(|x| anyhow!(x))?; + let state = plugin.state().lock().unwrap(); - let user_sk = plugin.state().lock().unwrap().user_sk; - let tower_net_addr = { - let state = plugin.state().lock().unwrap(); - if let Some(info) = state.towers.get(&tower_id) { - Ok(info.net_addr.clone()) - } else { - Err(anyhow!("Unknown tower id: {}", tower_id)) - } - }?; - - let get_subscription_info = format!("{}/get_subscription_info", tower_net_addr); - let signature = cryptography::sign("get subscription info".as_bytes(), &user_sk).unwrap(); - - let response: common_msgs::GetSubscriptionInfoResponse = process_post_response( - post_request( - &get_subscription_info, - &common_msgs::GetSubscriptionInfoRequest { signature }, - ) - .await, - ) - .await - .map_err(|e| { - if e.is_connection() { - plugin - .state() - .lock() - .unwrap() - .set_tower_status(tower_id, TowerStatus::TemporaryUnreachable); - } - to_cln_error(e) - })?; + let response = state + .get_appointment_receipt(params.tower_id, params.locator) + .map_err(|_| { + if state.towers.contains_key(¶ms.tower_id) { + anyhow!( + "Cannot find {} within {}. Did you send that appointment?", + params.locator, + params.tower_id + ) + } else { + anyhow!( + "Cannot find {} within the known towers. Have you registered?", + params.tower_id + ) + } + })?; Ok(json!(response)) } @@ -283,6 +333,21 @@ async fn retry_tower( } } +/// Forgets about a tower wiping out all local data associated to it. +async fn abandon_tower( + plugin: Plugin>>, + v: serde_json::Value, +) -> Result { + let tower_id = TowerId::try_from(v).map_err(|e| anyhow!(e))?; + let mut state = plugin.state().lock().unwrap(); + if state.towers.get(&tower_id).is_some() { + state.remove_tower(tower_id).unwrap(); + Ok(json!(format!("{} successfully abandoned", tower_id))) + } else { + Err(anyhow!("Unknown tower {}", tower_id)) + } +} + /// Sends an appointment to all registered towers for every new commitment transaction. /// /// The appointment is built using the data provided by the backend (dispute txid and penalty transaction). @@ -444,11 +509,19 @@ async fn main() -> Result<(), Error> { "registertower", "Registers the client public key (user id) with the tower.", register, + ).rpcmethod( + "getregistrationreceipt", + "Gets the latest registration receipt given a tower id.", + get_registration_receipt, ) .rpcmethod( "getappointment", "Gets appointment data from the tower given the tower id and the locator.", get_appointment, + ).rpcmethod( + "getappointmentreceipt", + "Gets a (local) appointment receipt given a tower id and an locator.", + get_appointment_receipt, ) .rpcmethod( "getsubscriptioninfo", @@ -466,6 +539,11 @@ async fn main() -> Result<(), Error> { "Retries to send pending appointment to an unreachable tower.", retry_tower, ) + .rpcmethod( + "abandontower", + "Forgets about a tower and wipes all local data.", + abandon_tower, + ) .hook("commitment_revocation", on_commitment_revocation); if let Some(plugin) = builder.start().await? { diff --git a/watchtower-plugin/src/retrier.rs b/watchtower-plugin/src/retrier.rs index 1f796fa..6decb63 100644 --- a/watchtower-plugin/src/retrier.rs +++ b/watchtower-plugin/src/retrier.rs @@ -36,10 +36,15 @@ impl Retrier { loop { let tower_id = unreachable_towers.recv().await.unwrap(); - self.wt_client - .lock() - .unwrap() - .set_tower_status(tower_id, crate::TowerStatus::TemporaryUnreachable); + { + // Not start a retry if the tower is flagged to be abandoned + let mut wt_client = self.wt_client.lock().unwrap(); + if wt_client.towers.get(&tower_id).is_none() { + log::info!("Skipping retrying abandoned tower {}", tower_id); + continue; + } + wt_client.set_tower_status(tower_id, crate::TowerStatus::TemporaryUnreachable); + } log::info!("Retrying tower {}", tower_id); match retry_notify( @@ -67,15 +72,13 @@ impl Retrier { // Notice we'll end up here after a permanent error. That is, either after finishing the backoff strategy // unsuccessfully or by manually raising such an error (like when facing a tower misbehavior) let mut wt_client = self.wt_client.lock().unwrap(); - if wt_client - .towers - .get(&tower_id) - .unwrap() - .status - .is_unreachable() - { - log::warn!("Setting {} as unreachable", tower_id); - wt_client.set_tower_status(tower_id, crate::TowerStatus::Unreachable); + if let Some(tower) = wt_client.towers.get_mut(&tower_id) { + if tower.status.is_unreachable() { + log::warn!("Setting {} as unreachable", tower_id); + wt_client.set_tower_status(tower_id, crate::TowerStatus::Unreachable); + } + } else { + log::info!("Skipping retrying abandoned tower {}", tower_id); } } } @@ -86,6 +89,10 @@ impl Retrier { // Create a new scope so we can get all the data only locking the WTClient once. let (appointments, net_addr, user_sk) = { let wt_client = self.wt_client.lock().unwrap(); + if wt_client.towers.get(&tower_id).is_none() { + return Err(Error::permanent("Tower was abandoned. Skipping retry")); + } + let appointments = wt_client .dbm .lock() @@ -410,7 +417,7 @@ mod tests { } #[tokio::test] - async fn test_retry_misbehaving() { + async fn test_manage_retry_misbehaving() { let tmp_path = &format!(".watchtower_{}/", get_random_user_id()); let (tx, rx) = unbounded_channel(); let wt_client = Arc::new(Mutex::new(WTClient::new(tmp_path.into(), tx.clone()).await)); @@ -474,6 +481,43 @@ mod tests { fs::remove_dir_all(tmp_path).await.unwrap(); } + #[tokio::test] + async fn test_manage_retry_abandoned() { + let tmp_path = &format!(".watchtower_{}/", get_random_user_id()); + let (tx, rx) = unbounded_channel(); + let wt_client = Arc::new(Mutex::new(WTClient::new(tmp_path.into(), tx.clone()).await)); + let server = MockServer::start(); + + // Add a tower with pending appointments + let (_, tower_pk) = cryptography::get_random_keypair(); + let tower_id = TowerId(tower_pk); + let receipt = get_random_registration_receipt(); + wt_client + .lock() + .unwrap() + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); + + // Remove the tower (to simulate it has been abandoned) + wt_client.lock().unwrap().remove_tower(tower_id).unwrap(); + + // Start the task and send the tower to the channel for retry + let wt_client_clone = wt_client.clone(); + let task = tokio::spawn(async move { + Retrier::new(wt_client_clone, MAX_ELAPSED_TIME, MAX_INTERVAL_TIME) + .manage_retry(rx) + .await + }); + + // Send the id and check how it gets removed + tx.send(tower_id).unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + assert!(!wt_client.lock().unwrap().towers.contains_key(&tower_id)); + + task.abort(); + fs::remove_dir_all(tmp_path).await.unwrap(); + } + #[tokio::test] async fn test_add_appointment() { let (tower_sk, tower_pk) = cryptography::get_random_keypair(); @@ -715,4 +759,36 @@ mod tests { fs::remove_dir_all(tmp_path).await.unwrap(); } + + #[tokio::test] + async fn test_add_appointment_abandoned() { + let (_, tower_pk) = cryptography::get_random_keypair(); + let tower_id = TowerId(tower_pk); + let tmp_path = &format!(".watchtower_{}/", get_random_user_id()); + let wt_client = Arc::new(Mutex::new( + WTClient::new(tmp_path.into(), unbounded_channel().0).await, + )); + let server = MockServer::start(); + + // The tower we'd like to retry sending appointments to has to exist within the plugin + let receipt = get_random_registration_receipt(); + wt_client + .lock() + .unwrap() + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); + + // Remove the tower (to simulate it has been abandoned) + wt_client.lock().unwrap().remove_tower(tower_id).unwrap(); + + // If there are no pending appointments the method will simply return + let r = Retrier::dummy(wt_client).add_appointment(tower_id).await; + + assert_eq!( + r, + Err(Error::permanent("Tower was abandoned. Skipping retry")) + ); + + fs::remove_dir_all(tmp_path).await.unwrap(); + } } diff --git a/watchtower-plugin/src/wt_client.rs b/watchtower-plugin/src/wt_client.rs index e2bf439..f75ca10 100644 --- a/watchtower-plugin/src/wt_client.rs +++ b/watchtower-plugin/src/wt_client.rs @@ -116,6 +116,17 @@ impl WTClient { Ok(()) } + /// Gets the latest registration receipt of a given tower. + pub fn get_registration_receipt( + &self, + tower_id: TowerId, + ) -> Result { + self.dbm + .lock() + .unwrap() + .load_registration_receipt(tower_id, self.user_id) + } + /// Loads a tower record from the database. pub fn load_tower_info(&self, tower_id: TowerId) -> Result { self.dbm.lock().unwrap().load_tower_record(tower_id) @@ -159,6 +170,18 @@ impl WTClient { } } + /// Gets an appointment receipt from the database (if found). + pub fn get_appointment_receipt( + &self, + tower_id: TowerId, + locator: Locator, + ) -> Result { + self.dbm + .lock() + .unwrap() + .load_appointment_receipt(tower_id, locator) + } + /// Adds a pending appointment to the tower record. pub fn add_pending_appointment(&mut self, tower_id: TowerId, appointment: &Appointment) { if let Some(tower) = self.towers.get_mut(&tower_id) { @@ -226,6 +249,18 @@ impl WTClient { log::error!("Cannot flag tower. Unknown tower_id: {}", tower_id); } } + + /// Removes a tower from the client (both memory and database). + /// + /// Any data associated to the tower will be deleted (i.e. links to appointments) + pub fn remove_tower(&mut self, tower_id: TowerId) -> Result<(), DBError> { + if self.towers.contains_key(&tower_id) { + self.towers.remove(&tower_id); + self.dbm.lock().unwrap().remove_tower_record(tower_id) + } else { + Err(DBError::NotFound) + } + } } #[cfg(test)] @@ -705,4 +740,166 @@ mod tests { assert_eq!(loaded_info.misbehaving_proof, Some(proof)); assert!(loaded_info.appointments.contains_key(&appointment.locator)); } + + #[tokio::test] + async fn test_remove_tower() { + let tmp_path = &format!(".watchtower_{}/", get_random_user_id()); + let mut wt_client = WTClient::new(tmp_path.into(), unbounded_channel().0).await; + + let receipt = get_random_registration_receipt(); + let (tower_sk, tower_pk) = cryptography::get_random_keypair(); + let tower_id = TowerId(tower_pk); + let tower_info = TowerInfo::empty( + "talaia.watch".into(), + receipt.available_slots(), + receipt.subscription_start(), + receipt.subscription_expiry(), + ); + + // Add the tower and check it is there + wt_client + .add_update_tower(tower_id, tower_info.net_addr.clone(), &receipt) + .unwrap(); + assert_eq!( + wt_client.towers.get(&tower_id), + Some(&TowerSummary::from(tower_info.clone())) + ); + assert_eq!(wt_client.load_tower_info(tower_id).unwrap(), tower_info); + + // Remove the tower and check it is not there anymore + wt_client.remove_tower(tower_id).unwrap(); + assert!(matches!( + wt_client.load_tower_info(tower_id), + Err(DBError::NotFound) + )); + assert!(!wt_client.towers.contains_key(&tower_id)); + + // Try again but this time with an associated appointment to check that it also gets removed + wt_client + .add_update_tower(tower_id, tower_info.net_addr, &receipt) + .unwrap(); + + let locator = generate_random_appointment(None).locator; + let registration_receipt = get_random_registration_receipt(); + let appointment_receipt = get_random_appointment_receipt(tower_sk); + + // If we call this on an unknown tower it will simply do nothing + wt_client.add_appointment_receipt( + tower_id, + locator, + registration_receipt.available_slots(), + &appointment_receipt, + ); + assert!(wt_client + .dbm + .lock() + .unwrap() + .appointment_receipt_exists(locator, tower_id)); + + // Remove and check both the tower and the appointment + wt_client.remove_tower(tower_id).unwrap(); + assert!(matches!( + wt_client.load_tower_info(tower_id), + Err(DBError::NotFound) + )); + assert!(!wt_client.towers.contains_key(&tower_id)); + assert!(!wt_client + .dbm + .lock() + .unwrap() + .appointment_receipt_exists(locator, tower_id)); + + fs::remove_dir_all(tmp_path).await.unwrap(); + } + + #[tokio::test] + async fn test_remove_tower_shared_appointment() { + // Lets test removing a tower that has associated data shared with another tower. + // For instance, having an appointment that was sent to two towers, and then deleting one of them + // should only remove the link between the tower and the appointment, but not delete the data. + let tmp_path = &format!(".watchtower_{}/", get_random_user_id()); + let mut wt_client = WTClient::new(tmp_path.into(), unbounded_channel().0).await; + + let receipt = get_random_registration_receipt(); + let (tower1_sk, tower1_pk) = cryptography::get_random_keypair(); + let tower1_id = TowerId(tower1_pk); + let (tower2_sk, tower2_pk) = cryptography::get_random_keypair(); + let tower2_id = TowerId(tower2_pk); + + let tower_info = TowerInfo::empty( + "talaia.watch".into(), + receipt.available_slots(), + receipt.subscription_start(), + receipt.subscription_expiry(), + ); + wt_client + .add_update_tower(tower1_id, tower_info.net_addr.clone(), &receipt) + .unwrap(); + wt_client + .add_update_tower(tower2_id, tower_info.net_addr, &receipt) + .unwrap(); + + let locator = generate_random_appointment(None).locator; + let registration_receipt = get_random_registration_receipt(); + let appointment_receipt_1 = get_random_appointment_receipt(tower1_sk); + let appointment_receipt_2 = get_random_appointment_receipt(tower2_sk); + + wt_client.add_appointment_receipt( + tower1_id, + locator, + registration_receipt.available_slots(), + &appointment_receipt_1, + ); + wt_client.add_appointment_receipt( + tower2_id, + locator, + registration_receipt.available_slots(), + &appointment_receipt_2, + ); + + // Check that the data exists in both towers + assert!(wt_client + .dbm + .lock() + .unwrap() + .appointment_receipt_exists(locator, tower1_id)); + assert!(wt_client + .dbm + .lock() + .unwrap() + .appointment_receipt_exists(locator, tower2_id)); + + // Remove tower1 and check that the appointment receipt can still be found for tower2 + wt_client.remove_tower(tower1_id).unwrap(); + assert!(matches!( + wt_client.load_tower_info(tower1_id), + Err(DBError::NotFound) + )); + + assert!(!wt_client + .dbm + .lock() + .unwrap() + .appointment_receipt_exists(locator, tower1_id)); + assert!(wt_client + .dbm + .lock() + .unwrap() + .appointment_receipt_exists(locator, tower2_id)); + + fs::remove_dir_all(tmp_path).await.unwrap(); + } + + #[tokio::test] + async fn test_remove_inexistent_tower() { + let tmp_path = &format!(".watchtower_{}/", get_random_user_id()); + let mut wt_client = WTClient::new(tmp_path.into(), unbounded_channel().0).await; + + assert!(matches!( + wt_client.remove_tower(get_random_user_id()), + Err(DBError::NotFound) + )); + + fs::remove_dir_all(tmp_path).await.unwrap(); + } }