27376140423

Committed 11 Jun 2026 08:42PM UTC coverage: 85.191% (+0.003%) from 85.188%

Build # 27376140423

Build Type

Pull #1638

github

Committed by

web-flow

Commit Message

Merge b89e663e8 into 4c2b7c897

Pull Request Pull Request #1638: Use time-based uuid instead of raw increment for session ids

Coverage Stats

30 of 37 new or added lines in 4 files covered. (81.08%)

1 existing line in 1 file now uncovered.

12547 of 14728 relevant lines covered (85.19%)

368.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.9

/payjoin-cli/src/db/v2.rs

use std::sync::Arc;

use payjoin::persist::SessionPersister;
use payjoin::receive::v2::SessionEvent as ReceiverSessionEvent;
use payjoin::send::v2::SessionEvent as SenderSessionEvent;
use payjoin::HpkePublicKey;
use rusqlite::params;

use super::*;

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SessionId(pub(crate) uuid::Uuid);

impl std::fmt::Display for SessionId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{}", self.0) }
}

#[derive(Clone, Debug)]
pub(crate) struct SenderPersister {
    db: Arc<Database>,
    session_id: SessionId,
}

impl SenderPersister {
    pub fn new(
        db: Arc<Database>,
        pj_uri: &str,
        receiver_pubkey: &HpkePublicKey,
    ) -> crate::db::Result<Self> {
        let conn = db.get_connection()?;
        let receiver_pubkey_bytes = receiver_pubkey.to_compressed_bytes();

        let (duplicate_uri, duplicate_rk): (bool, bool) = conn.query_row(
            "SELECT \
                EXISTS(SELECT 1 FROM send_sessions WHERE pj_uri = ?1), \
                EXISTS(SELECT 1 FROM send_sessions WHERE receiver_pubkey = ?2)",
            params![pj_uri, &receiver_pubkey_bytes],
            |row| Ok((row.get(0)?, row.get(1)?)),
        )?;

        if duplicate_uri {
            return Err(Error::DuplicateSendSession(DuplicateKind::Uri));
        }
        if duplicate_rk {
            return Err(Error::DuplicateSendSession(DuplicateKind::ReceiverPubkey));
        }

        let session_id = uuid::Uuid::now_v7();
        conn.execute(
            "INSERT INTO send_sessions (session_id, pj_uri, receiver_pubkey) VALUES (?1, ?2, ?3)",
            params![session_id.to_string(), pj_uri, &receiver_pubkey_bytes],
        )?;

        Ok(Self { db, session_id: SessionId(session_id) })
    }

    pub fn from_id(db: Arc<Database>, id: SessionId) -> Self { Self { db, session_id: id } }

    pub fn session_id(&self) -> SessionId { self.session_id.clone() }
}
impl SessionPersister for SenderPersister {
    type SessionEvent = SenderSessionEvent;
    type InternalStorageError = crate::db::error::Error;

    fn save_event(
        &self,
        event: SenderSessionEvent,
    ) -> std::result::Result<(), Self::InternalStorageError> {
        let conn = self.db.get_connection()?;
        let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?;

        conn.execute(
            "INSERT INTO send_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)",
            params![self.session_id.0.to_string(), event_data, now()],
        )?;

        Ok(())
    }

    fn load(
        &self,
    ) -> std::result::Result<Box<dyn Iterator<Item = SenderSessionEvent>>, Self::InternalStorageError>
    {
        let conn = self.db.get_connection()?;
        let mut stmt = conn.prepare(
            "SELECT event_data FROM send_session_events WHERE session_id = ?1 ORDER BY id ASC",
        )?;

        let event_rows = stmt.query_map(params![self.session_id.0.to_string()], |row| {
            let event_data: String = row.get(0)?;
            Ok(event_data)
        })?;

        let events: Vec<SenderSessionEvent> = event_rows
            .map(|row| {
                let event_data = row.expect("Failed to read event data from database");
                serde_json::from_str::<SenderSessionEvent>(&event_data)
                    .expect("Database corruption: failed to deserialize session event")
            })
            .collect();

        Ok(Box::new(events.into_iter()))
    }

    fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {
        let conn = self.db.get_connection()?;

        conn.execute(
            "UPDATE send_sessions SET completed_at = ?1 WHERE session_id = ?2",
            params![now(), self.session_id.0.to_string()],
        )?;

        Ok(())
    }
}

