fix(file tools) implement directory restrictions to file tools

python/packages/kagent-adk/src/kagent/adk/tools/README.md

@@ -35,7 +35,7 @@ agent = Agent(
     tools=[
         SkillsTool(skills_directory="./skills"),
         BashTool(skills_directory="./skills"),
-        ReadFileTool(),
+        ReadFileTool(skills_directory="./skills"),
         WriteFileTool(),
         EditFileTool(),
     ]

python/packages/kagent-adk/src/kagent/adk/tools/bash_tool.py

@@ -65,7 +65,7 @@ async def run_async(self, *, args: Dict[str, Any], tool_context: ToolContext) ->
 
         try:
             working_dir = get_session_path(session_id=tool_context.session.id)
-            result = await execute_command(command, working_dir)
+            result = await execute_command(command, working_dir, self.skills_directory)
             logger.info(f"Executed bash command: {command}, description: {description}")
             return result
         except Exception as e:

python/packages/kagent-adk/src/kagent/adk/tools/file_tools.py

@@ -29,11 +29,14 @@
 class ReadFileTool(BaseTool):
     """Read files with line numbers for precise editing."""
 
-    def __init__(self):
+    def __init__(self, skills_directory: str | Path):
         super().__init__(
             name="read_file",
             description=get_read_file_description(),
         )
+        self.skills_directory = Path(skills_directory).resolve()
+        if not self.skills_directory.exists():
+            raise ValueError(f"Skills directory does not exist: {self.skills_directory}")
 
     def _get_declaration(self) -> types.FunctionDeclaration:
         return types.FunctionDeclaration(
@@ -75,8 +78,8 @@ async def run_async(self, *, args: Dict[str, Any], tool_context: ToolContext) ->
                 path = working_dir / path
             path = path.resolve()
 
-            return read_file_content(path, offset, limit)
-        except (FileNotFoundError, IsADirectoryError, IOError) as e:
+            return read_file_content(path, offset, limit, allowed_root=[working_dir, Path(self.skills_directory)])
+        except (FileNotFoundError, IsADirectoryError, PermissionError, IOError) as e:
             return f"Error reading file {file_path_str}: {e}"
 
 
@@ -124,8 +127,8 @@ async def run_async(self, *, args: Dict[str, Any], tool_context: ToolContext) ->
                 path = working_dir / path
             path = path.resolve()
 
-            return write_file_content(path, content)
-        except IOError as e:
+            return write_file_content(path, content, allowed_root=working_dir)
+        except (PermissionError, IOError) as e:
             error_msg = f"Error writing file {file_path_str}: {e}"
             logger.error(error_msg)
             return error_msg
@@ -185,8 +188,8 @@ async def run_async(self, *, args: Dict[str, Any], tool_context: ToolContext) ->
                 path = working_dir / path
             path = path.resolve()
 
-            return edit_file_content(path, old_string, new_string, replace_all)
-        except (FileNotFoundError, IsADirectoryError, ValueError, IOError) as e:
+            return edit_file_content(path, old_string, new_string, replace_all, allowed_root=working_dir)
+        except (FileNotFoundError, IsADirectoryError, ValueError, PermissionError, IOError) as e:
             error_msg = f"Error editing file {file_path_str}: {e}"
             logger.error(error_msg)
             return error_msg

python/packages/kagent-adk/src/kagent/adk/tools/skills_plugin.py

@@ -37,7 +37,7 @@ def add_skills_tool_to_agent(skills_directory: str | Path, agent: BaseAgent) ->
         logger.debug(f"Added bash tool to agent: {agent.name}")
 
     if "read_file" not in existing_tool_names:
-        agent.tools.append(ReadFileTool())
+        agent.tools.append(ReadFileTool(skills_directory))
         logger.debug(f"Added read file tool to agent: {agent.name}")
 
     if "write_file" not in existing_tool_names:

python/packages/kagent-adk/src/kagent/adk/tools/skills_toolset.py

@@ -47,7 +47,7 @@ def __init__(self, skills_directory: str | Path):
 
         # Create skills tools
         self.skills_tool = SkillsTool(skills_directory)
-        self.read_file_tool = ReadFileTool()
+        self.read_file_tool = ReadFileTool(skills_directory)
         self.write_file_tool = WriteFileTool()
         self.edit_file_tool = EditFileTool()
         self.bash_tool = BashTool(skills_directory)

python/packages/kagent-openai/src/kagent/openai/tools/_tools.py

@@ -12,6 +12,7 @@
 from agents.exceptions import UserError
 from agents.run_context import RunContextWrapper
 from agents.tool import FunctionTool, function_tool
+import os
 
 from kagent.skills import (
     discover_skills,
@@ -55,8 +56,14 @@ def read_file(
         if not path.is_absolute():
             path = working_dir / path
 
-        return read_file_content(path, offset, limit)
-    except (FileNotFoundError, IsADirectoryError, OSError) as e:
+        allowed_dirs = [working_dir]
+
+        skills_directory = os.getenv("KAGENT_SKILLS_FOLDER", None)
+        if skills_directory:
+            allowed_dirs.append(Path(skills_directory))
+
+        return read_file_content(path, offset, limit, allowed_root=allowed_dirs)
+    except (FileNotFoundError, IsADirectoryError, PermissionError, OSError) as e:
         raise UserError(str(e)) from e
 
 
@@ -73,7 +80,7 @@ def write_file(wrapper: RunContextWrapper[SessionContext], file_path: str, conte
         if not path.is_absolute():
             path = working_dir / path
 
-        return write_file_content(path, content)
+        return write_file_content(path, content, allowed_root=working_dir)
     except OSError as e:
         raise UserError(str(e)) from e
 
@@ -97,7 +104,7 @@ def edit_file(
         if not path.is_absolute():
             path = working_dir / path
 
-        return edit_file_content(path, old_string, new_string, replace_all)
+        return edit_file_content(path, old_string, new_string, replace_all, allowed_root=working_dir)
     except (FileNotFoundError, IsADirectoryError, ValueError, OSError) as e:
         raise UserError(str(e)) from e
 

python/packages/kagent-skills/src/kagent/skills/shell.py

@@ -13,12 +13,33 @@
 # --- File Operation Tools ---
 
 
+def _validate_path(
+    file_path: Path,
+    allowed_roots: Path | list[Path] | None,
+) -> Path:
+    """Resolve the path and ensure it is within at least one allowed root directory."""
+    resolved = file_path.resolve()
+    if allowed_roots is None:
+        return resolved
+
+    roots = [allowed_roots] if isinstance(allowed_roots, Path) else allowed_roots
+    for root in roots:
+        if resolved.is_relative_to(root.resolve()):
+            return resolved
+
+    root_list = ", ".join(str(r.resolve()) for r in roots)
+    raise PermissionError(f"Access denied: {resolved} is outside the allowed directories: {root_list}")
+
+
 def read_file_content(
     file_path: Path,
     offset: int | None = None,
     limit: int | None = None,
+    allowed_root: Path | list[Path] | None = None,
 ) -> str:
     """Reads a file with line numbers, raising errors on failure."""
+    file_path = _validate_path(file_path, allowed_root)
+
     if not file_path.exists():
         raise FileNotFoundError(f"File not found: {file_path}")
 
@@ -45,8 +66,10 @@ def read_file_content(
     return "\n".join(result_lines)
 
 
-def write_file_content(file_path: Path, content: str) -> str:
+def write_file_content(file_path: Path, content: str, allowed_root: Path | None = None) -> str:
     """Writes content to a file, creating parent directories if needed."""
+    file_path = _validate_path(file_path, allowed_root)
+
     try:
         file_path.parent.mkdir(parents=True, exist_ok=True)
         file_path.write_text(content, encoding="utf-8")
@@ -61,11 +84,14 @@ def edit_file_content(
     old_string: str,
     new_string: str,
     replace_all: bool = False,
+    allowed_root: Path | None = None,
 ) -> str:
     """Performs an exact string replacement in a file."""
     if old_string == new_string:
         raise ValueError("old_string and new_string must be different")
 
+    file_path = _validate_path(file_path, allowed_root)
+
     if not file_path.exists():
         raise FileNotFoundError(f"File not found: {file_path}")
 
@@ -110,16 +136,13 @@ def _get_command_timeout_seconds(command: str) -> float:
         return 30.0  # 30 seconds for other commands
 
 
-async def execute_command(
-    command: str,
-    working_dir: Path,
-) -> str:
+async def execute_command(command: str, working_dir: Path, skills_dir: Path) -> str:
     """Executes a shell command in a sandboxed environment."""
     timeout = _get_command_timeout_seconds(command)
 
     env = os.environ.copy()
     # Add skills directory and working directory to PYTHONPATH
-    pythonpath_additions = [str(working_dir), "/skills"]
+    pythonpath_additions = [str(working_dir), str(skills_dir)]
     if "PYTHONPATH" in env:
         pythonpath_additions.append(env["PYTHONPATH"])
     env["PYTHONPATH"] = ":".join(pythonpath_additions)

python/packages/kagent-skills/src/kagent/tests/unittests/test_skill_execution.py

@@ -10,9 +10,11 @@
 
 from kagent.skills import (
     discover_skills,
+    edit_file_content,
     execute_command,
     load_skill_content,
     read_file_content,
+    write_file_content,
 )
 
 
@@ -159,6 +161,88 @@ async def mock_exec(*args, **kwargs):
     assert list(args).count(injection_payload) == 1
 
 
+# --- Path traversal tests ---
+
+
+def test_read_file_blocks_path_traversal(tmp_path):
+    """Reading a file outside the allowed root must raise PermissionError."""
+    outside_file = tmp_path.parent / "outside.txt"
+    outside_file.write_text("secret")
+
+    try:
+        with pytest.raises(PermissionError, match="outside the allowed director"):
+            read_file_content(outside_file, allowed_root=tmp_path)
+    finally:
+        outside_file.unlink(missing_ok=True)
+
+
+def test_read_file_blocks_relative_traversal(tmp_path):
+    """Relative paths like ../foo that escape the root must be blocked."""
+    (tmp_path / "subdir").mkdir()
+    outside = tmp_path.parent / "secret.txt"
+    outside.write_text("secret")
+
+    try:
+        with pytest.raises(PermissionError, match="outside the allowed director"):
+            read_file_content(
+                tmp_path / "subdir" / "../../secret.txt",
+                allowed_root=tmp_path,
+            )
+    finally:
+        outside.unlink(missing_ok=True)
+
+
+def test_read_file_allows_path_inside_root(tmp_path):
+    """Files inside the allowed root should work normally."""
+    f = tmp_path / "hello.txt"
+    f.write_text("hello world")
+    result = read_file_content(f, allowed_root=tmp_path)
+    assert "hello world" in result
+
+
+def test_read_file_allows_multiple_roots(tmp_path):
+    """Read should succeed when the file is inside any of the allowed roots."""
+    skills_dir = tmp_path / "skills"
+    skills_dir.mkdir()
+    skill_file = skills_dir / "script.py"
+    skill_file.write_text("print('hello')")
+
+    session_dir = tmp_path / "session"
+    session_dir.mkdir()
+
+    # File is in skills_dir, not session_dir — should still be allowed
+    result = read_file_content(skill_file, allowed_root=[session_dir, skills_dir])
+    assert "print('hello')" in result
+
+    # File outside both roots should be blocked
+    outside = tmp_path / "outside.txt"
+    outside.write_text("secret")
+    with pytest.raises(PermissionError, match="outside the allowed directories"):
+        read_file_content(outside, allowed_root=[session_dir, skills_dir])
+
+
+def test_write_file_blocks_path_traversal(tmp_path):
+    """Writing a file outside the allowed root must raise PermissionError."""
+    outside_path = tmp_path.parent / "evil.txt"
+    with pytest.raises(PermissionError, match="outside the allowed director"):
+        write_file_content(outside_path, "malicious", allowed_root=tmp_path)
+    assert not outside_path.exists()
+
+
+def test_edit_file_blocks_path_traversal(tmp_path):
+    """Editing a file outside the allowed root must raise PermissionError."""
+    outside_file = tmp_path.parent / "target.txt"
+    outside_file.write_text("original")
+
+    try:
+        with pytest.raises(PermissionError, match="outside the allowed director"):
+            edit_file_content(outside_file, "original", "hacked", allowed_root=tmp_path)
+        # File must not have been modified
+        assert outside_file.read_text() == "original"
+    finally:
+        outside_file.unlink(missing_ok=True)
+
+
 def test_skill_discovery_and_loading(skill_test_env: Path):
     """
     Tests the core logic of discovering a skill and loading its instructions.