Coverage for src/local_deep_research/database/encrypted_db.py: 94%
463 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Encrypted database management using SQLCipher.
3Handles per-user encrypted databases with browser-friendly authentication.
4"""
6import hmac
7import secrets
8import os
9import threading
10import time
11from pathlib import Path
12from typing import Any, Dict, Optional
14from loguru import logger
15from sqlalchemy import create_engine, event, text
16from sqlalchemy.engine import Engine
17from sqlalchemy.orm import Session, sessionmaker
18from sqlalchemy.pool import QueuePool, StaticPool
20from ..config.paths import get_data_directory, get_user_database_filename
21from ..settings.env_registry import get_env_setting
23# ``redact_secrets`` stays local to the exception handlers that use it. A
24# module-scope import would initialize the broad ``security`` package while
25# encrypted_db is still defining its bootstrap state, unnecessarily widening
26# this low-level module's dependency surface.
27#
28# Historically, ``security/__init__.py`` eagerly re-exported file-integrity
29# symbols, which created this cycle during database bootstrap:
30#
31# database/encrypted_db.py
32# → security/__init__.py
33# → security/file_integrity/integrity_manager.py
34# → database/session_context.py
35# → database/encrypted_db.py (mid-load, ``db_manager`` not yet
36# defined → ImportError)
37#
38# File-integrity exports are now lazy, and test_integrity_bootstrap_guard.py
39# locks in the current fail-loud session-context import. Keeping this import
40# local avoids coupling encrypted_db bootstrap to unrelated security package
41# initialization. Other files in PRs #4168/#4175/#4181 can import
42# ``redact_secrets`` at module scope because they are not on that bootstrap
43# path.
44from .sqlcipher_compat import get_sqlcipher_module
45from .pool_config import (
46 MAX_OVERFLOW,
47 POOL_PRE_PING,
48 POOL_RECYCLE_SECONDS,
49 POOL_SIZE,
50)
51from .sqlcipher_utils import (
52 set_sqlcipher_key,
53 set_sqlcipher_rekey,
54 apply_cipher_defaults_before_key,
55 apply_sqlcipher_pragmas,
56 apply_performance_pragmas,
57 verify_sqlcipher_connection,
58 create_database_salt,
59 get_salt_file_path,
60 has_per_database_salt,
61 get_key_from_password,
62 get_sqlcipher_version,
63 create_sqlcipher_connection,
64)
67class DatabaseInitializationError(Exception):
68 """Raised when a per-user database opens but its schema can't be initialised.
70 Distinct from credential / decryption failures (which return ``None``
71 from :py:meth:`DatabaseManager.open_user_database`) so callers — chiefly
72 the login route — can avoid penalising the user's lockout counter and
73 surface a different error message. The credentials are valid; the
74 database state isn't.
75 """
78def _best_effort_chmod(path, mode: int, *, warn: bool = False) -> None:
79 """Tighten permissions on ``path`` to ``mode``, never raising.
81 Permission hardening must NOT be able to break database creation. On
82 filesystems that don't support POSIX chmod — some Docker bind mounts and
83 network/FUSE volumes, notably Docker Desktop on macOS/Windows — os.chmod
84 can raise OSError even for the file's owner. In that case we leave the
85 path at its default (umask) mode rather than failing registration/login.
87 ``warn=True`` surfaces the failure at warning level: a silent downgrade
88 leaves the path group/world-readable, which matters for DB files holding
89 user data (the unencrypted fallback is plaintext). The owner-only
90 directory chmod is defense-in-depth, so it stays at debug.
91 """
92 try:
93 os.chmod(str(path), mode)
94 except OSError:
95 log = logger.warning if warn else logger.debug
96 log(
97 f"Could not set permissions {oct(mode)} on {path}; "
98 "left at filesystem default",
99 exc_info=True,
100 )
103def _remove_partial_user_db_files(db_path: Path) -> None:
104 """Remove a half-created user database and all of its sidecar files.
106 Called from ``create_user_database``'s failure paths. The encrypted
107 branch writes a per-database ``.salt`` file (via ``create_database_salt``)
108 *before* the DB itself, and SQLCipher/WAL leaves ``-wal``/``-shm``
109 sidecars. Removing only the ``.db`` — as these cleanup paths used to —
110 orphans the ``.salt``: a retry then hits ``create_database_salt``'s
111 ``FileExistsError`` (it refuses to overwrite to prevent data loss), so the
112 username stays permanently un-registerable even after the orphaned auth
113 row is cleaned up. Delete every artifact so the on-disk state is atomic
114 and the username is reusable on a subsequent retry.
116 ``create_user_database`` calls this from four sites: its three
117 failure-cleanup blocks (structure creation, engine build, migration — a
118 graceful error mid-creation) and its salt-creation recovery path (an
119 orphan left by a prior hard-killed create or by pre-#4934 cleanup that
120 removed only the ``.db``). In every case the ``db_path.exists()`` guard at
121 the top of ``create_user_database`` already ran and found no *pre-existing*
122 database, so any ``.db``/``.salt`` present now is this call's own partial
123 work — removing it cannot destroy live data, because a valid encrypted DB
124 always has both the ``.db`` and its ``.salt``.
126 Both the ``-wal``/``-shm`` (WAL mode, the default) and ``-journal``
127 (rollback-journal modes: DELETE/TRUNCATE/PERSIST) sidecars are removed so
128 the cleanup is correct regardless of the configured ``journal_mode``.
130 Never raises — cleanup must not mask the original creation error.
131 """
132 # Derive every artifact path up front, guarded: the "Never raises"
133 # contract has to hold even if a path helper is handed something odd
134 # (e.g. a nameless path, where ``with_name``/``with_suffix`` raise), so a
135 # derivation failure is logged and swallowed rather than propagating and
136 # masking the original creation error.
137 try:
138 db_path = Path(db_path)
139 artifacts = (
140 db_path,
141 get_salt_file_path(db_path),
142 db_path.with_name(db_path.name + "-wal"),
143 db_path.with_name(db_path.name + "-shm"),
144 db_path.with_name(db_path.name + "-journal"),
145 )
146 except Exception:
147 logger.warning(
148 f"Could not derive partial user DB artifact paths: {db_path!r}"
149 )
150 return
152 for path in artifacts:
153 try:
154 path.unlink(missing_ok=True)
155 except OSError:
156 # Best-effort: a leftover artifact is worth surfacing but must not
157 # mask the original creation error. The path is enough to
158 # diagnose; skip the exception detail per the sensitive-logging
159 # policy.
160 logger.warning(f"Could not remove partial user DB artifact {path}")
163class DatabaseManager:
164 """Manages encrypted SQLCipher databases for each user."""
166 def __init__(self):
167 self.connections: Dict[str, Engine] = {}
168 # username -> (salt, digest) proving which password opened the cached
169 # engine. `connections` is keyed by username ALONE, so a cache hit
170 # says nothing about the credential presented on THIS call; without a
171 # verifier, returning the cached engine silently accepts any password.
172 # See _password_matches_cached().
173 #
174 # A fast keyed digest is the right primitive here: this is an
175 # in-process equality check, not at-rest storage, and the plaintext
176 # password already lives in memory in session_password_store, so a
177 # slow KDF would add cost without reducing exposure.
178 self._password_verifiers: Dict[str, tuple[bytes, bytes]] = {}
179 self._verifier_key = secrets.token_bytes(32)
180 self._connections_lock = threading.RLock()
181 # Per-user locks serializing the cold-open (engine build + migration)
182 # so two concurrent first-opens of one user never run alembic against
183 # the same database file at once. Created lazily under
184 # _connections_lock; see open_user_database / _get_init_lock.
185 self._init_locks: Dict[str, threading.Lock] = {}
186 self._data_dir_override: Optional[Path] = None
187 self._initialized_data_dirs: set[Path] = set()
189 # Check SQLCipher availability
190 self.has_encryption = self._check_encryption_available()
192 # ----------------------------------------------------------------
193 # Pool class selection — see ADR-0004
194 #
195 # We use QueuePool (pool_size=20, max_overflow=40,
196 # pool_timeout=10) for production and StaticPool for tests.
197 #
198 # Why pool_size=20:
199 #
200 # 1. SQLCipher + WAL mode can leak file handles when connections
201 # close out of open-order. Fewer pooled connections = fewer
202 # opportunities for out-of-order closes during pool_recycle.
203 # See: https://github.com/sqlcipher/android-database-sqlcipher/issues/6
204 # See: https://github.com/dotnet/efcore/issues/35010
205 #
206 # 2. SQLite serializes all writes through a single file lock.
207 # Multiple pooled connections don't improve throughput — they
208 # just hold FDs (up to 3 per connection in WAL mode).
209 #
210 # 3. The cleanup scheduler periodically calls engine.dispose()
211 # to release all pooled connections, preventing long-lived
212 # handles from accumulating over days of idle operation.
213 #
214 # Why pool_size=20 and not 1: inject_current_user() creates a
215 # QueuePool session on every request via g.db_session. With the
216 # UI polling /api/research/<id>/status every 1-2s plus other
217 # API calls and before_request middleware, pool_size=1
218 # (max_overflow=2, so 3 total) is easily exhausted — causing
219 # 30-second timeouts and PendingRollbackError cascades.
220 # pool_size=20 + max_overflow=40 (60 total) provides ample
221 # headroom for concurrent requests and multiple browser tabs.
222 #
223 # Why not NullPool: SQLCipher's PRAGMA key adds ~0.2ms per
224 # connection open. With 20-30 queries per page load, NullPool
225 # adds a noticeable 4-6ms overhead vs QueuePool's ~1.5ms.
226 # ----------------------------------------------------------------
227 self._use_static_pool = bool(os.environ.get("TESTING"))
228 self._pool_class = StaticPool if self._use_static_pool else QueuePool
230 def _ensure_data_dir(self, path: Path) -> None:
231 """Create and harden one resolved encrypted-database directory."""
232 path = Path(path)
233 with self._connections_lock:
234 if path in self._initialized_data_dirs:
235 return
236 # Lazy import: avoid a security -> ... -> database import cycle at
237 # module load time (this can run during DatabaseManager
238 # construction, which happens early in app bootstrap).
239 from ..security.directory_creation import create_directory
241 create_directory(
242 path,
243 context="encrypted database storage",
244 )
245 # Restrict the per-user DB directory to the owner (0o700).
246 # This covers SQLite WAL/SHM sidecars and plaintext fallback temp
247 # files even when SQLite recreates them with umask permissions.
248 _best_effort_chmod(path, 0o700)
249 self._initialized_data_dirs.add(path)
251 @property
252 def data_dir(self) -> Path:
253 """Return the encrypted-DB directory for the live data root.
255 The default path is resolved on every access so a process that imports
256 ``db_manager`` before ``LDR_DATA_DIR`` is set does not stay bound to
257 the platform default. Explicit assignments use the setter below and
258 remain stable for tests and programmatic callers.
259 """
260 path = self._data_dir_override
261 if path is None:
262 path = get_data_directory() / "encrypted_databases"
263 self._ensure_data_dir(path)
264 return path
266 @data_dir.setter
267 def data_dir(self, value: Path) -> None:
268 """Set an explicit data-directory override for this manager."""
269 path = Path(value)
270 self._data_dir_override = path
271 self._ensure_data_dir(path)
273 def _get_pool_kwargs(self) -> Dict[str, Any]:
274 """Get pool configuration kwargs based on pool type.
276 StaticPool doesn't support pool_size or max_overflow.
277 QueuePool uses moderate sizing to handle concurrent web requests
278 while limiting FD usage. See ADR-0004 for rationale.
279 """
280 if self._use_static_pool:
281 return {}
282 return {
283 "pool_size": POOL_SIZE,
284 "max_overflow": MAX_OVERFLOW,
285 "pool_timeout": 10,
286 "pool_pre_ping": POOL_PRE_PING,
287 "pool_recycle": POOL_RECYCLE_SECONDS,
288 }
290 def _is_valid_encryption_key(self, password: str) -> bool:
291 """
292 Check if the provided password is valid (not None, empty, or whitespace-only).
294 Args:
295 password: The password to check
297 Returns:
298 True if the password is valid, False otherwise
299 """
300 return password is not None and password.strip() != ""
302 def is_user_connected(self, username: str) -> bool:
303 """Check if a user has an active database connection.
305 Thread-safe accessor for external callers.
307 Args:
308 username: The username to check
310 Returns:
311 True if the user has an active connection
312 """
313 with self._connections_lock:
314 return username in self.connections
316 def _check_encryption_available(self) -> bool:
317 """Check if SQLCipher is available for encryption."""
318 try:
319 import os as os_module
320 import tempfile
322 # Test if SQLCipher actually works, not just if it imports
323 with tempfile.NamedTemporaryFile(delete=False) as tmp:
324 tmp_path = tmp.name
326 try:
327 # Try to create a test encrypted database
328 sqlcipher_module = get_sqlcipher_module()
329 sqlcipher = sqlcipher_module.dbapi2
331 conn = sqlcipher.connect(tmp_path)
332 try:
333 cursor = conn.cursor()
334 # Use creation_mode=True since we're creating a new test database
335 apply_cipher_defaults_before_key(cursor)
336 # Use centralized key setting
337 set_sqlcipher_key(cursor, "testpass")
338 # Apply post-key pragmas (kdf_iter for new DB)
339 apply_sqlcipher_pragmas(cursor, creation_mode=True)
340 apply_performance_pragmas(cursor)
342 # Check SQLCipher version
343 version = get_sqlcipher_version(cursor)
344 if version: 344 ↛ 356line 344 didn't jump to line 356 because the condition on line 344 was always true
345 major = (
346 version.split(".")[0]
347 if "." in version
348 else version[0]
349 )
350 if major.isdigit() and int(major) < 4: 350 ↛ 351line 350 didn't jump to line 351 because the condition on line 350 was never true
351 logger.warning(
352 f"SQLCipher version {version} detected. "
353 "Version 4.x+ is recommended for proper PRAGMA ordering."
354 )
356 cursor.close()
357 # Now use the connection for table operations
358 conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)")
359 conn.execute("INSERT INTO test VALUES (1)")
360 result = conn.execute("SELECT * FROM test").fetchone()
362 if result != (1,): 362 ↛ 363line 362 didn't jump to line 363 because the condition on line 362 was never true
363 raise RuntimeError("SQLCipher encryption test failed")
364 logger.info(
365 "SQLCipher available and working - databases will be encrypted"
366 )
367 return True
368 finally:
369 from ..utilities.resource_utils import safe_close
371 safe_close(conn, "SQLCipher test connection")
372 except Exception:
373 logger.warning("SQLCipher module found but not working")
374 raise ImportError("SQLCipher not functional")
375 finally:
376 # Clean up test file
377 try:
378 os_module.unlink(tmp_path)
379 except OSError as e:
380 logger.debug(
381 f"Failed to clean up temp file {tmp_path}: {e}"
382 )
384 except ImportError:
385 # Check if user has explicitly allowed unencrypted databases.
386 # Registry handles deprecated LDR_ALLOW_UNENCRYPTED fallback automatically.
387 allow_unencrypted = get_env_setting(
388 "bootstrap.allow_unencrypted", False
389 )
391 if not allow_unencrypted:
392 # No ``password`` is in scope here (this runs at
393 # bootstrap, before any user authenticates), and the
394 # caught ``ImportError`` is purely about the SQLCipher
395 # package being missing. Still drop the traceback to
396 # keep this file's logging discipline uniform with the
397 # rest of the #4182 sweep: no exception traceback is
398 # written to the production log from this module. The
399 # install-hint message is the user-facing diagnostic;
400 # the traceback adds no value beyond it. ``warning``
401 # rather than ``error``/``exception`` so the custom
402 # "logger.exception in except blocks" lint passes
403 # without re-introducing a traceback render.
404 logger.warning(
405 "SECURITY ERROR: SQLCipher is not installed!\n"
406 "Your databases will NOT be encrypted.\n"
407 "To fix this:\n"
408 "1. Install SQLCipher: sudo apt install sqlcipher libsqlcipher-dev\n"
409 "2. Reinstall project: pdm install\n"
410 "Or use Docker with SQLCipher pre-installed.\n\n"
411 "To explicitly allow unencrypted databases (NOT RECOMMENDED):\n"
412 "export LDR_BOOTSTRAP_ALLOW_UNENCRYPTED=true"
413 )
414 raise RuntimeError(
415 "SQLCipher not available. Set LDR_BOOTSTRAP_ALLOW_UNENCRYPTED=true to proceed without encryption (NOT RECOMMENDED)"
416 )
417 logger.warning(
418 "WARNING: Running with UNENCRYPTED databases!\n"
419 "This means:\n"
420 "- Passwords don't protect data access\n"
421 "- API keys are stored in plain text\n"
422 "- Anyone with file access can read all data\n"
423 "Install SQLCipher for secure operation!"
424 )
425 return False
427 def _get_user_db_path(self, username: str) -> Path:
428 """Get the path for a user's encrypted database.
430 Raises:
431 ValueError: Propagated from ``get_user_database_filename`` if
432 ``username`` is empty/None -- that's the actual sha256
433 chokepoint, so this method fails closed the same way rather
434 than resolving to a phantom shared ``sha256("")`` filename.
435 Registration already enforces a non-empty username; this is
436 defense-in-depth at the path-resolution chokepoint.
437 """
438 return self.data_dir / get_user_database_filename(username)
440 def _apply_pragmas(self, connection, connection_record):
441 """Apply pragmas for optimal performance."""
442 # Check if this is SQLCipher or regular SQLite
443 is_encrypted = self.has_encryption
445 # Use centralized performance pragma application
447 apply_performance_pragmas(connection)
449 # SQLCipher-specific pragmas
450 if is_encrypted: 450 ↛ 451line 450 didn't jump to line 451 because the condition on line 450 was never true
451 from .sqlcipher_utils import get_sqlcipher_settings
453 settings = get_sqlcipher_settings()
454 pragmas = [
455 f"PRAGMA kdf_iter = {settings['kdf_iterations']}",
456 f"PRAGMA cipher_page_size = {settings['page_size']}",
457 ]
458 for pragma in pragmas:
459 try:
460 connection.execute(pragma)
461 except Exception as e:
462 logger.debug(f"Could not apply pragma '{pragma}': {e}")
463 else:
464 # Regular SQLite pragma
465 try:
466 connection.execute(
467 "PRAGMA mmap_size = 268435456"
468 ) # 256MB memory mapping
469 except Exception as e:
470 logger.debug(f"Could not apply mmap_size pragma: {e}")
472 @staticmethod
473 def _make_sqlcipher_connection(
474 db_path: Path,
475 password: str,
476 isolation_level: Optional[str] = "IMMEDIATE",
477 check_same_thread: bool = False,
478 ) -> Any:
479 """Create a properly initialized SQLCipher connection.
481 Follows the canonical SQLCipher initialization order: set key,
482 apply cipher pragmas, verify, then apply performance pragmas.
483 Cipher pragmas (page size, HMAC algorithm, KDF iterations) must
484 be configured before the first query (verification) because that
485 query triggers page decryption with the active cipher settings.
487 Args:
488 db_path: Path to the database file
489 password: The database encryption passphrase
490 isolation_level: SQLite isolation level (``""`` for deferred
491 transactions, ``None`` for autocommit)
492 check_same_thread: SQLite check_same_thread flag
494 Returns:
495 A raw ``sqlcipher3`` connection ready for use.
497 Raises:
498 ValueError: If the database key cannot be verified.
499 """
500 sqlcipher3 = get_sqlcipher_module()
501 conn = sqlcipher3.connect(
502 str(db_path),
503 isolation_level=isolation_level,
504 check_same_thread=check_same_thread,
505 )
506 cursor = conn.cursor()
508 try:
509 set_sqlcipher_key(cursor, password, db_path=db_path)
510 apply_sqlcipher_pragmas(cursor, creation_mode=False)
512 if not verify_sqlcipher_connection(cursor):
513 raise ValueError("Failed to verify database key") # noqa: TRY301 — cleanup in except before re-raise
515 apply_performance_pragmas(cursor)
516 except Exception:
517 try:
518 cursor.close()
519 except Exception: # noqa: BLE001
520 logger.warning("Failed to close cursor during cleanup")
521 from ..utilities.resource_utils import safe_close
523 safe_close(conn, "encrypted DB connection")
524 raise
526 cursor.close()
527 return conn
529 def create_user_database(self, username: str, password: str) -> Engine:
530 """Create a new encrypted database for a user."""
532 # Validate the encryption key
533 if not self._is_valid_encryption_key(password):
534 logger.error(
535 f"Invalid encryption key for user {username}: password is None or empty"
536 )
537 raise ValueError(
538 "Invalid encryption key: password cannot be None or empty"
539 )
541 db_path = self._get_user_db_path(username)
543 if db_path.exists():
544 raise ValueError(f"Database already exists for user {username}")
546 # Create connection string - use regular SQLite when SQLCipher not available
547 if self.has_encryption:
548 # Create directory if it doesn't exist
549 from ..security.directory_creation import create_directory
551 create_directory(
552 db_path.parent,
553 context="per-user encrypted database directory",
554 )
556 # Create per-database salt for new databases (v2 security improvement).
557 #
558 # A pre-existing salt here is an ORPHAN, not live data: the
559 # ``db_path.exists()`` guard above already established that no
560 # database file exists, and a valid encrypted DB always has BOTH
561 # the ``.db`` and its ``.salt``. Such an orphan is left by a create
562 # that died before completing — a hard process death (SIGKILL/OOM/
563 # power loss) mid-creation, or the pre-#4934 cleanup code that
564 # removed only the ``.db``. Without recovery, ``create_database_salt``
565 # raises ``FileExistsError`` (it refuses to overwrite to protect
566 # live data) on every retry, so the username stays permanently
567 # un-registerable. Remove the stale artifacts and create fresh —
568 # safe precisely because no database depends on this salt.
569 #
570 # Note: backups are keyed off the *source* DB's salt (they store no
571 # salt copy — see backup_service), so deleting a salt also strands
572 # any backups for it. That is not reachable here: this only fires
573 # when db_path is absent, and the app never removes a live .db while
574 # keeping its .salt (there is no in-app restore). Only out-of-band
575 # manual deletion of the .db could turn this into backup loss.
576 #
577 # SAFETY INVARIANT: this deletion is safe only while a salt-without-
578 # a-.db always means garbage. That holds today because create_user_
579 # database is the sole salt writer/caller and nothing restores or
580 # imports a .db from a salt. If an in-app restore/import that stages
581 # a .salt ahead of its .db is ever added, RE-EVALUATE this branch —
582 # it could delete a salt whose database is legitimately incoming.
583 try:
584 create_database_salt(db_path)
585 except FileExistsError:
586 logger.warning(
587 f"Removing orphaned salt (no matching database) for "
588 f"{username} before creating a fresh one"
589 )
590 _remove_partial_user_db_files(db_path)
591 create_database_salt(db_path)
592 logger.info(f"Created per-database salt for {username}")
594 # Create database structure using raw SQLCipher outside SQLAlchemy
595 try:
596 # Pre-derive key before closures to avoid capturing plaintext
597 # password. Kept inside the try so a derivation failure here
598 # also triggers the partial-file cleanup below, rather than
599 # orphaning the just-created salt file.
600 hex_key = get_key_from_password(password, db_path=db_path).hex()
602 conn = create_sqlcipher_connection(
603 db_path,
604 password=password,
605 creation_mode=True,
606 connect_kwargs={
607 # DEFERRED (empty string) so pure-SELECT transactions
608 # acquire only SQLite's SHARED lock, letting WAL-mode
609 # concurrent readers proceed while a writer is active.
610 # IMMEDIATE was previously set "defensively" and made
611 # every transaction (even reads) take a RESERVED lock,
612 # which was the single biggest contention source on
613 # the login-hang path. Race-prone check-then-insert
614 # call sites were made race-free at the application
615 # layer in the preceding prerequisite PR.
616 "isolation_level": "",
617 "check_same_thread": False,
618 },
619 )
620 _best_effort_chmod(db_path, 0o600, warn=True)
621 try:
622 # Get the CREATE TABLE statements from SQLAlchemy models
623 from sqlalchemy.dialects import sqlite
624 from sqlalchemy.schema import CreateIndex, CreateTable
626 from .models import Base
628 # Indexes must be emitted explicitly — SQLAlchemy compiles
629 # `index=True`/`unique=True` and `Index(...)` to separate
630 # CREATE [UNIQUE] INDEX statements, not inline.
631 sqlite_dialect = sqlite.dialect()
632 for table in Base.metadata.sorted_tables:
633 if table.name == "users":
634 continue
635 create_sql = str(
636 CreateTable(table, if_not_exists=True).compile(
637 dialect=sqlite_dialect
638 )
639 )
640 logger.debug(f"Creating table {table.name}")
641 conn.execute(create_sql)
642 for index in table.indexes:
643 index_sql = str(
644 CreateIndex(index, if_not_exists=True).compile(
645 dialect=sqlite_dialect
646 )
647 )
648 conn.execute(index_sql)
650 conn.commit()
651 finally:
652 from ..utilities.resource_utils import safe_close
654 safe_close(conn, "user DB setup connection")
656 logger.info(
657 f"Database structure created successfully for {username}"
658 )
660 except Exception as e:
661 # ``password`` is in lexical scope (function parameter) and
662 # is passed into ``create_sqlcipher_connection`` /
663 # ``set_sqlcipher_key`` above. A traceback rendered with
664 # loguru ``diagnose=True`` would dump frame locals — which
665 # includes the plaintext SQLCipher master password. Use
666 # ``logger.warning`` to drop the traceback chain and
667 # redact the password from str(e) for defense in depth.
668 # SQLCipher master passwords are unrecoverable (see
669 # TRUST.md §5) so the impact of a leak is permanent.
670 from ..security.log_sanitizer import redact_secrets
672 safe_msg = redact_secrets(str(e), password)
673 logger.warning(f"Error creating database structure: {safe_msg}")
674 # Remove the partial DB *and* its salt/WAL/journal sidecars so
675 # a retry of the same username isn't blocked by a leftover salt.
676 _remove_partial_user_db_files(db_path)
677 raise
679 # Small delay to ensure file is fully written
680 import time
682 time.sleep(0.1)
684 # Now create SQLAlchemy engine using custom connection creator
685 def create_engine_connection():
686 """Create a properly initialized SQLCipher connection."""
687 return create_sqlcipher_connection(
688 db_path,
689 hex_key=hex_key,
690 creation_mode=False,
691 connect_kwargs={
692 # DEFERRED (empty string) so pure-SELECT transactions
693 # acquire only SQLite's SHARED lock, letting WAL-mode
694 # concurrent readers proceed while a writer is active.
695 # IMMEDIATE was previously set "defensively" and made
696 # every transaction (even reads) take a RESERVED lock,
697 # which was the single biggest contention source on
698 # the login-hang path. Race-prone check-then-insert
699 # call sites were made race-free at the application
700 # layer in the preceding prerequisite PR.
701 "isolation_level": "",
702 "check_same_thread": False,
703 },
704 )
706 # Create engine with custom creator function and optimized cache.
707 # The .db and .salt already exist on disk (structure creation
708 # succeeded above). This step sits between the structure-creation
709 # and migration cleanup blocks; without its own cleanup a failure
710 # here would orphan those files, and unlike every other failure
711 # path it would NOT be self-healing — the retry trips the
712 # db_path.exists() guard and bricks the username. So clean up on
713 # any escape. (create_engine is lazy so this is near-unreachable,
714 # but it closes the one gap in the otherwise-uniform coverage.)
715 try:
716 engine = create_engine(
717 "sqlite://",
718 creator=create_engine_connection,
719 poolclass=self._pool_class,
720 echo=False,
721 query_cache_size=1000,
722 **self._get_pool_kwargs(),
723 )
724 except Exception:
725 _remove_partial_user_db_files(db_path)
726 raise
727 else:
728 logger.warning(
729 f"SQLCipher not available - creating UNENCRYPTED database for user {username}"
730 )
731 # Fall back to regular SQLite with query cache
732 engine = create_engine(
733 f"sqlite:///{db_path}",
734 connect_args={"check_same_thread": False, "timeout": 30},
735 poolclass=self._pool_class,
736 echo=False,
737 query_cache_size=1000,
738 **self._get_pool_kwargs(),
739 )
741 # For unencrypted databases, just apply pragmas
742 event.listen(engine, "connect", self._apply_pragmas)
744 # Tables have already been created using raw SQLCipher above
745 # No need to create them again with SQLAlchemy
747 # Initialize database tables using centralized initialization
748 from .initialize import initialize_database
750 # Mirror of the fail-loud change #3635 made to open_user_database.
751 # Previously this swallowed the exception with "tables exist but
752 # schema version not stamped — migrations will be retried on next
753 # process restart". That left a half-broken DB on disk: tables
754 # present, no alembic_version row. The next login then re-ran
755 # alembic, hit the same error, and (post-#3635) 503'd — so the
756 # user could register but never log in again. Better to fail
757 # registration loudly with the partial DB removed, so the real
758 # cause (e.g. world-writable migrations dir) gets fixed instead
759 # of producing a permanently-locked-out account.
760 try:
761 Session = sessionmaker(bind=engine)
762 with Session() as session:
763 initialize_database(engine, session)
764 except Exception as e:
765 # ``password`` is in scope and was passed into the engine
766 # creator closure above. Drop the traceback to avoid leaking
767 # frame locals under ``diagnose=True`` and redact the
768 # password from str(e) defensively.
769 from ..security.log_sanitizer import redact_secrets
771 safe_msg = redact_secrets(str(e), password)
772 logger.warning(
773 f"Database migration failed for {username} during creation"
774 f" — removing partial DB: {safe_msg}"
775 )
776 engine.dispose()
777 # Remove the partial DB *and* its salt/WAL/journal sidecars so a
778 # retry of the same username isn't blocked by a leftover salt.
779 _remove_partial_user_db_files(db_path)
780 raise
782 # Restrict DB file to owner-only (0o600). The encrypted branch
783 # chmod's right after create_sqlcipher_connection, but the
784 # unencrypted fallback creates the file lazily on first connect
785 # (during initialize_database above), so it is only guaranteed to
786 # exist here. This file holds PLAINTEXT user data — leaving it at
787 # umask-default perms (commonly 0o644) would expose it to other
788 # local accounts.
789 if not self.has_encryption and db_path.exists():
790 _best_effort_chmod(db_path, 0o600, warn=True)
792 # Store connection AFTER migrations complete. This password just proved
793 # itself by creating the database; publish the engine and its verifier
794 # together so the cache is never briefly trusted without one.
795 self._cache_connection(username, engine, password)
797 logger.info(f"Created encrypted database for user {username}")
798 return engine
800 def _compute_verifier_digest(self, salt: bytes, password: str) -> bytes:
801 """Keyed digest binding ``password`` to a per-entry ``salt``.
803 A fast keyed HMAC, deliberately not a slow KDF: this is an in-process
804 equality check, never persisted, and the plaintext password already
805 lives in memory in session_password_store, so a KDF would add cost
806 without reducing exposure.
808 NOTE: CodeQL py/weak-sensitive-data-hashing flags the hash below as
809 weak password hashing. It is a reviewed FALSE POSITIVE: this app stores
810 no password hashes at rest (SQLCipher's PBKDF2 is the real KDF on the
811 cold-open path); this keyed digest only proves, in memory, that the SAME
812 password opened the cached engine, so a slow KDF here would add login
813 latency without reducing exposure. The rule is temporarily excluded in
814 .github/codeql/codeql-config.yml (see the TEMPORARY note there); the
815 preferred long-term fix is a per-alert dismissal in code scanning.
816 """
817 return hmac.new(
818 self._verifier_key, salt + password.encode("utf-8"), "sha256"
819 ).digest()
821 def _make_verifier(self, password: str) -> tuple[bytes, bytes]:
822 """Build a ``(salt, digest)`` verifier for ``password``. Takes no lock."""
823 salt = secrets.token_bytes(16)
824 return salt, self._compute_verifier_digest(salt, password)
826 def _cache_connection(
827 self, username: str, engine: Engine, password: str
828 ) -> None:
829 """Publish an engine and the verifier for the password that opened it,
830 atomically.
832 Called only on paths that have ALREADY proved the password by opening
833 (or creating) the SQLCipher database with it. Storing the engine and
834 its verifier under a single lock keeps the invariant "a cached
835 connection always has a matching verifier" true by construction, so a
836 concurrent open can never observe a connection with a missing or stale
837 verifier.
838 """
839 verifier = self._make_verifier(password)
840 with self._connections_lock:
841 self.connections[username] = engine
842 self._password_verifiers[username] = verifier
844 def _record_password_verifier(self, username: str, password: str) -> None:
845 """Arm the verifier for a connection cached by other means.
847 Prefer :meth:`_cache_connection` when publishing the engine too — it
848 stores both atomically. This standalone form arms a verifier against an
849 already-cached engine. Caller must not hold ``_connections_lock``.
850 """
851 verifier = self._make_verifier(password)
852 with self._connections_lock:
853 self._password_verifiers[username] = verifier
855 def _verifier_matches(self, username: str, password: str) -> bool:
856 """Lock-free verifier comparison. **Caller must hold _connections_lock.**
858 Fails CLOSED: a missing verifier returns False so the caller falls
859 through to a real key-validating open. Returning True on a missing
860 verifier would reopen exactly the hole this closes.
861 """
862 entry = self._password_verifiers.get(username)
863 if entry is None:
864 return False
865 salt, expected = entry
866 return hmac.compare_digest(
867 self._compute_verifier_digest(salt, password), expected
868 )
870 def _cached_engine_trusted(self, username: str, password: str) -> bool:
871 """May the cached engine for ``username`` be returned to a caller
872 presenting ``password``? **Caller must hold _connections_lock.**
874 Unencrypted mode has no credential to verify — the cold path opens the
875 plain-SQLite database with any password too — so the verifier is
876 meaningless there and would only force needless cold opens against the
877 placeholder passwords the app uses (session_context /
878 database_middleware). Trust the cache directly. Encrypted mode requires
879 a verifier match.
880 """
881 if not self.has_encryption:
882 return True
883 return self._verifier_matches(username, password)
885 def _password_matches_cached(self, username: str, password: str) -> bool:
886 """Does ``password`` match the one that opened the cached engine?
888 Fails CLOSED (see :meth:`_verifier_matches`). Takes the lock; prefer
889 :meth:`_verifier_matches` when already holding it.
890 """
891 with self._connections_lock:
892 return self._verifier_matches(username, password)
894 def _get_init_lock(self, username: str) -> threading.Lock:
895 """Return the per-user cold-open lock, creating it on first use.
897 Guards the engine-build + migration in ``open_user_database`` so two
898 concurrent first-opens of the same user serialize instead of running
899 migrations against one database file simultaneously. Per-user (not the
900 global ``_connections_lock``) so different users still open in
901 parallel.
902 """
903 with self._connections_lock:
904 lock = self._init_locks.get(username)
905 if lock is None:
906 lock = threading.Lock()
907 self._init_locks[username] = lock
908 return lock
910 def open_user_database(
911 self, username: str, password: str
912 ) -> Optional[Engine]:
913 """Open an existing encrypted database for a user."""
914 open_start = time.perf_counter()
916 # Validate the encryption key
917 if not self._is_valid_encryption_key(password):
918 logger.error(
919 f"Invalid encryption key when opening database for user {username}: password is None or empty"
920 )
921 # TODO: Fix the root cause - research threads are not getting the correct password
922 logger.error(
923 "TODO: This usually means the research thread is not receiving the user's "
924 "password for database encryption. Need to ensure password is passed from "
925 "the main thread to research threads."
926 )
927 raise ValueError(
928 "Invalid encryption key: password cannot be None or empty"
929 )
931 # Check if already open.
932 #
933 # `connections` is keyed by username ALONE, so a hit proves only that
934 # SOMEONE opened this user's database -- not that the caller supplied
935 # the right password. Returning the engine on a bare hit made login
936 # accept ANY password for any user with a live cached connection,
937 # which is the normal state whenever they are logged in anywhere.
938 # Verify the credential against the one that actually opened it, and
939 # fall through to a real (key-validating) open when it does not match.
940 # Look up the engine and check its verifier under ONE lock acquisition
941 # so the engine cannot be evicted between the lookup and the verify
942 # (which would hand back a disposed, de-registered engine).
943 with self._connections_lock:
944 cached = self.connections.get(username)
945 if cached is not None and self._cached_engine_trusted(
946 username, password
947 ):
948 return cached
950 # Serialize the cold-open (engine build + migration) per user so two
951 # concurrent first-opens never run alembic against the same database
952 # file at once -- alembic's module-level proxy and the version-row
953 # UPDATE are not safe under concurrent migration of one DB.
954 init_lock = self._get_init_lock(username)
955 with init_lock:
956 # Re-check: another thread may have completed the cold-open while
957 # we were waiting for the per-user lock. One lock acquisition, as
958 # above, so the engine cannot be evicted between lookup and verify.
959 with self._connections_lock:
960 cached = self.connections.get(username)
961 if cached is not None and self._cached_engine_trusted(
962 username, password
963 ):
964 return cached
965 engine = self._open_user_database_cold(
966 username, password, open_start
967 )
969 # Phase 2 of the RAG vector-store cutover runs OUTSIDE init_lock (only
970 # on the path that actually did the cold-open — the early returns above
971 # belong to a thread that already ran it). It re-enters
972 # open_user_database() via get_user_db_session on a session-cache miss;
973 # doing it while still holding the non-reentrant init_lock would
974 # self-deadlock if a concurrent close_user_database() evicted the
975 # freshly-cached connection between the cache write and that re-entry.
976 if engine is not None:
977 self._run_phase2_rekey(username, password)
978 return engine
980 def _run_phase2_rekey(self, username: str, password: str) -> None:
981 """Re-key a user's legacy FAISS indices (uuid-keyed -> DocumentChunk.id
982 -keyed), once, at login. See vector_stores/legacy_rekey.py.
984 A fast no-op (one get_setting call) once the per-user marker
985 (research_library.rag_index_rekeyed_v2) is set, or when the user has no
986 legacy sidecars — the common case. Wrapped so a re-key problem never
987 blocks login; it logs its own errors. Must be called AFTER the init_lock
988 is released (see open_user_database) to avoid a re-entrant self-deadlock.
989 """
990 try:
991 from ..vector_stores.legacy_rekey import rekey_user_indexes
993 rekey_user_indexes(username, password)
994 except Exception as e:
995 # NO rendered traceback + redacted message — the password-redaction
996 # invariant (tests/security/test_password_redaction_invariant.py).
997 # `password` (the unrecoverable SQLCipher master key) is live in this
998 # frame and every frame of the rekey chain, so ANY traceback render
999 # leaks it under LDR_LOGURU_DIAGNOSE. logger.opt(exception=True) does
1000 # NOT suppress that (it renders identically to logger.exception) and
1001 # also evades the AST guardrail, so log a redacted message with no
1002 # stack, matching every other handler in this file.
1003 from ..security.log_sanitizer import redact_secrets
1005 safe_msg = redact_secrets(str(e), password)
1006 logger.warning(
1007 f"RAG index re-key failed for user {username}: {safe_msg}"
1008 )
1010 def _open_user_database_cold(
1011 self, username: str, password: str, open_start: float
1012 ) -> Optional[Engine]:
1013 """Build the engine and run migrations for a not-yet-cached user DB.
1015 Must be called while holding this user's init lock (acquired in
1016 ``open_user_database``) so concurrent first-opens of the same user do
1017 not migrate one database file simultaneously.
1018 """
1019 db_path = self._get_user_db_path(username)
1021 # Prevent timing attacks: always derive key before checking file existence
1022 # This ensures both existing and non-existent users take the same amount of time,
1023 # preventing username enumeration via timing analysis.
1024 # Pre-derive key before closures to avoid capturing plaintext password
1025 hex_key = get_key_from_password(password, db_path=db_path).hex()
1027 if not db_path.exists():
1028 logger.error(f"No database found for user {username}")
1029 return None
1031 # Warn if this is a legacy database without per-database salt
1032 if self.has_encryption and not has_per_database_salt(db_path): 1032 ↛ 1033line 1032 didn't jump to line 1033 because the condition on line 1032 was never true
1033 logger.warning(
1034 f"Database for user '{username}' uses the legacy shared salt "
1035 f"(deprecated). For improved security, consider creating a new "
1036 f"account to get a per-database salt. Legacy databases remain "
1037 f"fully functional but are less resistant to multi-target attacks."
1038 )
1040 # Create connection string - use regular SQLite when SQLCipher not available
1041 if self.has_encryption:
1043 def create_open_connection():
1044 """Create a properly initialized SQLCipher connection."""
1045 return create_sqlcipher_connection(
1046 db_path,
1047 hex_key=hex_key,
1048 creation_mode=False,
1049 connect_kwargs={
1050 # DEFERRED (empty string) so pure-SELECT transactions
1051 # acquire only SQLite's SHARED lock, letting WAL-mode
1052 # concurrent readers proceed while a writer is active.
1053 # IMMEDIATE was previously set "defensively" and made
1054 # every transaction (even reads) take a RESERVED lock,
1055 # which was the single biggest contention source on
1056 # the login-hang path. Race-prone check-then-insert
1057 # call sites were made race-free at the application
1058 # layer in the preceding prerequisite PR.
1059 "isolation_level": "",
1060 "check_same_thread": False,
1061 },
1062 )
1064 # Create engine with custom creator function and optimized cache
1065 engine = create_engine(
1066 "sqlite://",
1067 creator=create_open_connection,
1068 poolclass=self._pool_class,
1069 echo=False,
1070 query_cache_size=1000,
1071 **self._get_pool_kwargs(),
1072 )
1073 else:
1074 logger.warning(
1075 f"SQLCipher not available - opening UNENCRYPTED database for user {username}"
1076 )
1077 # Fall back to regular SQLite (no password protection!)
1078 engine = create_engine(
1079 f"sqlite:///{db_path}",
1080 connect_args={"check_same_thread": False, "timeout": 30},
1081 poolclass=self._pool_class,
1082 echo=False,
1083 query_cache_size=1000,
1084 **self._get_pool_kwargs(),
1085 )
1087 # For unencrypted databases, just apply pragmas
1088 event.listen(engine, "connect", self._apply_pragmas)
1090 try:
1091 # Test connection by running a simple query
1092 with engine.connect() as conn:
1093 conn.execute(text("SELECT 1"))
1095 # Run database initialization (creates missing tables and runs migrations)
1096 from .initialize import initialize_database
1098 # Create backup before migration to protect against schema change failures
1099 from .alembic_runner import needs_migration
1101 if needs_migration(engine):
1102 try:
1103 from .backup.backup_service import BackupService
1105 result = BackupService(
1106 username=username, password=password
1107 ).create_backup(force=True)
1108 if result.success: 1108 ↛ 1109line 1108 didn't jump to line 1109 because the condition on line 1108 was never true
1109 logger.info(
1110 f"Pre-migration backup created: {result.backup_path}"
1111 )
1112 else:
1113 logger.error(
1114 f"Pre-migration backup failed: {result.error}"
1115 )
1116 except Exception as e:
1117 # ``password`` is in scope and was passed into
1118 # ``BackupService`` above. Drop traceback + redact
1119 # so the upstream error (which can carry the
1120 # password through ``set_sqlcipher_key`` frames) is
1121 # not written to log sinks.
1122 from ..security.log_sanitizer import redact_secrets
1124 safe_msg = redact_secrets(str(e), password)
1125 logger.warning(
1126 f"Pre-migration backup failed — proceeding with migration: {safe_msg}"
1127 )
1129 # Init failures need to be distinguishable from credential
1130 # failures at the call site: the credentials worked (we
1131 # decrypted and ran SELECT 1), but the schema couldn't be
1132 # brought up. Re-raise as a typed error so the login route
1133 # can skip the lockout counter and surface a server-error
1134 # message instead of "Invalid username or password".
1135 try:
1136 initialize_database(engine)
1137 except Exception as init_err:
1138 elapsed_ms = (time.perf_counter() - open_start) * 1000
1139 # ``password`` is in scope (function parameter). The
1140 # initializer ran on an engine whose creator captured
1141 # ``hex_key`` derived from the password — a SQLAlchemy
1142 # ``OperationalError`` traceback could surface frames
1143 # holding the plaintext password under ``diagnose=True``.
1144 from ..security.log_sanitizer import redact_secrets
1146 safe_msg = redact_secrets(str(init_err), password)
1147 logger.warning(
1148 f"Database migration failed for {username} "
1149 f"after {elapsed_ms:.0f}ms — refusing login: {safe_msg}"
1150 )
1151 engine.dispose()
1152 # Sanitised message + ``from None``: ``init_err`` (its
1153 # str() and its chained traceback frames) can carry the
1154 # plaintext password, and the caller that catches this
1155 # typed error logs it with ``logger.exception``
1156 # (thread_local_session) — which would render the chain
1157 # and defeat the redaction applied just above. Break the
1158 # chain per ADR-0003; the redacted detail is already
1159 # logged at this site.
1160 raise DatabaseInitializationError(
1161 f"Database initialisation failed for {username}: {safe_msg}"
1162 ) from None
1164 # Store connection AFTER migrations complete. This password just
1165 # proved itself against SQLCipher; publish the engine and its
1166 # verifier together so the cache is never briefly trusted without
1167 # one.
1168 self._cache_connection(username, engine, password)
1170 # NOTE: the phase-2 RAG index re-key does NOT run here. It re-enters
1171 # open_user_database() (via get_user_db_session on a session-cache
1172 # miss), and this method runs while holding the non-reentrant
1173 # per-user init_lock — so a concurrent close_user_database() evicting
1174 # the just-cached connection would make that re-entrant open miss the
1175 # fast-path and self-deadlock on the lock this thread already holds.
1176 # open_user_database() runs it AFTER releasing the lock instead.
1178 elapsed_ms = (time.perf_counter() - open_start) * 1000
1179 if elapsed_ms > 100:
1180 logger.info(
1181 f"Opened encrypted database for user {username} "
1182 f"(cold-open wall clock: {elapsed_ms:.0f}ms)"
1183 )
1184 else:
1185 logger.info(
1186 f"Opened encrypted database for user {username} "
1187 f"({elapsed_ms:.0f}ms)"
1188 )
1189 return engine
1191 except DatabaseInitializationError:
1192 # Already logged + engine disposed at the raise site. Re-raise
1193 # past the catch-all below so callers see the typed error.
1194 raise
1195 except Exception as e:
1196 elapsed_ms = (time.perf_counter() - open_start) * 1000
1197 # Catches connection errors during ``engine.connect()`` —
1198 # the connection creator passes the password into
1199 # ``set_sqlcipher_key`` so a failure there can carry the
1200 # password in the traceback's frame locals. Drop the
1201 # traceback and redact str(e).
1202 from ..security.log_sanitizer import redact_secrets
1204 safe_msg = redact_secrets(str(e), password)
1205 logger.warning(
1206 f"Failed to open database for user {username} "
1207 f"after {elapsed_ms:.0f}ms: {safe_msg}"
1208 )
1209 engine.dispose()
1210 return None
1212 def get_session(self, username: str) -> Optional[Session]:
1213 """Bind a Session to ``username``'s ALREADY-open engine. Post-auth accessor.
1215 SECURITY INVARIANT -- this intentionally takes NO password and performs
1216 NO credential check, unlike ``open_user_database`` /
1217 ``create_thread_safe_session_for_metrics`` (which gate their cache hit
1218 through ``_cached_engine_trusted``). That is correct ONLY because this
1219 is a post-authentication accessor, never an authentication entry point:
1221 * It NEVER opens a database -- a cache miss returns ``None``. The
1222 engine it hands back was published to ``connections`` solely by a
1223 credential-verified create/open, so no unverified password can ever
1224 reach the cache via this method.
1225 * Every caller supplies a server-trusted ``username`` -- the signed
1226 Flask-session identity (``session["username"]`` / ``g.current_user``)
1227 or an internal server value (settings-snapshot thread, log drain)
1228 -- and most also gate on ``is_user_connected`` (the never-open-on-
1229 miss property above is the load-bearing guarantee, not that gate).
1230 The username is never an attacker-supplied login attempt paired with
1231 an unverified password, which is the shape that made the cached-
1232 connection bypass exploitable (see ``open_user_database``).
1234 DO NOT add a lazy open-on-miss here, and DO NOT call this from an
1235 auth-decision path: either turns this accessor into an unauthenticated
1236 bypass. Credential-taking entry points must go through
1237 ``open_user_database`` instead.
1238 """
1239 with self._connections_lock:
1240 if username not in self.connections:
1241 # Use debug level for this common scenario to reduce log noise
1242 logger.debug(f"No open database for user {username}")
1243 return None
1244 engine = self.connections[username]
1245 # Create session inside lock to prevent race with close_user_database()
1246 SessionLocal = sessionmaker(bind=engine)
1247 return SessionLocal()
1249 def get_connected_usernames(self) -> set:
1250 """Return a snapshot of usernames with open connections."""
1251 with self._connections_lock:
1252 return set(self.connections.keys())
1254 def _checkpoint_wal(self, engine, context: str = ""):
1255 """Checkpoint WAL before disposing engine to flush pending writes."""
1256 try:
1257 with engine.connect() as conn:
1258 # TRUNCATE returns (busy, log_pages, checkpointed_pages).
1259 # busy=1 means a reader/writer held a WAL lock and the WAL
1260 # was NOT truncated — surface that so we can tell from logs
1261 # whether the helper actually shrank the file.
1262 result = conn.execute(
1263 text("PRAGMA wal_checkpoint(TRUNCATE)")
1264 ).fetchone()
1265 if result and result[0] == 1: 1265 ↛ 1266line 1265 didn't jump to line 1266 because the condition on line 1265 was never true
1266 logger.debug(
1267 f"WAL checkpoint busy {context} — WAL not truncated"
1268 )
1269 except Exception:
1270 logger.debug(f"WAL checkpoint failed {context}", exc_info=True)
1272 def close_user_database(self, username: str):
1273 """Close a user's database connection."""
1274 with self._connections_lock:
1275 if username in self.connections:
1276 try:
1277 self._checkpoint_wal(
1278 self.connections[username], f"for {username}"
1279 )
1280 self.connections[username].dispose()
1281 except Exception:
1282 logger.warning(
1283 f"Failed to dispose engine for {username}",
1284 )
1285 del self.connections[username]
1286 # Drop the verifier with the connection it described, so a
1287 # stale entry can never outlive the engine it was recorded for.
1288 self._password_verifiers.pop(username, None)
1289 # Deliberately do NOT pop _init_locks[username] here. A
1290 # concurrent open_user_database may already hold a reference to
1291 # this user's lock (fetched via _get_init_lock) and be about to
1292 # enter its cold-open; dropping the entry would let a later open
1293 # create a *second* lock for the same user, so two cold-opens
1294 # could migrate one DB file at once -- the very race this lock
1295 # exists to prevent. The dict is bounded by the number of
1296 # distinct usernames (one small Lock each), so retaining it is
1297 # cheap; close_all_databases clears it wholesale at shutdown.
1298 logger.info(f"Closed database for user {username}")
1300 def close_all_databases(self):
1301 """Close all open user database connections and release file locks."""
1302 with self._connections_lock:
1303 for username, engine in list(self.connections.items()):
1304 try:
1305 self._checkpoint_wal(engine, f"for {username}")
1306 engine.dispose()
1307 except Exception:
1308 logger.debug(f"Error disposing engine for {username}")
1309 self.connections.clear()
1310 self._password_verifiers.clear()
1311 self._init_locks.clear()
1313 def check_database_integrity(self, username: str) -> bool:
1314 """Check integrity of a user's encrypted database."""
1315 with self._connections_lock:
1316 if username not in self.connections:
1317 return False
1318 engine = self.connections[username]
1320 try:
1321 with engine.connect() as conn:
1322 # Quick integrity check
1323 result = conn.execute(text("PRAGMA quick_check"))
1324 if result.fetchone()[0] != "ok":
1325 return False
1327 # SQLCipher integrity check
1328 result = conn.execute(text("PRAGMA cipher_integrity_check"))
1329 # If this returns any rows, there are HMAC failures
1330 failures = list(result)
1331 if failures:
1332 logger.error(
1333 f"Integrity check failed for {username}: {len(failures)} HMAC failures"
1334 )
1335 return False
1337 return True
1339 except Exception as e:
1340 # No ``password`` parameter on this method and the engine
1341 # was created via a hex-key closure (the plaintext password
1342 # is not retained on the engine itself), so the traceback
1343 # frames in this method do not hold a credential. Still
1344 # drop the traceback to stay uniform with the rest of the
1345 # #4182 sweep and to guard against a caller frame that
1346 # happens to hold one (e.g., a route handler that
1347 # retrieved the password from the session store).
1348 from ..security.log_sanitizer import redact_secrets
1350 safe_msg = redact_secrets(str(e), None)
1351 logger.warning(
1352 f"Integrity check error for user: {username}: {safe_msg}"
1353 )
1354 return False
1356 def change_password(
1357 self, username: str, old_password: str, new_password: str
1358 ) -> bool:
1359 """Change the encryption password for a user's database.
1361 This rekeys the SQLCipher database — no separate auth-DB
1362 password-hash update is needed because passwords are never
1363 stored. Login verification is done by attempting decryption.
1364 """
1365 if not self.has_encryption:
1366 logger.warning(
1367 "Cannot change password - SQLCipher not available (databases are unencrypted)"
1368 )
1369 return False
1371 db_path = self._get_user_db_path(username)
1373 if not db_path.exists():
1374 return False
1376 try:
1377 # Close existing connection if any
1378 self.close_user_database(username)
1380 # Open with old password
1381 engine = self.open_user_database(username, old_password)
1382 if not engine:
1383 return False
1385 # Rekey the database (only works with SQLCipher)
1386 with engine.connect() as conn:
1387 # Use centralized rekey function
1388 set_sqlcipher_rekey(conn, new_password, db_path=db_path)
1390 # Evict the cached engine and its verifier NOW, while still inside
1391 # the try. open_user_database() above cold-opened with old_password
1392 # and cached that engine together with a verifier for old_password.
1393 # The rekey just invalidated that engine's key (its creator closure
1394 # still derives the OLD hex key, so any freshly pooled connection is
1395 # mis-keyed) yet the cached verifier still matches old_password -- so
1396 # until this eviction a concurrent open_user_database(old_password)
1397 # would pass the verifier check and be handed the stale-key engine.
1398 # change_password holds no lock across the rekey, so evict here
1399 # rather than waiting for the finally. close_user_database is
1400 # idempotent (no-ops when the user is not cached), so the
1401 # finally-close below stays a backstop and is a no-op once this ran.
1402 self.close_user_database(username)
1404 logger.info(f"Password changed for user {username}")
1405 return True
1407 except Exception as e:
1408 # Both ``old_password`` and ``new_password`` are in lexical
1409 # scope. ``open_user_database`` / ``set_sqlcipher_rekey``
1410 # carry these into nested frames — a traceback rendered
1411 # with ``diagnose=True`` would leak them. Redact both and
1412 # drop the traceback chain.
1413 from ..security.log_sanitizer import redact_secrets
1415 safe_msg = redact_secrets(str(e), old_password, new_password)
1416 logger.warning(
1417 f"Failed to change password for user: {username}: {safe_msg}"
1418 )
1419 return False
1420 finally:
1421 # Close the connection
1422 self.close_user_database(username)
1424 def user_exists(self, username: str) -> bool:
1425 """Check if a user exists in the auth database."""
1426 from .auth_db import auth_db_session
1427 from .models.auth import User
1429 with auth_db_session() as session:
1430 user = session.query(User).filter_by(username=username).first()
1431 return user is not None
1433 def get_memory_usage(self) -> Dict[str, Any]:
1434 """Get memory usage statistics."""
1435 with self._connections_lock:
1436 num_connections = len(self.connections)
1437 return {
1438 "active_connections": num_connections,
1439 "active_sessions": 0, # Sessions are created on-demand, not tracked
1440 "estimated_memory_mb": num_connections
1441 * 3.5, # ~3.5MB per connection
1442 }
1444 def create_thread_safe_session_for_metrics(
1445 self, username: str, password: str
1446 ):
1447 """
1448 Create a new database session safe for use in background threads.
1450 Previously this method created a dedicated NullPool engine per
1451 (username, thread_id) pair, which leaked file descriptors under
1452 load (SQLCipher + WAL holds 3 FDs per active connection and
1453 orphaned engines accumulated when @thread_cleanup did not fire).
1455 It now routes through the shared per-user QueuePool engine at
1456 ``self.connections[username]``. That engine is already created
1457 with ``check_same_thread=False`` (so background threads are
1458 safe), is bounded by ``pool_size + max_overflow``, and is
1459 subject to the periodic ``dispose()`` workaround in
1460 ``connection_cleanup.py`` that mitigates the SQLCipher+WAL
1461 out-of-order-close FD leak.
1463 Args:
1464 username: The username
1465 password: The user's password (encryption key). Verified against
1466 the cached connection's recorded verifier on a cache hit, and
1467 used to open the user database on a cache miss or mismatch.
1469 Returns:
1470 A SQLAlchemy Session bound to the per-user QueuePool engine.
1471 """
1472 db_path = self._get_user_db_path(username)
1474 if not db_path.exists():
1475 raise ValueError(f"No database found for user {username}")
1477 # Trust the cached engine only when the presented password matches the
1478 # one that opened it -- the same fail-closed rule as
1479 # open_user_database. Returning a session on a bare cache hit here
1480 # would be a second instance of exactly the cached-connection password
1481 # bypass this change closes; falling through to open_user_database on a
1482 # mismatch re-validates the key against SQLCipher.
1483 with self._connections_lock:
1484 engine = self.connections.get(username)
1485 if engine is not None and not self._cached_engine_trusted(
1486 username, password
1487 ):
1488 engine = None
1490 if engine is None:
1491 # Cache miss (or a rejected cache hit) — open the user database.
1492 # This is idempotent: after the first call it just returns the
1493 # cached engine.
1494 engine = self.open_user_database(username, password)
1495 if engine is None: 1495 ↛ 1499line 1495 didn't jump to line 1499 because the condition on line 1495 was always true
1496 raise ValueError(f"Failed to open database for user {username}")
1498 # Use SQLAlchemy's default expire_on_commit=True.
1499 Session = sessionmaker(bind=engine)
1500 return Session()
1503# Global instance
1504db_manager = DatabaseManager()