#[derive(Clone)]
pub(crate) struct ReceiverPersister {
    db: Arc<Database>,
    session_id: SessionId,
}

impl ReceiverPersister {
    pub fn new(db: Arc<Database>) -> crate::db::Result<Self> {
        let conn = db.get_connection()?;

        let session_id = uuid::Uuid::now_v7();
        conn.execute(
            "INSERT INTO receive_sessions (session_id) VALUES (?1)",
            params![session_id.to_string()],
        )?;

        Ok(Self { db, session_id: SessionId(session_id) })
    }

    pub fn from_id(db: Arc<Database>, id: SessionId) -> Self { Self { db, session_id: id } }

    pub fn session_id(&self) -> SessionId { self.session_id.clone() }
}

impl SessionPersister for ReceiverPersister {
    type SessionEvent = ReceiverSessionEvent;
    type InternalStorageError = crate::db::error::Error;

    fn save_event(
        &self,
        event: ReceiverSessionEvent,
    ) -> std::result::Result<(), Self::InternalStorageError> {
        let conn = self.db.get_connection()?;
        let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?;

        conn.execute(
            "INSERT INTO receive_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)",
            params![self.session_id.0.to_string(), event_data, now()],
        )?;

        Ok(())
    }

    fn load(
        &self,
    ) -> std::result::Result<
        Box<dyn Iterator<Item = ReceiverSessionEvent>>,
        Self::InternalStorageError,
    > {
        let conn = self.db.get_connection()?;
        let mut stmt = conn.prepare(
            "SELECT event_data FROM receive_session_events WHERE session_id = ?1 ORDER BY id ASC",
        )?;

        let event_rows = stmt.query_map(params![self.session_id.0.to_string()], |row| {
            let event_data: String = row.get(0)?;
            Ok(event_data)
        })?;

        let events: Vec<ReceiverSessionEvent> = event_rows
            .map(|row| {
                let event_data = row.expect("Failed to read event data from database");
                serde_json::from_str::<ReceiverSessionEvent>(&event_data)
                    .expect("Database corruption: failed to deserialize session event")
            })
            .collect();

        Ok(Box::new(events.into_iter()))
    }

    fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {
        let conn = self.db.get_connection()?;

        conn.execute(
            "UPDATE receive_sessions SET completed_at = ?1 WHERE session_id = ?2",
            params![now(), self.session_id.0.to_string()],
        )?;

        Ok(())
    }
}

impl Database {
    pub(crate) fn get_recv_session_ids(&self) -> Result<Vec<SessionId>> {
        let conn = self.get_connection()?;
        let mut stmt =
            conn.prepare("SELECT session_id FROM receive_sessions WHERE completed_at IS NULL")?;

        let session_rows = stmt.query_map([], |row| {
            let session_id: String = row.get(0)?;
            let session_id = uuid::Uuid::parse_str(&session_id)
                .expect("Database corruption: invalid session_id UUID");
            Ok(SessionId(session_id))
        })?;

        let mut session_ids = Vec::new();
        for session_row in session_rows {
            let session_id = session_row?;
            session_ids.push(session_id);
        }

        Ok(session_ids)
    }

    pub(crate) fn get_send_session_ids(&self) -> Result<Vec<SessionId>> {
        let conn = self.get_connection()?;
        let mut stmt =
            conn.prepare("SELECT session_id FROM send_sessions WHERE completed_at IS NULL")?;

        let session_rows = stmt.query_map([], |row| {
            let session_id: String = row.get(0)?;
            let session_id = uuid::Uuid::parse_str(&session_id)
                .expect("Database corruption: invalid session_id UUID");
            Ok(SessionId(session_id))
        })?;

        let mut session_ids = Vec::new();
        for session_row in session_rows {
            let session_id = session_row?;
            session_ids.push(session_id);
        }

        Ok(session_ids)
    }

    pub(crate) fn get_send_session_receiver_pk(
        &self,
        session_id: &SessionId,
    ) -> Result<HpkePublicKey> {
        let conn = self.get_connection()?;
        let mut stmt =
            conn.prepare("SELECT receiver_pubkey FROM send_sessions WHERE session_id = ?1")?;
        let receiver_pubkey: Vec<u8> =
            stmt.query_row(params![session_id.0.to_string()], |row| row.get(0))?;
        Ok(HpkePublicKey::from_compressed_bytes(&receiver_pubkey).expect("Valid receiver pubkey"))
    }

