Coverage for agentos/system/file_ops.py: 20%

1""" 

2文件操作模块 — 带权限检查的文件系统读写。 

3 

4设计原则: 

5- 所有操作前检查权限 

6- 读操作可穿透任意路径 

7- 写操作区分沙箱/全盘 

8- 操作结果统一为 FileOpResult 

9""" 

10 

11from __future__ import annotations 

12 

13import mimetypes 

14import os 

15import shutil 

16from dataclasses import dataclass, field 

17from datetime import datetime 

18 

19from agentos.system.permissions import ( 

20 PermissionDenied, 

21 PermissionTier, 

22 SystemPermissionManager, 

23) 

24 

25 

26@dataclass 

27class FileListing: 

28 """文件/目录条目。""" 

29 

30 name: str 

31 path: str 

32 is_dir: bool 

33 size_bytes: int = 0 

34 modified_at: str = "" 

35 mime_type: str = "" 

36 

37 

38@dataclass 

39class FileOpResult: 

40 """文件操作结果。""" 

41 

42 success: bool 

43 action: str # read/write/delete/move/copy/mkdir/list 

44 path: str 

45 content: str = "" # 读取的内容 

46 listing: list[FileListing] = field(default_factory=list) 

47 error: str = "" 

48 bytes_written: int = 0 

49 

50 

51class FileOperator: 

52 """文件操作器 — 带权限检查的文件系统接口。""" 

53 

54 def __init__(self, perm_manager: SystemPermissionManager, session_id: str): 

55 self._pm = perm_manager 

56 self._sid = session_id 

57 

58 # ── 读取操作 ── 

59 

60 def read(self, file_path: str) -> FileOpResult: 

61 """读取文件内容。""" 

62 path = os.path.abspath(os.path.expanduser(file_path)) 

63 try: 

64 self._pm.require(self._sid, PermissionTier.READ, path) 

65 except PermissionDenied as e: 

66 return FileOpResult(success=False, action="read", path=path, error=str(e)) 

67 

68 try: 

69 # 自动检测是否为文本文件 

70 mime, _ = mimetypes.guess_type(path) 

71 if ( 

72 mime 

73 and mime.startswith("text/") 

74 or path.endswith( 

75 ( 

76 ".py", 

77 ".md", 

78 ".txt", 

79 ".json", 

80 ".yaml", 

81 ".yml", 

82 ".toml", 

83 ".cfg", 

84 ".ini", 

85 ".log", 

86 ".csv", 

87 ".xml", 

88 ".html", 

89 ".css", 

90 ".js", 

91 ".ts", 

92 ".sh", 

93 ".bash", 

94 ".env", 

95 ".gitignore", 

96 ) 

97 ) 

98 ): 

99 with open(path, encoding="utf-8", errors="replace") as f: 

100 content = f.read() 

101 return FileOpResult(success=True, action="read", path=path, content=content) 

102 else: 

103 # 二进制文件返回预览 

104 size = os.path.getsize(path) 

105 return FileOpResult( 

106 success=True, 

107 action="read", 

108 path=path, 

109 content=f"[Binary file, {self._format_size(size)}]", 

110 ) 

111 except Exception as e: 

112 return FileOpResult(success=False, action="read", path=path, error=str(e)) 

113 

114 def read_bytes(self, file_path: str, max_bytes: int = 1024 * 1024) -> FileOpResult: 

115 """读取二进制文件（限制大小）。""" 

116 path = os.path.abspath(os.path.expanduser(file_path)) 

117 try: 

118 self._pm.require(self._sid, PermissionTier.READ, path) 

119 except PermissionDenied as e: 

120 return FileOpResult(success=False, action="read", path=path, error=str(e)) 

121 

122 try: 

123 with open(path, "rb") as f: 

124 data = f.read(max_bytes) 

125 # Base64 编码返回 

126 import base64 

127 

128 encoded = base64.b64encode(data).decode("ascii") 

129 return FileOpResult( 

130 success=True, 

131 action="read", 

132 path=path, 

133 content=encoded, 

134 bytes_written=len(data), 

135 ) 

136 except Exception as e: 

137 return FileOpResult(success=False, action="read", path=path, error=str(e)) 

138 

139 # ── 列表操作 ── 

140 

141 def list_dir(self, dir_path: str, show_hidden: bool = False) -> FileOpResult: 

142 """列出目录内容。""" 

143 path = os.path.abspath(os.path.expanduser(dir_path)) 

144 try: 

145 self._pm.require(self._sid, PermissionTier.READ, path) 

