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 os 

14import shutil 

15import mimetypes 

16from dataclasses import dataclass, field 

17from datetime import datetime 

18 

19from agentos.system.permissions import ( 

20 SystemPermissionManager, 

21 PermissionTier, 

22 PermissionDenied, 

23) 

24 

25 

26@dataclass 

27class FileListing: 

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

29 name: str 

30 path: str 

31 is_dir: bool 

32 size_bytes: int = 0 

33 modified_at: str = "" 

34 mime_type: str = "" 

35 

36 

37@dataclass 

38class FileOpResult: 

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

40 success: bool 

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

42 path: str 

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

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

45 error: str = "" 

46 bytes_written: int = 0 

47 

48 

49class FileOperator: 

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

51 

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

53 self._pm = perm_manager 

54 self._sid = session_id 

55 

56 # ── 读取操作 ── 

57 

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

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

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

61 try: 

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

63 except PermissionDenied as e: 

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

65 

66 try: 

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

68 mime, _ = mimetypes.guess_type(path) 

69 if mime and mime.startswith("text/") or path.endswith((".py", ".md", ".txt", ".json", ".yaml", ".yml", ".toml", ".cfg", ".ini", ".log", ".csv", ".xml", ".html", ".css", ".js", ".ts", ".sh", ".bash", ".env", ".gitignore")): 

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

71 content = f.read() 

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

73 else: 

74 # 二进制文件返回预览 

75 size = os.path.getsize(path) 

76 return FileOpResult( 

77 success=True, action="read", path=path, 

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

79 ) 

80 except Exception as e: 

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

82 

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

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

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

86 try: 

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

88 except PermissionDenied as e: 

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

90 

91 try: 

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

93 data = f.read(max_bytes) 

94 # Base64 编码返回 

95 import base64 

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

97 return FileOpResult( 

98 success=True, action="read", path=path, 

99 content=encoded, bytes_written=len(data), 

100 ) 

101 except Exception as e: 

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

103 

104 # ── 列表操作 ── 

105 

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

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

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

109 try: 

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

111 except PermissionDenied as e: 

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

113 

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

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

116 

117 try: 

118 entries = [] 

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

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

121 continue 

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

123 stat = os.stat(full) 

124 mime, _ = mimetypes.guess_type(full) 

125 entries.append(FileListing( 

126 name=name, 

127 path=full, 

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

129 size_bytes=stat.st_size, 

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

131 mime_type=mime or ("inode/directory" if os.path.isdir(full) else "application/octet-stream"), 

132 )) 

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

134 except Exception as e: 

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

136 

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

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

139 import fnmatch 

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

141 try: 

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

143 except PermissionDenied as e: 

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

145 

146 results: list[FileListing] = [] 

147 try: 

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

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

150 if depth >= max_depth: 

151 dirnames.clear() 

152 continue 

153 # 跳过隐藏目录 

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

155 for fname in filenames: 

156 if fnmatch.fnmatch(fname, pattern): 

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

158 stat = os.stat(full) 

159 results.append(FileListing( 

160 name=fname, 

161 path=full, 

162 is_dir=False, 

163 size_bytes=stat.st_size, 

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

165 )) 

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

167 except Exception as e: 

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

169 

170 # ── 写入操作 ── 

171 

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

173 """写入文本文件。""" 

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

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

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

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

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

179 

180 try: 

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

182 except PermissionDenied as e: 

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

184 

185 try: 

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

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

188 f.write(content) 

189 return FileOpResult( 

190 success=True, action="write", path=path, 

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

192 ) 

193 except Exception as e: 

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

195 

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

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

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

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

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

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

202 

203 try: 

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

205 except PermissionDenied as e: 

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

207 

208 try: 

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

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

211 f.write(data) 

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

213 except Exception as e: 

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

215 

216 # ── 删除/移动 ── 

217 

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

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

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

221 try: 

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

223 except PermissionDenied as e: 

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

225 

226 try: 

227 if os.path.isdir(path): 

228 shutil.rmtree(path) 

229 else: 

230 os.remove(path) 

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

232 except Exception as e: 

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

234 

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

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

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

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

239 try: 

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

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

242 except PermissionDenied as e: 

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

244 

245 try: 

246 shutil.move(src_path, dst_path) 

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

248 except Exception as e: 

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

250 

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

252 """创建目录。""" 

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

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

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

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

257 

258 try: 

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

260 except PermissionDenied as e: 

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

262 

263 try: 

264 os.makedirs(path, exist_ok=True) 

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

266 except Exception as e: 

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

268 

269 # ── 文件信息 ── 

270 

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

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

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

274 try: 

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

276 except PermissionDenied as e: 

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

278 

279 try: 

280 st = os.stat(path) 

281 info_lines = [ 

282 f"路径: {path}", 

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

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

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

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

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

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

289 ] 

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

291 except Exception as e: 

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

293 

294 # ── 工具 ── 

295 

296 @staticmethod 

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

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

299 if size < 1024: 

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

301 size /= 1024 

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