Coverage for src/local_deep_research/config/paths.py: 100%

59 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-09-06 15:42 +0000

1""" 

2Centralized path configuration for Local Deep Research. 

3Handles database location using platformdirs for proper user data storage. 

4""" 

5 

6import hashlib 

7import os 

8from pathlib import Path 

9 

10import platformdirs 

11from loguru import logger 

12 

13 

14def get_data_directory() -> Path: 

15 """ 

16 Get the appropriate data directory for storing application data. 

17 Uses platformdirs to get platform-specific user data directory. 

18 

19 Environment variable: 

20 LDR_DATA_DIR: Override the default data directory location. 

21 All subdirectories (research_outputs, cache, logs, database) 

22 will be created under this directory. 

23 

24 Returns: 

25 Path to data directory 

26 """ 

27 # Check for explicit override via environment variable 

28 custom_path = os.getenv("LDR_DATA_DIR") 

29 if custom_path: 

30 if not Path(custom_path).is_absolute(): 

31 raise ValueError("LDR_DATA_DIR must be an absolute path") 

32 data_dir = Path(custom_path).resolve() 

33 # Reject only control characters that are never valid in a path and 

34 # could break a SQL/ATTACH statement (null byte, newline, carriage 

35 # return). Quotes are intentionally NOT rejected here: legitimate POSIX 

36 # home directories contain apostrophes (e.g. /home/O'Brien/ldr), and the 

37 # actual ATTACH-DATABASE sink (backup_service) already validates the 

38 # full interpolated path against a stricter denylist before use. 

39 # The offending path is not echoed, mirroring the redaction below. 

40 path_str = str(data_dir) 

41 if any(c in path_str for c in ("\0", "\n", "\r")): 

42 raise ValueError( 

43 "LDR_DATA_DIR contains unsafe control characters " 

44 "(null byte, newline, or carriage return)" 

45 ) 

46 logger.debug( 

47 f"Using custom data directory from LDR_DATA_DIR: {data_dir}" 

48 ) 

49 return data_dir 

50 

51 # Use platformdirs for platform-specific user data directory 

52 # Windows: C:\Users\Username\AppData\Local\local-deep-research 

53 # macOS: ~/Library/Application Support/local-deep-research 

54 # Linux: ~/.local/share/local-deep-research 

55 data_dir = Path(platformdirs.user_data_dir("local-deep-research")) 

56 # Log only the directory pattern, not the full path which may contain username 

57 logger.debug( 

58 f"Using platformdirs data directory pattern: .../{data_dir.name}" 

59 ) 

60 

61 return data_dir 

62 

63 

64def _ensure_dir(subdir: str, label: str | None = None) -> Path: 

65 """Create (if needed) and return ``<data_dir>/<subdir>``. 

66 

67 Args: 

68 subdir: Path segment below the data directory. May be multi-segment 

69 (e.g. ``"encrypted_databases/backups"``). 

70 label: If provided, emit a debug log ``"Using {label} directory: 

71 {path}"``. ``None`` suppresses logging. 

72 

73 Returns: 

74 The ensured directory path. 

75 """ 

76 path = get_data_directory() / subdir 

77 path.mkdir(parents=True, exist_ok=True) # Safe: bootstrap data subdirectory 

78 if label is not None: 

79 logger.debug(f"Using {label} directory: {path}") 

80 return path 

81 

82 

83def get_research_outputs_directory() -> Path: 

84 """ 

85 Get the directory for storing research outputs (reports, etc.). 

86 

87 Returns: 

88 Path to research outputs directory 

89 """ 

90 return _ensure_dir("research_outputs", label="research outputs") 

91 

92 

93def get_journal_data_directory() -> Path: 

94 """Get the directory for downloaded journal quality data files. 

95 

96 Contains openalex_sources.json.gz, doaj_journals.json, and the 

97 compiled journal_reference.db. Fetched on first use from 

98 OpenAlex and DOAJ APIs. 

99 

100 Returns: 

101 Path to journal data directory 

102 """ 

