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