Coverage for src/taipanstack/utils/filesystem.py: 100%

1""" 

2Safe filesystem operations. 

3 

4Provides secure wrappers around file operations with path validation, 

5atomic writes, and proper error handling using Result types. 

6""" 

7 

8import os 

9import shutil 

10import stat 

11import tempfile 

12from dataclasses import dataclass 

13from pathlib import Path 

14from typing import TypeAlias 

15 

16from taipanstack.core.result import Err, Ok, Result 

17from taipanstack.security.guards import ( 

18 TRAVERSAL_REGEX, 

19 SecurityError, 

20 guard_path_traversal, 

21) 

22from taipanstack.security.sanitizers import sanitize_filename 

23 

24 

25@dataclass(frozen=True) 

26class FileNotFoundErr: 

27 """Error when file is not found.""" 

28 

29 path: Path 

30 

31 @property 

32 def message(self) -> str: 

33 """Get the error message.""" 

34 return f"File not found: {self.path}" 

35 

36 

37@dataclass(frozen=True) 

38class NotAFileErr: 

39 """Error when path is not a file.""" 

40 

41 path: Path 

42 

43 @property 

44 def message(self) -> str: 

45 """Get the error message.""" 

46 return f"Not a file: {self.path}" 

47 

48 

49def _validate_path( 

50 path: Path | str, 

51 base_dir: Path | str | None = None, 

52 *, 

53 allow_symlinks: bool = False, 

54) -> Path: 

55 """Validate path for traversal. 

56 

57 If base_dir is None, we only check for explicit traversal patterns 

58 to allow absolute paths (required for tests and some use cases), 

59 but still prevent '..' attacks. 

60 """ 

61 path = Path(path) 

62 if base_dir is not None: 

63 return guard_path_traversal(path, base_dir, allow_symlinks=allow_symlinks) 

64 

65 # Check for explicit traversal patterns 

66 path_str = str(path).lower() 

67 if TRAVERSAL_REGEX.search(path_str): 

68 raise SecurityError( 

69 "Path traversal pattern detected", 

70 guard_name="path_traversal", 

71 value=path_str[:50], 

72 ) 

73 return path 

74 

75 

76@dataclass(frozen=True) 

77class FileTooLargeErr: 

78 """Error when file exceeds size limit.""" 

79 

80 path: Path 

81 size: int 

82 max_size: int 

83 

84 @property 

85 def message(self) -> str: 

86 """Get the error message.""" 

87 return f"File too large: {self.size} bytes (max: {self.max_size})" 

88 

89 

90@dataclass(frozen=True) 

91class WriteOptions: 

92 """Options for safe_write. 

93 

94 Attributes: 

95 base_dir: Base directory to constrain to. 

96 encoding: File encoding. 

97 create_parents: Create parent directories if needed. 

98 backup: Create backup of existing file. 

99 atomic: Use atomic write. 

100 

101 """ 

102 

103 base_dir: Path | str | None = None 

104 encoding: str = "utf-8" 

105 create_parents: bool = True 

106 backup: bool = True 

107 atomic: bool = True 

108 

109 

110# Union type for safe_read errors 

111ReadFileError: TypeAlias = ( 

112 FileNotFoundErr | NotAFileErr | FileTooLargeErr | SecurityError 

113) 

114 

115 

116def _check_file_size( 

117 path: Path, file_size: int, max_size_bytes: int | None 

118) -> Result[None, ReadFileError]: 

119 if max_size_bytes is not None and file_size > max_size_bytes: 

120 return Err( 

121 FileTooLargeErr(path=path, size=file_size, max_size=max_size_bytes), 

122 ) 

123 return Ok(None) 

124 

125 

126def _check_read_path_and_size( 

127 path: Path, 

128 max_size_bytes: int | None, 

129) -> Result[None, ReadFileError]: 

130 try: 

131 st = path.stat() 

132 except (FileNotFoundError, OSError): 

133 return Err(FileNotFoundErr(path=path)) 

134 

135 if not stat.S_ISREG(st.st_mode): 

136 return Err(NotAFileErr(path=path)) 

137 

138 return _check_file_size(path, st.st_size, max_size_bytes) 

139 

140 

141def safe_read( 

142 path: Path | str, 

143 *, 

144 base_dir: Path | str | None = None, 

145 encoding: str = "utf-8", 

146 max_size_bytes: int | None = 10 * 1024 * 1024, # 10MB default 

147) -> Result[str, ReadFileError]: 

148 """Read a file safely with path validation. 

149 

150 Args: 

151 path: Path to the file to read. 

152 base_dir: Base directory to constrain to. 

153 encoding: File encoding. 

154 max_size_bytes: Maximum file size to read (None for no limit). 

155 

156 Returns: 

157 Ok(str): File contents on success. 

158 Err(ReadFileError): Error details on failure. 

159 

160 Example: 

161 >>> result = safe_read("config.json") 

162 >>> if isinstance(result, Ok): 

163 ... data = json.loads(result.unwrap()) 

164 ... else: 

165 ... err = result.unwrap_err() 

166 ... if isinstance(err, FileNotFoundErr): 

167 ... print(f"Missing: {err.path}") 

168 ... elif isinstance(err, FileTooLargeErr): 

169 ... print(f"Too big: {err.size} bytes") 

170 

171 """ 

172 path = Path(path) 

173 

174 # Validate path 

175 try: 

176 path = _validate_path(path, base_dir) 

177 except SecurityError as e: 

178 return Err(e) 

179 

180 path_check = _check_read_path_and_size(path, max_size_bytes) 

181 if isinstance(path_check, Err): 

182 return path_check 

183 