103 return _ensure_dir("journal_data") 

104 

105 

106def get_cache_directory() -> Path: 

107 """ 

108 Get the directory for storing cache files (search cache, etc.). 

109 

110 Returns: 

111 Path to cache directory 

112 """ 

113 return _ensure_dir("cache", label="cache") 

114 

115 

116def get_logs_directory() -> Path: 

117 """ 

118 Get the directory for storing log files. 

119 

120 Returns: 

121 Path to logs directory 

122 """ 

123 return _ensure_dir("logs", label="logs") 

124 

125 

126def get_encrypted_database_path() -> Path: 

127 """Get the path to the encrypted databases directory. 

128 

129 Returns: 

130 Path to the encrypted databases directory 

131 """ 

132 return _ensure_dir("encrypted_databases") 

133 

134 

135def get_user_database_filename(username: str) -> str: 

136 """Get the database filename for a specific user. 

137 

138 Args: 

139 username: The username to generate a filename for 

140 

141 Returns: 

142 The database filename (not full path) for the user 

143 

144 Raises: 

145 ValueError: If username is empty/None. Registration already enforces 

146 a non-empty username, but this is the actual hash chokepoint: 

147 sha256("") is a well-known constant, so silently accepting an 

148 empty username would let two different "no username" call sites 

149 collide into one shared, deterministic database filename instead 

150 of failing loudly. Defense-in-depth -- see #5481. 

151 """ 

152 if not username: 

153 raise ValueError( 

154 "Cannot compute a user database filename for an empty username" 

155 ) 

156 # Use username hash to avoid filesystem issues with special characters 

157 username_hash = hashlib.sha256(username.encode()).hexdigest()[:16] 

158 return f"ldr_user_{username_hash}.db" 

159 

160 

161def get_library_directory() -> Path: 

162 """ 

163 Get the directory for storing library files (documents, PDFs, etc.). 

164 

165 Returns: 

166 Path to library directory 

167 """ 

168 return _ensure_dir("library", label="library") 

169 

170 

171def get_config_directory() -> Path: 

172 """ 

173 Get the directory for storing configuration files. 

174 

175 Returns: 

176 Path to config directory 

177 """ 

178 return _ensure_dir("config", label="config") 

179 

180 

181def get_models_directory() -> Path: 

182 """ 

183 Get the directory for storing downloaded models. 

184 

185 Returns: 

186 Path to models directory 

187 """ 

188 return _ensure_dir("models", label="models") 

189 

190 

191def get_backup_directory() -> Path: 

192 """Get the base backup directory for all users.""" 

193 return _ensure_dir("encrypted_databases/backups") 

194 

195 

196def get_user_backup_directory(username: str) -> Path: 

197 """Get backup directory for a specific user. 

198 

199 Raises: 

200 ValueError: If username is empty/None. Same fail-closed guard as 

201 get_user_database_filename: sha256("") is a well-known constant, 

202 so silently accepting an empty username would collapse every 

203 "no username" call site into one shared, deterministic backup 

204 directory. Defense-in-depth -- see #5481. 

205 """ 

206 if not username: 

207 raise ValueError( 

208 "Cannot compute a user backup directory for an empty username" 

209 ) 

210 from local_deep_research.security.directory_creation import create_directory 

211 

212 username_hash = hashlib.sha256(username.encode()).hexdigest()[:16] 

213 backup_directory = get_backup_directory() 

214 user_backup_dir = create_directory( 

215 backup_directory / username_hash, 

216 context="user backup directory", 

217 root=backup_directory, 

218 mode=0o700, 

219 ) 

220 # Enforce 0o700 regardless of umask (mkdir mode is umask-masked) 

221 user_backup_dir.chmod(0o700) 

222 return user_backup_dir 

223 

224 

225# Convenience functions for backward compatibility 

226def get_data_dir() -> str: 

227 """Get data directory as string for backward compatibility.""" 

228 return str(get_data_directory())