Coverage for src/local_deep_research/storage/file.py: 97%

1"""File-based report storage implementation.""" 

2 

3import os 

4from pathlib import Path 

5from typing import Dict, Any, List, Optional 

6from loguru import logger 

7 

8from .base import ReportStorage 

9from ..config.paths import ( 

10 get_research_outputs_directory, 

11 restrict_directory_to_owner, 

12) 

13 

14 

15class FileReportStorage(ReportStorage): 

16 """Store reports as files on disk.""" 

17 

18 def __init__(self, base_dir: Optional[Path] = None): 

19 """Initialize file storage. 

20 

21 Args: 

22 base_dir: Base directory for storing reports. 

23 If None, uses default research outputs directory. 

24 

25 Raises: 

26 DirectoryNotPrivateError: On POSIX, another OS account owns the 

27 directory or its filesystem did not keep mode 0700. 

28 OSError: The directory could not be created or restricted. 

29 DirectoryCreationSecurityError: The path contains a null byte or 

30 a ``..`` segment. 

31 """ 

32 # Lazy import: mirrors the other lazy security imports in this 

33 # module (e.g. file_write_verifier) and avoids pulling security 

34 # into this module's top-level import graph. 

35 from ..security.directory_creation import create_directory 

36 

37 self.base_dir = create_directory( 

38 base_dir or get_research_outputs_directory(), 

39 context="file report storage base directory", 

40 mode=0o700, 

41 ) 

42 # The directory may already exist at 0755 from an older install. 

43 # Protect legacy backups, and refuse the directory unless it is 

44 # private, before any new write. 

45 restrict_directory_to_owner(self.base_dir) 

46 

47 def _contained_path(self, research_id: str, filename: str) -> Path: 

48 """Return ``base_dir / filename`` resolved, or raise ValueError. 

49 

50 Both sides are resolved with resolve_allow_missing(), never a bare 

51 non-strict resolve(), which can leave part of the path unresolved 

52 (a symlink chain whose resolved form outgrows PATH_MAX) and so hide 

53 an escape. 

54 """ 

55 from ..security.path_validator import resolve_allow_missing 

56 

57 try: 

58 candidate = resolve_allow_missing(self.base_dir / filename) 

59 base = resolve_allow_missing(self.base_dir) 

60 except OSError: 

61 raise ValueError( 

62 f"Unresolvable report path for research_id: {research_id!r}" 

63 ) 

64 if not candidate.is_relative_to(base): 

65 raise ValueError( 

66 f"Path traversal attempt with research_id: {research_id!r}" 

67 ) 

68 return candidate 

69 

70 def _get_report_path(self, research_id: str) -> Path: 

71 """Get the file path for a report.""" 

72 return self._contained_path(research_id, f"{research_id}.md") 

73 

74 def _get_metadata_path(self, research_id: str) -> Path: 

75 """Get the file path for report metadata.""" 

76 return self._contained_path(research_id, f"{research_id}_metadata.json") 

77 

78 def save_report( 

79 self, 

80 research_id: str, 

81 content: str, 

82 metadata: Optional[Dict[str, Any]] = None, 

83 username: Optional[str] = None, 

84 ) -> bool: 

85 """Save report to file.""" 

86 try: 

87 from ..security.file_write_verifier import ( 

88 write_file_verified, 

89 write_json_verified, 

90 ) 

91 

92 report_path = self._get_report_path(research_id) 

93 

94 # Save content 

95 write_file_verified( 

96 report_path, 

97 content, 

98 "report.enable_file_backup", 

99 context="file storage backup", 

100 permissions=0o600 if os.name == "posix" else None, 

101 ) 

102 

103 # Save metadata if provided 

104 if metadata: 

105 metadata_path = self._get_metadata_path(research_id) 

106 write_json_verified( 

107 metadata_path, 

108 metadata, 

109 "report.enable_file_backup", 

110 context="file storage metadata", 

111 permissions=0o600 if os.name == "posix" else None, 

112 ) 

113 

114 logger.info( 

115 f"Saved report for research {research_id} to {report_path}" 

116 ) 

117 return True 

118 

119 except Exception: 

120 logger.exception("Error saving report to file") 

121 return False 

122 

123 def get_report( 

124 self, research_id: str, username: Optional[str] = None 

125 ) -> Optional[str]: 

126 """Get report from file.""" 

127 try: 

128 report_path = self._get_report_path(research_id) 

129 

130 if not report_path.exists(): 

131 return None 

132 

133 with open(report_path, "r", encoding="utf-8") as f: 

134 return f.read() 

135 

136 except Exception: 

137 logger.exception("Error reading report from file") 

138 return None 

139 

140 def get_report_with_metadata( 

141 self, research_id: str, username: Optional[str] = None 

142 ) -> Optional[Dict[str, Any]]: 

143 """Get report with metadata from files.""" 

144 try: 

145 content = self.get_report(research_id) 

146 if not content: 

147 return None 

148 

149 result = {"content": content, "metadata": {}} 

150 

151 # Try to load metadata 

152 metadata_path = self._get_metadata_path(research_id) 

153 if metadata_path.exists(): 

154 import json 

155 

156 with open(metadata_path, "r", encoding="utf-8") as f: 

157 result["metadata"] = json.load(f) 

158 

159 return result 

160 

161 except Exception: 

162 logger.exception("Error getting report with metadata") 

163 return None 

164 

165 def list_reports( 

166 self, username: Optional[str] = None 

167 ) -> List[Dict[str, Any]]: 

168 """List reports from file system.""" 

169 try: 

170 from datetime import datetime, timezone 

171 

172 reports = [] 

173 for report_path in self.base_dir.glob("*.md"): 

174 research_id = report_path.stem 

175 metadata = self._load_metadata(research_id) 

176 reports.append( 

177 { 

178 "id": research_id, 

179 "query": metadata.get("query") if metadata else None, 

180 "mode": metadata.get("mode", "unknown") 

181 if metadata 

182 else "unknown", 

183 "created_at": datetime.fromtimestamp( 

184 report_path.stat().st_mtime, tz=timezone.utc 

185 ).isoformat(), 

186 "completed_at": metadata.get("completed_at") 

187 if metadata 

188 else None, 

189 } 

190 ) 

191 return reports 

192 except Exception: 

193 logger.exception("Error listing reports from file system") 

194 return [] 

195 

196 def _load_metadata(self, research_id: str) -> Optional[Dict[str, Any]]: 

197 """Load metadata JSON for a report if it exists.""" 

198 try: 

199 metadata_path = self._get_metadata_path(research_id) 

200 if metadata_path.exists(): 

201 import json 

202 

203 with open(metadata_path, "r", encoding="utf-8") as f: 

204 return json.load(f) 

205 except Exception: 

206 logger.debug(f"Failed to load metadata for {research_id}") 

207 return None 

208 

209 def delete_report( 

210 self, research_id: str, username: Optional[str] = None 

211 ) -> bool: 

212 """Delete report files.""" 

213 try: 

214 report_path = self._get_report_path(research_id) 

215 metadata_path = self._get_metadata_path(research_id) 

216 

217 deleted = False 

218 

219 if report_path.exists(): 

220 report_path.unlink() 

221 deleted = True 

222 

223 if metadata_path.exists(): 

224 metadata_path.unlink() 

225 

226 return deleted 

227 

228 except Exception: 

229 logger.exception("Error deleting report files") 

230 return False 

231 

232 def report_exists( 

233 self, research_id: str, username: Optional[str] = None 

234 ) -> bool: 

235 """Check if report file exists.""" 

236 return self._get_report_path(research_id).exists()