    pub(crate) fn get_inactive_send_session_ids(&self) -> Result<Vec<(SessionId, u64)>> {
        let conn = self.get_connection()?;
        let mut stmt = conn.prepare(
            "SELECT session_id, completed_at FROM send_sessions WHERE completed_at IS NOT NULL",
        )?;
        let session_rows = stmt.query_map([], |row| {
            let session_id: String = row.get(0)?;
            let session_id = uuid::Uuid::parse_str(&session_id)
                .expect("Database corruption: invalid session_id UUID");
            let completed_at: u64 = row.get(1)?;
            Ok((SessionId(session_id), completed_at))
        })?;

        let mut session_ids = Vec::new();
        for session_row in session_rows {
            let (session_id, completed_at) = session_row?;
            session_ids.push((session_id, completed_at));
        }
        Ok(session_ids)
    }

    pub(crate) fn get_inactive_recv_session_ids(&self) -> Result<Vec<(SessionId, u64)>> {
        let conn = self.get_connection()?;
        let mut stmt = conn.prepare(
            "SELECT session_id, completed_at FROM receive_sessions WHERE completed_at IS NOT NULL",
        )?;
        let session_rows = stmt.query_map([], |row| {
            let session_id: String = row.get(0)?;
            let session_id = uuid::Uuid::parse_str(&session_id)
                .expect("Database corruption: invalid session_id UUID");
            let completed_at: u64 = row.get(1)?;
            Ok((SessionId(session_id), completed_at))
        })?;

        let mut session_ids = Vec::new();
        for session_row in session_rows {
            let (session_id, completed_at) = session_row?;
            session_ids.push((session_id, completed_at));
        }
        Ok(session_ids)
    }
}

#[cfg(all(test, feature = "v2"))]
mod tests {
    use std::sync::Arc;

    use payjoin::HpkeKeyPair;

    use super::*;

    fn create_test_db() -> Arc<Database> {
        // Use an in-memory database for tests
        let manager = r2d2_sqlite::SqliteConnectionManager::memory()
            .with_init(|conn| conn.execute_batch("PRAGMA locking_mode = EXCLUSIVE;"));
        let pool = r2d2::Pool::new(manager).expect("pool creation should succeed");
        let conn = pool.get().expect("connection should succeed");
        Database::init_schema(&conn).expect("schema init should succeed");
        Arc::new(Database(pool))
    }

    fn make_receiver_pubkey() -> payjoin::HpkePublicKey { HpkeKeyPair::gen_keypair().1 }

    // Second call with the same URI (same active session) should return DuplicateSendSession(Uri).
    #[test]
    fn test_duplicate_uri_returns_error() {
        let db = create_test_db();
        let rk1 = make_receiver_pubkey();
        let rk2 = make_receiver_pubkey();
        let uri = "bitcoin:addr1?pj=https://example.com/BBBBBBBB";

        SenderPersister::new(db.clone(), uri, &rk1).expect("first session should succeed");

        let err = SenderPersister::new(db, uri, &rk2).expect_err("duplicate URI should fail");
        assert!(
            matches!(err, Error::DuplicateSendSession(DuplicateKind::Uri)),
            "expected DuplicateSendSession(Uri), got: {err:?}"
        );
    }

    // Same receiver pubkey under a different URI should return DuplicateSendSession(ReceiverPubkey).
    #[test]
    fn test_duplicate_rk_returns_error() {
        let db = create_test_db();
        let rk = make_receiver_pubkey();
        let uri1 = "bitcoin:addr1?pj=https://example.com/CCCCCCCC";
        let uri2 = "bitcoin:addr1?pj=https://example.com/DDDDDDDD";

        SenderPersister::new(db.clone(), uri1, &rk).expect("first session should succeed");

        let err = SenderPersister::new(db, uri2, &rk).expect_err("duplicate RK should fail");
        assert!(
            matches!(err, Error::DuplicateSendSession(DuplicateKind::ReceiverPubkey)),
            "expected DuplicateSendSession(ReceiverPubkey), got: {err:?}"
        );
    }

