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_read_path_and_size( 

117 path: Path, 

118 max_size_bytes: int | None, 

119) -> Result[None, ReadFileError]: 

120 try: 

121 st = path.stat() 

122 except (FileNotFoundError, OSError): 

123 return Err(FileNotFoundErr(path=path)) 

124 

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

126 return Err(NotAFileErr(path=path)) 

127 

128 if max_size_bytes is not None: 

129 file_size = st.st_size 

130 if file_size > max_size_bytes: 

131 return Err( 

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

133 ) 

134 return Ok(None) 

135 

136 

137def safe_read( 

138 path: Path | str, 

139 *, 

140 base_dir: Path | str | None = None, 

141 encoding: str = "utf-8", 

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

143) -> Result[str, ReadFileError]: 

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

145 

146 Args: 

147 path: Path to the file to read. 

148 base_dir: Base directory to constrain to. 

149 encoding: File encoding. 

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

151 

152 Returns: 

153 Ok(str): File contents on success. 

154 Err(ReadFileError): Error details on failure. 

155 

156 Example: 

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

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

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

160 ... else: 

161 ... err = result.unwrap_err() 

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

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

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

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

166 

167 """ 

168 path = Path(path) 

169 

170 # Validate path 

171 try: 

172 path = _validate_path(path, base_dir) 

173 except SecurityError as e: 

174 return Err(e) 

175 

176 path_check = _check_read_path_and_size(path, max_size_bytes) 

177 if isinstance(path_check, Err): 

178 return path_check 

179 

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

181 

182 

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

184 """Validate the path for safe_write.""" 

185 if opts.base_dir is not None: 

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

187 # For new files, validate the parent 

188 if not path.exists(): 

189 parent = path.parent 

190 guard_path_traversal(parent, base) 

191 else: 

192 guard_path_traversal(path, base) 

193 else: 

194 _validate_path(path) 

195 

196 

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

198 """Sanitize the filename for safe_write.""" 

199 safe_name = sanitize_filename(path.name) 

200 if safe_name != path.name: 

201 raise SecurityError( 

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

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

204 guard_name="sanitize_filename", 

205 value=path.name, 

206 ) 

207 return path.parent / safe_name 

208 

209 

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

211 """Perform an atomic write operation.""" 

212 # Write to temp file first, then rename 

213 _fd, temp_path = tempfile.mkstemp( 

214 dir=path.parent, 

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

216 suffix=".tmp", 

217 ) 

218 try: 

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

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

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

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

223 f.write(content) 

224 f.flush() 

225 os.fsync(_fd) 

226 

227 temp_file = Path(temp_path) 

228 # Preserve permissions if original exists 

229 if path.exists(): 

230 shutil.copymode(path, temp_file) 

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

232 if path.exists(): 

233 path.unlink() 

234 temp_file.rename(path) 

235 except BaseException: 

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

237 # __exit__ if the exception happens inside the block. 

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

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

240 raise 

241 

242 

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

244 if create_parents: 

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

246 

247 

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

249 if backup and path.is_file(): 

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

251 shutil.copy2(path, backup_path) 

252 

253 

254def safe_write( 

255 path: Path | str, 

256 content: str, 

257 *, 

258 options: WriteOptions | None = None, 

259) -> Path: 

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

261 

262 Args: 

263 path: Path to write to. 

264 content: Content to write. 

265 options: Write options. 

266 

267 Returns: 

268 Path to the written file. 

269 

270 Raises: 

271 SecurityError: If path validation fails. 

272 

273 """ 

274 opts = options or WriteOptions() 

275 path = Path(path) 

276 

277 _validate_safe_write_path(path, opts) 

278 path = _sanitize_write_path(path) 

279 

280 _prepare_write_dir(path, opts.create_parents) 

281 _create_write_backup(path, opts.backup) 

282 

283 # Write file 

284 if opts.atomic: 

285 _perform_atomic_write(path, content, opts) 

286 else: 

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

288 

289 return path.resolve() 

290 

291 

292def ensure_dir( 

293 path: Path | str, 

294 *, 

295 base_dir: Path | str | None = None, 

296 mode: int = 0o755, 

297) -> Path: 

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

299 

300 Args: 

301 path: Path to the directory. 

302 base_dir: Base directory to constrain to. 

303 mode: Directory permissions. 

304 

305 Returns: 

306 Path to the directory. 

307 

308 Raises: 

309 SecurityError: If path validation fails. 

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

311 paths. 

312 

313 """ 

314 path = Path(path) 

315 

316 # Validate path 

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

318 resolved_path = path.resolve() 

319 

320 # Identify missing parent directories from root to leaf 

321 paths_to_create: list[Path] = [] 

322 current_path = resolved_path 

323 

324 while not current_path.is_dir(): 

325 if current_path.exists(): 

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

327 paths_to_create.insert(0, current_path) 

328 parent = current_path.parent 

329 if parent == current_path: 

330 break 

331 current_path = parent 

332 

333 # Iterate through parents and create them with specific mode 

334 for p in paths_to_create: 

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

336 

337 return resolved_path