Coverage for src/local_deep_research/database/sqlcipher_utils.py: 96%
184 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2SQLCipher utility functions for consistent database operations.
4This module centralizes all SQLCipher-specific operations to ensure
5consistent password handling and PRAGMA settings across the codebase.
6"""
8import os
9import secrets
10import threading
11import time
12from hashlib import pbkdf2_hmac
13from pathlib import Path
14from typing import Any, Dict, Optional, Union
16from loguru import logger
18from ..settings.env_registry import get_env_setting
19from ..utilities.type_utils import to_bool
21# Lock to protect cipher_default_* global state during creation
22_cipher_default_lock = threading.Lock()
24# Salt file constants
25SALT_FILE_SUFFIX = ".salt"
26SALT_SIZE = 32 # 256 bits
29def get_salt_file_path(db_path: Union[str, Path]) -> Path:
30 """
31 Get the path to the salt file for a database.
33 Args:
34 db_path: Path to the database file
36 Returns:
37 Path to the corresponding .salt file
38 """
39 return Path(db_path).with_suffix(Path(db_path).suffix + SALT_FILE_SUFFIX)
42def get_salt_for_database(db_path: Union[str, Path]) -> bytes:
43 """
44 Get the salt for a database file.
46 For new databases (v2+): reads from the .salt file alongside the database.
47 For legacy databases (v1): returns LEGACY_PBKDF2_SALT for backwards compatibility.
49 Args:
50 db_path: Path to the database file
52 Returns:
53 The salt bytes to use for key derivation
54 """
55 salt_file = get_salt_file_path(db_path)
57 try:
58 salt = salt_file.read_bytes()
59 except FileNotFoundError:
60 # v1: Legacy salt for backwards compatibility
61 logger.warning(
62 f"Database '{Path(db_path).name}' uses the legacy shared salt "
63 f"(deprecated). Consider creating a new database to benefit from "
64 f"per-database salt security. See issue #1439 for migration details."
65 )
66 return LEGACY_PBKDF2_SALT
68 # v2: Per-database random salt
69 if len(salt) != SALT_SIZE:
70 raise ValueError(
71 f"Salt file {salt_file} has unexpected size ({len(salt)} bytes), "
72 f"expected {SALT_SIZE}. The salt file may be corrupted."
73 )
74 return salt
77def create_database_salt(db_path: Union[str, Path]) -> bytes:
78 """
79 Create and store a new random salt for a database.
81 This should be called when creating a new database.
82 The salt is stored in a .salt file alongside the database.
84 WARNING: If this salt file is deleted, the associated database becomes
85 permanently unreadable. Always back up .salt files alongside their .db files.
87 Args:
88 db_path: Path to the database file
90 Returns:
91 The newly generated salt bytes
93 Raises:
94 FileExistsError: If a salt file already exists for this database
95 """
96 salt_file = get_salt_file_path(db_path)
98 if salt_file.exists():
99 raise FileExistsError(
100 f"Salt file already exists: {salt_file}. "
101 f"Refusing to overwrite to prevent data loss."
102 )
104 salt = secrets.token_bytes(SALT_SIZE)
106 # Ensure parent directory exists.
107 # Lazy import: this runs during early database bootstrap, so avoid a
108 # top-level security -> ... -> database import cycle.
109 from ..security.directory_creation import create_directory
111 create_directory(salt_file.parent, context="database salt file directory")
113 # Write salt file with owner-only permissions (0o600)
114 fd = os.open(str(salt_file), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
115 try:
116 os.write(fd, salt)
117 finally:
118 os.close(fd)
120 logger.info(f"Created new database salt file: {salt_file}")
121 return salt
124def has_per_database_salt(db_path: Union[str, Path]) -> bool:
125 """
126 Check if a database has a per-database salt file (v2).
128 Args:
129 db_path: Path to the database file
131 Returns:
132 True if the database has a .salt file, False otherwise
133 """
134 return get_salt_file_path(db_path).exists()
137def _get_key_from_password(
138 password: str, salt: bytes, kdf_iterations: int
139) -> bytes:
140 """
141 Generates an encryption key from the user's password and salt.
143 Args:
144 password: The password.
145 salt: The salt bytes to use for key derivation.
146 kdf_iterations: Number of PBKDF2 iterations.
148 Returns:
149 The generated key.
150 """
151 logger.debug(
152 f"Generating DB encryption key with {kdf_iterations} iterations..."
153 )
155 start = time.perf_counter()
156 key = pbkdf2_hmac(
157 "sha512",
158 password.encode(),
159 salt,
160 kdf_iterations,
161 )
162 elapsed_ms = (time.perf_counter() - start) * 1000
164 if elapsed_ms > 500:
165 logger.info(
166 f"PBKDF2 key derivation took {elapsed_ms:.0f}ms "
167 f"({kdf_iterations} iterations)"
168 )
169 else:
170 logger.debug(
171 f"PBKDF2 key derivation took {elapsed_ms:.0f}ms "
172 f"({kdf_iterations} iterations)"
173 )
174 return key
177def get_key_from_password(
178 password: str, db_path: Optional[Union[str, Path]] = None
179) -> bytes:
180 """
181 Wrapper that gets salt and settings, then calls the key derivation.
183 Args:
184 password: The password.
185 db_path: Optional path to the database file. If provided, uses
186 per-database salt. If not provided, uses legacy salt.
188 Returns:
189 The derived encryption key bytes.
190 """
191 if db_path is not None:
192 salt = get_salt_for_database(db_path)
193 else:
194 salt = LEGACY_PBKDF2_SALT
196 settings = get_sqlcipher_settings()
197 return _get_key_from_password(password, salt, settings["kdf_iterations"])
200def set_sqlcipher_key(
201 cursor_or_conn: Any,
202 password: str,
203 db_path: Optional[Union[str, Path]] = None,
204) -> None:
205 """
206 Set the SQLCipher encryption key using hexadecimal encoding.
208 This avoids SQL injection and escaping issues with special characters.
210 Args:
211 cursor_or_conn: SQLCipher cursor or connection object
212 password: The password to use for encryption
213 db_path: Optional path to the database file. If provided, uses
214 per-database salt. If not provided, uses legacy salt.
215 """
216 key = get_key_from_password(password, db_path=db_path) # gitleaks:allow
217 cursor_or_conn.execute(f"PRAGMA key = \"x'{key.hex()}'\"")
220def set_sqlcipher_key_from_hex(cursor_or_conn: Any, hex_key: str) -> None:
221 """
222 Set the SQLCipher encryption key from a pre-derived hex key string.
224 Used by connection closures to avoid capturing plaintext passwords.
226 Args:
227 cursor_or_conn: SQLCipher cursor or connection object
228 hex_key: Pre-derived hex key string (from get_key_from_password().hex())
229 """
230 cursor_or_conn.execute(f"PRAGMA key = \"x'{hex_key}'\"") # gitleaks:allow
233def set_sqlcipher_rekey(
234 cursor_or_conn: Any,
235 new_password: str,
236 db_path: Optional[Union[str, Path]] = None,
237) -> None:
238 """
239 Change the SQLCipher encryption key using hexadecimal encoding.
241 Uses the same PBKDF2 key derivation as set_sqlcipher_key() to ensure
242 consistency when re-opening databases after password change.
244 Args:
245 cursor_or_conn: SQLCipher cursor or connection object
246 new_password: The new password to use for encryption
247 db_path: Optional path to the database file. If provided, uses
248 per-database salt. If not provided, uses legacy salt.
249 """
250 # Use the same key derivation as set_sqlcipher_key for consistency
251 key = get_key_from_password(new_password, db_path=db_path) # gitleaks:allow
253 # The hex encoding already prevents injection since it only contains [0-9a-f]
254 safe_sql = f"PRAGMA rekey = \"x'{key.hex()}'\""
256 try:
257 # Try SQLAlchemy connection (needs text() wrapper)
258 from sqlalchemy import text
260 cursor_or_conn.execute(text(safe_sql))
261 except TypeError:
262 # Raw SQLCipher connection - use string directly
263 cursor_or_conn.execute(safe_sql)
266# Default SQLCipher configuration (can be overridden by settings)
267DEFAULT_KDF_ITERATIONS = 256000
268DEFAULT_PAGE_SIZE = 16384 # 16KB pages for maximum performance with caching
269DEFAULT_HMAC_ALGORITHM = "HMAC_SHA512"
270DEFAULT_KDF_ALGORITHM = "PBKDF2_HMAC_SHA512"
272# Valid page sizes (powers of 2 within the SQLite range).
273# IntegerSetting validates min/max but not that the value is a power of 2,
274# so we check against this set as an additional safeguard.
275VALID_PAGE_SIZES = frozenset({512, 1024, 2048, 4096, 8192, 16384, 32768, 65536})
276MAX_KDF_ITERATIONS = 1_000_000
278# Production minimum KDF iterations. Relaxed automatically in test/CI environments.
279MIN_KDF_ITERATIONS_PRODUCTION = 100_000
280MIN_KDF_ITERATIONS_TESTING = 1
283def _get_min_kdf_iterations() -> int:
284 """Get minimum KDF iterations, relaxed for test/CI environments.
286 Only relaxes when PYTEST_CURRENT_TEST (set automatically by pytest) or
287 LDR_TEST_MODE (project-specific) is set. Generic env vars like CI or
288 TESTING are NOT checked to avoid accidentally weakening production
289 encryption in Docker/CD pipelines that set CI=true.
291 PYTEST_CURRENT_TEST is presence-based: pytest sets it to a descriptive
292 non-boolean string (e.g. "tests/foo.py::test_bar (call)"), so its mere
293 presence signals a test run. LDR_TEST_MODE is parsed as a proper boolean
294 so an explicit LDR_TEST_MODE=0 / =false does NOT relax the floor — a bare
295 truthiness check would treat any non-empty string (including "false") as
296 enabled and silently weaken encryption for someone trying to disable it.
297 """
298 is_testing = bool(os.environ.get("PYTEST_CURRENT_TEST")) or to_bool(
299 os.environ.get("LDR_TEST_MODE")
300 )
301 return (
302 MIN_KDF_ITERATIONS_TESTING
303 if is_testing
304 else MIN_KDF_ITERATIONS_PRODUCTION
305 )
308# Legacy salt for backwards compatibility with databases created before v2.
309# New databases use per-database random salts stored in .salt files.
310# WARNING: Do NOT change this value - it would break all existing legacy databases!
311LEGACY_PBKDF2_SALT = b"no salt"
313# Alias for backwards compatibility with code that references the old name
314PBKDF2_PLACEHOLDER_SALT = LEGACY_PBKDF2_SALT
317def get_sqlcipher_settings() -> dict:
318 """
319 Get SQLCipher settings from environment variables or use defaults.
321 These settings cannot be changed after database creation, so they
322 must be configured via environment variables only.
324 Settings are read via the env settings registry, which handles
325 canonical env var names (LDR_DB_CONFIG_*) with automatic fallback
326 to deprecated names (LDR_DB_*) and deprecation warnings.
328 Returns:
329 Dictionary with SQLCipher configuration
330 """
331 # HMAC algorithm - registry validates against allowed values
332 hmac_algorithm = get_env_setting(
333 "db_config.hmac_algorithm", DEFAULT_HMAC_ALGORITHM
334 )
336 # KDF algorithm - registry validates against allowed values
337 kdf_algorithm = get_env_setting(
338 "db_config.kdf_algorithm", DEFAULT_KDF_ALGORITHM
339 )
341 # Page size - registry validates range, we also check power-of-2
342 page_size = get_env_setting("db_config.page_size", DEFAULT_PAGE_SIZE)
343 if page_size not in VALID_PAGE_SIZES:
344 logger.warning(
345 f"Invalid page_size value '{page_size}', using default "
346 f"'{DEFAULT_PAGE_SIZE}'. Valid values: {sorted(VALID_PAGE_SIZES)}"
347 )
348 page_size = DEFAULT_PAGE_SIZE
350 # KDF iterations - registry validates basic range, then apply CI-aware minimum
351 kdf_iterations = get_env_setting(
352 "db_config.kdf_iterations", DEFAULT_KDF_ITERATIONS
353 )
354 min_kdf = _get_min_kdf_iterations()
355 if not (min_kdf <= kdf_iterations <= MAX_KDF_ITERATIONS):
356 logger.warning(
357 f"KDF iterations value '{kdf_iterations}' outside safe range "
358 f"[{min_kdf}, {MAX_KDF_ITERATIONS}], using default "
359 f"'{DEFAULT_KDF_ITERATIONS}'."
360 )
361 kdf_iterations = DEFAULT_KDF_ITERATIONS
363 return {
364 "kdf_iterations": kdf_iterations,
365 "page_size": page_size,
366 "hmac_algorithm": hmac_algorithm,
367 "kdf_algorithm": kdf_algorithm,
368 }
371def warn_if_weak_kdf_with_existing_databases(
372 data_dir: Union[str, Path],
373) -> bool:
374 """Warn loudly when the effective SQLCipher KDF is below the production
375 floor *and* user databases already exist in ``data_dir``.
377 The KDF iteration count is derived from the environment at open time
378 (see :func:`get_sqlcipher_settings`) and is NOT stored with the
379 database. So if the floor is relaxed (test mode) on a deployment that
380 already holds real data, newly created databases get a weak at-rest work
381 factor, and — separately — any existing database that was created at a
382 *higher* KDF can no longer be opened: its on-disk key is unchanged, but
383 the server now derives a different, weaker key, so decryption fails and
384 login returns a generic "Invalid username or password" 401 that is
385 indistinguishable from a wrong password (the KDF-mismatch symptom class
386 behind PR #4775).
388 Scope: this only catches the "server is now weak" direction, and only a
389 *sub-floor* effective KDF. It cannot detect the inverse (a production-KDF
390 server opening databases created weak) or general in-range KDF drift,
391 because the creation-time KDF is not persisted to compare against — only
392 a sub-floor effective KDF is observable from the environment alone.
394 Intentionally silent on a fresh deployment (no pre-existing databases →
395 nothing to mismatch) and when the effective KDF is at/above the floor
396 (e.g. the 256000 default) — that is harmless.
398 Returns ``True`` if a warning was emitted (used by tests).
399 """
400 effective_kdf = get_sqlcipher_settings()["kdf_iterations"]
401 if effective_kdf >= MIN_KDF_ITERATIONS_PRODUCTION:
402 return False
404 # Only count + truthiness are used, so no need to sort the glob result.
405 existing = list(Path(data_dir).glob("ldr_user_*.db"))
406 if not existing:
407 return False
409 logger.warning(
410 "SQLCipher KDF is configured to {} iterations — below the production "
411 "floor of {} (reached only in test mode, e.g. LDR_TEST_MODE) — while "
412 "{} user database(s) already exist in {}. New databases created now "
413 "get this weak at-rest work factor. Separately, any of those existing "
414 "databases that was created at a higher KDF can no longer be opened: "
415 "its on-disk key is unchanged, but the server now derives a different, "
416 "weaker key, so decryption fails and login returns a generic 'Invalid "
417 "username or password' 401 for every affected user (the KDF-mismatch "
418 "symptom class behind PR #4775). If this is a production deployment, "
419 "unset LDR_TEST_MODE (and any low LDR_DB_CONFIG_KDF_ITERATIONS).",
420 effective_kdf,
421 MIN_KDF_ITERATIONS_PRODUCTION,
422 len(existing),
423 data_dir,
424 )
425 return True
428def apply_cipher_defaults_before_key(
429 cursor_or_conn: Any,
430) -> None:
431 """
432 Apply cipher_default_* pragmas BEFORE PRAGMA key for new database creation.
434 Per SQLCipher 4.x docs, cipher_default_* pragmas set the defaults that
435 apply when a key is set on a NEW database. These MUST be called before
436 PRAGMA key.
438 For EXISTING databases, cipher_page_size/cipher_hmac_algorithm/
439 cipher_kdf_algorithm are set AFTER the key via apply_sqlcipher_pragmas().
441 Args:
442 cursor_or_conn: SQLCipher cursor or connection object
443 """
444 settings = get_sqlcipher_settings()
446 logger.debug(
447 f"Applying cipher_default_* pragmas for new DB: settings={settings}"
448 )
450 cursor_or_conn.execute(
451 f"PRAGMA cipher_default_page_size = {settings['page_size']}"
452 )
453 cursor_or_conn.execute(
454 f"PRAGMA cipher_default_hmac_algorithm = {settings['hmac_algorithm']}"
455 )
456 cursor_or_conn.execute(
457 f"PRAGMA cipher_default_kdf_algorithm = {settings['kdf_algorithm']}"
458 )
461def apply_sqlcipher_pragmas(
462 cursor_or_conn: Any,
463 creation_mode: bool = False,
464) -> None:
465 """
466 Apply SQLCipher PRAGMA settings that are set AFTER the key.
468 For SQLCipher 4.x:
469 - New databases: cipher_default_* are set before key via
470 apply_cipher_defaults_before_key(). This function only sets kdf_iter.
471 - Existing databases: cipher_page_size, cipher_hmac_algorithm,
472 cipher_kdf_algorithm MUST be set AFTER the key (not before).
473 This function handles that.
475 Args:
476 cursor_or_conn: SQLCipher cursor or connection object
477 creation_mode: If True, only sets kdf_iter (defaults already applied).
478 If False, sets cipher_* settings + kdf_iter for existing DB.
479 """
480 settings = get_sqlcipher_settings()
482 if not creation_mode:
483 # For existing databases: cipher_* pragmas go AFTER the key
484 cursor_or_conn.execute(
485 f"PRAGMA cipher_page_size = {settings['page_size']}"
486 )
487 cursor_or_conn.execute(
488 f"PRAGMA cipher_hmac_algorithm = {settings['hmac_algorithm']}"
489 )
490 cursor_or_conn.execute(
491 f"PRAGMA cipher_kdf_algorithm = {settings['kdf_algorithm']}"
492 )
494 # kdf_iter can be set after the key (applies to future derivation)
495 cursor_or_conn.execute(f"PRAGMA kdf_iter = {settings['kdf_iterations']}")
497 # cipher_memory_security is a runtime PRAGMA. ON zeroes SQLCipher buffers
498 # and calls mlock() to prevent swap; OFF skips this. Defaulting to OFF
499 # because the password already sits unprotected in Flask session, db_manager,
500 # and thread-local storage — mlock on SQLCipher's buffers alone doesn't help.
501 # Users can opt in with LDR_DB_CONFIG_CIPHER_MEMORY_SECURITY=ON + IPC_LOCK.
502 # Applied on every connection (not just creation) so env var overrides work.
503 mem_security = get_env_setting("db_config.cipher_memory_security", "OFF")
504 # Defense-in-depth: the env registry's EnumSetting already restricts
505 # this to {ON, OFF}, but an f-string interpolation into PRAGMA is a
506 # SQL-injection primitive if that registry check is ever bypassed
507 # (e.g. a refactor that swaps `get_env_setting` for raw `os.getenv`
508 # in a future patch). Assert explicitly so the call site fails loud
509 # instead of becoming an injection vector.
510 if mem_security not in ("ON", "OFF"): 510 ↛ 511line 510 didn't jump to line 511 because the condition on line 510 was never true
511 raise ValueError(
512 f"cipher_memory_security must be ON or OFF, got: {mem_security!r}"
513 )
514 cursor_or_conn.execute(f"PRAGMA cipher_memory_security = {mem_security}")
517def apply_performance_pragmas(cursor_or_conn: Any) -> None:
518 """
519 Apply performance-related PRAGMA settings from environment variables.
521 Settings are read via the env settings registry, which handles
522 canonical env var names (LDR_DB_CONFIG_*) with automatic fallback
523 to deprecated names (LDR_DB_*) and deprecation warnings.
525 Args:
526 cursor_or_conn: SQLCipher cursor or connection object
527 """
528 # Default values that are always applied
529 cursor_or_conn.execute("PRAGMA temp_store = MEMORY")
530 cursor_or_conn.execute("PRAGMA busy_timeout = 10000") # 10 second timeout
532 # SQLite defaults foreign_keys to OFF — without this every
533 # ondelete="CASCADE"/"SET NULL" declared on an FK is inert,
534 # including for raw-SQL DELETEs issued by Query.delete().
535 cursor_or_conn.execute("PRAGMA foreign_keys = ON")
537 # Cache size - registry validates min/max range. Force int before
538 # f-string interpolation so a non-numeric value surfaces as TypeError
539 # at this point rather than becoming an injection vector downstream.
540 cache_mb = int(get_env_setting("db_config.cache_size_mb", 64))
541 cache_pages = -(cache_mb * 1024) # Negative for KB cache size
542 cursor_or_conn.execute(f"PRAGMA cache_size = {cache_pages}")
544 # Journal mode - registry validates against allowed values. Belt-and-
545 # braces: assert the allowlist explicitly at the f-string call site.
546 journal_mode = get_env_setting("db_config.journal_mode", "WAL")
547 if journal_mode not in ( 547 ↛ 555line 547 didn't jump to line 555 because the condition on line 547 was never true
548 "DELETE",
549 "TRUNCATE",
550 "PERSIST",
551 "MEMORY",
552 "WAL",
553 "OFF",
554 ):
555 raise ValueError(f"Invalid journal_mode: {journal_mode!r}")
556 cursor_or_conn.execute(f"PRAGMA journal_mode = {journal_mode}")
558 # Synchronous mode - same defense-in-depth pattern.
559 sync_mode = get_env_setting("db_config.synchronous", "NORMAL")
560 if sync_mode not in ("OFF", "NORMAL", "FULL", "EXTRA"): 560 ↛ 561line 560 didn't jump to line 561 because the condition on line 560 was never true
561 raise ValueError(f"Invalid synchronous mode: {sync_mode!r}")
562 cursor_or_conn.execute(f"PRAGMA synchronous = {sync_mode}")
564 # WAL autocheckpoint frame threshold. SQLite's default of 1000 frames
565 # paired with our 16 KB page size means the WAL can grow to ~16 MB
566 # before SQLite triggers a PASSIVE checkpoint at commit. Lowering the
567 # threshold bounds the WAL high-water-mark on disk for users who never
568 # log out (the explicit TRUNCATE checkpoint runs on dispose, not here).
569 wal_autocheckpoint = get_env_setting("db_config.wal_autocheckpoint", 250)
570 cursor_or_conn.execute(f"PRAGMA wal_autocheckpoint = {wal_autocheckpoint}")
573def verify_sqlcipher_connection(cursor_or_conn: Any) -> bool:
574 """
575 Verify that the SQLCipher connection is working correctly.
577 Args:
578 cursor_or_conn: SQLCipher cursor or connection object
580 Returns:
581 True if the connection is valid, False otherwise
582 """
583 try:
584 cursor_or_conn.execute("SELECT 1")
585 result = (
586 cursor_or_conn.fetchone()
587 if hasattr(cursor_or_conn, "fetchone")
588 else cursor_or_conn.execute("SELECT 1").fetchone()
589 )
590 is_valid = result == (1,)
591 if not is_valid:
592 logger.error(
593 f"SQLCipher verification failed: result {result} != (1,)"
594 )
595 return is_valid
596 except Exception:
597 logger.exception("SQLCipher verification failed")
598 return False
601def get_sqlcipher_version(cursor_or_conn: Any) -> Optional[str]:
602 """
603 Get the SQLCipher version string.
605 Args:
606 cursor_or_conn: SQLCipher cursor or connection object
608 Returns:
609 Version string (e.g. "4.6.1 community") or None if unavailable
610 """
611 try:
612 cursor_or_conn.execute("PRAGMA cipher_version")
613 result = cursor_or_conn.fetchone()
614 return result[0] if result else None
615 except Exception:
616 logger.debug("Could not query SQLCipher version", exc_info=True)
617 return None
620def create_sqlcipher_connection(
621 db_path: Union[str, Path],
622 password: Optional[str] = None,
623 creation_mode: bool = False,
624 connect_kwargs: Optional[Dict[str, Any]] = None,
625 hex_key: Optional[str] = None,
626) -> Any:
627 """
628 Create a properly configured SQLCipher connection.
630 Implements the full PRAGMA sequence with proper error cleanup:
631 - Creation: cipher_default_* -> key -> kdf_iter -> performance -> verify
632 - Existing: key -> cipher_* + kdf_iter -> performance -> verify
634 Uses per-database salt if a .salt file exists alongside the database,
635 otherwise falls back to legacy salt for backwards compatibility.
637 Args:
638 db_path: Path to the database file
639 password: The password for encryption (mutually exclusive with hex_key)
640 creation_mode: If True, set cipher_default_* before key (new DB)
641 connect_kwargs: Extra kwargs passed to sqlcipher3.connect()
642 hex_key: Pre-derived hex key (skips PBKDF2 derivation)
644 Returns:
645 SQLCipher connection object
647 Raises:
648 ImportError: If sqlcipher3 is not available
649 ValueError: If the connection cannot be established
650 """
651 from .sqlcipher_compat import get_sqlcipher_module
653 try:
654 sqlcipher3 = get_sqlcipher_module()
655 except ImportError:
656 raise ImportError(
657 "sqlcipher3 is not available for encrypted databases. "
658 "Ensure SQLCipher system library is installed, then run: pdm install"
659 )
661 conn = sqlcipher3.connect(str(db_path), **(connect_kwargs or {}))
662 try:
663 cursor = conn.cursor()
665 if creation_mode:
666 with _cipher_default_lock:
667 apply_cipher_defaults_before_key(cursor)
669 # Set encryption key (uses per-database salt when password + db_path)
670 if hex_key:
671 set_sqlcipher_key_from_hex(cursor, hex_key)
672 elif password: 672 ↛ 675line 672 didn't jump to line 675 because the condition on line 672 was always true
673 set_sqlcipher_key(cursor, password, db_path=db_path)
674 else:
675 raise ValueError("Either password or hex_key must be provided") # noqa: TRY301 — except does connection cleanup before re-raise
677 # Apply post-key pragmas (cipher_* for existing, kdf_iter for both)
678 apply_sqlcipher_pragmas(cursor, creation_mode=creation_mode)
680 # Apply performance settings
681 apply_performance_pragmas(cursor)
683 # Verify connection works
684 if not verify_sqlcipher_connection(cursor):
685 raise ValueError( # noqa: TRY301 — except does connection cleanup before re-raise
686 "Failed to establish encrypted database connection"
687 )
689 cursor.close()
690 return conn
691 except Exception:
692 from ..utilities.resource_utils import safe_close
694 safe_close(conn, "SQLCipher connection")
695 raise
698# Backwards compatibility alias — old name still importable
699apply_cipher_settings_before_key = apply_cipher_defaults_before_key