184 return Ok(path.read_text(encoding=encoding)) 

185 

186 

187def _validate_safe_write_path(path: Path, opts: WriteOptions) -> None: 

188 """Validate the path for safe_write.""" 

189 if opts.base_dir is not None: 

190 base = Path(opts.base_dir).resolve() 

191 # For new files, validate the parent 

192 if not path.exists(): 

193 parent = path.parent 

194 guard_path_traversal(parent, base) 

195 else: 

196 guard_path_traversal(path, base) 

197 else: 

198 _validate_path(path) 

199 

200 

201def _sanitize_write_path(path: Path) -> Path: 

202 """Sanitize the filename for safe_write.""" 

203 safe_name = sanitize_filename(path.name) 

204 if safe_name != path.name: 

205 raise SecurityError( 

206 f"Unsafe or invalid characters in filename: '{path.name}'. " 

207 f"Expected safe name: '{safe_name}'", 

208 guard_name="sanitize_filename", 

209 value=path.name, 

210 ) 

211 return path.parent / safe_name 

212 

213 

214def _perform_atomic_write(path: Path, content: str, opts: WriteOptions) -> None: 

215 """Perform an atomic write operation.""" 

216 # Write to temp file first, then rename 

217 _fd, temp_path = tempfile.mkstemp( 

218 dir=path.parent, 

219 prefix=f".{path.name}.", 

220 suffix=".tmp", 

221 ) 

222 try: 

223 # Write directly to the returned file descriptor to prevent TOCTOU 

224 # We MUST close the file descriptor before renaming/modifying it, 

225 # otherwise Windows will throw a PermissionError (WinError 32). 

226 with os.fdopen(_fd, "w", encoding=opts.encoding) as f: 

227 f.write(content) 

228 f.flush() 

229 os.fsync(_fd) 

230 

231 temp_file = Path(temp_path) 

232 # Preserve permissions if original exists 

233 if path.exists(): 

234 shutil.copymode(path, temp_file) 

235 # On Windows, we need to remove the target first if it exists 

236 if path.exists(): 

237 path.unlink() 

238 temp_file.rename(path) 

239 except BaseException: 

240 # Clean up temp file on error; _fd is already managed by the context manager's 

241 # __exit__ if the exception happens inside the block. 

242 # If it happens before/after, we unlink. 

243 Path(temp_path).unlink(missing_ok=True) 

244 raise 

245 

246 

247def _prepare_write_dir(path: Path, create_parents: bool) -> None: 

248 if create_parents: 

249 path.parent.mkdir(parents=True, exist_ok=True) 

250 

251 

252def _create_write_backup(path: Path, backup: bool) -> None: 

253 if backup and path.is_file(): 

254 backup_path = path.with_suffix(f"{path.suffix}.bak") 

255 shutil.copy2(path, backup_path) 

256 

257 

258def safe_write( 

259 path: Path | str, 

260 content: str, 

261 *, 

262 options: WriteOptions | None = None, 

263) -> Path: 

264 """Write to a file safely with path validation. 

265 

266 Args: 

267 path: Path to write to. 

268 content: Content to write. 

269 options: Write options. 

270 

271 Returns: 

272 Path to the written file. 

273 

274 Raises: 

275 SecurityError: If path validation fails. 

276 

277 """ 

278 opts = options or WriteOptions() 

279 path = Path(path) 

280 

281 _validate_safe_write_path(path, opts) 

282 path = _sanitize_write_path(path) 

283 

284 _prepare_write_dir(path, opts.create_parents) 

285 _create_write_backup(path, opts.backup) 

286 

287 # Write file 

288 if opts.atomic: 

289 _perform_atomic_write(path, content, opts) 

290 else: 

291 path.write_text(content, encoding=opts.encoding) 

292 

293 return path.resolve() 

294 

295 

296def ensure_dir( 

297 path: Path | str, 

298 *, 

299 base_dir: Path | str | None = None, 

300 mode: int = 0o755, 

301) -> Path: 

302 """Ensure a directory exists, creating it if needed. 

303 

304 Args: 

305 path: Path to the directory. 

306 base_dir: Base directory to constrain to. 

307 mode: Directory permissions. 

308 

309 Returns: 

310 Path to the directory. 

311 

312 Raises: 

313 SecurityError: If path validation fails. 

314 FileExistsError: If a file already exists at the given path or intermediate 

315 paths. 

316 

317 """ 

318 path = Path(path) 

319 

320 # Validate path 

321 path = _validate_path(path, base_dir, allow_symlinks=True) 

322 resolved_path = path.resolve() 

323 

324 paths_to_create = _get_missing_parents(resolved_path) 

325 

326 # Iterate through parents and create them with specific mode 

327 for p in paths_to_create: 

328 p.mkdir(mode=mode, exist_ok=True) 

329 

330 return resolved_path 

331 

332 

333def _get_missing_parents(resolved_path: Path) -> list[Path]: 

334 """Identify missing parent directories from root to leaf. 

335 

336 Args: 

337 resolved_path: The fully resolved path to check. 

338 

339 Returns: 

340 List of paths to create in order from root to leaf. 

341 

342 Raises: 

343 FileExistsError: If a file exists in the path that is not a directory. 

344 

345 """ 

346 paths_to_create: list[Path] = [] 

347 current_path = resolved_path 

348 

349 while not current_path.is_dir(): 

350 if current_path.exists(): 

351 raise FileExistsError(f"Path exists but is not a directory: {current_path}") 

352 paths_to_create.insert(0, current_path) 

353 parent = current_path.parent 

354 if parent == current_path: 

355 break 

356 current_path = parent 

357 

358 return paths_to_create