Coverage for src/local_deep_research/database/encrypted_db.py: 92%
406 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Encrypted database management using SQLCipher.
3Handles per-user encrypted databases with browser-friendly authentication.
4"""
6import os
7import threading
8import time
9from pathlib import Path
10from typing import Any, Dict, Optional
12from loguru import logger
13from sqlalchemy import create_engine, event, text
14from sqlalchemy.engine import Engine
15from sqlalchemy.orm import Session, sessionmaker
16from sqlalchemy.pool import QueuePool, StaticPool
18from ..config.paths import get_data_directory, get_user_database_filename
19from ..settings.env_registry import get_env_setting
21# ``redact_secrets`` is imported lazily inside each except handler that
22# uses it. A top-level import would trip a circular-import chain:
23#
24# security/__init__.py
25# → security/file_integrity/integrity_manager.py
26# → database/session_context.py
27# → database/encrypted_db.py (mid-load, ``db_manager`` not yet
28# defined → ImportError that
29# integrity_manager catches and
30# sets ``_has_session_context`` to
31# False for the rest of the run)
32#
33# This file sits on the import path that ``security/__init__.py``
34# re-enters via the file_integrity submodule. Other files in
35# PRs #4168/#4175/#4181 can import ``redact_secrets`` at the top
36# because they are not on that path.
37from .sqlcipher_compat import get_sqlcipher_module
38from .pool_config import POOL_PRE_PING, POOL_RECYCLE_SECONDS
39from .sqlcipher_utils import (
40 set_sqlcipher_key,
41 set_sqlcipher_rekey,
42 apply_cipher_defaults_before_key,
43 apply_sqlcipher_pragmas,
44 apply_performance_pragmas,
45 verify_sqlcipher_connection,
46 create_database_salt,
47 get_salt_file_path,
48 has_per_database_salt,
49 get_key_from_password,
50 get_sqlcipher_version,
51 create_sqlcipher_connection,
52)
55class DatabaseInitializationError(Exception):
56 """Raised when a per-user database opens but its schema can't be initialised.
58 Distinct from credential / decryption failures (which return ``None``
59 from :py:meth:`DatabaseManager.open_user_database`) so callers — chiefly
60 the login route — can avoid penalising the user's lockout counter and
61 surface a different error message. The credentials are valid; the
62 database state isn't.
63 """
66def _best_effort_chmod(path, mode: int, *, warn: bool = False) -> None:
67 """Tighten permissions on ``path`` to ``mode``, never raising.
69 Permission hardening must NOT be able to break database creation. On
70 filesystems that don't support POSIX chmod — some Docker bind mounts and
71 network/FUSE volumes, notably Docker Desktop on macOS/Windows — os.chmod
72 can raise OSError even for the file's owner. In that case we leave the
73 path at its default (umask) mode rather than failing registration/login.
75 ``warn=True`` surfaces the failure at warning level: a silent downgrade
76 leaves the path group/world-readable, which matters for DB files holding
77 user data (the unencrypted fallback is plaintext). The owner-only
78 directory chmod is defense-in-depth, so it stays at debug.
79 """
80 try:
81 os.chmod(str(path), mode)
82 except OSError:
83 log = logger.warning if warn else logger.debug
84 log(
85 f"Could not set permissions {oct(mode)} on {path}; "
86 "left at filesystem default",
87 exc_info=True,
88 )
91def _remove_partial_user_db_files(db_path: Path) -> None:
92 """Remove a half-created user database and all of its sidecar files.
94 Called from ``create_user_database``'s failure paths. The encrypted
95 branch writes a per-database ``.salt`` file (via ``create_database_salt``)
96 *before* the DB itself, and SQLCipher/WAL leaves ``-wal``/``-shm``
97 sidecars. Removing only the ``.db`` — as these cleanup paths used to —
98 orphans the ``.salt``: a retry then hits ``create_database_salt``'s
99 ``FileExistsError`` (it refuses to overwrite to prevent data loss), so the
100 username stays permanently un-registerable even after the orphaned auth
101 row is cleaned up. Delete every artifact so the on-disk state is atomic
102 and the username is reusable on a subsequent retry.
104 ``create_user_database`` calls this from four sites: its three
105 failure-cleanup blocks (structure creation, engine build, migration — a
106 graceful error mid-creation) and its salt-creation recovery path (an
107 orphan left by a prior hard-killed create or by pre-#4934 cleanup that
108 removed only the ``.db``). In every case the ``db_path.exists()`` guard at
109 the top of ``create_user_database`` already ran and found no *pre-existing*
110 database, so any ``.db``/``.salt`` present now is this call's own partial
111 work — removing it cannot destroy live data, because a valid encrypted DB
112 always has both the ``.db`` and its ``.salt``.
114 Both the ``-wal``/``-shm`` (WAL mode, the default) and ``-journal``
115 (rollback-journal modes: DELETE/TRUNCATE/PERSIST) sidecars are removed so
116 the cleanup is correct regardless of the configured ``journal_mode``.
118 Never raises — cleanup must not mask the original creation error.
119 """
120 # Derive every artifact path up front, guarded: the "Never raises"
121 # contract has to hold even if a path helper is handed something odd
122 # (e.g. a nameless path, where ``with_name``/``with_suffix`` raise), so a
123 # derivation failure is logged and swallowed rather than propagating and
124 # masking the original creation error.
125 try:
126 db_path = Path(db_path)
127 artifacts = (
128 db_path,
129 get_salt_file_path(db_path),
130 db_path.with_name(db_path.name + "-wal"),
131 db_path.with_name(db_path.name + "-shm"),
132 db_path.with_name(db_path.name + "-journal"),
133 )
134 except Exception:
135 logger.warning(
136 f"Could not derive partial user DB artifact paths: {db_path!r}"
137 )
138 return
140 for path in artifacts:
141 try:
142 path.unlink(missing_ok=True)
143 except OSError:
144 # Best-effort: a leftover artifact is worth surfacing but must not
145 # mask the original creation error. The path is enough to
146 # diagnose; skip the exception detail per the sensitive-logging
147 # policy.
148 logger.warning(f"Could not remove partial user DB artifact {path}")
151class DatabaseManager:
152 """Manages encrypted SQLCipher databases for each user."""
154 def __init__(self):
155 self.connections: Dict[str, Engine] = {}
156 self._connections_lock = threading.RLock()
157 # Per-user locks serializing the cold-open (engine build + migration)
158 # so two concurrent first-opens of one user never run alembic against
159 # the same database file at once. Created lazily under
160 # _connections_lock; see open_user_database / _get_init_lock.
161 self._init_locks: Dict[str, threading.Lock] = {}
162 self.data_dir = get_data_directory() / "encrypted_databases"
163 self.data_dir.mkdir(parents=True, exist_ok=True)
164 # Restrict the per-user DB directory to the owner (0o700). Belt-and-
165 # suspenders with the per-file 0o600 chmod: it also covers SQLite's
166 # WAL/SHM sidecars and any temp files (which the unencrypted fallback
167 # writes in PLAINTEXT and which SQLite may recreate at umask perms on
168 # each connect), so no sibling-readable user data is exposed to other
169 # local accounts regardless of per-file modes.
170 _best_effort_chmod(self.data_dir, 0o700)
172 # Check SQLCipher availability
173 self.has_encryption = self._check_encryption_available()
175 # ----------------------------------------------------------------
176 # Pool class selection — see ADR-0004
177 #
178 # We use QueuePool (pool_size=20, max_overflow=40,
179 # pool_timeout=10) for production and StaticPool for tests.
180 #
181 # Why pool_size=20:
182 #
183 # 1. SQLCipher + WAL mode can leak file handles when connections
184 # close out of open-order. Fewer pooled connections = fewer
185 # opportunities for out-of-order closes during pool_recycle.
186 # See: https://github.com/sqlcipher/android-database-sqlcipher/issues/6
187 # See: https://github.com/dotnet/efcore/issues/35010
188 #
189 # 2. SQLite serializes all writes through a single file lock.
190 # Multiple pooled connections don't improve throughput — they
191 # just hold FDs (up to 3 per connection in WAL mode).
192 #
193 # 3. The cleanup scheduler periodically calls engine.dispose()
194 # to release all pooled connections, preventing long-lived
195 # handles from accumulating over days of idle operation.
196 #
197 # Why pool_size=20 and not 1: inject_current_user() creates a
198 # QueuePool session on every request via g.db_session. With the
199 # UI polling /api/research/<id>/status every 1-2s plus other
200 # API calls and before_request middleware, pool_size=1
201 # (max_overflow=2, so 3 total) is easily exhausted — causing
202 # 30-second timeouts and PendingRollbackError cascades.
203 # pool_size=20 + max_overflow=40 (60 total) provides ample
204 # headroom for concurrent requests and multiple browser tabs.
205 #
206 # Why not NullPool: SQLCipher's PRAGMA key adds ~0.2ms per
207 # connection open. With 20-30 queries per page load, NullPool
208 # adds a noticeable 4-6ms overhead vs QueuePool's ~1.5ms.
209 # ----------------------------------------------------------------
210 self._use_static_pool = bool(os.environ.get("TESTING"))
211 self._pool_class = StaticPool if self._use_static_pool else QueuePool
213 def _get_pool_kwargs(self) -> Dict[str, Any]:
214 """Get pool configuration kwargs based on pool type.
216 StaticPool doesn't support pool_size or max_overflow.
217 QueuePool uses moderate sizing to handle concurrent web requests
218 while limiting FD usage. See ADR-0004 for rationale.
219 """
220 if self._use_static_pool:
221 return {}
222 return {
223 "pool_size": 20,
224 "max_overflow": 40,
225 "pool_timeout": 10,
226 "pool_pre_ping": POOL_PRE_PING,
227 "pool_recycle": POOL_RECYCLE_SECONDS,
228 }
230 def _is_valid_encryption_key(self, password: str) -> bool:
231 """
232 Check if the provided password is valid (not None, empty, or whitespace-only).
234 Args:
235 password: The password to check
237 Returns:
238 True if the password is valid, False otherwise
239 """
240 return password is not None and password.strip() != ""
242 def is_user_connected(self, username: str) -> bool:
243 """Check if a user has an active database connection.
245 Thread-safe accessor for external callers.
247 Args:
248 username: The username to check
250 Returns:
251 True if the user has an active connection
252 """
253 with self._connections_lock:
254 return username in self.connections
256 def _check_encryption_available(self) -> bool:
257 """Check if SQLCipher is available for encryption."""
258 try:
259 import os as os_module
260 import tempfile
262 # Test if SQLCipher actually works, not just if it imports
263 with tempfile.NamedTemporaryFile(delete=False) as tmp:
264 tmp_path = tmp.name
266 try:
267 # Try to create a test encrypted database
268 sqlcipher_module = get_sqlcipher_module()
269 sqlcipher = sqlcipher_module.dbapi2
271 conn = sqlcipher.connect(tmp_path)
272 try:
273 cursor = conn.cursor()
274 # Use creation_mode=True since we're creating a new test database
275 apply_cipher_defaults_before_key(cursor)
276 # Use centralized key setting
277 set_sqlcipher_key(cursor, "testpass")
278 # Apply post-key pragmas (kdf_iter for new DB)
279 apply_sqlcipher_pragmas(cursor, creation_mode=True)
280 apply_performance_pragmas(cursor)
282 # Check SQLCipher version
283 version = get_sqlcipher_version(cursor)
284 if version: 284 ↛ 296line 284 didn't jump to line 296 because the condition on line 284 was always true
285 major = (
286 version.split(".")[0]
287 if "." in version
288 else version[0]
289 )
290 if major.isdigit() and int(major) < 4: 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true
291 logger.warning(
292 f"SQLCipher version {version} detected. "
293 "Version 4.x+ is recommended for proper PRAGMA ordering."
294 )
296 cursor.close()
297 # Now use the connection for table operations
298 conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)")
299 conn.execute("INSERT INTO test VALUES (1)")
300 result = conn.execute("SELECT * FROM test").fetchone()
302 if result != (1,): 302 ↛ 303line 302 didn't jump to line 303 because the condition on line 302 was never true
303 raise RuntimeError("SQLCipher encryption test failed")
304 logger.info(
305 "SQLCipher available and working - databases will be encrypted"
306 )
307 return True
308 finally:
309 from ..utilities.resource_utils import safe_close
311 safe_close(conn, "SQLCipher test connection")
312 except Exception:
313 logger.warning("SQLCipher module found but not working")
314 raise ImportError("SQLCipher not functional")
315 finally:
316 # Clean up test file
317 try:
318 os_module.unlink(tmp_path)
319 except OSError as e:
320 logger.debug(
321 f"Failed to clean up temp file {tmp_path}: {e}"
322 )
324 except ImportError:
325 # Check if user has explicitly allowed unencrypted databases.
326 # Registry handles deprecated LDR_ALLOW_UNENCRYPTED fallback automatically.
327 allow_unencrypted = get_env_setting(
328 "bootstrap.allow_unencrypted", False
329 )
331 if not allow_unencrypted:
332 # No ``password`` is in scope here (this runs at
333 # bootstrap, before any user authenticates), and the
334 # caught ``ImportError`` is purely about the SQLCipher
335 # package being missing. Still drop the traceback to
336 # keep this file's logging discipline uniform with the
337 # rest of the #4182 sweep: no exception traceback is
338 # written to the production log from this module. The
339 # install-hint message is the user-facing diagnostic;
340 # the traceback adds no value beyond it. ``warning``
341 # rather than ``error``/``exception`` so the custom
342 # "logger.exception in except blocks" lint passes
343 # without re-introducing a traceback render.
344 logger.warning(
345 "SECURITY ERROR: SQLCipher is not installed!\n"
346 "Your databases will NOT be encrypted.\n"
347 "To fix this:\n"
348 "1. Install SQLCipher: sudo apt install sqlcipher libsqlcipher-dev\n"
349 "2. Reinstall project: pdm install\n"
350 "Or use Docker with SQLCipher pre-installed.\n\n"
351 "To explicitly allow unencrypted databases (NOT RECOMMENDED):\n"
352 "export LDR_BOOTSTRAP_ALLOW_UNENCRYPTED=true"
353 )
354 raise RuntimeError(
355 "SQLCipher not available. Set LDR_BOOTSTRAP_ALLOW_UNENCRYPTED=true to proceed without encryption (NOT RECOMMENDED)"
356 )
357 logger.warning(
358 "WARNING: Running with UNENCRYPTED databases!\n"
359 "This means:\n"
360 "- Passwords don't protect data access\n"
361 "- API keys are stored in plain text\n"
362 "- Anyone with file access can read all data\n"
363 "Install SQLCipher for secure operation!"
364 )
365 return False
367 def _get_user_db_path(self, username: str) -> Path:
368 """Get the path for a user's encrypted database."""
369 return self.data_dir / get_user_database_filename(username)
371 def _apply_pragmas(self, connection, connection_record):
372 """Apply pragmas for optimal performance."""
373 # Check if this is SQLCipher or regular SQLite
374 is_encrypted = self.has_encryption
376 # Use centralized performance pragma application
378 apply_performance_pragmas(connection)
380 # SQLCipher-specific pragmas
381 if is_encrypted: 381 ↛ 382line 381 didn't jump to line 382 because the condition on line 381 was never true
382 from .sqlcipher_utils import get_sqlcipher_settings
384 settings = get_sqlcipher_settings()
385 pragmas = [
386 f"PRAGMA kdf_iter = {settings['kdf_iterations']}",
387 f"PRAGMA cipher_page_size = {settings['page_size']}",
388 ]
389 for pragma in pragmas:
390 try:
391 connection.execute(pragma)
392 except Exception as e:
393 logger.debug(f"Could not apply pragma '{pragma}': {e}")
394 else:
395 # Regular SQLite pragma
396 try:
397 connection.execute(
398 "PRAGMA mmap_size = 268435456"
399 ) # 256MB memory mapping
400 except Exception as e:
401 logger.debug(f"Could not apply mmap_size pragma: {e}")
403 @staticmethod
404 def _make_sqlcipher_connection(
405 db_path: Path,
406 password: str,
407 isolation_level: Optional[str] = "IMMEDIATE",
408 check_same_thread: bool = False,
409 ) -> Any:
410 """Create a properly initialized SQLCipher connection.
412 Follows the canonical SQLCipher initialization order: set key,
413 apply cipher pragmas, verify, then apply performance pragmas.
414 Cipher pragmas (page size, HMAC algorithm, KDF iterations) must
415 be configured before the first query (verification) because that
416 query triggers page decryption with the active cipher settings.
418 Args:
419 db_path: Path to the database file
420 password: The database encryption passphrase
421 isolation_level: SQLite isolation level (``""`` for deferred
422 transactions, ``None`` for autocommit)
423 check_same_thread: SQLite check_same_thread flag
425 Returns:
426 A raw ``sqlcipher3`` connection ready for use.
428 Raises:
429 ValueError: If the database key cannot be verified.
430 """
431 sqlcipher3 = get_sqlcipher_module()
432 conn = sqlcipher3.connect(
433 str(db_path),
434 isolation_level=isolation_level,
435 check_same_thread=check_same_thread,
436 )
437 cursor = conn.cursor()
439 try:
440 set_sqlcipher_key(cursor, password, db_path=db_path)
441 apply_sqlcipher_pragmas(cursor, creation_mode=False)
443 if not verify_sqlcipher_connection(cursor):
444 raise ValueError("Failed to verify database key") # noqa: TRY301 — cleanup in except before re-raise
446 apply_performance_pragmas(cursor)
447 except Exception:
448 try:
449 cursor.close()
450 except Exception: # noqa: BLE001
451 logger.warning("Failed to close cursor during cleanup")
452 from ..utilities.resource_utils import safe_close
454 safe_close(conn, "encrypted DB connection")
455 raise
457 cursor.close()
458 return conn
460 def create_user_database(self, username: str, password: str) -> Engine:
461 """Create a new encrypted database for a user."""
463 # Validate the encryption key
464 if not self._is_valid_encryption_key(password):
465 logger.error(
466 f"Invalid encryption key for user {username}: password is None or empty"
467 )
468 raise ValueError(
469 "Invalid encryption key: password cannot be None or empty"
470 )
472 db_path = self._get_user_db_path(username)
474 if db_path.exists():
475 raise ValueError(f"Database already exists for user {username}")
477 # Create connection string - use regular SQLite when SQLCipher not available
478 if self.has_encryption:
479 # Create directory if it doesn't exist
480 db_path.parent.mkdir(parents=True, exist_ok=True)
482 # Create per-database salt for new databases (v2 security improvement).
483 #
484 # A pre-existing salt here is an ORPHAN, not live data: the
485 # ``db_path.exists()`` guard above already established that no
486 # database file exists, and a valid encrypted DB always has BOTH
487 # the ``.db`` and its ``.salt``. Such an orphan is left by a create
488 # that died before completing — a hard process death (SIGKILL/OOM/
489 # power loss) mid-creation, or the pre-#4934 cleanup code that
490 # removed only the ``.db``. Without recovery, ``create_database_salt``
491 # raises ``FileExistsError`` (it refuses to overwrite to protect
492 # live data) on every retry, so the username stays permanently
493 # un-registerable. Remove the stale artifacts and create fresh —
494 # safe precisely because no database depends on this salt.
495 #
496 # Note: backups are keyed off the *source* DB's salt (they store no
497 # salt copy — see backup_service), so deleting a salt also strands
498 # any backups for it. That is not reachable here: this only fires
499 # when db_path is absent, and the app never removes a live .db while
500 # keeping its .salt (there is no in-app restore). Only out-of-band
501 # manual deletion of the .db could turn this into backup loss.
502 #
503 # SAFETY INVARIANT: this deletion is safe only while a salt-without-
504 # a-.db always means garbage. That holds today because create_user_
505 # database is the sole salt writer/caller and nothing restores or
506 # imports a .db from a salt. If an in-app restore/import that stages
507 # a .salt ahead of its .db is ever added, RE-EVALUATE this branch —
508 # it could delete a salt whose database is legitimately incoming.
509 try:
510 create_database_salt(db_path)
511 except FileExistsError:
512 logger.warning(
513 f"Removing orphaned salt (no matching database) for "
514 f"{username} before creating a fresh one"
515 )
516 _remove_partial_user_db_files(db_path)
517 create_database_salt(db_path)
518 logger.info(f"Created per-database salt for {username}")
520 # Create database structure using raw SQLCipher outside SQLAlchemy
521 try:
522 # Pre-derive key before closures to avoid capturing plaintext
523 # password. Kept inside the try so a derivation failure here
524 # also triggers the partial-file cleanup below, rather than
525 # orphaning the just-created salt file.
526 hex_key = get_key_from_password(password, db_path=db_path).hex()
528 conn = create_sqlcipher_connection(
529 db_path,
530 password=password,
531 creation_mode=True,
532 connect_kwargs={
533 # DEFERRED (empty string) so pure-SELECT transactions
534 # acquire only SQLite's SHARED lock, letting WAL-mode
535 # concurrent readers proceed while a writer is active.
536 # IMMEDIATE was previously set "defensively" and made
537 # every transaction (even reads) take a RESERVED lock,
538 # which was the single biggest contention source on
539 # the login-hang path. Race-prone check-then-insert
540 # call sites were made race-free at the application
541 # layer in the preceding prerequisite PR.
542 "isolation_level": "",
543 "check_same_thread": False,
544 },
545 )
546 _best_effort_chmod(db_path, 0o600, warn=True)
547 try:
548 # Get the CREATE TABLE statements from SQLAlchemy models
549 from sqlalchemy.dialects import sqlite
550 from sqlalchemy.schema import CreateIndex, CreateTable
552 from .models import Base
554 # Indexes must be emitted explicitly — SQLAlchemy compiles
555 # `index=True`/`unique=True` and `Index(...)` to separate
556 # CREATE [UNIQUE] INDEX statements, not inline.
557 sqlite_dialect = sqlite.dialect()
558 for table in Base.metadata.sorted_tables:
559 if table.name == "users":
560 continue
561 create_sql = str(
562 CreateTable(table, if_not_exists=True).compile(
563 dialect=sqlite_dialect
564 )
565 )
566 logger.debug(f"Creating table {table.name}")
567 conn.execute(create_sql)
568 for index in table.indexes:
569 index_sql = str(
570 CreateIndex(index, if_not_exists=True).compile(
571 dialect=sqlite_dialect
572 )
573 )
574 conn.execute(index_sql)
576 conn.commit()
577 finally:
578 from ..utilities.resource_utils import safe_close
580 safe_close(conn, "user DB setup connection")
582 logger.info(
583 f"Database structure created successfully for {username}"
584 )
586 except Exception as e:
587 # ``password`` is in lexical scope (function parameter) and
588 # is passed into ``create_sqlcipher_connection`` /
589 # ``set_sqlcipher_key`` above. A traceback rendered with
590 # loguru ``diagnose=True`` would dump frame locals — which
591 # includes the plaintext SQLCipher master password. Use
592 # ``logger.warning`` to drop the traceback chain and
593 # redact the password from str(e) for defense in depth.
594 # SQLCipher master passwords are unrecoverable (see
595 # TRUST.md §5) so the impact of a leak is permanent.
596 from ..security.log_sanitizer import redact_secrets
598 safe_msg = redact_secrets(str(e), password)
599 logger.warning(f"Error creating database structure: {safe_msg}")
600 # Remove the partial DB *and* its salt/WAL/journal sidecars so
601 # a retry of the same username isn't blocked by a leftover salt.
602 _remove_partial_user_db_files(db_path)
603 raise
605 # Small delay to ensure file is fully written
606 import time
608 time.sleep(0.1)
610 # Now create SQLAlchemy engine using custom connection creator
611 def create_engine_connection():
612 """Create a properly initialized SQLCipher connection."""
613 return create_sqlcipher_connection(
614 db_path,
615 hex_key=hex_key,
616 creation_mode=False,
617 connect_kwargs={
618 # DEFERRED (empty string) so pure-SELECT transactions
619 # acquire only SQLite's SHARED lock, letting WAL-mode
620 # concurrent readers proceed while a writer is active.
621 # IMMEDIATE was previously set "defensively" and made
622 # every transaction (even reads) take a RESERVED lock,
623 # which was the single biggest contention source on
624 # the login-hang path. Race-prone check-then-insert
625 # call sites were made race-free at the application
626 # layer in the preceding prerequisite PR.
627 "isolation_level": "",
628 "check_same_thread": False,
629 },
630 )
632 # Create engine with custom creator function and optimized cache.
633 # The .db and .salt already exist on disk (structure creation
634 # succeeded above). This step sits between the structure-creation
635 # and migration cleanup blocks; without its own cleanup a failure
636 # here would orphan those files, and unlike every other failure
637 # path it would NOT be self-healing — the retry trips the
638 # db_path.exists() guard and bricks the username. So clean up on
639 # any escape. (create_engine is lazy so this is near-unreachable,
640 # but it closes the one gap in the otherwise-uniform coverage.)
641 try:
642 engine = create_engine(
643 "sqlite://",
644 creator=create_engine_connection,
645 poolclass=self._pool_class,
646 echo=False,
647 query_cache_size=1000,
648 **self._get_pool_kwargs(),
649 )
650 except Exception:
651 _remove_partial_user_db_files(db_path)
652 raise
653 else:
654 logger.warning(
655 f"SQLCipher not available - creating UNENCRYPTED database for user {username}"
656 )
657 # Fall back to regular SQLite with query cache
658 engine = create_engine(
659 f"sqlite:///{db_path}",
660 connect_args={"check_same_thread": False, "timeout": 30},
661 poolclass=self._pool_class,
662 echo=False,
663 query_cache_size=1000,
664 **self._get_pool_kwargs(),
665 )
667 # For unencrypted databases, just apply pragmas
668 event.listen(engine, "connect", self._apply_pragmas)
670 # Tables have already been created using raw SQLCipher above
671 # No need to create them again with SQLAlchemy
673 # Initialize database tables using centralized initialization
674 from .initialize import initialize_database
676 # Mirror of the fail-loud change #3635 made to open_user_database.
677 # Previously this swallowed the exception with "tables exist but
678 # schema version not stamped — migrations will be retried on next
679 # process restart". That left a half-broken DB on disk: tables
680 # present, no alembic_version row. The next login then re-ran
681 # alembic, hit the same error, and (post-#3635) 503'd — so the
682 # user could register but never log in again. Better to fail
683 # registration loudly with the partial DB removed, so the real
684 # cause (e.g. world-writable migrations dir) gets fixed instead
685 # of producing a permanently-locked-out account.
686 try:
687 Session = sessionmaker(bind=engine)
688 with Session() as session:
689 initialize_database(engine, session)
690 except Exception as e:
691 # ``password`` is in scope and was passed into the engine
692 # creator closure above. Drop the traceback to avoid leaking
693 # frame locals under ``diagnose=True`` and redact the
694 # password from str(e) defensively.
695 from ..security.log_sanitizer import redact_secrets
697 safe_msg = redact_secrets(str(e), password)
698 logger.warning(
699 f"Database migration failed for {username} during creation"
700 f" — removing partial DB: {safe_msg}"
701 )
702 engine.dispose()
703 # Remove the partial DB *and* its salt/WAL/journal sidecars so a
704 # retry of the same username isn't blocked by a leftover salt.
705 _remove_partial_user_db_files(db_path)
706 raise
708 # Restrict DB file to owner-only (0o600). The encrypted branch
709 # chmod's right after create_sqlcipher_connection, but the
710 # unencrypted fallback creates the file lazily on first connect
711 # (during initialize_database above), so it is only guaranteed to
712 # exist here. This file holds PLAINTEXT user data — leaving it at
713 # umask-default perms (commonly 0o644) would expose it to other
714 # local accounts.
715 if not self.has_encryption and db_path.exists():
716 _best_effort_chmod(db_path, 0o600, warn=True)
718 # Store connection AFTER migrations complete
719 with self._connections_lock:
720 self.connections[username] = engine
722 logger.info(f"Created encrypted database for user {username}")
723 return engine
725 def _get_init_lock(self, username: str) -> threading.Lock:
726 """Return the per-user cold-open lock, creating it on first use.
728 Guards the engine-build + migration in ``open_user_database`` so two
729 concurrent first-opens of the same user serialize instead of running
730 migrations against one database file simultaneously. Per-user (not the
731 global ``_connections_lock``) so different users still open in
732 parallel.
733 """
734 with self._connections_lock:
735 lock = self._init_locks.get(username)
736 if lock is None:
737 lock = threading.Lock()
738 self._init_locks[username] = lock
739 return lock
741 def open_user_database(
742 self, username: str, password: str
743 ) -> Optional[Engine]:
744 """Open an existing encrypted database for a user."""
745 open_start = time.perf_counter()
747 # Validate the encryption key
748 if not self._is_valid_encryption_key(password):
749 logger.error(
750 f"Invalid encryption key when opening database for user {username}: password is None or empty"
751 )
752 # TODO: Fix the root cause - research threads are not getting the correct password
753 logger.error(
754 "TODO: This usually means the research thread is not receiving the user's "
755 "password for database encryption. Need to ensure password is passed from "
756 "the main thread to research threads."
757 )
758 raise ValueError(
759 "Invalid encryption key: password cannot be None or empty"
760 )
762 # Check if already open
763 with self._connections_lock:
764 if username in self.connections:
765 return self.connections[username]
767 # Serialize the cold-open (engine build + migration) per user so two
768 # concurrent first-opens never run alembic against the same database
769 # file at once -- alembic's module-level proxy and the version-row
770 # UPDATE are not safe under concurrent migration of one DB.
771 init_lock = self._get_init_lock(username)
772 with init_lock:
773 # Re-check: another thread may have completed the cold-open while
774 # we were waiting for the per-user lock.
775 with self._connections_lock:
776 if username in self.connections:
777 return self.connections[username]
778 engine = self._open_user_database_cold(
779 username, password, open_start
780 )
782 # Phase 2 of the RAG vector-store cutover runs OUTSIDE init_lock (only
783 # on the path that actually did the cold-open — the early returns above
784 # belong to a thread that already ran it). It re-enters
785 # open_user_database() via get_user_db_session on a session-cache miss;
786 # doing it while still holding the non-reentrant init_lock would
787 # self-deadlock if a concurrent close_user_database() evicted the
788 # freshly-cached connection between the cache write and that re-entry.
789 if engine is not None:
790 self._run_phase2_rekey(username, password)
791 return engine
793 def _run_phase2_rekey(self, username: str, password: str) -> None:
794 """Re-key a user's legacy FAISS indices (uuid-keyed -> DocumentChunk.id
795 -keyed), once, at login. See vector_stores/legacy_rekey.py.
797 A fast no-op (one get_setting call) once the per-user marker
798 (research_library.rag_index_rekeyed_v2) is set, or when the user has no
799 legacy sidecars — the common case. Wrapped so a re-key problem never
800 blocks login; it logs its own errors. Must be called AFTER the init_lock
801 is released (see open_user_database) to avoid a re-entrant self-deadlock.
802 """
803 try:
804 from ..vector_stores.legacy_rekey import rekey_user_indexes
806 rekey_user_indexes(username, password)
807 except Exception as e:
808 # NO rendered traceback + redacted message — the password-redaction
809 # invariant (tests/security/test_password_redaction_invariant.py).
810 # `password` (the unrecoverable SQLCipher master key) is live in this
811 # frame and every frame of the rekey chain, so ANY traceback render
812 # leaks it under LDR_LOGURU_DIAGNOSE. logger.opt(exception=True) does
813 # NOT suppress that (it renders identically to logger.exception) and
814 # also evades the AST guardrail, so log a redacted message with no
815 # stack, matching every other handler in this file.
816 from ..security.log_sanitizer import redact_secrets
818 safe_msg = redact_secrets(str(e), password)
819 logger.warning(
820 f"RAG index re-key failed for user {username}: {safe_msg}"
821 )
823 def _open_user_database_cold(
824 self, username: str, password: str, open_start: float
825 ) -> Optional[Engine]:
826 """Build the engine and run migrations for a not-yet-cached user DB.
828 Must be called while holding this user's init lock (acquired in
829 ``open_user_database``) so concurrent first-opens of the same user do
830 not migrate one database file simultaneously.
831 """
832 db_path = self._get_user_db_path(username)
834 # Prevent timing attacks: always derive key before checking file existence
835 # This ensures both existing and non-existent users take the same amount of time,
836 # preventing username enumeration via timing analysis.
837 # Pre-derive key before closures to avoid capturing plaintext password
838 hex_key = get_key_from_password(password, db_path=db_path).hex()
840 if not db_path.exists():
841 logger.error(f"No database found for user {username}")
842 return None
844 # Warn if this is a legacy database without per-database salt
845 if self.has_encryption and not has_per_database_salt(db_path): 845 ↛ 846line 845 didn't jump to line 846 because the condition on line 845 was never true
846 logger.warning(
847 f"Database for user '{username}' uses the legacy shared salt "
848 f"(deprecated). For improved security, consider creating a new "
849 f"account to get a per-database salt. Legacy databases remain "
850 f"fully functional but are less resistant to multi-target attacks."
851 )
853 # Create connection string - use regular SQLite when SQLCipher not available
854 if self.has_encryption:
856 def create_open_connection():
857 """Create a properly initialized SQLCipher connection."""
858 return create_sqlcipher_connection(
859 db_path,
860 hex_key=hex_key,
861 creation_mode=False,
862 connect_kwargs={
863 # DEFERRED (empty string) so pure-SELECT transactions
864 # acquire only SQLite's SHARED lock, letting WAL-mode
865 # concurrent readers proceed while a writer is active.
866 # IMMEDIATE was previously set "defensively" and made
867 # every transaction (even reads) take a RESERVED lock,
868 # which was the single biggest contention source on
869 # the login-hang path. Race-prone check-then-insert
870 # call sites were made race-free at the application
871 # layer in the preceding prerequisite PR.
872 "isolation_level": "",
873 "check_same_thread": False,
874 },
875 )
877 # Create engine with custom creator function and optimized cache
878 engine = create_engine(
879 "sqlite://",
880 creator=create_open_connection,
881 poolclass=self._pool_class,
882 echo=False,
883 query_cache_size=1000,
884 **self._get_pool_kwargs(),
885 )
886 else:
887 logger.warning(
888 f"SQLCipher not available - opening UNENCRYPTED database for user {username}"
889 )
890 # Fall back to regular SQLite (no password protection!)
891 engine = create_engine(
892 f"sqlite:///{db_path}",
893 connect_args={"check_same_thread": False, "timeout": 30},
894 poolclass=self._pool_class,
895 echo=False,
896 query_cache_size=1000,
897 **self._get_pool_kwargs(),
898 )
900 # For unencrypted databases, just apply pragmas
901 event.listen(engine, "connect", self._apply_pragmas)
903 try:
904 # Test connection by running a simple query
905 with engine.connect() as conn:
906 conn.execute(text("SELECT 1"))
908 # Run database initialization (creates missing tables and runs migrations)
909 from .initialize import initialize_database
911 # Create backup before migration to protect against schema change failures
912 from .alembic_runner import needs_migration
914 if needs_migration(engine):
915 try:
916 from .backup.backup_service import BackupService
918 result = BackupService(
919 username=username, password=password
920 ).create_backup(force=True)
921 if result.success: 921 ↛ 922line 921 didn't jump to line 922 because the condition on line 921 was never true
922 logger.info(
923 f"Pre-migration backup created: {result.backup_path}"
924 )
925 else:
926 logger.error(
927 f"Pre-migration backup failed: {result.error}"
928 )
929 except Exception as e:
930 # ``password`` is in scope and was passed into
931 # ``BackupService`` above. Drop traceback + redact
932 # so the upstream error (which can carry the
933 # password through ``set_sqlcipher_key`` frames) is
934 # not written to log sinks.
935 from ..security.log_sanitizer import redact_secrets
937 safe_msg = redact_secrets(str(e), password)
938 logger.warning(
939 f"Pre-migration backup failed — proceeding with migration: {safe_msg}"
940 )
942 # Init failures need to be distinguishable from credential
943 # failures at the call site: the credentials worked (we
944 # decrypted and ran SELECT 1), but the schema couldn't be
945 # brought up. Re-raise as a typed error so the login route
946 # can skip the lockout counter and surface a server-error
947 # message instead of "Invalid username or password".
948 try:
949 initialize_database(engine)
950 except Exception as init_err:
951 elapsed_ms = (time.perf_counter() - open_start) * 1000
952 # ``password`` is in scope (function parameter). The
953 # initializer ran on an engine whose creator captured
954 # ``hex_key`` derived from the password — a SQLAlchemy
955 # ``OperationalError`` traceback could surface frames
956 # holding the plaintext password under ``diagnose=True``.
957 from ..security.log_sanitizer import redact_secrets
959 safe_msg = redact_secrets(str(init_err), password)
960 logger.warning(
961 f"Database migration failed for {username} "
962 f"after {elapsed_ms:.0f}ms — refusing login: {safe_msg}"
963 )
964 engine.dispose()
965 # Sanitised message + ``from None``: ``init_err`` (its
966 # str() and its chained traceback frames) can carry the
967 # plaintext password, and the caller that catches this
968 # typed error logs it with ``logger.exception``
969 # (thread_local_session) — which would render the chain
970 # and defeat the redaction applied just above. Break the
971 # chain per ADR-0003; the redacted detail is already
972 # logged at this site.
973 raise DatabaseInitializationError(
974 f"Database initialisation failed for {username}: {safe_msg}"
975 ) from None
977 # Store connection AFTER migrations complete
978 with self._connections_lock:
979 self.connections[username] = engine
981 # NOTE: the phase-2 RAG index re-key does NOT run here. It re-enters
982 # open_user_database() (via get_user_db_session on a session-cache
983 # miss), and this method runs while holding the non-reentrant
984 # per-user init_lock — so a concurrent close_user_database() evicting
985 # the just-cached connection would make that re-entrant open miss the
986 # fast-path and self-deadlock on the lock this thread already holds.
987 # open_user_database() runs it AFTER releasing the lock instead.
989 elapsed_ms = (time.perf_counter() - open_start) * 1000
990 if elapsed_ms > 100:
991 logger.info(
992 f"Opened encrypted database for user {username} "
993 f"(cold-open wall clock: {elapsed_ms:.0f}ms)"
994 )
995 else:
996 logger.info(
997 f"Opened encrypted database for user {username} "
998 f"({elapsed_ms:.0f}ms)"
999 )
1000 return engine
1002 except DatabaseInitializationError:
1003 # Already logged + engine disposed at the raise site. Re-raise
1004 # past the catch-all below so callers see the typed error.
1005 raise
1006 except Exception as e:
1007 elapsed_ms = (time.perf_counter() - open_start) * 1000
1008 # Catches connection errors during ``engine.connect()`` —
1009 # the connection creator passes the password into
1010 # ``set_sqlcipher_key`` so a failure there can carry the
1011 # password in the traceback's frame locals. Drop the
1012 # traceback and redact str(e).
1013 from ..security.log_sanitizer import redact_secrets
1015 safe_msg = redact_secrets(str(e), password)
1016 logger.warning(
1017 f"Failed to open database for user {username} "
1018 f"after {elapsed_ms:.0f}ms: {safe_msg}"
1019 )
1020 engine.dispose()
1021 return None
1023 def get_session(self, username: str) -> Optional[Session]:
1024 """Create a new session for a user's database."""
1025 with self._connections_lock:
1026 if username not in self.connections:
1027 # Use debug level for this common scenario to reduce log noise
1028 logger.debug(f"No open database for user {username}")
1029 return None
1030 engine = self.connections[username]
1031 # Create session inside lock to prevent race with close_user_database()
1032 SessionLocal = sessionmaker(bind=engine)
1033 return SessionLocal()
1035 def get_connected_usernames(self) -> set:
1036 """Return a snapshot of usernames with open connections."""
1037 with self._connections_lock:
1038 return set(self.connections.keys())
1040 def _checkpoint_wal(self, engine, context: str = ""):
1041 """Checkpoint WAL before disposing engine to flush pending writes."""
1042 try:
1043 with engine.connect() as conn:
1044 # TRUNCATE returns (busy, log_pages, checkpointed_pages).
1045 # busy=1 means a reader/writer held a WAL lock and the WAL
1046 # was NOT truncated — surface that so we can tell from logs
1047 # whether the helper actually shrank the file.
1048 result = conn.execute(
1049 text("PRAGMA wal_checkpoint(TRUNCATE)")
1050 ).fetchone()
1051 if result and result[0] == 1: 1051 ↛ 1052line 1051 didn't jump to line 1052 because the condition on line 1051 was never true
1052 logger.debug(
1053 f"WAL checkpoint busy {context} — WAL not truncated"
1054 )
1055 except Exception:
1056 logger.debug(f"WAL checkpoint failed {context}", exc_info=True)
1058 def close_user_database(self, username: str):
1059 """Close a user's database connection."""
1060 with self._connections_lock:
1061 if username in self.connections:
1062 try:
1063 self._checkpoint_wal(
1064 self.connections[username], f"for {username}"
1065 )
1066 self.connections[username].dispose()
1067 except Exception:
1068 logger.warning(
1069 f"Failed to dispose engine for {username}",
1070 )
1071 del self.connections[username]
1072 # Deliberately do NOT pop _init_locks[username] here. A
1073 # concurrent open_user_database may already hold a reference to
1074 # this user's lock (fetched via _get_init_lock) and be about to
1075 # enter its cold-open; dropping the entry would let a later open
1076 # create a *second* lock for the same user, so two cold-opens
1077 # could migrate one DB file at once -- the very race this lock
1078 # exists to prevent. The dict is bounded by the number of
1079 # distinct usernames (one small Lock each), so retaining it is
1080 # cheap; close_all_databases clears it wholesale at shutdown.
1081 logger.info(f"Closed database for user {username}")
1083 def close_all_databases(self):
1084 """Close all open user database connections and release file locks."""
1085 with self._connections_lock:
1086 for username, engine in list(self.connections.items()):
1087 try:
1088 self._checkpoint_wal(engine, f"for {username}")
1089 engine.dispose()
1090 except Exception:
1091 logger.debug(f"Error disposing engine for {username}")
1092 self.connections.clear()
1093 self._init_locks.clear()
1095 def check_database_integrity(self, username: str) -> bool:
1096 """Check integrity of a user's encrypted database."""
1097 with self._connections_lock:
1098 if username not in self.connections:
1099 return False
1100 engine = self.connections[username]
1102 try:
1103 with engine.connect() as conn:
1104 # Quick integrity check
1105 result = conn.execute(text("PRAGMA quick_check"))
1106 if result.fetchone()[0] != "ok":
1107 return False
1109 # SQLCipher integrity check
1110 result = conn.execute(text("PRAGMA cipher_integrity_check"))
1111 # If this returns any rows, there are HMAC failures
1112 failures = list(result)
1113 if failures:
1114 logger.error(
1115 f"Integrity check failed for {username}: {len(failures)} HMAC failures"
1116 )
1117 return False
1119 return True
1121 except Exception as e:
1122 # No ``password`` parameter on this method and the engine
1123 # was created via a hex-key closure (the plaintext password
1124 # is not retained on the engine itself), so the traceback
1125 # frames in this method do not hold a credential. Still
1126 # drop the traceback to stay uniform with the rest of the
1127 # #4182 sweep and to guard against a caller frame that
1128 # happens to hold one (e.g., a route handler that
1129 # retrieved the password from the session store).
1130 from ..security.log_sanitizer import redact_secrets
1132 safe_msg = redact_secrets(str(e), None)
1133 logger.warning(
1134 f"Integrity check error for user: {username}: {safe_msg}"
1135 )
1136 return False
1138 def change_password(
1139 self, username: str, old_password: str, new_password: str
1140 ) -> bool:
1141 """Change the encryption password for a user's database.
1143 This rekeys the SQLCipher database — no separate auth-DB
1144 password-hash update is needed because passwords are never
1145 stored. Login verification is done by attempting decryption.
1146 """
1147 if not self.has_encryption:
1148 logger.warning(
1149 "Cannot change password - SQLCipher not available (databases are unencrypted)"
1150 )
1151 return False
1153 db_path = self._get_user_db_path(username)
1155 if not db_path.exists():
1156 return False
1158 try:
1159 # Close existing connection if any
1160 self.close_user_database(username)
1162 # Open with old password
1163 engine = self.open_user_database(username, old_password)
1164 if not engine:
1165 return False
1167 # Rekey the database (only works with SQLCipher)
1168 with engine.connect() as conn:
1169 # Use centralized rekey function
1170 set_sqlcipher_rekey(conn, new_password, db_path=db_path)
1172 logger.info(f"Password changed for user {username}")
1173 return True
1175 except Exception as e:
1176 # Both ``old_password`` and ``new_password`` are in lexical
1177 # scope. ``open_user_database`` / ``set_sqlcipher_rekey``
1178 # carry these into nested frames — a traceback rendered
1179 # with ``diagnose=True`` would leak them. Redact both and
1180 # drop the traceback chain.
1181 from ..security.log_sanitizer import redact_secrets
1183 safe_msg = redact_secrets(str(e), old_password, new_password)
1184 logger.warning(
1185 f"Failed to change password for user: {username}: {safe_msg}"
1186 )
1187 return False
1188 finally:
1189 # Close the connection
1190 self.close_user_database(username)
1192 def user_exists(self, username: str) -> bool:
1193 """Check if a user exists in the auth database."""
1194 from .auth_db import auth_db_session
1195 from .models.auth import User
1197 with auth_db_session() as session:
1198 user = session.query(User).filter_by(username=username).first()
1199 return user is not None
1201 def get_memory_usage(self) -> Dict[str, Any]:
1202 """Get memory usage statistics."""
1203 with self._connections_lock:
1204 num_connections = len(self.connections)
1205 return {
1206 "active_connections": num_connections,
1207 "active_sessions": 0, # Sessions are created on-demand, not tracked
1208 "estimated_memory_mb": num_connections
1209 * 3.5, # ~3.5MB per connection
1210 }
1212 def create_thread_safe_session_for_metrics(
1213 self, username: str, password: str
1214 ):
1215 """
1216 Create a new database session safe for use in background threads.
1218 Previously this method created a dedicated NullPool engine per
1219 (username, thread_id) pair, which leaked file descriptors under
1220 load (SQLCipher + WAL holds 3 FDs per active connection and
1221 orphaned engines accumulated when @thread_cleanup did not fire).
1223 It now routes through the shared per-user QueuePool engine at
1224 ``self.connections[username]``. That engine is already created
1225 with ``check_same_thread=False`` (so background threads are
1226 safe), is bounded by ``pool_size + max_overflow``, and is
1227 subject to the periodic ``dispose()`` workaround in
1228 ``connection_cleanup.py`` that mitigates the SQLCipher+WAL
1229 out-of-order-close FD leak.
1231 Args:
1232 username: The username
1233 password: The user's password (encryption key), used only
1234 to open the user database on cache miss.
1236 Returns:
1237 A SQLAlchemy Session bound to the per-user QueuePool engine.
1238 """
1239 db_path = self._get_user_db_path(username)
1241 if not db_path.exists():
1242 raise ValueError(f"No database found for user {username}")
1244 with self._connections_lock:
1245 engine = self.connections.get(username)
1247 if engine is None: 1247 ↛ 1250line 1247 didn't jump to line 1250 because the condition on line 1247 was never true
1248 # Cache miss — open the user database. This is idempotent:
1249 # after the first call it just returns the cached engine.
1250 engine = self.open_user_database(username, password)
1251 if engine is None:
1252 raise ValueError(f"Failed to open database for user {username}")
1254 # Use SQLAlchemy's default expire_on_commit=True.
1255 Session = sessionmaker(bind=engine)
1256 return Session()
1259# Global instance
1260db_manager = DatabaseManager()