@@ -0,0 +1,112 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests that the file/search tools stay confined to LOCAL_REPOS_DIR_PATH.
+
+These tools are driven by an LLM that processes untrusted input (issue bodies,
+release diffs), so a path that escapes the managed repositories directory must
+be rejected even though it is absolute.
+"""
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+# Set GITHUB_TOKEN before import since settings.py validates it at import time.
+os.environ.setdefault("GITHUB_TOKEN", "test-token")
+
+# E402 is required because GITHUB_TOKEN must be set before this import.
+from adk_documentation import tools  # noqa: E402
+
+
+class PathConfinementTest(unittest.TestCase):
+
+  @classmethod
+  def setUpClass(cls):
+    cls.sandbox = tempfile.mkdtemp(prefix="adk_repos_")
+
+  def setUp(self):
+    self.repo = os.path.join(self.sandbox, "adk-docs")
+    os.makedirs(self.repo, exist_ok=True)
+    self.inside_file = os.path.join(self.repo, "index.md")
+    with open(self.inside_file, "w", encoding="utf-8") as f:
+      f.write("hello")
+
+    # Dynamically mock the constant to isolate setting from sys.modules cache
+    self.patcher = mock.patch(
+        "adk_documentation.tools.LOCAL_REPOS_DIR_PATH", self.sandbox
+    )
+    self.patcher.start()
+
+  def tearDown(self):
+    self.patcher.stop()
+
+  def test_read_inside_sandbox_succeeds(self):
+    res = tools.read_local_git_repo_file_content(self.inside_file)
+    self.assertEqual(res["status"], "success")
+
+  def test_read_outside_sandbox_is_denied(self):
+    for path in ("/etc/passwd", "/proc/self/environ"):
+      res = tools.read_local_git_repo_file_content(path)
+      self.assertEqual(res["status"], "error", path)
+      self.assertIn("Access denied", res["error_message"])
+
+  def test_read_symlink_escape_is_denied(self):
+    link = os.path.join(self.repo, "sneaky")
+    if not os.path.lexists(link):
+      os.symlink("/etc/passwd", link)
+    res = tools.read_local_git_repo_file_content(link)
+    self.assertEqual(res["status"], "error")
+    self.assertIn("Access denied", res["error_message"])
+
+  def test_list_and_search_outside_sandbox_are_denied(self):
+    self.assertEqual(tools.list_directory_contents("/etc")["status"], "error")
+    self.assertEqual(
+        tools.search_local_git_repo("/etc", "root")["status"], "error"
+    )
+
+  def test_clone_or_pull_outside_sandbox_is_denied(self):
+    res = tools.clone_or_pull_repo(
+        repo_owner="google",
+        repo_name="adk-docs",
+        local_path="/etc",
+    )
+    self.assertEqual(res["status"], "error")
+    self.assertIn("Access denied", res["error_message"])
+
+  def test_create_pr_rejects_path_traversal_in_changes(self):
+    # Reach the file-writing step with the git/network calls stubbed out, then
+    # assert a traversing change key is rejected before any write.
+    with (
+        mock.patch.object(tools, "_run_git_command"),
+        mock.patch.object(tools, "post_request"),
+    ):
+      res = tools.create_pull_request_from_changes(
+          repo_owner="google",
+          repo_name="adk-docs",
+          local_path=self.repo,
+          base_branch="main",
+          changes={"../../../../tmp/evil.txt": "owned"},
+          commit_message="m",
+          pr_title="t",
+          pr_body="b",
+      )
+    self.assertEqual(res["status"], "error")
+    self.assertIn("escapes the repository", res["error_message"])
+    self.assertFalse(os.path.exists("/tmp/evil.txt"))
+
+
+if __name__ == "__main__":
+  unittest.main()
@@ -22,6 +22,7 @@
 from typing import Optional
 
 from adk_documentation.settings import GITHUB_BASE_URL
+from adk_documentation.settings import LOCAL_REPOS_DIR_PATH
 from adk_documentation.utils import error_response
 from adk_documentation.utils import get_paginated_request
 from adk_documentation.utils import get_request
@@ -30,6 +31,44 @@
 import requests
 
 
+def _resolve_within_repos_dir(path: str) -> Optional[str]:
+  """Resolves `path` and verifies it stays within LOCAL_REPOS_DIR_PATH.
+
+  Returns the resolved absolute path if it is inside the sandbox, else None.
+  """
+  allowed_root = os.path.realpath(LOCAL_REPOS_DIR_PATH)
+  resolved = os.path.realpath(path)
+  if resolved == allowed_root or resolved.startswith(allowed_root + os.sep):
+    return resolved
+  return None
+
+
+def _confine_path(
+    path: str, param_name: str
+) -> tuple[Optional[str], Optional[Dict[str, Any]]]:
+  """Resolves `path` and verifies it stays within LOCAL_REPOS_DIR_PATH.
+
+  The file/search tools are only meant to operate on the repositories cloned
+  under LOCAL_REPOS_DIR_PATH. Because these tools are driven by an LLM that
+  processes untrusted input (issue bodies, release diffs), an `os.path.isabs`
+  check alone lets a crafted instruction read or write arbitrary paths (e.g.
+  the credentials file pointed to by GOOGLE_APPLICATION_CREDENTIALS, or
+  /proc/self/environ). Resolve symlinks and `..` segments first, then require
+  the result to stay inside the managed directory.
+
+  Returns:
+      A tuple of (safe_path, error_response_dict). If the path is outside the
+      sandbox, safe_path is None and error_response_dict contains the error.
+  """
+  safe_path = _resolve_within_repos_dir(path)
+  if safe_path is None:
+    return None, error_response(
+        f"Access denied: {param_name} is outside the managed repositories "
+        f"directory ({LOCAL_REPOS_DIR_PATH}): {path}"
+    )
+  return safe_path, None
+
+
 def list_releases(repo_owner: str, repo_name: str) -> Dict[str, Any]:
   """Lists all releases for a repository.
 
