pipedash_core/infrastructure/secrets/
sqlite.rs1use std::collections::HashMap;
2use std::sync::Arc;
3
4use aes_gcm::{
5 aead::{
6 Aead,
7 KeyInit,
8 },
9 Aes256Gcm,
10 Nonce,
11};
12use async_trait::async_trait;
13use sqlx::{
14 Pool,
15 Sqlite,
16};
17use tokio::sync::RwLock;
18
19use crate::domain::{
20 DomainError,
21 DomainResult,
22};
23use crate::infrastructure::TokenStore;
24
25const SQLITE_KDF_SALT: &[u8] = b"pipedash-sqlite-vault-v1";
26const BACKUP_KDF_SALT: &[u8] = b"pipedash-backup-salt-v1";
27
28pub struct SqliteTokenStore {
29 pool: Pool<Sqlite>,
30 encryption_key: [u8; 32],
31 cache: Arc<RwLock<HashMap<i64, String>>>,
32}
33
34impl SqliteTokenStore {
35 pub async fn new(pool: Pool<Sqlite>, vault_password: Option<String>) -> DomainResult<Self> {
36 let password = vault_password
37 .or_else(|| std::env::var("PIPEDASH_VAULT_PASSWORD").ok())
38 .ok_or_else(|| {
39 DomainError::InvalidConfig(
40 "PIPEDASH_VAULT_PASSWORD environment variable required for SQLite token encryption".into(),
41 )
42 })?;
43
44 let encryption_key = Self::derive_encryption_key(&password);
45
46 let store = Self {
47 pool,
48 encryption_key,
49 cache: Arc::new(RwLock::new(HashMap::new())),
50 };
51
52 store.load_to_cache().await?;
53
54 tracing::info!("SQLite token store initialized");
55 Ok(store)
56 }
57
58 pub async fn with_pool(
59 pool: Pool<Sqlite>, vault_password: Option<String>,
60 ) -> DomainResult<Self> {
61 Self::new(pool, vault_password).await
62 }
63
64 fn derive_encryption_key(password: &str) -> [u8; 32] {
65 use argon2::{
66 Argon2,
67 ParamsBuilder,
68 };
69
70 let mut output = [0u8; 32];
71
72 let params = ParamsBuilder::new()
73 .m_cost(65536)
74 .t_cost(3)
75 .p_cost(1)
76 .output_len(32)
77 .build()
78 .expect("Invalid Argon2 parameters");
79
80 let argon2 = Argon2::new(argon2::Algorithm::Argon2id, argon2::Version::V0x13, params);
81
82 argon2
83 .hash_password_into(password.as_bytes(), SQLITE_KDF_SALT, &mut output)
84 .expect("Failed to derive encryption key");
85
86 output
87 }
88
89 fn derive_backup_key(password: &str) -> [u8; 32] {
90 use argon2::{
91 Argon2,
92 ParamsBuilder,
93 };
94
95 let mut output = [0u8; 32];
96
97 let params = ParamsBuilder::new()
98 .m_cost(65536)
99 .t_cost(3)
100 .p_cost(1)
101 .output_len(32)
102 .build()
103 .expect("Invalid Argon2 parameters");
104
105 let argon2 = Argon2::new(argon2::Algorithm::Argon2id, argon2::Version::V0x13, params);
106
107 argon2
108 .hash_password_into(password.as_bytes(), BACKUP_KDF_SALT, &mut output)
109 .expect("Failed to derive backup key");
110
111 output
112 }
113
114 async fn encrypt_token(&self, plaintext: &str) -> DomainResult<(Vec<u8>, Vec<u8>)> {
115 let cipher = Aes256Gcm::new_from_slice(&self.encryption_key)
116 .map_err(|e| DomainError::InternalError(format!("Failed to create cipher: {}", e)))?;
117
118 let nonce_bytes: [u8; 12] = rand::random();
119 let nonce = Nonce::from(nonce_bytes);
120
121 let ciphertext = cipher
122 .encrypt(&nonce, plaintext.as_bytes())
123 .map_err(|e| DomainError::InternalError(format!("Encryption failed: {}", e)))?;
124
125 Ok((nonce_bytes.to_vec(), ciphertext))
126 }
127
128 async fn decrypt_token(&self, nonce: &[u8], ciphertext: &[u8]) -> DomainResult<String> {
129 if nonce.len() != 12 {
130 return Err(DomainError::InvalidConfig("Invalid nonce length".into()));
131 }
132
133 let cipher = Aes256Gcm::new_from_slice(&self.encryption_key)
134 .map_err(|e| DomainError::InternalError(format!("Failed to create cipher: {}", e)))?;
135
136 let nonce_array: [u8; 12] = nonce
137 .try_into()
138 .map_err(|_| DomainError::InvalidConfig("Invalid nonce".into()))?;
139 let nonce = Nonce::from(nonce_array);
140
141 let plaintext = cipher.decrypt(&nonce, ciphertext).map_err(|_| {
142 DomainError::AuthenticationFailed("Token decryption failed - wrong password?".into())
143 })?;
144
145 String::from_utf8(plaintext)
146 .map_err(|e| DomainError::DatabaseError(format!("Invalid UTF-8 in token: {}", e)))
147 }
148
149 async fn load_to_cache(&self) -> DomainResult<()> {
150 let rows = sqlx::query_as::<_, (i64, Vec<u8>, Vec<u8>)>(
151 "SELECT id, encrypted_token, token_nonce FROM providers
152 WHERE encrypted_token IS NOT NULL AND token_nonce IS NOT NULL",
153 )
154 .fetch_all(&self.pool)
155 .await
156 .map_err(|e| DomainError::DatabaseError(format!("Failed to load tokens: {}", e)))?;
157
158 let mut cache = self.cache.write().await;
159 cache.clear();
160
161 let mut decryption_errors = Vec::new();
162
163 for (provider_id, ciphertext, nonce) in rows {
164 match self.decrypt_token(&nonce, &ciphertext).await {
165 Ok(token) => {
166 cache.insert(provider_id, token);
167 }
168 Err(e) => {
169 tracing::warn!(
170 provider_id = provider_id,
171 error = %e,
172 "Failed to decrypt token for provider"
173 );
174 decryption_errors.push((provider_id, e));
175 }
176 }
177 }
178
179 if !decryption_errors.is_empty() {
180 let failed_count = decryption_errors.len();
181 return Err(DomainError::AuthenticationFailed(format!(
182 "Failed to decrypt {} token(s) - wrong vault password",
183 failed_count
184 )));
185 }
186
187 tracing::debug!(count = cache.len(), "Loaded tokens to cache");
188 Ok(())
189 }
190}
191
192#[async_trait]
193impl TokenStore for SqliteTokenStore {
194 async fn store_token(&self, provider_id: i64, token: &str) -> DomainResult<()> {
195 tracing::debug!(
196 "[SqliteTokenStore] Storing token for provider {} (token length: {})",
197 provider_id,
198 token.len()
199 );
200
201 let (nonce, ciphertext) = self.encrypt_token(token).await?;
202 tracing::debug!(
203 "[SqliteTokenStore] Token encrypted: nonce={} bytes, ciphertext={} bytes",
204 nonce.len(),
205 ciphertext.len()
206 );
207
208 let result = sqlx::query(
209 "UPDATE providers SET encrypted_token = ?1, token_nonce = ?2, updated_at = datetime('now')
210 WHERE id = ?3",
211 )
212 .bind(&ciphertext)
213 .bind(&nonce)
214 .bind(provider_id)
215 .execute(&self.pool)
216 .await
217 .map_err(|e| DomainError::DatabaseError(format!("Failed to store token: {}", e)))?;
218
219 tracing::info!(
220 "[SqliteTokenStore] Token stored for provider {} (rows affected: {})",
221 provider_id,
222 result.rows_affected()
223 );
224
225 if result.rows_affected() == 0 {
226 tracing::warn!(
227 "[SqliteTokenStore] No rows affected when storing token for provider {} - provider may not exist!",
228 provider_id
229 );
230 }
231
232 let mut cache = self.cache.write().await;
233 cache.insert(provider_id, token.to_string());
234
235 tracing::debug!(provider_id = provider_id, "Token stored successfully");
236 Ok(())
237 }
238
239 async fn get_token(&self, provider_id: i64) -> DomainResult<String> {
240 {
241 let cache = self.cache.read().await;
242 if let Some(token) = cache.get(&provider_id) {
243 return Ok(token.clone());
244 }
245 }
246
247 let row = sqlx::query_as::<_, (Vec<u8>, Vec<u8>)>(
248 "SELECT encrypted_token, token_nonce FROM providers
249 WHERE id = ?1 AND encrypted_token IS NOT NULL",
250 )
251 .bind(provider_id)
252 .fetch_optional(&self.pool)
253 .await
254 .map_err(|e| DomainError::DatabaseError(format!("Failed to get token: {}", e)))?
255 .ok_or_else(|| {
256 DomainError::DatabaseError(format!("Token not found for provider {}", provider_id))
257 })?;
258
259 let token = self.decrypt_token(&row.1, &row.0).await?;
260
261 {
262 let mut cache = self.cache.write().await;
263 cache.insert(provider_id, token.clone());
264 }
265
266 Ok(token)
267 }
268
269 async fn delete_token(&self, provider_id: i64) -> DomainResult<()> {
270 sqlx::query(
271 "UPDATE providers SET encrypted_token = NULL, token_nonce = NULL, updated_at = datetime('now')
272 WHERE id = ?1",
273 )
274 .bind(provider_id)
275 .execute(&self.pool)
276 .await
277 .map_err(|e| DomainError::DatabaseError(format!("Failed to delete token: {}", e)))?;
278
279 let mut cache = self.cache.write().await;
280 cache.remove(&provider_id);
281
282 tracing::debug!(provider_id = provider_id, "Token deleted successfully");
283 Ok(())
284 }
285
286 async fn get_all_tokens(&self) -> DomainResult<HashMap<i64, String>> {
287 let cache = self.cache.read().await;
288 Ok(cache.clone())
289 }
290
291 async fn export_encrypted(&self, password: &str) -> DomainResult<Vec<u8>> {
292 let tokens = self.get_all_tokens().await?;
293 let json = serde_json::to_vec(&tokens).map_err(|e| {
294 DomainError::DatabaseError(format!("Failed to serialize tokens: {}", e))
295 })?;
296
297 let backup_key = Self::derive_backup_key(password);
298 let cipher = Aes256Gcm::new_from_slice(&backup_key)
299 .map_err(|e| DomainError::InternalError(format!("Failed to create cipher: {}", e)))?;
300
301 let nonce_bytes: [u8; 12] = rand::random();
302 let nonce = Nonce::from(nonce_bytes);
303
304 let ciphertext = cipher
305 .encrypt(&nonce, json.as_ref())
306 .map_err(|e| DomainError::InternalError(format!("Encryption failed: {}", e)))?;
307
308 let mut result = nonce_bytes.to_vec();
309 result.extend(ciphertext);
310 Ok(result)
311 }
312
313 async fn import_encrypted(&self, data: &[u8], password: &str) -> DomainResult<()> {
314 tracing::info!(
315 "[SqliteTokenStore] Starting import_encrypted (data size: {} bytes, password length: {})",
316 data.len(),
317 password.len()
318 );
319
320 if data.len() < 12 {
321 return Err(DomainError::InvalidConfig("Invalid backup data".into()));
322 }
323
324 let nonce_bytes: [u8; 12] = data[..12]
325 .try_into()
326 .map_err(|_| DomainError::InvalidConfig("Invalid nonce".into()))?;
327 let ciphertext = &data[12..];
328
329 tracing::debug!(
330 "[SqliteTokenStore] Extracted nonce (12 bytes) and ciphertext ({} bytes)",
331 ciphertext.len()
332 );
333
334 let backup_key = Self::derive_backup_key(password);
335 let cipher = Aes256Gcm::new_from_slice(&backup_key)
336 .map_err(|e| DomainError::InternalError(format!("Failed to create cipher: {}", e)))?;
337
338 let nonce = Nonce::from(nonce_bytes);
339
340 tracing::debug!("[SqliteTokenStore] Attempting to decrypt token blob...");
341 let plaintext = cipher.decrypt(&nonce, ciphertext).map_err(|_| {
342 DomainError::AuthenticationFailed("Decryption failed (wrong password?)".into())
343 })?;
344 tracing::debug!(
345 "[SqliteTokenStore] Decryption successful ({} bytes)",
346 plaintext.len()
347 );
348
349 tracing::debug!("[SqliteTokenStore] Parsing JSON tokens...");
350 let tokens: HashMap<i64, String> = serde_json::from_slice(&plaintext)
351 .map_err(|e| DomainError::DatabaseError(format!("Failed to parse tokens: {}", e)))?;
352
353 tracing::info!(
354 "[SqliteTokenStore] Parsed {} tokens from encrypted blob",
355 tokens.len()
356 );
357
358 let mut stored_count = 0;
359 for (provider_id, token) in tokens {
360 tracing::debug!(
361 "[SqliteTokenStore] Importing token for provider {} (length: {})",
362 provider_id,
363 token.len()
364 );
365 self.store_token(provider_id, &token).await?;
366 stored_count += 1;
367 }
368
369 tracing::info!(
370 "[SqliteTokenStore] Import completed: {} tokens stored",
371 stored_count
372 );
373
374 Ok(())
375 }
376
377 async fn warmup(&self) -> DomainResult<()> {
378 tracing::info!("SQLite token store warmup complete (instant)");
379 Ok(())
380 }
381}
382
383#[cfg(test)]
384mod tests {
385 use super::*;
386
387 async fn create_test_pool() -> Pool<Sqlite> {
388 use sqlx::sqlite::SqlitePoolOptions;
389
390 let pool = SqlitePoolOptions::new()
391 .max_connections(1)
392 .connect(":memory:")
393 .await
394 .expect("Failed to create in-memory pool");
395
396 sqlx::query(
397 "CREATE TABLE providers (
398 id INTEGER PRIMARY KEY AUTOINCREMENT,
399 name TEXT NOT NULL UNIQUE,
400 provider_type TEXT NOT NULL,
401 token_encrypted TEXT NOT NULL DEFAULT '',
402 config_json TEXT NOT NULL DEFAULT '{}',
403 encrypted_token BLOB,
404 token_nonce BLOB,
405 created_at TEXT NOT NULL DEFAULT (datetime('now')),
406 updated_at TEXT NOT NULL DEFAULT (datetime('now'))
407 )",
408 )
409 .execute(&pool)
410 .await
411 .expect("Failed to create table");
412
413 sqlx::query(
414 "INSERT INTO providers (name, provider_type) VALUES ('test-provider', 'github')",
415 )
416 .execute(&pool)
417 .await
418 .expect("Failed to insert provider");
419
420 pool
421 }
422
423 #[tokio::test]
424 async fn test_sqlite_basic() {
425 let pool = create_test_pool().await;
426 let store = SqliteTokenStore::new(pool, Some("test-password".to_string()))
427 .await
428 .unwrap();
429
430 store.store_token(1, "test-token").await.unwrap();
431 let token = store.get_token(1).await.unwrap();
432 assert_eq!(token, "test-token");
433 }
434
435 #[tokio::test]
436 async fn test_sqlite_cache() {
437 let pool = create_test_pool().await;
438 let store = SqliteTokenStore::new(pool, Some("test-password".to_string()))
439 .await
440 .unwrap();
441
442 store.store_token(1, "cached-token").await.unwrap();
443
444 let token = store.get_token(1).await.unwrap();
445 assert_eq!(token, "cached-token");
446 }
447
448 #[tokio::test]
449 async fn test_sqlite_delete() {
450 let pool = create_test_pool().await;
451 let store = SqliteTokenStore::new(pool, Some("test-password".to_string()))
452 .await
453 .unwrap();
454
455 store.store_token(1, "to-delete").await.unwrap();
456 store.delete_token(1).await.unwrap();
457
458 assert!(store.get_token(1).await.is_err());
459 }
460
461 #[tokio::test]
462 async fn test_sqlite_export_import() {
463 let pool = create_test_pool().await;
464 let store = SqliteTokenStore::new(pool.clone(), Some("test-password".to_string()))
465 .await
466 .unwrap();
467
468 store.store_token(1, "export-token").await.unwrap();
469
470 let encrypted = store.export_encrypted("backup-password").await.unwrap();
471
472 store.delete_token(1).await.unwrap();
473
474 store
475 .import_encrypted(&encrypted, "backup-password")
476 .await
477 .unwrap();
478
479 let token = store.get_token(1).await.unwrap();
480 assert_eq!(token, "export-token");
481 }
482
483 #[tokio::test]
484 async fn test_sqlite_wrong_backup_password() {
485 let pool = create_test_pool().await;
486 let store = SqliteTokenStore::new(pool, Some("test-password".to_string()))
487 .await
488 .unwrap();
489
490 store.store_token(1, "secret").await.unwrap();
491 let encrypted = store.export_encrypted("correct-password").await.unwrap();
492
493 let result = store.import_encrypted(&encrypted, "wrong-password").await;
494 assert!(result.is_err());
495 }
496}