diff --git a/teos-common/proto/common/teos/v2/user.proto b/teos-common/proto/common/teos/v2/user.proto index 83be03d..84e54d7 100644 --- a/teos-common/proto/common/teos/v2/user.proto +++ b/teos-common/proto/common/teos/v2/user.proto @@ -12,8 +12,9 @@ message RegisterRequest { bytes user_id = 1; uint32 available_slots = 2; - uint32 subscription_expiry = 3; - string subscription_signature = 4; + uint32 subscription_start = 3; + uint32 subscription_expiry = 4; + string subscription_signature = 5; } message GetSubscriptionInfoRequest { diff --git a/teos-common/src/receipts.rs b/teos-common/src/receipts.rs index a2d4e11..7679648 100644 --- a/teos-common/src/receipts.rs +++ b/teos-common/src/receipts.rs @@ -6,20 +6,40 @@ use bitcoin::secp256k1::SecretKey; use crate::{cryptography, UserId}; -#[derive(Serialize, Debug)] +/// Proof that a user has registered with a tower. This serves two purposes: +/// +/// - First, the user is able to prove that the tower agreed on providing a service. If a tower refuses to accept appointments +/// from a user (claiming the subscription has expired) but the expiry time has still not passed and the tower cannot +/// provide the relevant appointments signed by the user, it means it is cheating. +/// - Second, it serves as proof, alongside an appointment receipt, that an appointment was not fulfilled. A registration receipt +/// specifies a subscription period (`subscription_start` - `subscription_expiry`) and the appointment a `start_block` so inclusion +/// can be proved. +/// +/// TODO: / DISCUSS: In order to minimize the amount of receipts the user has to store, the tower could batch subscription receipts +/// as long as the user info is still known. That is, if a user has a subscription with range (S, E) and the user renews the subscription +/// before the tower wipes their data, then the tower can create a new receipt with (S, E') for E' > E instead of a second receipt (E, E'). +// Notice this only applies as long as there is no gap between the two subscriptions. +#[derive(Serialize, Debug, Eq, PartialEq, Clone)] pub struct RegistrationReceipt { user_id: UserId, available_slots: u32, + subscription_start: u32, subscription_expiry: u32, #[serde(skip)] signature: Option, } impl RegistrationReceipt { - pub fn new(user_id: UserId, available_slots: u32, subscription_expiry: u32) -> Self { + pub fn new( + user_id: UserId, + available_slots: u32, + subscription_start: u32, + subscription_expiry: u32, + ) -> Self { RegistrationReceipt { user_id, available_slots, + subscription_start, subscription_expiry, signature: None, } @@ -28,12 +48,14 @@ impl RegistrationReceipt { pub fn with_signature( user_id: UserId, available_slots: u32, + subscription_start: u32, subscription_expiry: u32, signature: String, ) -> Self { RegistrationReceipt { user_id, available_slots, + subscription_start, subscription_expiry, signature: Some(signature), } @@ -47,6 +69,10 @@ impl RegistrationReceipt { self.available_slots } + pub fn subscription_start(&self) -> u32 { + self.subscription_start + } + pub fn subscription_expiry(&self) -> u32 { self.subscription_expiry } @@ -59,6 +85,7 @@ impl RegistrationReceipt { let mut ser = Vec::new(); ser.extend_from_slice(&self.user_id.to_vec()); ser.extend_from_slice(&self.available_slots.to_be_bytes()); + ser.extend_from_slice(&self.subscription_start.to_be_bytes()); ser.extend_from_slice(&self.subscription_expiry.to_be_bytes()); ser @@ -78,6 +105,9 @@ impl RegistrationReceipt { } } +/// Proof that a certain state was backed up with the tower. +/// +/// Appointment receipts can be used alongside a registration receipt that covers it, and on chain data (a breach not being reacted with a penalty), to prove a tower has not reacted to a channel breach. #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct AppointmentReceipt { user_signature: String, diff --git a/teos-common/src/test_utils.rs b/teos-common/src/test_utils.rs index b131b51..9eaed7a 100644 --- a/teos-common/src/test_utils.rs +++ b/teos-common/src/test_utils.rs @@ -53,7 +53,26 @@ pub fn generate_random_appointment(dispute_txid: Option<&Txid>) -> Appointment { } pub fn get_random_registration_receipt() -> RegistrationReceipt { - RegistrationReceipt::new(get_random_user_id(), get_random_int(), get_random_int()) + let (sk, _) = cryptography::get_random_keypair(); + let start = get_random_int(); + let mut receipt = + RegistrationReceipt::new(get_random_user_id(), get_random_int(), start, start + 420); + receipt.sign(&sk); + + receipt +} + +pub fn get_registration_receipt_from_previous(r: &RegistrationReceipt) -> RegistrationReceipt { + let (sk, _) = cryptography::get_random_keypair(); + let mut receipt = RegistrationReceipt::new( + r.user_id(), + r.available_slots() + 1 + get_random_int::() as u32, + r.subscription_start(), + r.subscription_expiry() + 1 + get_random_int::() as u32, + ); + receipt.sign(&sk); + + receipt } pub fn get_random_appointment_receipt(tower_sk: SecretKey) -> AppointmentReceipt { diff --git a/teos/src/api/internal.rs b/teos/src/api/internal.rs index 6324ba5..5c48bee 100644 --- a/teos/src/api/internal.rs +++ b/teos/src/api/internal.rs @@ -77,6 +77,7 @@ impl PublicTowerServices for Arc { Ok(receipt) => Ok(Response::new(common_msgs::RegisterResponse { user_id: req_data.user_id, available_slots: receipt.available_slots(), + subscription_start: receipt.subscription_start(), subscription_expiry: receipt.subscription_expiry(), subscription_signature: receipt.signature().unwrap(), })), diff --git a/teos/src/dbm.rs b/teos/src/dbm.rs index 500677a..fe6b9a0 100644 --- a/teos/src/dbm.rs +++ b/teos/src/dbm.rs @@ -27,6 +27,7 @@ const TABLES: [&str; 5] = [ "CREATE TABLE IF NOT EXISTS users ( user_id INT PRIMARY KEY, available_slots INT NOT NULL, + subscription_start INT NOT NULL, subscription_expiry INT NOT NULL )", "CREATE TABLE IF NOT EXISTS appointments ( @@ -94,13 +95,14 @@ impl DBM { /// Stores a user ([UserInfo]) into the database. pub(crate) fn store_user(&self, user_id: UserId, user_info: &UserInfo) -> Result<(), Error> { let query = - "INSERT INTO users (user_id, available_slots, subscription_expiry) VALUES (?1, ?2, ?3)"; + "INSERT INTO users (user_id, available_slots, subscription_start, subscription_expiry) VALUES (?1, ?2, ?3, ?4)"; match self.store_data( query, params![ user_id.to_vec(), user_info.available_slots, + user_info.subscription_start, user_info.subscription_expiry, ], ) { @@ -118,11 +120,12 @@ impl DBM { /// Updates an existing user ([UserInfo]) in the database. pub(crate) fn update_user(&self, user_id: UserId, user_info: &UserInfo) { let query = - "UPDATE users SET available_slots=(?1), subscription_expiry=(?2) WHERE user_id=(?3)"; + "UPDATE users SET available_slots=(?1), subscription_start=(?2), subscription_expiry=(?3) WHERE user_id=(?4)"; match self.update_data( query, params![ user_info.available_slots, + user_info.subscription_start, user_info.subscription_expiry, user_id.to_vec(), ], @@ -169,11 +172,17 @@ impl DBM { let raw_userid: Vec = row.get(0).unwrap(); let user_id = UserId::from_slice(&raw_userid).unwrap(); let slots = row.get(1).unwrap(); - let expiry = row.get(2).unwrap(); + let start = row.get(2).unwrap(); + let expiry = row.get(3).unwrap(); users.insert( user_id, - UserInfo::with_appointments(slots, expiry, self.load_user_appointments(user_id)), + UserInfo::with_appointments( + slots, + start, + expiry, + self.load_user_appointments(user_id), + ), ); } @@ -566,7 +575,8 @@ mod tests { use crate::test_utils::{ generate_dummy_appointment, generate_dummy_appointment_with_user, generate_uuid, - get_random_tracker, get_random_tx, + get_random_tracker, get_random_tx, AVAILABLE_SLOTS, SUBSCRIPTION_EXPIRY, + SUBSCRIPTION_START, }; impl DBM { @@ -583,14 +593,16 @@ mod tests { let key = user_id.to_vec(); let mut stmt = self .connection - .prepare("SELECT available_slots, subscription_expiry FROM users WHERE user_id=(?)") + .prepare("SELECT * FROM users WHERE user_id=(?)") .unwrap(); let user = stmt .query_row([&key], |row| { - let slots = row.get(0).unwrap(); - let expiry = row.get(1).unwrap(); + let slots = row.get(1).unwrap(); + let start = row.get(2).unwrap(); + let expiry = row.get(3).unwrap(); Ok(UserInfo::with_appointments( slots, + start, expiry, self.load_user_appointments(user_id), )) @@ -613,13 +625,13 @@ mod tests { let dbm = DBM::in_memory().unwrap(); let user_id = get_random_user_id(); - let mut user = UserInfo::new(21, 42); + let mut user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); assert!(matches!(dbm.store_user(user_id, &user), Ok { .. })); assert_eq!(dbm.load_user(user_id).unwrap(), user); // User info should be updatable but only via the update_user method - user = UserInfo::new(42, 21); + user = UserInfo::new(AVAILABLE_SLOTS * 2, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); assert!(matches!( dbm.store_user(user_id, &user), Err(Error::AlreadyExists) @@ -631,7 +643,7 @@ mod tests { let dbm = DBM::in_memory().unwrap(); let user_id = get_random_user_id(); - let mut user = UserInfo::new(21, 42); + let mut user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); @@ -660,7 +672,7 @@ mod tests { let dbm = DBM::in_memory().unwrap(); let user_id = get_random_user_id(); - let mut user = UserInfo::new(21, 42); + let mut user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); assert_eq!(dbm.load_user(user_id).unwrap(), user); @@ -677,7 +689,11 @@ mod tests { for i in 1..11 { let user_id = get_random_user_id(); - let user = UserInfo::new(i, i * 2); + let user = UserInfo::new( + AVAILABLE_SLOTS + i, + SUBSCRIPTION_START + i, + SUBSCRIPTION_EXPIRY + i, + ); users.insert(user_id, user.clone()); dbm.store_user(user_id, &user).unwrap(); @@ -710,7 +726,7 @@ mod tests { let mut rest = HashSet::new(); for i in 1..100 { let user_id = get_random_user_id(); - let user = UserInfo::new(21, 42); + let user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); if i % 2 == 0 { @@ -737,7 +753,7 @@ mod tests { // Add the user and link an appointment (this is usually done once the appointment) // is added after the user creation, but for the test purpose it can be done all at once. - let info = UserInfo::new(21, 42); + let info = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(appointment.user_id, &info).unwrap(); // Appointment only @@ -785,7 +801,7 @@ mod tests { // In order to add an appointment we need the associated user to be present let user_id = get_random_user_id(); - let user = UserInfo::new(21, 42); + let user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); let (uuid, appointment) = generate_dummy_appointment_with_user(user_id, None); @@ -830,7 +846,7 @@ mod tests { let dbm = DBM::in_memory().unwrap(); let user_id = get_random_user_id(); - let user = UserInfo::new(21, 42); + let user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); let (uuid, appointment) = generate_dummy_appointment_with_user(user_id, None); @@ -863,7 +879,11 @@ mod tests { for i in 1..11 { let user_id = get_random_user_id(); - let user = UserInfo::new(i, i * 2); + let user = UserInfo::new( + AVAILABLE_SLOTS + i, + SUBSCRIPTION_START + i, + SUBSCRIPTION_EXPIRY + i, + ); dbm.store_user(user_id, &user).unwrap(); let (uuid, appointment) = generate_dummy_appointment_with_user(user_id, None); @@ -876,7 +896,7 @@ mod tests { // If an appointment has an associated tracker, it should not be loaded since it is seen // as a triggered appointment let user_id = get_random_user_id(); - let user = UserInfo::new(21, 42); + let user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); let (uuid, appointment) = generate_dummy_appointment_with_user(user_id, None); @@ -900,7 +920,11 @@ mod tests { for i in 1..11 { let user_id = get_random_user_id(); - let user = UserInfo::new(i, i * 2); + let user = UserInfo::new( + AVAILABLE_SLOTS + i, + SUBSCRIPTION_START + i, + SUBSCRIPTION_EXPIRY + i, + ); dbm.store_user(user_id, &user).unwrap(); // Let some appointments belong to a specific dispute tx and some with random ones. @@ -923,7 +947,7 @@ mod tests { // If an appointment has an associated tracker, it should not be loaded since it is seen // as a triggered appointment let user_id = get_random_user_id(); - let user = UserInfo::new(21, 42); + let user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); // Generate an appointment for our dispute tx, thus it gets the same locator as the ones generated above. @@ -950,7 +974,11 @@ mod tests { .set_limit(Limit::SQLITE_LIMIT_VARIABLE_NUMBER, limit); let user_id = get_random_user_id(); - let mut user = UserInfo::new(500, 42); + let mut user = UserInfo::new( + AVAILABLE_SLOTS + 123, + SUBSCRIPTION_START, + SUBSCRIPTION_EXPIRY, + ); dbm.store_user(user_id, &user).unwrap(); let mut rest = HashSet::new(); @@ -994,7 +1022,7 @@ mod tests { // The confirmation status doesn't really matter here, it can be any of {ConfirmedIn, InMempoolSince}. let tracker = get_random_tracker(appointment.user_id, ConfirmationStatus::ConfirmedIn(21)); - let info = UserInfo::new(21, 42); + let info = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); // Add the user b/c of FK restrictions dbm.store_user(appointment.user_id, &info).unwrap(); @@ -1040,7 +1068,7 @@ mod tests { // In order to add an appointment we need the associated user to be present let user_id = get_random_user_id(); - let user = UserInfo::new(21, 42); + let user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); let (uuid, appointment) = generate_dummy_appointment_with_user(user_id, None); @@ -1069,7 +1097,7 @@ mod tests { // In order to add a tracker we need the associated appointment to be present (which // at the same time requires an associated user to be present) let user_id = get_random_user_id(); - let user = UserInfo::new(21, 42); + let user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); let (uuid, appointment) = generate_dummy_appointment_with_user(user_id, None); @@ -1086,7 +1114,7 @@ mod tests { let dbm = DBM::in_memory().unwrap(); let user_id = get_random_user_id(); - let user = UserInfo::new(21, 42); + let user = UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY); dbm.store_user(user_id, &user).unwrap(); let (uuid, appointment) = generate_dummy_appointment_with_user(user_id, None); @@ -1134,7 +1162,11 @@ mod tests { for i in 1..11 { let user_id = get_random_user_id(); - let user = UserInfo::new(i, i * 2); + let user = UserInfo::new( + AVAILABLE_SLOTS + i, + SUBSCRIPTION_START + i, + SUBSCRIPTION_EXPIRY + i, + ); dbm.store_user(user_id, &user).unwrap(); let (uuid, appointment) = generate_dummy_appointment_with_user(user_id, None); @@ -1160,7 +1192,11 @@ mod tests { for i in 1..11 { let user_id = get_random_user_id(); - let user = UserInfo::new(i, i * 2); + let user = UserInfo::new( + AVAILABLE_SLOTS + i, + SUBSCRIPTION_START + i, + SUBSCRIPTION_EXPIRY + i, + ); dbm.store_user(user_id, &user).unwrap(); let tracker = get_random_tracker(user_id, status); diff --git a/teos/src/gatekeeper.rs b/teos/src/gatekeeper.rs index a758dee..d1e3764 100644 --- a/teos/src/gatekeeper.rs +++ b/teos/src/gatekeeper.rs @@ -21,7 +21,9 @@ use crate::extended_appointment::{ExtendedAppointment, UUID}; pub(crate) struct UserInfo { /// Number of appointment slots available for a given user. pub(crate) available_slots: u32, - /// Block height where the user subscription will expire. + /// Block height where the user subscription starts. + pub(crate) subscription_start: u32, + /// Block height where the user subscription expires. pub(crate) subscription_expiry: u32, /// Map of appointment ids and the how many slots they take from the subscription. pub(crate) appointments: HashMap, @@ -29,9 +31,10 @@ pub(crate) struct UserInfo { impl UserInfo { /// Creates a new [UserInfo] instance. - pub fn new(available_slots: u32, subscription_expiry: u32) -> Self { + pub fn new(available_slots: u32, subscription_start: u32, subscription_expiry: u32) -> Self { UserInfo { available_slots, + subscription_start, subscription_expiry, appointments: HashMap::new(), } @@ -40,11 +43,13 @@ impl UserInfo { /// Creates a new [UserInfo] instance with some associated appointments. pub fn with_appointments( available_slots: u32, + subscription_start: u32, subscription_expiry: u32, appointments: HashMap, ) -> Self { UserInfo { available_slots, + subscription_start, subscription_expiry, appointments, } @@ -179,6 +184,7 @@ impl Gatekeeper { None => { let user_info = UserInfo::new( self.subscription_slots, + block_count, block_count + self.subscription_duration, ); self.dbm @@ -195,6 +201,7 @@ impl Gatekeeper { Ok(RegistrationReceipt::new( user_id, user_info.available_slots, + user_info.subscription_start, user_info.subscription_expiry, )) } @@ -473,7 +480,11 @@ mod tests { // The data should have been also added to the database assert_eq!( gatekeeper.dbm.lock().unwrap().load_user(user_id).unwrap(), - UserInfo::new(receipt.available_slots(), receipt.subscription_expiry()) + UserInfo::new( + receipt.available_slots(), + receipt.subscription_start(), + receipt.subscription_expiry() + ) ); // Let generate a new block and add the user again to check that both the slots and expiry are updated. @@ -497,6 +508,7 @@ mod tests { gatekeeper.dbm.lock().unwrap().load_user(user_id).unwrap(), UserInfo::new( updated_receipt.available_slots(), + updated_receipt.subscription_start(), updated_receipt.subscription_expiry() ) ); @@ -520,6 +532,7 @@ mod tests { gatekeeper.dbm.lock().unwrap().load_user(user_id).unwrap(), UserInfo::new( updated_receipt.available_slots(), + updated_receipt.subscription_start(), updated_receipt.subscription_expiry() ) ); diff --git a/teos/src/responder.rs b/teos/src/responder.rs index 7605277..0a7a41f 100644 --- a/teos/src/responder.rs +++ b/teos/src/responder.rs @@ -576,7 +576,8 @@ mod tests { use crate::test_utils::{ create_carrier, generate_dummy_appointment_with_user, generate_uuid, get_random_breach, get_random_tracker, get_random_tx, store_appointment_and_fks_to_db, Blockchain, - MockedServerQuery, DURATION, EXPIRY_DELTA, SLOTS, START_HEIGHT, + MockedServerQuery, AVAILABLE_SLOTS, DURATION, EXPIRY_DELTA, SLOTS, START_HEIGHT, + SUBSCRIPTION_EXPIRY, SUBSCRIPTION_START, }; use teos_common::dbm::Error as DBError; @@ -1118,7 +1119,10 @@ mod tests { .dbm .lock() .unwrap() - .store_user(user_id, &UserInfo::new(21, 42)) + .store_user( + user_id, + &UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY), + ) .unwrap(); // Transactions are flagged to be rebroadcast when they've been in mempool for longer than CONFIRMATIONS_BEFORE_RETRY @@ -1163,7 +1167,10 @@ mod tests { .dbm .lock() .unwrap() - .store_user(user_id, &UserInfo::new(21, 42)) + .store_user( + user_id, + &UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY), + ) .unwrap(); // Transactions are flagged to be rebroadcast when they've been in mempool for longer than CONFIRMATIONS_BEFORE_RETRY @@ -1272,7 +1279,10 @@ mod tests { .dbm .lock() .unwrap() - .store_user(user_id, &UserInfo::new(21, 42)) + .store_user( + user_id, + &UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY), + ) .unwrap(); // Transactions are rebroadcast once they've been in mempool for CONFIRMATIONS_BEFORE_RETRY or they've been reorged out @@ -1339,7 +1349,10 @@ mod tests { .dbm .lock() .unwrap() - .store_user(user_id, &UserInfo::new(21, 42)) + .store_user( + user_id, + &UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY), + ) .unwrap(); // Transactions are rebroadcast once they've been in mempool for CONFIRMATIONS_BEFORE_RETRY or they've been reorged out @@ -1400,7 +1413,10 @@ mod tests { .dbm .lock() .unwrap() - .store_user(user_id, &UserInfo::new(21, 42)) + .store_user( + user_id, + &UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY), + ) .unwrap(); // Add some trackers both to memory and to the database @@ -1454,7 +1470,10 @@ mod tests { .dbm .lock() .unwrap() - .store_user(user_id, &UserInfo::new(21, 42)) + .store_user( + user_id, + &UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY), + ) .unwrap(); // Delete trackers removes data from the trackers, tx_tracker_map maps, the database. The deletion of the later is @@ -1504,7 +1523,14 @@ mod tests { // Users will also be updated once the data is deleted. // We can made up the numbers here just to check they are updated. target_trackers.insert(uuid); - updated_users.insert(appointment.user_id, UserInfo::new(i, 42)); + updated_users.insert( + appointment.user_id, + UserInfo::new( + AVAILABLE_SLOTS + i, + SUBSCRIPTION_START + i, + SUBSCRIPTION_EXPIRY + i, + ), + ); } } @@ -1826,7 +1852,10 @@ mod tests { .dbm .lock() .unwrap() - .store_user(user_id, &UserInfo::new(21, 42)) + .store_user( + user_id, + &UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY), + ) .unwrap(); let mut reorged = Vec::new(); diff --git a/teos/src/test_utils.rs b/teos/src/test_utils.rs index 417b263..5ecd833 100644 --- a/teos/src/test_utils.rs +++ b/teos/src/test_utils.rs @@ -52,6 +52,10 @@ pub(crate) const DURATION: u32 = 500; pub(crate) const EXPIRY_DELTA: u32 = 42; pub(crate) const START_HEIGHT: usize = 100; +pub(crate) const AVAILABLE_SLOTS: u32 = 21; +pub(crate) const SUBSCRIPTION_START: u32 = START_HEIGHT as u32; +pub(crate) const SUBSCRIPTION_EXPIRY: u32 = SUBSCRIPTION_START + 42; + #[derive(Clone, Default, Debug)] pub(crate) struct Blockchain { pub blocks: Vec, @@ -349,8 +353,11 @@ pub(crate) fn store_appointment_and_fks_to_db( uuid: UUID, appointment: &ExtendedAppointment, ) { - dbm.store_user(appointment.user_id, &UserInfo::new(21, 42)) - .unwrap(); + dbm.store_user( + appointment.user_id, + &UserInfo::new(AVAILABLE_SLOTS, SUBSCRIPTION_START, SUBSCRIPTION_EXPIRY), + ) + .unwrap(); dbm.store_appointment(uuid, appointment).unwrap(); } diff --git a/teos/src/watcher.rs b/teos/src/watcher.rs index 6af373a..0b3a63c 100644 --- a/teos/src/watcher.rs +++ b/teos/src/watcher.rs @@ -892,7 +892,8 @@ mod tests { create_carrier, create_responder, create_watcher, generate_dummy_appointment, generate_dummy_appointment_with_user, generate_uuid, get_last_n_blocks, get_random_breach, get_random_tx, store_appointment_and_fks_to_db, BitcoindMock, Blockchain, MockOptions, - MockedServerQuery, DURATION, EXPIRY_DELTA, SLOTS, START_HEIGHT, + MockedServerQuery, AVAILABLE_SLOTS, DURATION, EXPIRY_DELTA, SLOTS, START_HEIGHT, + SUBSCRIPTION_EXPIRY, SUBSCRIPTION_START, }; use teos_common::cryptography::{get_random_bytes, get_random_keypair}; use teos_common::dbm::Error as DBError; @@ -1798,7 +1799,14 @@ mod tests { // Users will also be updated once the data is deleted. // We can made up the numbers here just to check they are updated. target_appointments.insert(uuid); - updated_users.insert(appointment.user_id, UserInfo::new(i, 42)); + updated_users.insert( + appointment.user_id, + UserInfo::new( + AVAILABLE_SLOTS + i, + SUBSCRIPTION_START + i, + SUBSCRIPTION_EXPIRY + i, + ), + ); } } diff --git a/watchtower-plugin/src/dbm.rs b/watchtower-plugin/src/dbm.rs index 58f2d5c..e2e9d03 100755 --- a/watchtower-plugin/src/dbm.rs +++ b/watchtower-plugin/src/dbm.rs @@ -10,7 +10,7 @@ use bitcoin::secp256k1::SecretKey; use teos_common::appointment::{Appointment, Locator}; use teos_common::dbm::{DatabaseConnection, DatabaseManager, Error}; use teos_common::receipts::{AppointmentReceipt, RegistrationReceipt}; -use teos_common::TowerId; +use teos_common::{TowerId, UserId}; use crate::{AppointmentStatus, MisbehaviorProof, TowerInfo, TowerStatus, TowerSummary}; @@ -18,8 +18,7 @@ const TABLES: [&str; 8] = [ "CREATE TABLE IF NOT EXISTS towers ( tower_id INT PRIMARY KEY, net_addr TEXT NOT NULL, - available_slots INT NOT NULL, - subscription_expiry INT NOT NULL + available_slots INT NOT NULL )", "CREATE TABLE IF NOT EXISTS appointments ( locator INT PRIMARY KEY, @@ -49,10 +48,12 @@ const TABLES: [&str; 8] = [ ON DELETE CASCADE )", "CREATE TABLE IF NOT EXISTS registration_receipts ( - tower_id INT PRIMARY KEY, + tower_id INT NOT NULL, available_slots INT NOT NULL, + subscription_start INT NOT NULL, subscription_expiry INT NOT NULL, signature BLOB NOT NULL, + PRIMARY KEY (tower_id, subscription_expiry), FOREIGN KEY(tower_id) REFERENCES towers(tower_id) ON DELETE CASCADE @@ -139,24 +140,30 @@ impl DBM { .map_err(|_| Error::NotFound) } - /// Stores a tower record into the database. + /// Stores a tower record into the database alongside the corresponding registration receipt. + /// + /// This function MUST be guarded against inserting duplicate (tower_id, subscription_expiry) pairs. + /// This is currently done in WTClient::add_update_tower. pub fn store_tower_record( - &self, + &mut self, tower_id: TowerId, net_addr: &str, receipt: &RegistrationReceipt, ) -> Result<(), Error> { - let query = - "INSERT OR REPLACE INTO towers (tower_id, net_addr, available_slots, subscription_expiry) VALUES (?1, ?2, ?3, ?4)"; - self.store_data( - query, - params![ - tower_id.to_vec(), - net_addr, - receipt.available_slots(), - receipt.subscription_expiry() - ], + let tx = self.get_mut_connection().transaction().unwrap(); + tx.execute( + "INSERT INTO towers (tower_id, net_addr, available_slots) + VALUES (?1, ?2, ?3) + ON CONFLICT (tower_id) DO UPDATE SET net_addr = ?2, available_slots = ?3", + params![tower_id.to_vec(), net_addr, receipt.available_slots()], ) + .map_err(Error::Unknown)?; + tx.execute( + "INSERT INTO registration_receipts (tower_id, available_slots, subscription_start, subscription_expiry, signature) + VALUES (?1, ?2, ?3, ?4, ?5)", + params![tower_id.to_vec(), receipt.available_slots(), receipt.subscription_start(), receipt.subscription_expiry(), receipt.signature()]).map_err( Error::Unknown)?; + + tx.commit().map_err(Error::Unknown) } /// Loads a tower record from the database. @@ -167,17 +174,23 @@ impl DBM { pub fn load_tower_record(&self, tower_id: TowerId) -> Result { let mut stmt = self .connection - .prepare("SELECT net_addr, available_slots, subscription_expiry FROM towers WHERE tower_id = ?") + .prepare("SELECT t.net_addr, t.available_slots, r.subscription_start, r.subscription_expiry + FROM towers as t, registration_receipts as r + WHERE t.tower_id = r.tower_id AND t.tower_id = ?1 AND r.subscription_expiry = (SELECT MAX(subscription_expiry) + FROM registration_receipts + WHERE tower_id = ?1)") .unwrap(); let mut tower = stmt .query_row([tower_id.to_vec()], |row| { let net_addr: String = row.get(0).unwrap(); let available_slots: u32 = row.get(1).unwrap(); - let subscription_expiry: u32 = row.get(2).unwrap(); + let subscription_start: u32 = row.get(2).unwrap(); + let subscription_expiry: u32 = row.get(3).unwrap(); Ok(TowerInfo::new( net_addr, available_slots, + subscription_start, subscription_expiry, self.load_appointment_receipts(tower_id), self.load_appointments(tower_id, AppointmentStatus::Pending), @@ -196,10 +209,55 @@ impl DBM { Ok(tower) } + /// Loads the latest registration receipt for a given tower. + /// + /// Latests is determined by the one with the `subscription_expiry` further into the future. + pub fn load_registration_receipt( + &self, + tower_id: TowerId, + user_id: UserId, + ) -> Result { + let mut stmt = self + .connection + .prepare( + "SELECT * + FROM registration_receipts + WHERE tower_id = ?1 AND subscription_expiry = (SELECT MAX(subscription_expiry) + FROM registration_receipts + WHERE tower_id = ?1)", + ) + .unwrap(); + + let receipt = stmt + .query_row([tower_id.to_vec()], |row| { + let slots: u32 = row.get(1).unwrap(); + let start: u32 = row.get(2).unwrap(); + let expiry: u32 = row.get(3).unwrap(); + let signature: String = row.get(4).unwrap(); + + Ok(RegistrationReceipt::with_signature( + user_id, slots, start, expiry, signature, + )) + }) + .map_err(|_| Error::NotFound)?; + + Ok(receipt) + } + /// Loads all tower records from the database. pub fn load_towers(&self) -> HashMap { let mut towers = HashMap::new(); - let mut stmt = self.connection.prepare("SELECT * FROM towers").unwrap(); + let mut stmt = self + .connection + .prepare("SELECT tw.tower_id, tw.net_addr, tw.available_slots, rr.subscription_start, rr.subscription_expiry + FROM towers AS tw + JOIN registration_receipts AS rr + JOIN (SELECT tower_id, MAX(subscription_expiry) AS max_se + FROM registration_receipts + GROUP BY tower_id) AS max_rrs ON (tw.tower_id = rr.tower_id) + AND (rr.tower_id = max_rrs.tower_id) + AND (rr.subscription_expiry = max_rrs.max_se)") + .unwrap(); let mut rows = stmt.query([]).unwrap(); while let Ok(Some(row)) = rows.next() { @@ -207,12 +265,14 @@ impl DBM { let tower_id = TowerId::from_slice(&raw_towerid).unwrap(); let net_addr: String = row.get(1).unwrap(); let available_slots: u32 = row.get(2).unwrap(); - let subscription_expiry: u32 = row.get(3).unwrap(); + let start: u32 = row.get(3).unwrap(); + let expiry: u32 = row.get(4).unwrap(); let mut tower = TowerSummary::with_appointments( net_addr, available_slots, - subscription_expiry, + start, + expiry, self.load_appointment_locators(tower_id, AppointmentStatus::Pending), self.load_appointment_locators(tower_id, AppointmentStatus::Invalid), ); @@ -239,7 +299,8 @@ impl DBM { ) -> Result<(), SqliteError> { let tx = self.get_mut_connection().transaction().unwrap(); tx.execute( - "INSERT INTO appointment_receipts (locator, tower_id, start_block, user_signature, tower_signature) VALUES (?1, ?2, ?3, ?4, ?5)", + "INSERT INTO appointment_receipts (locator, tower_id, start_block, user_signature, tower_signature) + VALUES (?1, ?2, ?3, ?4, ?5)", params![ locator.to_vec(), tower_id.to_vec(), @@ -464,7 +525,8 @@ impl DBM { ) -> Result<(), SqliteError> { let tx = self.get_mut_connection().transaction().unwrap(); tx.execute( - "INSERT INTO appointment_receipts (tower_id, locator, start_block, user_signature, tower_signature) VALUES (?1, ?2, ?3, ?4, ?5)", + "INSERT INTO appointment_receipts (tower_id, locator, start_block, user_signature, tower_signature) + VALUES (?1, ?2, ?3, ?4, ?5)", params![ tower_id.to_vec(), proof.locator.to_vec(), @@ -500,11 +562,13 @@ impl DBM { }) .map(|(locator, recovered_id)| { let mut receipt_stmt = self - .connection - .prepare( - "SELECT start_block, user_signature, tower_signature FROM appointment_receipts WHERE locator = ?1 AND tower_id = ?2", - ) - .unwrap(); + .connection + .prepare( + "SELECT start_block, user_signature, tower_signature + FROM appointment_receipts + WHERE locator = ?1 AND tower_id = ?2", + ) + .unwrap(); let receipt = receipt_stmt .query_row([locator.to_vec(), tower_id.to_vec()], |row| { let start_block = row.get::<_, u32>(0).unwrap(); @@ -518,7 +582,8 @@ impl DBM { }) .unwrap(); MisbehaviorProof::new(locator, receipt, recovered_id) - }).map_err(|_| Error::NotFound) + }) + .map_err(|_| Error::NotFound) } /// Checks whether a misbehaving proof exists for a given tower. @@ -537,6 +602,7 @@ mod tests { use teos_common::test_utils::{ generate_random_appointment, get_random_registration_receipt, get_random_user_id, + get_registration_receipt_from_previous, }; impl DBM { @@ -567,7 +633,7 @@ mod tests { #[test] fn test_store_load_tower_record() { - let dbm = DBM::in_memory().unwrap(); + let mut dbm = DBM::in_memory().unwrap(); // In order to add a tower record we need to associated registration receipt. let tower_id = get_random_user_id(); @@ -577,6 +643,7 @@ mod tests { let tower_info = TowerInfo::new( net_addr.into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), HashMap::new(), Vec::new(), @@ -589,6 +656,72 @@ mod tests { assert_eq!(dbm.load_tower_record(tower_id).unwrap(), tower_info); } + #[test] + fn test_load_registration_receipt() { + let mut dbm = DBM::in_memory().unwrap(); + + // Registration receipts are stored alongside tower records when the register command is called + let tower_id = get_random_user_id(); + let net_addr = "talaia.watch"; + let receipt = get_random_registration_receipt(); + + // Check the receipt was stored + dbm.store_tower_record(tower_id, net_addr, &receipt) + .unwrap(); + assert_eq!( + dbm.load_registration_receipt(tower_id, receipt.user_id()) + .unwrap(), + receipt + ); + + // Add another receipt for the same tower with a higher expiry and check this last one is loaded + let middle_receipt = get_registration_receipt_from_previous(&receipt); + let latest_receipt = get_registration_receipt_from_previous(&middle_receipt); + + dbm.store_tower_record(tower_id, net_addr, &latest_receipt) + .unwrap(); + assert_eq!( + dbm.load_registration_receipt(tower_id, latest_receipt.user_id()) + .unwrap(), + latest_receipt + ); + + // Add a final one with a lower expiry and check the last is still loaded + dbm.store_tower_record(tower_id, net_addr, &middle_receipt) + .unwrap(); + assert_eq!( + dbm.load_registration_receipt(tower_id, latest_receipt.user_id()) + .unwrap(), + latest_receipt + ); + } + + #[test] + fn test_load_same_registration_receipt() { + let mut dbm = DBM::in_memory().unwrap(); + + // Registration receipts are stored alongside tower records when the register command is called + let tower_id = get_random_user_id(); + let net_addr = "talaia.watch"; + let receipt = get_random_registration_receipt(); + + // Store it once + dbm.store_tower_record(tower_id, net_addr, &receipt) + .unwrap(); + assert_eq!( + dbm.load_registration_receipt(tower_id, receipt.user_id()) + .unwrap(), + receipt + ); + + // Store the same again, this should fail due to UNIQUE PK constrains. + // Notice store_tower_record is guarded against this by WTClient::add_update_tower though. + assert!(matches!( + dbm.store_tower_record(tower_id, net_addr, &receipt), + Err { .. } + )); + } + #[test] fn test_load_nonexistent_tower_record() { let dbm = DBM::in_memory().unwrap(); @@ -603,26 +736,33 @@ mod tests { #[test] fn test_store_load_towers() { - let dbm = DBM::in_memory().unwrap(); + let mut dbm = DBM::in_memory().unwrap(); let mut towers = HashMap::new(); // In order to add a tower record we need to associated registration receipt. - for _ in 0..5 { + for _ in 0..10 { let tower_id = get_random_user_id(); let net_addr = "talaia.watch"; + let mut receipt = get_random_registration_receipt(); + dbm.store_tower_record(tower_id, net_addr, &receipt) + .unwrap(); + + // Add not only one registration receipt to test if the tower retrieves the one with furthest expiry date. + for _ in 0..10 { + receipt = get_registration_receipt_from_previous(&receipt); + dbm.store_tower_record(tower_id, net_addr, &receipt) + .unwrap(); + } - let receipt = get_random_registration_receipt(); towers.insert( tower_id, TowerSummary::new( net_addr.into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ), ); - - dbm.store_tower_record(tower_id, net_addr, &receipt) - .unwrap(); } assert_eq!(dbm.load_towers(), towers); @@ -647,6 +787,7 @@ mod tests { let mut tower_summary = TowerSummary::new( net_addr.into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ); dbm.store_tower_record(tower_id, net_addr, &receipt) @@ -694,6 +835,7 @@ mod tests { let tower_summary = TowerSummary::new( net_addr.into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ); dbm.store_tower_record(tower_id, net_addr, &receipt) @@ -760,6 +902,7 @@ mod tests { let mut tower_summary = TowerSummary::new( net_addr.into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ) .with_status(TowerStatus::TemporaryUnreachable); @@ -889,6 +1032,7 @@ mod tests { let mut tower_summary = TowerSummary::new( net_addr.into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ); dbm.store_tower_record(tower_id, net_addr, &receipt) @@ -951,6 +1095,7 @@ mod tests { let tower_summary = TowerSummary::new( net_addr.into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ); dbm.store_tower_record(tower_id, net_addr, &receipt) @@ -999,6 +1144,7 @@ mod tests { let tower_summary = TowerSummary::new( net_addr.into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ); dbm.store_tower_record(tower_id, net_addr, &receipt) diff --git a/watchtower-plugin/src/lib.rs b/watchtower-plugin/src/lib.rs index 433486d..e23563d 100755 --- a/watchtower-plugin/src/lib.rs +++ b/watchtower-plugin/src/lib.rs @@ -35,6 +35,20 @@ pub enum AppointmentStatus { Invalid, } +/// Errors related to updating a subscription +#[derive(Debug, PartialEq, Eq)] +pub enum SubscriptionError { + Expiry, + Slots, +} + +impl SubscriptionError { + /// Whether the error is related to the expiry time or not. + pub fn is_expiry(&self) -> bool { + *self == SubscriptionError::Expiry + } +} + impl fmt::Display for TowerStatus { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!( @@ -78,6 +92,7 @@ impl TowerStatus { pub struct TowerSummary { pub net_addr: String, pub available_slots: u32, + subscription_start: u32, pub subscription_expiry: u32, pub status: TowerStatus, #[serde(serialize_with = "teos_common::ser::serialize_locators")] @@ -88,10 +103,16 @@ pub struct TowerSummary { impl TowerSummary { /// Creates a new [TowerSummary] instance. - pub fn new(net_addr: String, available_slots: u32, subscription_expiry: u32) -> Self { + pub fn new( + net_addr: String, + available_slots: u32, + subscription_start: u32, + subscription_expiry: u32, + ) -> Self { Self { net_addr, available_slots, + subscription_start, subscription_expiry, status: TowerStatus::Reachable, pending_appointments: HashSet::new(), @@ -103,6 +124,7 @@ impl TowerSummary { pub fn with_appointments( net_addr: String, available_slots: u32, + subscription_start: u32, subscription_expiry: u32, pending_appointments: HashSet, invalid_appointments: HashSet, @@ -110,6 +132,7 @@ impl TowerSummary { Self { net_addr, available_slots, + subscription_start, subscription_expiry, status: TowerStatus::Reachable, pending_appointments, @@ -129,6 +152,7 @@ impl From for TowerSummary { TowerSummary::with_appointments( info.net_addr, info.available_slots, + info.subscription_start, info.subscription_expiry, info.pending_appointments .iter() @@ -148,6 +172,7 @@ impl From for TowerSummary { pub struct TowerInfo { pub net_addr: String, pub available_slots: u32, + pub subscription_start: u32, pub subscription_expiry: u32, pub status: TowerStatus, #[serde(serialize_with = "crate::ser::serialize_receipts")] @@ -165,6 +190,7 @@ impl TowerInfo { pub fn new( net_addr: String, available_slots: u32, + subscription_start: u32, subscription_expiry: u32, appointments: HashMap, pending_appointments: Vec, @@ -173,6 +199,7 @@ impl TowerInfo { Self { net_addr, available_slots, + subscription_start, subscription_expiry, status: TowerStatus::Reachable, appointments, @@ -230,6 +257,10 @@ mod tests { TowerStatus::Misbehaving, ]; + const AVAILABLE_SLOTS: u32 = 21; + const SUBSCRIPTION_START: u32 = 100; + const SUBSCRIPTION_EXPIRY: u32 = SUBSCRIPTION_START + 42; + mod tower_status { use super::*; @@ -277,17 +308,20 @@ mod tests { #[test] fn test_new() { let net_addr: String = "addr".into(); - let available_slots = 21; - let subscription_expiry = 42; - let tower_summary = - TowerSummary::new(net_addr.clone(), available_slots, subscription_expiry); + let tower_summary = TowerSummary::new( + net_addr.clone(), + AVAILABLE_SLOTS, + SUBSCRIPTION_START, + SUBSCRIPTION_EXPIRY, + ); assert_eq!( tower_summary, TowerSummary { net_addr, - available_slots, - subscription_expiry, + available_slots: AVAILABLE_SLOTS, + subscription_start: SUBSCRIPTION_START, + subscription_expiry: SUBSCRIPTION_EXPIRY, status: TowerStatus::Reachable, pending_appointments: HashSet::new(), invalid_appointments: HashSet::new(), @@ -298,8 +332,7 @@ mod tests { #[test] fn test_with_appointments() { let net_addr: String = "addr".into(); - let available_slots = 21; - let subscription_expiry = 42; + let pending_appointments = HashSet::from_iter([generate_random_appointment(None).locator]); let invalid_appointments = @@ -307,8 +340,9 @@ mod tests { let tower_summary = TowerSummary::with_appointments( net_addr.clone(), - available_slots, - subscription_expiry, + AVAILABLE_SLOTS, + SUBSCRIPTION_START, + SUBSCRIPTION_EXPIRY, pending_appointments.clone(), invalid_appointments.clone(), ); @@ -316,8 +350,9 @@ mod tests { tower_summary, TowerSummary { net_addr, - available_slots, - subscription_expiry, + available_slots: AVAILABLE_SLOTS, + subscription_start: SUBSCRIPTION_START, + subscription_expiry: SUBSCRIPTION_EXPIRY, status: TowerStatus::Reachable, pending_appointments, invalid_appointments, @@ -327,7 +362,12 @@ mod tests { #[test] fn test_with_status() { - let mut tower_summary = TowerSummary::new("addr".into(), 21, 42); + let mut tower_summary = TowerSummary::new( + "addr".into(), + AVAILABLE_SLOTS, + SUBSCRIPTION_START, + SUBSCRIPTION_EXPIRY, + ); let unreachable_tower = tower_summary.clone().with_status(TowerStatus::Unreachable); tower_summary.status = TowerStatus::Unreachable; @@ -341,10 +381,16 @@ mod tests { use teos_common::test_utils::{generate_random_appointment, get_random_user_id}; impl TowerInfo { - pub fn empty(net_addr: String, available_slots: u32, subscription_expiry: u32) -> Self { + pub fn empty( + net_addr: String, + available_slots: u32, + subscription_start: u32, + subscription_expiry: u32, + ) -> Self { TowerInfo::new( net_addr, available_slots, + subscription_start, subscription_expiry, HashMap::new(), Vec::new(), @@ -357,8 +403,9 @@ mod tests { fn test_new() { let tower_info = TowerInfo::new( "addr".into(), - 21, - 42, + AVAILABLE_SLOTS, + SUBSCRIPTION_START, + SUBSCRIPTION_EXPIRY, HashMap::new(), Vec::new(), Vec::new(), @@ -370,7 +417,12 @@ mod tests { #[test] fn test_with_status() { - let mut tower_info = TowerInfo::empty("addr".into(), 21, 42); + let mut tower_info = TowerInfo::empty( + "addr".into(), + AVAILABLE_SLOTS, + SUBSCRIPTION_START, + SUBSCRIPTION_EXPIRY, + ); let unreachable_tower = tower_info.clone().with_status(TowerStatus::Unreachable); tower_info.status = TowerStatus::Unreachable; @@ -379,12 +431,17 @@ mod tests { #[test] fn test_set_misbehaving_proof() { - let mut tower_info = TowerInfo::empty("addr".into(), 21, 42); + let mut tower_info = TowerInfo::empty( + "addr".into(), + AVAILABLE_SLOTS, + SUBSCRIPTION_START, + SUBSCRIPTION_EXPIRY, + ); assert_eq!(tower_info.misbehaving_proof, None); let appointment_receipt = AppointmentReceipt::with_signature( "user_signature".into(), - 21, + SUBSCRIPTION_START + 1, "tower_signature".into(), ); let proof = MisbehaviorProof::new( diff --git a/watchtower-plugin/src/main.rs b/watchtower-plugin/src/main.rs index 8df6863..b4d14e5 100755 --- a/watchtower-plugin/src/main.rs +++ b/watchtower-plugin/src/main.rs @@ -50,6 +50,10 @@ async fn register( let tower_id = params.tower_id; let user_id = plugin.state().lock().unwrap().user_id; + // TODO: The user should pick the start_time or, at least, check the returned start time against it's known block height. + // Otherwise the tower could just generate a subscription starting far in the future. For this we need to access lightning RPC + // which is not available in the current version of `cln-plugin` (but already on master). Add it for the next release. + // FIXME: This is a workaround. Ideally, `cln_plugin::options::Value` will implement `as_u64` so we can simply call and unwrap // given that we are certain the option exists. let port = params.port.unwrap_or( @@ -83,6 +87,7 @@ async fn register( RegistrationReceipt::with_signature( user_id, r.available_slots, + r.subscription_start, r.subscription_expiry, r.subscription_signature, ) @@ -104,16 +109,24 @@ async fn register( )); } - log::info!( - "Registration succeeded. Available slots: {}", - receipt.available_slots() - ); - plugin .state() .lock() .unwrap() - .add_update_tower(tower_id, tower_net_addr, &receipt); + .add_update_tower(tower_id, tower_net_addr, &receipt).map_err(|e| { + if e.is_expiry() { + anyhow!("Registration receipt contains a subscription expiry that is not higher than the one we are currently registered for") + } else { + anyhow!("Registration receipt does not contain more slots than the ones we are currently registered for") + } + })?; + + log::info!( + "Registration succeeded. Available slots: {}. Subscription period (block height range): ({}-{})", + receipt.available_slots(), + receipt.subscription_start(), + receipt.subscription_expiry() + ); Ok(json!(receipt)) } diff --git a/watchtower-plugin/src/retrier.rs b/watchtower-plugin/src/retrier.rs index ccb8b49..1f796fa 100644 --- a/watchtower-plugin/src/retrier.rs +++ b/watchtower-plugin/src/retrier.rs @@ -206,7 +206,8 @@ mod tests { wt_client .lock() .unwrap() - .add_update_tower(tower_id, server.base_url(), &receipt); + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); // Add appointment to pending let appointment = generate_random_appointment(None); @@ -276,11 +277,11 @@ mod tests { 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, - "http://unreachable.tower".into(), - &receipt, - ); + wt_client + .lock() + .unwrap() + .add_update_tower(tower_id, "http://unreachable.tower".into(), &receipt) + .unwrap(); // Add appointment to pending let appointment = generate_random_appointment(None); @@ -344,7 +345,8 @@ mod tests { wt_client .lock() .unwrap() - .add_update_tower(tower_id, server.base_url(), &receipt); + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); // Add appointment to pending let appointment = generate_random_appointment(None); @@ -421,7 +423,8 @@ mod tests { wt_client .lock() .unwrap() - .add_update_tower(tower_id, server.base_url(), &receipt); + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); // Add appointment to pending let appointment = generate_random_appointment(None); @@ -486,7 +489,8 @@ mod tests { wt_client .lock() .unwrap() - .add_update_tower(tower_id, server.base_url(), &receipt); + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); // Add appointment to pending let appointment = generate_random_appointment(None); @@ -532,7 +536,8 @@ mod tests { wt_client .lock() .unwrap() - .add_update_tower(tower_id, server.base_url(), &receipt); + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); // If there are no pending appointments the method will simply return let r = Retrier::dummy(wt_client).add_appointment(tower_id).await; @@ -557,7 +562,8 @@ mod tests { wt_client .lock() .unwrap() - .add_update_tower(tower_id, server.base_url(), &receipt); + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); // Add appointment to pending let appointment = generate_random_appointment(None); @@ -598,11 +604,11 @@ mod tests { // 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, - "http://unreachable.tower".into(), - &receipt, - ); + wt_client + .lock() + .unwrap() + .add_update_tower(tower_id, "http://unreachable.tower".into(), &receipt) + .unwrap(); // Add some pending appointments and try again (with an unreachable tower). let appointment = generate_random_appointment(None); @@ -631,7 +637,8 @@ mod tests { wt_client .lock() .unwrap() - .add_update_tower(tower_id, server.base_url(), &receipt); + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); let api_mock = server.mock(|when, then| { when.method(POST).path("/add_appointment"); @@ -672,7 +679,8 @@ mod tests { wt_client .lock() .unwrap() - .add_update_tower(tower_id, server.base_url(), &receipt); + .add_update_tower(tower_id, server.base_url(), &receipt) + .unwrap(); let api_mock = server.mock(|when, then| { when.method(POST).path("/add_appointment"); diff --git a/watchtower-plugin/src/wt_client.rs b/watchtower-plugin/src/wt_client.rs index 7d30248..e2bf439 100644 --- a/watchtower-plugin/src/wt_client.rs +++ b/watchtower-plugin/src/wt_client.rs @@ -13,7 +13,7 @@ use teos_common::receipts::{AppointmentReceipt, RegistrationReceipt}; use teos_common::{TowerId, UserId}; use crate::dbm::DBM; -use crate::{MisbehaviorProof, TowerInfo, TowerStatus, TowerSummary}; +use crate::{MisbehaviorProof, SubscriptionError, TowerInfo, TowerStatus, TowerSummary}; /// Represents the watchtower client that is being used as the CoreLN plugin state. #[derive(Clone)] @@ -79,7 +79,25 @@ impl WTClient { tower_id: TowerId, tower_net_addr: String, receipt: &RegistrationReceipt, - ) { + ) -> Result<(), SubscriptionError> { + if let Some(tower) = self.towers.get(&tower_id) { + // TODO: For now we're forcing updates to increase both slots and expiry. This is not mandatory and may + // be changed in the future, but the tower is currently set to do this anyway so let's keep it simple. + if receipt.subscription_expiry() <= tower.subscription_expiry { + return Err(SubscriptionError::Expiry); + } else { + let previous_receipt = self + .dbm + .lock() + .unwrap() + .load_registration_receipt(tower_id, self.user_id) + .unwrap(); + if receipt.available_slots() <= previous_receipt.available_slots() { + return Err(SubscriptionError::Slots); + } + } + } + self.dbm .lock() .unwrap() @@ -90,9 +108,12 @@ impl WTClient { TowerSummary::new( tower_net_addr, receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ), ); + + Ok(()) } /// Loads a tower record from the database. @@ -217,6 +238,7 @@ mod tests { use teos_common::test_utils::{ generate_random_appointment, get_random_appointment_receipt, get_random_registration_receipt, get_random_user_id, + get_registration_receipt_from_previous, }; #[tokio::test] @@ -226,15 +248,18 @@ mod tests { WTClient::new(tmp_path.path().to_path_buf(), unbounded_channel().0).await; // Adding a new tower will add a summary to towers and the full data to the - let receipt = get_random_registration_receipt(); + let mut receipt = get_random_registration_receipt(); let tower_id = get_random_user_id(); let tower_info = TowerInfo::empty( "talaia.watch".into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ); - wt_client.add_update_tower(tower_id, tower_info.net_addr.clone(), &receipt); + 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())) @@ -242,13 +267,17 @@ mod tests { assert_eq!(wt_client.load_tower_info(tower_id).unwrap(), tower_info); // Calling the method again with updated information should also updated the records in memory and the database - let receipt = get_random_registration_receipt(); + receipt = get_registration_receipt_from_previous(&receipt); + let updated_tower_info = TowerInfo::empty( "talaia.watch".into(), receipt.available_slots(), + receipt.subscription_start(), receipt.subscription_expiry(), ); - wt_client.add_update_tower(tower_id, updated_tower_info.net_addr.clone(), &receipt); + wt_client + .add_update_tower(tower_id, updated_tower_info.net_addr.clone(), &receipt) + .unwrap(); assert_eq!( wt_client.towers.get(&tower_id), @@ -258,6 +287,37 @@ mod tests { wt_client.load_tower_info(tower_id).unwrap(), updated_tower_info ); + + // If we try to update without increasing both the end_time and the slots, this will fail + let receipt_same_slots = RegistrationReceipt::new( + receipt.user_id(), + receipt.available_slots(), + receipt.subscription_start(), + receipt.subscription_expiry() + 1, + ); + let receipt_same_expiry = RegistrationReceipt::new( + receipt.user_id(), + receipt.available_slots() + 1, + receipt.subscription_start(), + receipt.subscription_expiry(), + ); + + assert!(matches!( + wt_client.add_update_tower(tower_id, updated_tower_info.net_addr.clone(), &receipt), + Err(SubscriptionError::Expiry) + )); + assert!(matches!( + wt_client.add_update_tower( + tower_id, + updated_tower_info.net_addr.clone(), + &receipt_same_slots + ), + Err(SubscriptionError::Slots) + )); + assert!(matches!( + wt_client.add_update_tower(tower_id, updated_tower_info.net_addr, &receipt_same_expiry), + Err(SubscriptionError::Expiry) + )); } #[tokio::test] @@ -274,7 +334,9 @@ mod tests { // If the tower is known, the status will be updated. let receipt = get_random_registration_receipt(); let tower_id = get_random_user_id(); - wt_client.add_update_tower(tower_id, "talaia.watch".into(), &receipt); + wt_client + .add_update_tower(tower_id, "talaia.watch".into(), &receipt) + .unwrap(); for status in [ TowerStatus::Reachable, @@ -315,12 +377,15 @@ mod tests { let tower_info = TowerInfo::new( tower_net_addr.into(), registration_receipt.available_slots(), + registration_receipt.subscription_start(), registration_receipt.subscription_expiry(), HashMap::from([(locator, appointment_receipt.signature().unwrap())]), Vec::new(), Vec::new(), ); - wt_client.add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt); + wt_client + .add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt) + .unwrap(); wt_client.add_appointment_receipt( tower_id, locator, @@ -356,13 +421,16 @@ mod tests { let tower_info = TowerInfo::new( tower_net_addr.into(), registration_receipt.available_slots(), + registration_receipt.subscription_start(), registration_receipt.subscription_expiry(), HashMap::new(), vec![appointment.clone()], Vec::new(), ); - wt_client.add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt); + wt_client + .add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt) + .unwrap(); wt_client.add_pending_appointment(tower_id, &appointment); assert!(wt_client.towers.contains_key(&tower_id)); @@ -393,7 +461,9 @@ mod tests { wt_client.remove_pending_appointment(tower_id, appointment.locator); // Add the tower to the state and try again - wt_client.add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt); + wt_client + .add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt) + .unwrap(); wt_client.add_pending_appointment(tower_id, &appointment); wt_client.remove_pending_appointment(tower_id, appointment.locator); @@ -433,13 +503,16 @@ mod tests { let tower_info = TowerInfo::new( tower_net_addr.into(), registration_receipt.available_slots(), + registration_receipt.subscription_start(), registration_receipt.subscription_expiry(), HashMap::new(), Vec::new(), vec![appointment.clone()], ); - wt_client.add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt); + wt_client + .add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt) + .unwrap(); wt_client.add_invalid_appointment(tower_id, &appointment); assert!(wt_client.towers.contains_key(&tower_id)); @@ -462,7 +535,9 @@ mod tests { let registration_receipt = get_random_registration_receipt(); let appointment = generate_random_appointment(None); - wt_client.add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt); + wt_client + .add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt) + .unwrap(); wt_client.add_pending_appointment(tower_id, &appointment); // Check that the appointment can be moved from pending to invalid @@ -516,12 +591,16 @@ mod tests { let registration_receipt = get_random_registration_receipt(); let appointment = generate_random_appointment(None); - wt_client.add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt); - wt_client.add_update_tower( - another_tower_id, - tower_net_addr.into(), - ®istration_receipt, - ); + wt_client + .add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt) + .unwrap(); + wt_client + .add_update_tower( + another_tower_id, + tower_net_addr.into(), + ®istration_receipt, + ) + .unwrap(); wt_client.add_pending_appointment(tower_id, &appointment); wt_client.add_pending_appointment(another_tower_id, &appointment); @@ -610,7 +689,9 @@ mod tests { // // Add the tower to the state and try again let registration_receipt = get_random_registration_receipt(); - wt_client.add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt); + wt_client + .add_update_tower(tower_id, tower_net_addr.into(), ®istration_receipt) + .unwrap(); wt_client.flag_misbehaving_tower(tower_id, proof.clone()); // Check data in memory