146 except PermissionDenied as e: 

147 return FileOpResult(success=False, action="list", path=path, error=str(e)) 

148 

149 if not os.path.isdir(path): 

150 return FileOpResult(success=False, action="list", path=path, error=f"不是目录: {path}") 

151 

152 try: 

153 entries = [] 

154 for name in sorted(os.listdir(path)): 

155 if not show_hidden and name.startswith("."): 

156 continue 

157 full = os.path.join(path, name) 

158 stat = os.stat(full) 

159 mime, _ = mimetypes.guess_type(full) 

160 entries.append( 

161 FileListing( 

162 name=name, 

163 path=full, 

164 is_dir=os.path.isdir(full), 

165 size_bytes=stat.st_size, 

166 modified_at=datetime.fromtimestamp(stat.st_mtime).isoformat(), 

167 mime_type=mime 

168 or ( 

169 "inode/directory" if os.path.isdir(full) else "application/octet-stream" 

170 ), 

171 ) 

172 ) 

173 return FileOpResult(success=True, action="list", path=path, listing=entries) 

174 except Exception as e: 

175 return FileOpResult(success=False, action="list", path=path, error=str(e)) 

176 

177 def search(self, root_dir: str, pattern: str, max_depth: int = 5) -> FileOpResult: 

178 """递归搜索文件（类似 find + glob）。""" 

179 import fnmatch 

180 

181 path = os.path.abspath(os.path.expanduser(root_dir)) 

182 try: 

183 self._pm.require(self._sid, PermissionTier.READ, path) 

184 except PermissionDenied as e: 

185 return FileOpResult(success=False, action="list", path=path, error=str(e)) 

186 

187 results: list[FileListing] = [] 

188 try: 

189 for dirpath, dirnames, filenames in os.walk(path): 

190 depth = dirpath[len(path) :].count(os.sep) 

191 if depth >= max_depth: 

192 dirnames.clear() 

193 continue 

194 # 跳过隐藏目录 

195 dirnames[:] = [d for d in dirnames if not d.startswith(".")] 

196 for fname in filenames: 

197 if fnmatch.fnmatch(fname, pattern): 

198 full = os.path.join(dirpath, fname) 

199 stat = os.stat(full) 

200 results.append( 

201 FileListing( 

202 name=fname, 

203 path=full, 

204 is_dir=False, 

205 size_bytes=stat.st_size, 

206 modified_at=datetime.fromtimestamp(stat.st_mtime).isoformat(), 

207 ) 

208 ) 

209 return FileOpResult(success=True, action="list", path=path, listing=results) 

210 except Exception as e: 

211 return FileOpResult(success=False, action="list", path=path, error=str(e)) 

212 

213 # ── 写入操作 ── 

214 

215 def write(self, file_path: str, content: str) -> FileOpResult: 

216 """写入文本文件。""" 

217 path = os.path.abspath(os.path.expanduser(file_path)) 

218 # 判断需要沙箱还是全盘权限 

219 sandbox_paths = ["/tmp/agentos/", "/home/marvis/Marvis/"] 

220 needs_full = not any(path.startswith(sp) for sp in sandbox_paths) 

221 tier = PermissionTier.WRITE_ALL if needs_full else PermissionTier.WRITE_SANDBOX 

222 

223 try: 

224 self._pm.require(self._sid, tier, path) 

225 except PermissionDenied as e: 

226 return FileOpResult(success=False, action="write", path=path, error=str(e)) 

227 

228 try: 

229 os.makedirs(os.path.dirname(path) or ".", exist_ok=True) 

230 with open(path, "w", encoding="utf-8") as f: 

231 f.write(content) 

232 return FileOpResult( 

233 success=True, 

234 action="write", 

235 path=path, 

236 bytes_written=len(content.encode("utf-8")), 

237 ) 

238 except Exception as e: 

239 return FileOpResult(success=False, action="write", path=path, error=str(e)) 

240 

241 def write_bytes(self, file_path: str, data: bytes) -> FileOpResult: 

242 """写入二进制文件。""" 

243 path = os.path.abspath(os.path.expanduser(file_path)) 

244 sandbox_paths = ["/tmp/agentos/", "/home/marvis/Marvis/"] 

245 needs_full = not any(path.startswith(sp) for sp in sandbox_paths) 

246 tier = PermissionTier.WRITE_ALL if needs_full else PermissionTier.WRITE_SANDBOX 

247 

248 try: 

249 self._pm.require(self._sid, tier, path) 

250 except PermissionDenied as e: 

251 return FileOpResult(success=False, action="write", path=path, error=str(e)) 