@@ -135,6 +174,10 @@ def clone_or_pull_repo(
       A dictionary indicating the status of the operation, output message, and
       the head commit hash.
   """
+  local_path, err = _confine_path(local_path, "local_path")
+  if err:
+    return err
+
   repo_url = f"git@github.com:{repo_owner}/{repo_name}.git"
 
   try:
@@ -188,6 +231,10 @@ def read_local_git_repo_file_content(file_path: str) -> Dict[str, Any]:
         f"file_path must be an absolute path, got: {file_path}"
     )
 
+  file_path, err = _confine_path(file_path, "file_path")
+  if err:
+    return err
+
   try:
     dir_path = os.path.dirname(file_path)
     head_commit_sha = _find_head_commit_sha(dir_path)
@@ -229,6 +276,9 @@ def list_directory_contents(directory_path: str) -> Dict[str, Any]:
   print(
       f"Attempting to recursively list contents of directory: {directory_path}"
   )
+  directory_path, err = _confine_path(directory_path, "directory_path")
+  if err:
+    return err
   if not os.path.isdir(directory_path):
     return error_response(f"Error: Directory not found at {directory_path}")
 
@@ -276,6 +326,9 @@ def search_local_git_repo(
       f"Attempting to search for pattern: {pattern} in directory:"
       f" {directory_path}, with extensions: {extensions}"
   )
+  directory_path, err = _confine_path(directory_path, "directory_path")
+  if err:
+    return err
   try:
     grep_process = _git_grep(directory_path, pattern, extensions, ignored_dirs)
     if grep_process.returncode > 1:
@@ -350,8 +403,18 @@ def create_pull_request_from_changes(
     if not changes:
       return error_response("No changes provided to apply.")
 
+    repo_root = os.path.realpath(local_path)
     for relative_path, new_content in changes.items():
-      full_path = os.path.join(local_path, relative_path)
+      full_path = os.path.realpath(os.path.join(local_path, relative_path))
+      # Confine writes to the repository: a crafted relative_path such as
+      # "../../etc/x" would otherwise escape `local_path`.
+      if not (
+          full_path == repo_root or full_path.startswith(repo_root + os.sep)
+      ):
+        return error_response(
+            "Access denied: change path escapes the repository:"
+            f" {relative_path}"
+        )
       os.makedirs(os.path.dirname(full_path), exist_ok=True)
       with open(full_path, "w", encoding="utf-8") as f:
         f.write(new_content)