GlcoinCore/rust-teos
1
0
Fork 0
mirror of https://github.com/talaia-labs/rust-teos.git synced 2026-08-13 12:33:22 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add start_time to registration receipt

Browse source
This commit is contained in:
Sergi Delgado Segura 2022-07-08 16:41:02 +02:00
parent 3175acafcc
commit c14543d6ea
14 changed files with 597 additions and 148 deletions
Download patch file Download diff file Expand all files Collapse all files

5
teos-common/proto/common/teos/v2/user.proto
View file

@ -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 {

34
teos-common/src/receipts.rs
View file

@ -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<String>,
}
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,

21
teos-common/src/test_utils.rs
View file

@ -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::<u8>() as u32,
r.subscription_start(),
r.subscription_expiry() + 1 + get_random_int::<u8>() as u32,
);
receipt.sign(&sk);
receipt
}
pub fn get_random_appointment_receipt(tower_sk: SecretKey) -> AppointmentReceipt {

1
teos/src/api/internal.rs
View file

@ -77,6 +77,7 @@ impl PublicTowerServices for Arc<InternalAPI> {
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(),
})),

92
teos/src/dbm.rs
View file

@ -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<u8> = 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);

19
teos/src/gatekeeper.rs
View file

@ -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<UUID, u32>,
@ -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<UUID, u32>,
) -> 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()
)
);

47
teos/src/responder.rs
View file

@ -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();

11
teos/src/test_utils.rs
View file

@ -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<Block>,
@ -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();
}

12
teos/src/watcher.rs
View file

@ -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,
),
);
}
}

218
watchtower-plugin/src/dbm.rs
View file

@ -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<TowerInfo, Error> {
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<RegistrationReceipt, Error> {
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<TowerId, TowerSummary> {
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)

97
watchtower-plugin/src/lib.rs
View file

@ -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<Locator>,
invalid_appointments: HashSet<Locator>,
@ -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<TowerInfo> 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<TowerInfo> 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<Locator, String>,
pending_appointments: Vec<Appointment>,
@ -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(

25
watchtower-plugin/src/main.rs
View file

@ -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))
}

44
watchtower-plugin/src/retrier.rs
View file

@ -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");

119
watchtower-plugin/src/wt_client.rs
View file

@ -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(), &registration_receipt);
wt_client
.add_update_tower(tower_id, tower_net_addr.into(), &registration_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(), &registration_receipt);
wt_client
.add_update_tower(tower_id, tower_net_addr.into(), &registration_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(), &registration_receipt);
wt_client
.add_update_tower(tower_id, tower_net_addr.into(), &registration_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(), &registration_receipt);
wt_client
.add_update_tower(tower_id, tower_net_addr.into(), &registration_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(), &registration_receipt);
wt_client
.add_update_tower(tower_id, tower_net_addr.into(), &registration_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(), &registration_receipt);
wt_client.add_update_tower(
another_tower_id,
tower_net_addr.into(),
&registration_receipt,
);
wt_client
.add_update_tower(tower_id, tower_net_addr.into(), &registration_receipt)
.unwrap();
wt_client
.add_update_tower(
another_tower_id,
tower_net_addr.into(),
&registration_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(), &registration_receipt);
wt_client
.add_update_tower(tower_id, tower_net_addr.into(), &registration_receipt)
.unwrap();
wt_client.flag_misbehaving_tower(tower_id, proof.clone());
// Check data in memory