252 

253 try: 

254 os.makedirs(os.path.dirname(path) or ".", exist_ok=True) 

255 with open(path, "wb") as f: 

256 f.write(data) 

257 return FileOpResult(success=True, action="write", path=path, bytes_written=len(data)) 

258 except Exception as e: 

259 return FileOpResult(success=False, action="write", path=path, error=str(e)) 

260 

261 # ── 删除/移动 ── 

262 

263 def delete(self, target_path: str) -> FileOpResult: 

264 """删除文件或目录（高风险，需 WRITE_ALL 权限）。""" 

265 path = os.path.abspath(os.path.expanduser(target_path)) 

266 try: 

267 self._pm.require(self._sid, PermissionTier.WRITE_ALL, path) 

268 except PermissionDenied as e: 

269 return FileOpResult(success=False, action="delete", path=path, error=str(e)) 

270 

271 try: 

272 if os.path.isdir(path): 

273 shutil.rmtree(path) 

274 else: 

275 os.remove(path) 

276 return FileOpResult(success=True, action="delete", path=path) 

277 except Exception as e: 

278 return FileOpResult(success=False, action="delete", path=path, error=str(e)) 

279 

280 def move(self, src: str, dst: str) -> FileOpResult: 

281 """移动/重命名文件。""" 

282 src_path = os.path.abspath(os.path.expanduser(src)) 

283 dst_path = os.path.abspath(os.path.expanduser(dst)) 

284 try: 

285 self._pm.require(self._sid, PermissionTier.WRITE_ALL, src_path) 

286 self._pm.require(self._sid, PermissionTier.WRITE_ALL, dst_path) 

287 except PermissionDenied as e: 

288 return FileOpResult(success=False, action="move", path=src_path, error=str(e)) 

289 

290 try: 

291 shutil.move(src_path, dst_path) 

292 return FileOpResult(success=True, action="move", path=dst_path) 

293 except Exception as e: 

294 return FileOpResult(success=False, action="move", path=src_path, error=str(e)) 

295 

296 def mkdir(self, dir_path: str) -> FileOpResult: 

297 """创建目录。""" 

298 path = os.path.abspath(os.path.expanduser(dir_path)) 

299 sandbox_paths = ["/tmp/agentos/", "/home/marvis/Marvis/"] 

300 needs_full = not any(path.startswith(sp) for sp in sandbox_paths) 

301 tier = PermissionTier.WRITE_ALL if needs_full else PermissionTier.WRITE_SANDBOX 

302 

303 try: 

304 self._pm.require(self._sid, tier, path) 

305 except PermissionDenied as e: 

306 return FileOpResult(success=False, action="mkdir", path=path, error=str(e)) 

307 

308 try: 

309 os.makedirs(path, exist_ok=True) 

310 return FileOpResult(success=True, action="mkdir", path=path) 

311 except Exception as e: 

312 return FileOpResult(success=False, action="mkdir", path=path, error=str(e)) 

313 

314 # ── 文件信息 ── 

315 

316 def stat(self, file_path: str) -> FileOpResult: 

317 """获取文件/目录详细信息。""" 

318 path = os.path.abspath(os.path.expanduser(file_path)) 

319 try: 

320 self._pm.require(self._sid, PermissionTier.READ, path) 

321 except PermissionDenied as e: 

322 return FileOpResult(success=False, action="read", path=path, error=str(e)) 

323 

324 try: 

325 st = os.stat(path) 

326 info_lines = [ 

327 f"路径: {path}", 

328 f"类型: {'目录' if os.path.isdir(path) else '文件'}", 

329 f"大小: {self._format_size(st.st_size)}", 

330 f"权限: {oct(st.st_mode)[-3:]}", 

331 f"修改时间: {datetime.fromtimestamp(st.st_mtime).isoformat()}", 

332 f"创建时间: {datetime.fromtimestamp(st.st_ctime).isoformat()}", 

333 f"inode: {st.st_ino}", 

334 ] 

335 return FileOpResult( 

336 success=True, action="read", path=path, content="\n".join(info_lines) 

337 ) 

338 except Exception as e: 

339 return FileOpResult(success=False, action="read", path=path, error=str(e)) 

340 

341 # ── 工具 ── 

342 

343 @staticmethod 

344 def _format_size(size: int) -> str: 

345 for unit in ["B", "KB", "MB", "GB", "TB"]: 

346 if size < 1024: 

347 return f"{size:.1f} {unit}" 

348 size /= 1024 

349 return f"{size:.1f} PB"