Coverage for src / local_deep_research / database / encryption_check.py: 98%

1""" 

2Check for SQLCipher availability and provide fallback options. 

3""" 

4 

5import warnings 

6from typing import Optional, Tuple 

7 

8from loguru import logger 

9 

10 

11def check_sqlcipher_available() -> Tuple[bool, Optional[str]]: 

12 """ 

13 Check if SQLCipher is available for use. 

14 

15 Returns: 

16 Tuple of (is_available, error_message) 

17 """ 

18 try: 

19 # Try to import sqlcipher3 (covers both sqlcipher3 and sqlcipher3-binary) 

20 import importlib.util 

21 

22 if importlib.util.find_spec("sqlcipher3") is not None: 

23 return True, None 

24 except ImportError: 

25 pass 

26 

27 # Check if sqlcipher command is available 

28 import subprocess 

29 

30 try: 

31 result = subprocess.run( 

32 ["sqlcipher", "--version"], 

33 capture_output=True, 

34 text=True, 

35 timeout=5, 

36 ) 

37 if result.returncode == 0: 37 ↛ 45line 37 didn't jump to line 45 because the condition on line 37 was always true

38 return ( 

39 False, 

40 "SQLCipher is installed but Python bindings are missing. Reinstall the project with: pdm install", 

41 ) 

42 except (subprocess.TimeoutExpired, FileNotFoundError): 

43 pass 

44 

45 return ( 

46 False, 

47 "SQLCipher is not installed. See docs/SQLCIPHER_INSTALL.md for installation instructions.", 

48 ) 

49 

50 

51def warn_if_no_encryption(): 

52 """Warn user if encryption is not available.""" 

53 available, message = check_sqlcipher_available() 

54 

55 if not available: 

56 warning_msg = ( 

57 "\n" + "=" * 60 + "\n" 

58 "⚠️ SECURITY WARNING: Database Encryption Not Available\n" 

59 + "=" 

60 * 60 

61 + "\n" 

62 f"{message}\n\n" 

63 "Your data will be stored in UNENCRYPTED SQLite databases.\n" 

64 "This means:\n" 

65 " - API keys are stored in plain text\n" 

66 " - Research data is not encrypted at rest\n" 

67 " - Anyone with file system access can read your data\n\n" 

68 "For production use, we strongly recommend installing SQLCipher.\n" 

69 + "=" 

70 * 60 

71 ) 

72 

73 warnings.warn(warning_msg, category=UserWarning, stacklevel=2) 

74 logger.warning( 

75 "Running without database encryption - SQLCipher not available" 

76 ) 

77 

78 return available 

79 

80 

81def get_connection_string(db_path: str, password: Optional[str] = None) -> str: 

82 """ 

83 Get the appropriate connection string based on SQLCipher availability. 

84 

85 Args: 

86 db_path: Path to the database file 

87 password: Password for encryption (ignored if SQLCipher not available) 

88 

89 Returns: 

90 SQLAlchemy connection string 

91 """ 

92 available, _ = check_sqlcipher_available() 

93 

94 if available and password: 

95 # Use encrypted connection 

96 return f"sqlite+pysqlcipher://:{password}@/{db_path}" 

97 else: 

98 # Fall back to regular SQLite 

99 if password: 

100 logger.warning( 

101 "Password provided but SQLCipher not available - using unencrypted database" 

102 ) 

103 return f"sqlite:///{db_path}"