[bugf][autonomous-loop][confine the built-in file tools to the agent workspace]

Author: ayaangazali

swarms/structs/autonomous_loop_utils.py

@@ -640,6 +640,43 @@ def respond_to_user_tool(
     return f"Message sent to user: {message}"
 
 
+def _resolve_in_workspace(agent: Any, path: str) -> str:
+    """Resolve a model-supplied path, refusing anything outside the workspace.
+
+    The file tools take ``file_path`` straight from the model, so any prompt
+    injection reaching the autonomous loop can otherwise read, overwrite or
+    delete anywhere the host process can. ``realpath`` is what makes the check
+    hold: it collapses ``..`` segments and follows symlinks before the
+    containment test, so neither can be used to step outside.
+
+    Args:
+        agent: The agent instance, used for the workspace directory
+        path: Model-supplied path, relative to the workspace or absolute
+
+    Returns:
+        str: The resolved absolute path, guaranteed inside the workspace
+
+    Raises:
+        ValueError: If the path resolves outside the agent workspace
+    """
+    workspace = os.path.realpath(agent._get_agent_workspace_dir())
+    full_path = (
+        path
+        if path and os.path.isabs(path)
+        else os.path.join(workspace, path or "")
+    )
+    resolved = os.path.realpath(full_path)
+
+    if resolved != workspace and not resolved.startswith(
+        workspace + os.sep
+    ):
+        raise ValueError(
+            f"Path is outside the agent workspace and was refused: {path}"
+        )
+
+    return resolved
+
+
 def create_file_tool(
     agent: Any, file_path: str, content: str, **kwargs
 ) -> str:
@@ -656,12 +693,7 @@ def create_file_tool(
         str: Path to the created file or error message
     """
     try:
-        # Resolve path - if relative, use agent workspace
-        if not os.path.isabs(file_path):
-            workspace_dir = agent._get_agent_workspace_dir()
-            full_path = os.path.join(workspace_dir, file_path)
-        else:
-            full_path = file_path
+        full_path = _resolve_in_workspace(agent, file_path)
 
         # Create parent directories if they don't exist
         os.makedirs(os.path.dirname(full_path), exist_ok=True)
@@ -715,12 +747,7 @@ def update_file_tool(
         str: Success message or error message
     """
     try:
-        # Resolve path - if relative, use agent workspace
-        if not os.path.isabs(file_path):
-            workspace_dir = agent._get_agent_workspace_dir()
-            full_path = os.path.join(workspace_dir, file_path)
-        else:
-            full_path = file_path
+        full_path = _resolve_in_workspace(agent, file_path)
 
         # Check if file exists
         if not os.path.exists(full_path):
@@ -769,12 +796,7 @@ def read_file_tool(agent: Any, file_path: str, **kwargs) -> str:
         str: File contents or error message
     """
     try:
-        # Resolve path - if relative, use agent workspace
-        if not os.path.isabs(file_path):
-            workspace_dir = agent._get_agent_workspace_dir()
-            full_path = os.path.join(workspace_dir, file_path)
-        else:
-            full_path = file_path
+        full_path = _resolve_in_workspace(agent, file_path)
 
         # Check if file exists
         if not os.path.exists(full_path):
@@ -819,17 +841,7 @@ def list_directory_tool(
         str: Formatted list of directory contents
     """
     try:
-        # Resolve path - if relative or empty, use agent workspace
-        if not directory_path or not os.path.isabs(directory_path):
-            workspace_dir = agent._get_agent_workspace_dir()
-            if directory_path:
-                full_path = os.path.join(
-                    workspace_dir, directory_path
-                )
-            else:
-                full_path = workspace_dir
-        else:
-            full_path = directory_path
+        full_path = _resolve_in_workspace(agent, directory_path)
 
         # Check if directory exists
         if not os.path.exists(full_path):
@@ -888,12 +900,7 @@ def delete_file_tool(agent: Any, file_path: str, **kwargs) -> str:
         str: Success message or error message
     """
     try:
-        # Resolve path - if relative, use agent workspace
-        if not os.path.isabs(file_path):
-            workspace_dir = agent._get_agent_workspace_dir()
-            full_path = os.path.join(workspace_dir, file_path)
-        else:
-            full_path = file_path
+        full_path = _resolve_in_workspace(agent, file_path)
 
         # Check if file exists
         if not os.path.exists(full_path):
@@ -1129,16 +1136,7 @@ def grep_tool(
         str: Matching lines or error message
     """
     try:
-        # Resolve path
-        if not path or not os.path.isabs(path):
-            workspace_dir = agent._get_agent_workspace_dir()
-            full_path = (
-                os.path.join(workspace_dir, path)
-                if path
-                else workspace_dir
-            )
-        else:
-            full_path = path
+        full_path = _resolve_in_workspace(agent, path)
 
         if not os.path.exists(full_path):
             return f"Error: Path does not exist: {full_path}"

tests/structs/test_autonomous_loop_file_confinement.py

@@ -0,0 +1,84 @@
+"""The autonomous-loop file tools must stay inside the agent workspace.
+
+The model picks `file_path` itself, so any prompt injection reaching the loop
+(a fetched page, a file the agent was asked to summarise, a sub-agent's task
+string) otherwise becomes arbitrary local file read, overwrite or deletion
+under the host process's privileges.
+
+No agent, no network: the workspace is a tmp_path and the agent is a mock that
+only answers _get_agent_workspace_dir.
+"""
+
+import os
+from unittest.mock import MagicMock
+
+import pytest
+
+from swarms.structs.autonomous_loop_utils import (
+    create_file_tool,
+    delete_file_tool,
+    grep_tool,
+    list_directory_tool,
+    read_file_tool,
+    update_file_tool,
+)
+
+
+@pytest.fixture
+def workspace(tmp_path):
+    """A workspace with a secret sitting just outside it."""
+    ws = tmp_path / "agent_workspace" / "agents" / "worker-1"
+    ws.mkdir(parents=True)
+    (ws / "inside.txt").write_text("INSIDE-OK\n")
+    (tmp_path / "outside_secret.txt").write_text("SECRET-OUTSIDE\n")
+
+    agent = MagicMock()
+    agent._get_agent_workspace_dir.return_value = str(ws)
+    return agent, ws, tmp_path / "outside_secret.txt"
+
+
+def test_every_file_tool_refuses_to_escape(workspace):
+    agent, ws, secret = workspace
+    relative_escape = os.path.relpath(secret, ws)
+
+    # signatures differ (grep takes the pattern first), so call each one
+    for label, call in (
+        (
+            "read_file relative",
+            lambda: read_file_tool(agent, relative_escape),
+        ),
+        (
+            "read_file absolute",
+            lambda: read_file_tool(agent, str(secret)),
+        ),
+        ("grep", lambda: grep_tool(agent, "SECRET", relative_escape)),
+        (
+            "list_directory",
+            lambda: list_directory_tool(agent, str(secret.parent)),
+        ),
+        ("delete_file", lambda: delete_file_tool(agent, str(secret))),
+    ):
+        result = str(call())
+        assert (
+            "outside the agent workspace" in result
+        ), f"{label} did not refuse the escape: {result[:120]}"
+        assert "SECRET-OUTSIDE" not in result
+
+    # writes must not land outside either, by relative or absolute path
+    create_file_tool(agent, relative_escape, "overwritten")
+    update_file_tool(agent, str(secret), "overwritten")
+    assert secret.read_text() == "SECRET-OUTSIDE\n"
+
+
+def test_legitimate_workspace_access_still_works(workspace):
+    agent, ws, _ = workspace
+
+    assert "INSIDE-OK" in str(read_file_tool(agent, "inside.txt"))
+    # an absolute path inside the workspace is still fine: the check is
+    # containment, not "relative paths only"
+    assert "INSIDE-OK" in str(
+        read_file_tool(agent, str(ws / "inside.txt"))
+    )
+    assert "inside.txt" in str(list_directory_tool(agent, ""))
+    assert "Error" not in str(create_file_tool(agent, "new.txt", "x"))
+    assert (ws / "new.txt").exists()