Coverage for src/local_deep_research/database/backup/backup_service.py: 90%
234 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""Core backup service for encrypted database backups.
3Uses sqlcipher_export() for safe atomic backups that preserve encryption
4and work correctly with WAL mode.
5"""
7import os
8import shutil
9import threading
10import time
11from dataclasses import dataclass
12from datetime import UTC, datetime, timedelta
13from pathlib import Path
14from typing import Optional
16from loguru import logger
18from ...utilities.resource_utils import safe_close
20from ...config.paths import (
21 get_encrypted_database_path,
22 get_user_backup_directory,
23 get_user_database_filename,
24)
25from ..sqlcipher_utils import (
26 apply_sqlcipher_pragmas,
27 create_sqlcipher_connection,
28 get_key_from_password,
29 get_sqlcipher_settings,
30 set_sqlcipher_key,
31 verify_sqlcipher_connection,
32)
34# Module-level per-user locks to prevent concurrent backup operations for the
35# same user across different BackupService instances. Entries deliberately
36# retain stable identity for the process lifetime: removing a lock between a
37# caller's lookup and acquire would let a later caller create a second lock and
38# bypass the serialization.
39_user_locks: dict[str, threading.Lock] = {}
40_user_locks_lock = threading.Lock()
42# Characters that must never appear in a backup path. ATTACH DATABASE takes a
43# string literal and cannot be parameterized in SQLite/SQLCipher, so the path is
44# interpolated into SQL. A single quote IS allowed — it is escaped by doubling
45# (per the SQLite literal grammar) so an apostrophe in the data-dir path (a
46# home dir named O'Brien, say) doesn't break backups. Backslash, double-quote,
47# NUL and control chars are still rejected (never valid in a real backup path).
48_UNSAFE_BACKUP_PATH_CHARS = frozenset('"\\\0\n\r\t')
51def _get_user_lock(username: str) -> threading.Lock:
52 """Get or create a lock for a specific user.
54 Thread-safe lazy initialization of per-user locks.
56 Args:
57 username: The username to get lock for
59 Returns:
60 A threading.Lock for the specified user
61 """
62 with _user_locks_lock:
63 if username not in _user_locks:
64 _user_locks[username] = threading.Lock()
65 return _user_locks[username]
68def is_safe_glob_result(path: Path, base_dir: Path) -> bool:
69 """Validate that a glob result is safe to use.
71 Rejects symlinks and paths that resolve outside the expected
72 base directory, preventing path-traversal via crafted filenames
73 or symlink attacks.
75 This is a best-effort pre-check, not atomic with the subsequent
76 stat()/unlink()/open() on the path (a classic TOCTOU window). It is
77 acceptable here because the backup directory is per-user and
78 server-owned with 0o600 backups — an attacker who could swap a file
79 for a symlink already has write access to that directory.
81 Args:
82 path: The path returned by glob to validate.
83 base_dir: The directory the path must reside within.
85 Returns:
86 True if the path is a regular (non-symlink) file that resolves
87 to a location within base_dir.
88 """
89 return not path.is_symlink() and path.resolve().is_relative_to(
90 base_dir.resolve()
91 )
94def pop_user_lock(username: str) -> None:
95 """Compatibility no-op: per-user backup locks keep stable identity.
97 User-close cleanup can race both a held lock and a caller that has looked
98 up the lock but not acquired it yet. Removing the registry entry in either
99 case permits a later backup operation to create a second lock and enter the
100 same user's backup critical section concurrently.
102 Keep one small lock per distinct username for the process lifetime. The
103 function remains as a no-op for existing user-close cleanup callers.
104 """
105 del username
108@dataclass
109class BackupResult:
110 """Result of a backup operation."""
112 success: bool
113 backup_path: Optional[Path] = None
114 error: Optional[str] = None
115 size_bytes: int = 0
118class BackupService:
119 """Service for creating and managing encrypted database backups.
121 Uses sqlcipher_export() for safe backups that:
122 - Work correctly with WAL mode
123 - Preserve encryption with the same key
124 - Create atomic copies via ATTACH + export + DETACH
125 - Never corrupt the source database
126 """
128 def __init__(
129 self,
130 username: str,
131 password: str,
132 max_backups: int = 1,
133 max_age_days: int = 7,
134 ):
135 """Initialize backup service.
137 Args:
138 username: User's username
139 password: User's password (for encryption)
140 max_backups: Maximum number of backup files to keep
141 max_age_days: Delete backups older than this many days
142 """
143 self.username = username
144 self.password = password
145 self.max_backups = max_backups
146 self.max_age_days = max_age_days
148 # Get paths
149 self.db_filename = get_user_database_filename(username)
150 self.db_path = get_encrypted_database_path() / self.db_filename
151 self.backup_dir = get_user_backup_directory(username)
153 def create_backup(self, force: bool = False) -> BackupResult:
154 """Create an encrypted backup of the user's database.
156 Uses sqlcipher_export() to create a safe, atomic backup that inherits
157 the encryption key from the source database. The backup is created
158 with a .tmp suffix and atomically renamed to prevent race conditions
159 with cleanup operations.
161 By default, only one backup per calendar day is created to prevent
162 a corrupted database from rapidly overwriting all good backups.
163 Use force=True to bypass this check (used by pre-migration backups).
165 This method is protected by a per-user lock to prevent concurrent
166 backup operations for the same user.
168 Args:
169 force: If True, skip the daily limit check.
171 Returns:
172 BackupResult with success status and backup path
173 """
174 # Acquire per-user lock to prevent concurrent backup operations
175 with _get_user_lock(self.username):
176 # Skip if a backup already exists for today (unless forced)
177 if not force:
178 today = datetime.now(UTC).strftime("%Y%m%d")
179 # After globbing, verify each result is a direct child of backup_dir
180 existing_today = [
181 f
182 for f in self.backup_dir.glob(f"ldr_backup_{today}_*.db")
183 if is_safe_glob_result(f, self.backup_dir)
184 ]
185 if existing_today:
186 latest = max(existing_today, key=lambda p: p.name)
187 logger.debug(
188 f"Backup already exists for today ({latest.name}), "
189 "skipping"
190 )
191 return BackupResult(
192 success=True,
193 backup_path=latest,
194 size_bytes=latest.stat().st_size
195 if latest.exists()
196 else 0,
197 )
199 start = time.perf_counter()
200 result = self._create_backup_impl()
201 elapsed_ms = (time.perf_counter() - start) * 1000
202 size_info = (
203 f"{result.size_bytes / (1024 * 1024):.1f}MB"
204 if result.size_bytes
205 else "unknown size"
206 )
207 if elapsed_ms > 1000:
208 logger.info(
209 f"Backup for user {self.username} "
210 f"({size_info}) took {elapsed_ms:.0f}ms"
211 )
212 else:
213 logger.debug(
214 f"Backup for user {self.username} "
215 f"({size_info}) took {elapsed_ms:.0f}ms"
216 )
217 return result
219 def _create_backup_impl(self) -> BackupResult:
220 """Internal implementation of backup creation (must be called with lock held)."""
221 if not self.db_path.exists():
222 return BackupResult(
223 success=False,
224 error=f"Database not found: {self.db_path}",
225 )
227 # Check available disk space
228 try:
229 db_size = self.db_path.stat().st_size
230 free_space = shutil.disk_usage(self.backup_dir).free
231 # Require at least 2x the database size as free space
232 if free_space < db_size * 2:
233 return BackupResult(
234 success=False,
235 error=f"Insufficient disk space. Need {db_size * 2} bytes, have {free_space}",
236 )
237 except OSError as e:
238 # Fail closed - don't proceed with backup if we can't verify disk space
239 logger.warning("Could not check disk space, skipping backup")
240 return BackupResult(
241 success=False,
242 error=f"Could not verify disk space: {e}",
243 )
245 # Generate backup filename with timestamp
246 # Use .tmp suffix during creation to prevent cleanup race conditions
247 timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
248 backup_filename = f"ldr_backup_{timestamp}.db"
249 backup_path = self.backup_dir / backup_filename
250 temp_path = self.backup_dir / f"ldr_backup_{timestamp}.db.tmp"
252 try:
253 # Create connection to source database
254 conn = create_sqlcipher_connection(str(self.db_path), self.password)
255 cursor = conn.cursor()
257 # Set busy timeout so concurrent writers don't cause instant failure
258 cursor.execute("PRAGMA busy_timeout = 10000")
260 try:
261 # Use sqlcipher_export() to create an encrypted backup
262 # VACUUM INTO doesn't preserve encryption in SQLCipher
263 # Security: the ATTACH path below is interpolated (not
264 # parameterizable). Single quotes are escaped (see below); any
265 # other unsafe char is rejected. Don't log the full path — it can
266 # contain a username — only the offending characters.
267 temp_path_str = str(temp_path)
268 bad_chars = _UNSAFE_BACKUP_PATH_CHARS & set(temp_path_str)
269 if bad_chars:
270 raise ValueError(
271 "Backup path contains characters not allowed in a "
272 "SQLCipher ATTACH statement: "
273 + ", ".join(sorted(repr(c) for c in bad_chars))
274 )
276 # Get the hex key for ATTACH (same key derivation as source)
277 hex_key = get_key_from_password(
278 self.password, db_path=self.db_path
279 ).hex()
281 # Defensive: ensure hex_key is strictly hexadecimal
282 if not hex_key or not all( 282 ↛ 285line 282 didn't jump to line 285 because the condition on line 282 was never true
283 c in "0123456789abcdef" for c in hex_key
284 ):
285 raise ValueError("Derived key is not valid hex")
287 # Attach backup database with encryption (using temp path).
288 # ATTACH DATABASE can't be parameterized in SQLite/SQLCipher, so
289 # the path is interpolated as a string literal — escape any
290 # single quote by doubling it per the SQLite literal grammar
291 # (all other unsafe chars were rejected above).
292 attach_path = temp_path_str.replace("'", "''")
293 cursor.execute(
294 f"ATTACH DATABASE '{attach_path}' AS backup KEY \"x'{hex_key}'\""
295 )
297 try:
298 # Apply cipher settings to the backup database (must match source)
299 # Note: PRAGMA statements do not support parameter binding
300 # in SQLite — f-string is required. Values are validated
301 # upstream by get_sqlcipher_settings() against allow-lists.
302 settings = get_sqlcipher_settings()
303 page_size = int(settings["page_size"])
304 kdf_iter = int(settings["kdf_iterations"])
305 hmac_alg = str(settings["hmac_algorithm"])
306 cursor.execute(
307 f"PRAGMA backup.cipher_page_size = {page_size}"
308 )
309 cursor.execute(
310 f"PRAGMA backup.cipher_hmac_algorithm = {hmac_alg}"
311 )
312 cursor.execute(f"PRAGMA backup.kdf_iter = {kdf_iter}")
314 # Export all data to the backup database
315 cursor.execute("SELECT sqlcipher_export('backup')")
316 finally:
317 # Always detach to release the backup file handle
318 try:
319 cursor.execute("DETACH DATABASE backup")
320 except Exception:
321 logger.warning(
322 "DETACH failed (connection will release on close)"
323 )
324 finally:
325 safe_close(cursor, "backup cursor")
326 safe_close(conn, "backup connection")
328 # Verify the backup is valid (still using temp path)
329 if not self._verify_backup(temp_path):
330 # Delete corrupted backup
331 if temp_path.exists():
332 temp_path.unlink()
333 return BackupResult(
334 success=False,
335 error="Backup verification failed - backup was corrupted",
336 )
338 # Set restrictive permissions (owner read/write only)
339 # SECURITY: Backup files contain sensitive user data
340 os.chmod(temp_path, 0o600)
342 # Get backup size before rename
343 backup_size = temp_path.stat().st_size
345 # Atomic rename from .tmp to final .db
346 # This ensures cleanup won't see/delete partially created backups
347 temp_path.rename(backup_path)
349 logger.info(
350 f"Created backup for user: {backup_path.name} ({backup_size} bytes)"
351 )
353 # Cleanup old backups (safe now - new backup is finalized)
354 self._cleanup_old_backups()
356 return BackupResult(
357 success=True,
358 backup_path=backup_path,
359 size_bytes=backup_size,
360 )
362 except Exception as e:
363 logger.exception("Backup creation failed")
364 # Clean up any partial backup (temp file)
365 if temp_path.exists():
366 try:
367 temp_path.unlink()
368 except OSError:
369 pass
370 # Also clean up final path in case rename partially succeeded
371 if backup_path.exists(): 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 try:
373 backup_path.unlink()
374 except OSError:
375 pass
376 return BackupResult(
377 success=False,
378 error=str(e),
379 )
381 def _verify_backup(self, backup_path: Path) -> bool:
382 """Verify that a backup file is valid and readable.
384 Args:
385 backup_path: Path to the backup file
387 Returns:
388 True if backup is valid, False otherwise
389 """
390 if not backup_path.exists():
391 return False
393 if backup_path.stat().st_size == 0:
394 logger.warning("Backup file is empty (0 bytes)")
395 return False
397 try:
398 # Import SQLCipher module
399 from ..sqlcipher_compat import get_sqlcipher_module
401 sqlcipher3 = get_sqlcipher_module()
403 # Open the backup with the same password
404 conn = sqlcipher3.connect(str(backup_path))
405 cursor = conn.cursor()
407 try:
408 # Set encryption key using the SOURCE database's salt
409 # (backup was encrypted with the source DB's per-database salt)
410 set_sqlcipher_key(cursor, self.password, db_path=self.db_path)
411 apply_sqlcipher_pragmas(cursor, creation_mode=False)
413 # Run quick integrity check
414 cursor.execute("PRAGMA quick_check")
415 result = cursor.fetchone()
417 if result and result[0] == "ok":
418 # Additional verification: try to read a table
419 if verify_sqlcipher_connection(cursor): 419 ↛ 422line 419 didn't jump to line 422 because the condition on line 419 was always true
420 return True
422 logger.warning(f"Backup integrity check failed: {result}")
423 return False
425 finally:
426 safe_close(cursor, "backup cursor")
427 safe_close(conn, "backup connection")
429 except Exception:
430 logger.warning("Backup verification failed")
431 return False
433 def _cleanup_old_backups(self) -> int:
434 """Remove old backups based on age and count limits.
436 Also cleans up stale .tmp files from interrupted backups.
438 Returns:
439 Number of backups deleted
440 """
441 deleted_count = 0
442 cutoff_time = datetime.now(UTC) - timedelta(days=self.max_age_days)
443 stale_tmp_cutoff = datetime.now(UTC) - timedelta(hours=1)
445 try:
446 # Clean up stale .tmp files from interrupted/crashed backups
447 for tmp_file in [
448 f
449 for f in self.backup_dir.glob("ldr_backup_*.db.tmp")
450 if is_safe_glob_result(f, self.backup_dir)
451 ]:
452 try:
453 mtime = datetime.fromtimestamp(
454 tmp_file.stat().st_mtime, tz=UTC
455 )
456 if mtime < stale_tmp_cutoff:
457 tmp_file.unlink()
458 logger.info(
459 f"Cleaned up stale temp file: {tmp_file.name}"
460 )
461 except (OSError, FileNotFoundError):
462 pass
464 # Get all backup files sorted by modification time (newest first)
465 def _safe_mtime(p: Path) -> float:
466 try:
467 return p.stat().st_mtime
468 except FileNotFoundError:
469 return 0.0
471 # After globbing, verify each result is a direct child of backup_dir
472 backups = [
473 p
474 for p in sorted(
475 self.backup_dir.glob("ldr_backup_*.db"),
476 key=_safe_mtime,
477 reverse=True,
478 )
479 if p.exists() and is_safe_glob_result(p, self.backup_dir)
480 ]
482 for i, backup in enumerate(backups):
483 should_delete = False
485 # Delete if beyond max count
486 if i >= self.max_backups:
487 should_delete = True
488 reason = f"exceeds max count ({self.max_backups})"
490 # Delete if too old
491 else:
492 try:
493 mtime = datetime.fromtimestamp(
494 backup.stat().st_mtime, tz=UTC
495 )
496 if mtime < cutoff_time:
497 should_delete = True
498 reason = f"older than {self.max_age_days} days"
499 except FileNotFoundError:
500 continue
502 if should_delete:
503 try:
504 backup.unlink()
505 deleted_count += 1
506 logger.debug(
507 f"Deleted old backup {backup.name}: {reason}"
508 )
509 except OSError:
510 logger.warning(f"Could not delete backup {backup.name}")
512 except Exception:
513 logger.exception("Error during backup cleanup")
515 if deleted_count > 0:
516 logger.info(f"Cleaned up {deleted_count} old backups")
518 return deleted_count
520 def list_backups(self) -> list[dict]:
521 """List all backups for this user.
523 Returns:
524 List of backup info dictionaries with path, size, and timestamp
525 """
526 backups = []
528 try:
530 def _safe_mtime_list(p: Path) -> float:
531 try:
532 return p.stat().st_mtime
533 except FileNotFoundError:
534 return 0.0
536 # After globbing, verify each result is a direct child of backup_dir
537 for backup_file in sorted(
538 [
539 f
540 for f in self.backup_dir.glob("ldr_backup_*.db")
541 if is_safe_glob_result(f, self.backup_dir)
542 ],
543 key=_safe_mtime_list,
544 reverse=True,
545 ):
546 try:
547 stat = backup_file.stat()
548 except FileNotFoundError:
549 continue
550 backups.append(
551 {
552 "filename": backup_file.name,
553 "path": backup_file.name, # Only expose filename, not full server path
554 "size_bytes": stat.st_size,
555 "created_at": datetime.fromtimestamp(
556 stat.st_mtime, tz=UTC
557 ).isoformat(),
558 }
559 )
560 except Exception:
561 logger.exception("Error listing backups")
563 return backups
565 def purge_and_refresh(self) -> "BackupResult":
566 """Delete all existing backups and create a fresh one.
568 Used after a password change to replace old-key backups with a
569 new backup encrypted under the current password. Old backups
570 encrypted with a previous password are a security risk (NIST
571 SP 800-57, OWASP A02) because they remain decryptable with the
572 old (potentially compromised) password.
574 Returns:
575 BackupResult from the fresh backup creation
576 """
577 # Hold per-user lock for the entire purge+create operation to
578 # prevent a concurrent backup from writing an old-key backup
579 # between the purge and the fresh backup creation.
580 with _get_user_lock(self.username):
581 # Delete all existing backup files
582 for info in self.list_backups():
583 try:
584 (self.backup_dir / info["path"]).unlink()
585 logger.debug(f"Purged old-key backup: {info['filename']}")
586 except OSError:
587 logger.warning(
588 f"Could not delete backup {info['filename']}"
589 )
591 # Also clean up any stale .tmp files
592 for tmp_file in [
593 f
594 for f in self.backup_dir.glob("ldr_backup_*.db.tmp")
595 if is_safe_glob_result(f, self.backup_dir)
596 ]:
597 try:
598 tmp_file.unlink()
599 except OSError:
600 logger.warning(
601 f"Could not delete stale tmp file {tmp_file.name}"
602 )
604 # Create fresh backup with current password (lock already held)
605 return self._create_backup_impl()
607 def get_latest_backup(self) -> Optional[Path]:
608 """Get the path to the most recent backup.
610 Returns:
611 Path to latest backup, or None if no backups exist
612 """
613 try:
615 def _safe_mtime_latest(p: Path) -> float:
616 try:
617 return p.stat().st_mtime
618 except FileNotFoundError:
619 return 0.0
621 # After globbing, verify each result is a direct child of backup_dir
622 backups = [
623 p
624 for p in sorted(
625 self.backup_dir.glob("ldr_backup_*.db"),
626 key=_safe_mtime_latest,
627 reverse=True,
628 )
629 if p.exists() and is_safe_glob_result(p, self.backup_dir)
630 ]
631 return backups[0] if backups else None
632 except Exception:
633 logger.exception("Error finding latest backup")
634 return None