    // After a session is marked completed, a new session with the same URI must still be rejected
    // to prevent address reuse, HPKE receiver-key reuse
    #[test]
    fn test_completed_session_blocks_reuse() {
        let db = create_test_db();
        let rk1 = make_receiver_pubkey();
        let rk2 = make_receiver_pubkey();
        let uri = "bitcoin:addr1?pj=https://example.com/EEEEEEEE";

        let persister =
            SenderPersister::new(db.clone(), uri, &rk1).expect("first session should succeed");

        // Mark the session as completed
        use payjoin::persist::SessionPersister;
        persister.close().expect("close should succeed");

        // A new session with the same URI must be rejected even after completion
        let err = SenderPersister::new(db, uri, &rk2)
            .expect_err("reuse of a completed session URI must be rejected");
        assert!(
            matches!(err, Error::DuplicateSendSession(DuplicateKind::Uri)),
            "expected DuplicateSendSession(Uri), got: {err:?}"
        );
    }
}

1	use std::sync::Arc;
2
3	use payjoin::persist::SessionPersister;
4	use payjoin::receive::v2::SessionEvent as ReceiverSessionEvent;
5	use payjoin::send::v2::SessionEvent as SenderSessionEvent;
6	use payjoin::HpkePublicKey;
7	use rusqlite::params;
8
9	use super::*;
10
11	#[derive(Debug, Clone, PartialEq, Eq)]
12	pub(crate) struct SessionId(pub(crate) uuid::Uuid);
13
14	impl std::fmt::Display for SessionId {
15	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{}", self.0) }	14✔
16	}
17
18	#[derive(Clone, Debug)]
19	pub(crate) struct SenderPersister {
20	db: Arc<Database>,
21	session_id: SessionId,
22	}
23
24	impl SenderPersister {
25	pub fn new(	8✔
26	db: Arc<Database>,	8✔
27	pj_uri: &str,	8✔
28	receiver_pubkey: &HpkePublicKey,	8✔
29	) -> crate::db::Result<Self> {	8✔
30	let conn = db.get_connection()?;	8✔
31	let receiver_pubkey_bytes = receiver_pubkey.to_compressed_bytes();	8✔
32
33	let (duplicate_uri, duplicate_rk): (bool, bool) = conn.query_row(	8✔
34	"SELECT \	8✔
35	EXISTS(SELECT 1 FROM send_sessions WHERE pj_uri = ?1), \	8✔
36	EXISTS(SELECT 1 FROM send_sessions WHERE receiver_pubkey = ?2)",	8✔
37	params![pj_uri, &receiver_pubkey_bytes],	8✔
38	\|row\| Ok((row.get(0)?, row.get(1)?)),	8✔
39	)?;	×
40
41	if duplicate_uri {	8✔
42	return Err(Error::DuplicateSendSession(DuplicateKind::Uri));	2✔
43	}	6✔
44	if duplicate_rk {	6✔
45	return Err(Error::DuplicateSendSession(DuplicateKind::ReceiverPubkey));	1✔
46	}	5✔
47
48	let session_id = uuid::Uuid::now_v7();	5✔
49	conn.execute(	5✔
50	"INSERT INTO send_sessions (session_id, pj_uri, receiver_pubkey) VALUES (?1, ?2, ?3)",	5✔
51	params![session_id.to_string(), pj_uri, &receiver_pubkey_bytes],	5✔
52	)?;	5✔
53
54	Ok(Self { db, session_id: SessionId(session_id) })	5✔
55	}	8✔
56
57	pub fn from_id(db: Arc<Database>, id: SessionId) -> Self { Self { db, session_id: id } }	2✔
58
59	pub fn session_id(&self) -> SessionId { self.session_id.clone() }	5✔
60	}
61	impl SessionPersister for SenderPersister {
62	type SessionEvent = SenderSessionEvent;
63	type InternalStorageError = crate::db::error::Error;
64
65	fn save_event(	6✔
66	&self,	6✔
67	event: SenderSessionEvent,	6✔
68	) -> std::result::Result<(), Self::InternalStorageError> {	6✔
69	let conn = self.db.get_connection()?;	6✔
70	let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?;	6✔
71
72	conn.execute(	6✔
73	"INSERT INTO send_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)",	6✔
74	params![self.session_id.0.to_string(), event_data, now()],	6✔
75	)?;	6✔
76
77	Ok(())	6✔
78	}	6✔
79
80	fn load(	2✔
81	&self,	2✔
82	) -> std::result::Result<Box<dyn Iterator<Item = SenderSessionEvent>>, Self::InternalStorageError>	2✔
83	{
84	let conn = self.db.get_connection()?;	2✔
85	let mut stmt = conn.prepare(	2✔
86	"SELECT event_data FROM send_session_events WHERE session_id = ?1 ORDER BY id ASC",	2✔
87	)?;	2✔
88
89	let event_rows = stmt.query_map(params![self.session_id.0.to_string()], \|row\| {	3✔
90	let event_data: String = row.get(0)?;	3✔
91	Ok(event_data)	3✔
92	})?;	3✔
93
94	let events: Vec<SenderSessionEvent> = event_rows	2✔
95	.map(\|row\| {	3✔
96	let event_data = row.expect("Failed to read event data from database");	3✔
97	serde_json::from_str::<SenderSessionEvent>(&event_data)	3✔
98	.expect("Database corruption: failed to deserialize session event")	3✔
99	})	3✔
100	.collect();	2✔
101
102	Ok(Box::new(events.into_iter()))	2✔
103	}	2✔
104
105	fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {	3✔
106	let conn = self.db.get_connection()?;	3✔
107
108	conn.execute(	3✔
109	"UPDATE send_sessions SET completed_at = ?1 WHERE session_id = ?2",	3✔
110	params![now(), self.session_id.0.to_string()],	3✔
111	)?;	3✔
112
113	Ok(())	3✔
114	}	3✔
115	}
116
117	#[derive(Clone)]
118	pub(crate) struct ReceiverPersister {
119	db: Arc<Database>,
120	session_id: SessionId,
121	}
122
123	impl ReceiverPersister {
124	pub fn new(db: Arc<Database>) -> crate::db::Result<Self> {	3✔
125	let conn = db.get_connection()?;	3✔
126
127	let session_id = uuid::Uuid::now_v7();	3✔
128	conn.execute(	3✔
129	"INSERT INTO receive_sessions (session_id) VALUES (?1)",	3✔
130	params![session_id.to_string()],	3✔
131	)?;	3✔
132
133	Ok(Self { db, session_id: SessionId(session_id) })	3✔
134	}	3✔
135
136	pub fn from_id(db: Arc<Database>, id: SessionId) -> Self { Self { db, session_id: id } }	4✔
137
138	pub fn session_id(&self) -> SessionId { self.session_id.clone() }	3✔
139	}
140
141	impl SessionPersister for ReceiverPersister {
142	type SessionEvent = ReceiverSessionEvent;
143	type InternalStorageError = crate::db::error::Error;
144
145	fn save_event(	15✔
146	&self,	15✔
147	event: ReceiverSessionEvent,	15✔
148	) -> std::result::Result<(), Self::InternalStorageError> {	15✔
149	let conn = self.db.get_connection()?;	15✔
150	let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?;	15✔
151
152	conn.execute(	15✔
153	"INSERT INTO receive_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)",	15✔
154	params![self.session_id.0.to_string(), event_data, now()],	15✔
155	)?;	15✔
156
157	Ok(())	15✔
158	}	15✔
159
160	fn load(	4✔
161	&self,	4✔
162	) -> std::result::Result<	4✔
163	Box<dyn Iterator<Item = ReceiverSessionEvent>>,	4✔
164	Self::InternalStorageError,	4✔
165	> {	4✔
166	let conn = self.db.get_connection()?;	4✔
167	let mut stmt = conn.prepare(	4✔
168	"SELECT event_data FROM receive_session_events WHERE session_id = ?1 ORDER BY id ASC",	4✔
169	)?;	4✔
170
171	let event_rows = stmt.query_map(params![self.session_id.0.to_string()], \|row\| {	24✔
172	let event_data: String = row.get(0)?;	24✔
173	Ok(event_data)	24✔
174	})?;	24✔
175
176	let events: Vec<ReceiverSessionEvent> = event_rows	4✔
177	.map(\|row\| {	24✔
178	let event_data = row.expect("Failed to read event data from database");	24✔
179	serde_json::from_str::<ReceiverSessionEvent>(&event_data)	24✔
180	.expect("Database corruption: failed to deserialize session event")	24✔
181	})	24✔
182	.collect();	4✔
183
184	Ok(Box::new(events.into_iter()))	4✔
185	}	4✔
186
187	fn close(&self) -> std::result::Result<(), Self::InternalStorageError> {	2✔
188	let conn = self.db.get_connection()?;	2✔
189
190	conn.execute(	2✔
191	"UPDATE receive_sessions SET completed_at = ?1 WHERE session_id = ?2",	2✔
192	params![now(), self.session_id.0.to_string()],	2✔
193	)?;	2✔
194
195	Ok(())	2✔
196	}	2✔
197	}
198
199	impl Database {
200	pub(crate) fn get_recv_session_ids(&self) -> Result<Vec<SessionId>> {	6✔
201	let conn = self.get_connection()?;	6✔
202	let mut stmt =	6✔
203	conn.prepare("SELECT session_id FROM receive_sessions WHERE completed_at IS NULL")?;	6✔
204
205	let session_rows = stmt.query_map([], \|row\| {	6✔
206	let session_id: String = row.get(0)?;	3✔
207	let session_id = uuid::Uuid::parse_str(&session_id)	3✔
208	.expect("Database corruption: invalid session_id UUID");	3✔
209	Ok(SessionId(session_id))	3✔
210	})?;	3✔
211
212	let mut session_ids = Vec::new();	6✔
213	for session_row in session_rows {	6✔
214	let session_id = session_row?;	3✔
215	session_ids.push(session_id);	3✔
216	}
217
218	Ok(session_ids)	6✔
219	}	6✔
220
221	pub(crate) fn get_send_session_ids(&self) -> Result<Vec<SessionId>> {	9✔
222	let conn = self.get_connection()?;	9✔
223	let mut stmt =	9✔
224	conn.prepare("SELECT session_id FROM send_sessions WHERE completed_at IS NULL")?;	9✔
225
226	let session_rows = stmt.query_map([], \|row\| {	9✔
227	let session_id: String = row.get(0)?;	2✔
228	let session_id = uuid::Uuid::parse_str(&session_id)	2✔
229	.expect("Database corruption: invalid session_id UUID");	2✔
230	Ok(SessionId(session_id))	2✔
231	})?;	2✔
232
233	let mut session_ids = Vec::new();	9✔
234	for session_row in session_rows {	9✔
235	let session_id = session_row?;	2✔
236	session_ids.push(session_id);	2✔
237	}
238
239	Ok(session_ids)	9✔
240	}	9✔
241
242	pub(crate) fn get_send_session_receiver_pk(	1✔
243	&self,	1✔
244	session_id: &SessionId,	1✔
245	) -> Result<HpkePublicKey> {	1✔
246	let conn = self.get_connection()?;	1✔
247	let mut stmt =	1✔
248	conn.prepare("SELECT receiver_pubkey FROM send_sessions WHERE session_id = ?1")?;	1✔
249	let receiver_pubkey: Vec<u8> =	1✔
250	stmt.query_row(params![session_id.0.to_string()], \|row\| row.get(0))?;	1✔
251	Ok(HpkePublicKey::from_compressed_bytes(&receiver_pubkey).expect("Valid receiver pubkey"))	1✔
252	}	1✔
253
254	pub(crate) fn get_inactive_send_session_ids(&self) -> Result<Vec<(SessionId, u64)>> {	×
255	let conn = self.get_connection()?;	×
256	let mut stmt = conn.prepare(	×
257	"SELECT session_id, completed_at FROM send_sessions WHERE completed_at IS NOT NULL",	×
258	)?;	×
259	let session_rows = stmt.query_map([], \|row\| {	×
NEW 260	let session_id: String = row.get(0)?;	×
NEW 261	let session_id = uuid::Uuid::parse_str(&session_id)	×
NEW 262	.expect("Database corruption: invalid session_id UUID");	×
263	let completed_at: u64 = row.get(1)?;	×
264	Ok((SessionId(session_id), completed_at))	×
265	})?;	×
266
267	let mut session_ids = Vec::new();	×
268	for session_row in session_rows {	×
269	let (session_id, completed_at) = session_row?;	×
270	session_ids.push((session_id, completed_at));	×
271	}
272	Ok(session_ids)	×
273	}	×
274
275	pub(crate) fn get_inactive_recv_session_ids(&self) -> Result<Vec<(SessionId, u64)>> {	×
276	let conn = self.get_connection()?;	×
277	let mut stmt = conn.prepare(	×
278	"SELECT session_id, completed_at FROM receive_sessions WHERE completed_at IS NOT NULL",	×
279	)?;	×
280	let session_rows = stmt.query_map([], \|row\| {	×
NEW 281	let session_id: String = row.get(0)?;	×
NEW 282	let session_id = uuid::Uuid::parse_str(&session_id)	×
NEW 283	.expect("Database corruption: invalid session_id UUID");	×
284	let completed_at: u64 = row.get(1)?;	×
285	Ok((SessionId(session_id), completed_at))	×
286	})?;	×
287
288	let mut session_ids = Vec::new();	×
289	for session_row in session_rows {	×
290	let (session_id, completed_at) = session_row?;	×
291	session_ids.push((session_id, completed_at));	×
292	}
293	Ok(session_ids)	×
294	}	×
295	}
296
297	#[cfg(all(test, feature = "v2"))]
298	mod tests {
299	use std::sync::Arc;
300
301	use payjoin::HpkeKeyPair;
302
303	use super::*;
304
305	fn create_test_db() -> Arc<Database> {	3✔
306	// Use an in-memory database for tests
307	let manager = r2d2_sqlite::SqliteConnectionManager::memory()	3✔
308	.with_init(\|conn\| conn.execute_batch("PRAGMA locking_mode = EXCLUSIVE;"));	30✔
309	let pool = r2d2::Pool::new(manager).expect("pool creation should succeed");	3✔
310	let conn = pool.get().expect("connection should succeed");	3✔
311	Database::init_schema(&conn).expect("schema init should succeed");	3✔
312	Arc::new(Database(pool))	3✔
313	}	3✔
314
315	fn make_receiver_pubkey() -> payjoin::HpkePublicKey { HpkeKeyPair::gen_keypair().1 }	5✔
316
317	// Second call with the same URI (same active session) should return DuplicateSendSession(Uri).
318	#[test]
319	fn test_duplicate_uri_returns_error() {	1✔
320	let db = create_test_db();	1✔
321	let rk1 = make_receiver_pubkey();	1✔
322	let rk2 = make_receiver_pubkey();	1✔
323	let uri = "bitcoin:addr1?pj=https://example.com/BBBBBBBB";	1✔
324
325	SenderPersister::new(db.clone(), uri, &rk1).expect("first session should succeed");	1✔
326
327	let err = SenderPersister::new(db, uri, &rk2).expect_err("duplicate URI should fail");	1✔
328	assert!(	1✔
329	matches!(err, Error::DuplicateSendSession(DuplicateKind::Uri)),	1✔
330	"expected DuplicateSendSession(Uri), got: {err:?}"
331	);
332	}	1✔
333
334	// Same receiver pubkey under a different URI should return DuplicateSendSession(ReceiverPubkey).
335	#[test]
336	fn test_duplicate_rk_returns_error() {	1✔
337	let db = create_test_db();	1✔
338	let rk = make_receiver_pubkey();	1✔
339	let uri1 = "bitcoin:addr1?pj=https://example.com/CCCCCCCC";	1✔
340	let uri2 = "bitcoin:addr1?pj=https://example.com/DDDDDDDD";	1✔
341
342	SenderPersister::new(db.clone(), uri1, &rk).expect("first session should succeed");	1✔
343
344	let err = SenderPersister::new(db, uri2, &rk).expect_err("duplicate RK should fail");	1✔
345	assert!(	1✔
346	matches!(err, Error::DuplicateSendSession(DuplicateKind::ReceiverPubkey)),	1✔
347	"expected DuplicateSendSession(ReceiverPubkey), got: {err:?}"
348	);
349	}	1✔
350
351	// After a session is marked completed, a new session with the same URI must still be rejected
352	// to prevent address reuse, HPKE receiver-key reuse
353	#[test]
354	fn test_completed_session_blocks_reuse() {	1✔
355	let db = create_test_db();	1✔
356	let rk1 = make_receiver_pubkey();	1✔
357	let rk2 = make_receiver_pubkey();	1✔
358	let uri = "bitcoin:addr1?pj=https://example.com/EEEEEEEE";	1✔
359
360	let persister =	1✔
361	SenderPersister::new(db.clone(), uri, &rk1).expect("first session should succeed");	1✔
362
363	// Mark the session as completed
364	use payjoin::persist::SessionPersister;
365	persister.close().expect("close should succeed");	1✔
366
367	// A new session with the same URI must be rejected even after completion
368	let err = SenderPersister::new(db, uri, &rk2)	1✔
369	.expect_err("reuse of a completed session URI must be rejected");	1✔
370	assert!(	1✔
371	matches!(err, Error::DuplicateSendSession(DuplicateKind::Uri)),	1✔
372	"expected DuplicateSendSession(Uri), got: {err:?}"
373	);
374	}	1✔
375	}

payjoin / rust-payjoin / 27376140423

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous