|
28 | 28 | """ |
29 | 29 |
|
30 | 30 | import json |
| 31 | +import os |
| 32 | +from pathlib import Path |
| 33 | +from unittest.mock import MagicMock |
31 | 34 |
|
32 | 35 | import pytest |
33 | 36 |
|
34 | 37 | from swarms import Agent |
35 | 38 | from swarms.agents.autonomous_loop import AutonomousAgentLoop |
36 | | -from swarms.structs.autonomous_loop_utils import MAX_SUBTASK_LOOPS |
| 39 | +from swarms.structs.autonomous_loop_utils import ( |
| 40 | + MAX_SUBTASK_LOOPS, |
| 41 | + create_file_tool, |
| 42 | + delete_file_tool, |
| 43 | + grep_tool, |
| 44 | + list_directory_tool, |
| 45 | + read_file_tool, |
| 46 | + update_file_tool, |
| 47 | +) |
37 | 48 |
|
38 | 49 |
|
39 | 50 | # -------------------------------------------------------------------------- |
@@ -228,10 +239,13 @@ class TestBatchedToolCalls: |
228 | 239 | """Every tool call in a response runs, whatever its position.""" |
229 | 240 |
|
230 | 241 | def test_call_after_subtask_done_still_executes( |
231 | | - self, monkeypatch, tmp_path |
| 242 | + self, monkeypatch |
232 | 243 | ): |
233 | 244 | agent = build_agent() |
234 | | - target = tmp_path / "written.txt" |
| 245 | + # Inside the agent's workspace: the file tools refuse paths outside it. |
| 246 | + target = ( |
| 247 | + Path(agent._get_agent_workspace_dir()) / "written.txt" |
| 248 | + ) |
235 | 249 |
|
236 | 250 | script_llm( |
237 | 251 | agent, |
@@ -273,10 +287,11 @@ def test_call_after_subtask_done_still_executes( |
273 | 287 | assert status_of(agent, "step1") == "completed" |
274 | 288 |
|
275 | 289 | def test_call_after_complete_task_still_executes( |
276 | | - self, monkeypatch, tmp_path |
| 290 | + self, monkeypatch |
277 | 291 | ): |
278 | 292 | agent = build_agent() |
279 | | - target = tmp_path / "late.txt" |
| 293 | + # Inside the agent's workspace: the file tools refuse paths outside it. |
| 294 | + target = Path(agent._get_agent_workspace_dir()) / "late.txt" |
280 | 295 |
|
281 | 296 | script_llm( |
282 | 297 | agent, |
@@ -365,10 +380,14 @@ def exploding_think(*args, **kwargs): |
365 | 380 | assert "RuntimeError" in text |
366 | 381 |
|
367 | 382 | def test_malformed_arguments_do_not_abort_the_iteration( |
368 | | - self, monkeypatch, tmp_path |
| 383 | + self, monkeypatch |
369 | 384 | ): |
370 | 385 | agent = build_agent() |
371 | | - target = tmp_path / "after_bad_json.txt" |
| 386 | + # Inside the agent's workspace: the file tools refuse paths outside it. |
| 387 | + target = ( |
| 388 | + Path(agent._get_agent_workspace_dir()) |
| 389 | + / "after_bad_json.txt" |
| 390 | + ) |
372 | 391 |
|
373 | 392 | bad = { |
374 | 393 | "type": "function", |
@@ -1108,5 +1127,77 @@ def boom(agent, history): |
1108 | 1127 | assert len(loop._transcript) == 1 |
1109 | 1128 |
|
1110 | 1129 |
|
| 1130 | +class TestFileToolsStayInTheWorkspace: |
| 1131 | + """The model picks file_path itself, so any prompt injection reaching the |
| 1132 | + loop otherwise becomes arbitrary local file read, overwrite or deletion |
| 1133 | + under the host process's privileges. |
| 1134 | + """ |
| 1135 | + |
| 1136 | + @pytest.fixture |
| 1137 | + def workspace(self, tmp_path): |
| 1138 | + """A workspace with a secret sitting just outside it.""" |
| 1139 | + ws = tmp_path / "agent_workspace" / "agents" / "worker-1" |
| 1140 | + ws.mkdir(parents=True) |
| 1141 | + (ws / "inside.txt").write_text("INSIDE-OK\n") |
| 1142 | + (tmp_path / "outside_secret.txt").write_text( |
| 1143 | + "SECRET-OUTSIDE\n" |
| 1144 | + ) |
| 1145 | + |
| 1146 | + agent = MagicMock() |
| 1147 | + agent._get_agent_workspace_dir.return_value = str(ws) |
| 1148 | + return agent, ws, tmp_path / "outside_secret.txt" |
| 1149 | + |
| 1150 | + def test_every_file_tool_refuses_to_escape(self, workspace): |
| 1151 | + agent, ws, secret = workspace |
| 1152 | + relative_escape = os.path.relpath(secret, ws) |
| 1153 | + |
| 1154 | + for label, call in ( |
| 1155 | + ( |
| 1156 | + "read_file relative", |
| 1157 | + lambda: read_file_tool(agent, relative_escape), |
| 1158 | + ), |
| 1159 | + ( |
| 1160 | + "read_file absolute", |
| 1161 | + lambda: read_file_tool(agent, str(secret)), |
| 1162 | + ), |
| 1163 | + ( |
| 1164 | + "grep", |
| 1165 | + lambda: grep_tool(agent, "SECRET", relative_escape), |
| 1166 | + ), |
| 1167 | + ( |
| 1168 | + "list_directory", |
| 1169 | + lambda: list_directory_tool( |
| 1170 | + agent, str(secret.parent) |
| 1171 | + ), |
| 1172 | + ), |
| 1173 | + ( |
| 1174 | + "delete_file", |
| 1175 | + lambda: delete_file_tool(agent, str(secret)), |
| 1176 | + ), |
| 1177 | + ): |
| 1178 | + result = str(call()) |
| 1179 | + assert ( |
| 1180 | + "outside the agent workspace" in result |
| 1181 | + ), f"{label} did not refuse the escape: {result[:120]}" |
| 1182 | + assert "SECRET-OUTSIDE" not in result |
| 1183 | + |
| 1184 | + create_file_tool(agent, relative_escape, "overwritten") |
| 1185 | + update_file_tool(agent, str(secret), "overwritten") |
| 1186 | + assert secret.read_text() == "SECRET-OUTSIDE\n" |
| 1187 | + |
| 1188 | + def test_legitimate_workspace_access_still_works(self, workspace): |
| 1189 | + agent, ws, _ = workspace |
| 1190 | + |
| 1191 | + assert "INSIDE-OK" in str(read_file_tool(agent, "inside.txt")) |
| 1192 | + assert "INSIDE-OK" in str( |
| 1193 | + read_file_tool(agent, str(ws / "inside.txt")) |
| 1194 | + ) |
| 1195 | + assert "inside.txt" in str(list_directory_tool(agent, "")) |
| 1196 | + assert "Error" not in str( |
| 1197 | + create_file_tool(agent, "new.txt", "x") |
| 1198 | + ) |
| 1199 | + assert (ws / "new.txt").exists() |
| 1200 | + |
| 1201 | + |
1111 | 1202 | if __name__ == "__main__": |
1112 | 1203 | pytest.main([__file__, "-q", "-p", "no:randomly"]) |
0 commit comments