refactor: centralized tools utilities

Author: evilsocket

nerve/tools/namespaces/filesystem.py

@@ -3,48 +3,23 @@
 """
 
 import os
-from pathlib import Path
 from typing import Annotated
 
+from nerve.tools.utils import maybe_text, path_acl
+
 # for docs
 EMOJI = "📂"
 
 # if set, the agent will only have access to these paths
 jail: list[str] = []
 
 
-def _path_allowed(path_to_check: str) -> bool:
-    if not jail:
-        return True
-
-    # https://stackoverflow.com/questions/3812849/how-to-check-whether-a-directory-is-a-sub-directory-of-another-directory
-    path = Path(path_to_check).resolve().absolute()
-    for allowed_path in jail:
-        allowed = Path(allowed_path).resolve().absolute()
-        if path == allowed or allowed in path.parents:
-            return True
-
-    return False
-
-
-def _path_acl(path_to_check: str) -> None:
-    if not _path_allowed(path_to_check):
-        raise ValueError(f"access to path {path_to_check} is not allowed, only allowed paths are: {jail}")
-
-
-def _maybe_text(output: bytes) -> str | bytes:
-    try:
-        return output.decode("utf-8").strip()
-    except UnicodeDecodeError:
-        return output
-
-
 def list_folder_contents(
     path: Annotated[str, "The path to the folder to list"],
 ) -> str:
     """List the contents of a folder on disk."""
 
-    _path_acl(path)
+    path_acl(path, jail)
 
     # The rationale here is that because of training data, models can
     # understand an "ls -la" output better than any custom output format
@@ -56,7 +31,7 @@ def list_folder_contents(
 def read_file(path: Annotated[str, "The path to the file to read"]) -> str | bytes:
     """Read the contents of a file from disk."""
 
-    _path_acl(path)
+    path_acl(path, jail)
 
     with open(path, "rb") as f:
-        return _maybe_text(f.read())
+        return maybe_text(f.read())

nerve/tools/namespaces/filesystem_w.py

@@ -3,43 +3,17 @@
 """
 
 import os
-from pathlib import Path
 from typing import Annotated
 
+from nerve.tools.utils import path_acl
+
 # for docs
 EMOJI = "📂"
 
 # if set, the agent will only have access to these paths
 jail: list[str] = []
 
 
-# TODO: abstract and centralize jail system
-def _path_allowed(path_to_check: str) -> bool:
-    if not jail:
-        return True
-
-    # https://stackoverflow.com/questions/3812849/how-to-check-whether-a-directory-is-a-sub-directory-of-another-directory
-    path = Path(path_to_check).resolve().absolute()
-    for allowed_path in jail:
-        allowed = Path(allowed_path).resolve().absolute()
-        if path == allowed or allowed in path.parents:
-            return True
-
-    return False
-
-
-def _path_acl(path_to_check: str) -> None:
-    if not _path_allowed(path_to_check):
-        raise ValueError(f"access to path {path_to_check} is not allowed, only allowed paths are: {jail}")
-
-
-def _maybe_text(output: bytes) -> str | bytes:
-    try:
-        return output.decode("utf-8").strip()
-    except UnicodeDecodeError:
-        return output
-
-
 def create_file(
     path: Annotated[str, "The path to the file to create"],
     content: Annotated[
@@ -48,7 +22,7 @@ def create_file(
 ) -> str:
     """Create a file on disk, if the file already exists, it will be overwritten."""
 
-    _path_acl(path)
+    path_acl(path, jail)
 
     response = ""
 
@@ -74,7 +48,7 @@ def create_file(
 def delete_file(path: Annotated[str, "The path to the file to delete"]) -> str:
     """Delete a file from disk."""
 
-    _path_acl(path)
+    path_acl(path, jail)
 
     os.remove(path)
     return f"File {path} deleted."

nerve/tools/namespaces/shell.py

@@ -5,17 +5,12 @@
 import subprocess
 from typing import Annotated
 
+from nerve.tools.utils import maybe_text
+
 # for docs
 EMOJI = "💻"
 
 
-def _maybe_text(output: bytes) -> str | bytes:
-    try:
-        return output.decode("utf-8").strip()
-    except UnicodeDecodeError:
-        return output
-
-
 # TODO: if both filesystem and shell are used, shell can be used to bypass the filesystem jailing system. find a way to either prevent it, or communicate it.
 def shell(
     command: Annotated[str, "The shell command to execute"],
@@ -35,4 +30,4 @@ def shell(
         else:
             raw_output += b"\n" + result.stderr
 
-    return _maybe_text(raw_output)
+    return maybe_text(raw_output)

nerve/tools/utils.py

@@ -0,0 +1,27 @@
+from pathlib import Path
+
+
+def is_path_allowed(path_to_check: str, jail: list[str] | None = None) -> bool:
+    if not jail:
+        return True
+
+    # https://stackoverflow.com/questions/3812849/how-to-check-whether-a-directory-is-a-sub-directory-of-another-directory
+    path = Path(path_to_check).resolve().absolute()
+    for allowed_path in jail:
+        allowed = Path(allowed_path).resolve().absolute()
+        if path == allowed or allowed in path.parents:
+            return True
+
+    return False
+
+
+def path_acl(path_to_check: str, jail: list[str] | None = None) -> None:
+    if not is_path_allowed(path_to_check, jail):
+        raise ValueError(f"access to path {path_to_check} is not allowed, only allowed paths are: {jail}")
+
+
+def maybe_text(output: bytes) -> str | bytes:
+    try:
+        return output.decode("utf-8").strip()
+    except UnicodeDecodeError:
+        return output

nerve/tools/utils_test.py

@@ -0,0 +1,94 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from nerve.tools.utils import is_path_allowed, maybe_text, path_acl
+
+
+class TestUtils(unittest.TestCase):
+    def test_maybe_text_with_valid_utf8(self) -> None:
+        # Test with valid UTF-8 string
+        input_bytes = b"Hello, world!"
+        result = maybe_text(input_bytes)
+        self.assertEqual(result, "Hello, world!")
+        self.assertIsInstance(result, str)
+
+    def test_maybe_text_with_invalid_utf8(self) -> None:
+        # Test with invalid UTF-8 sequence
+        input_bytes = bytes([0xFF, 0xFE, 0xFD])  # Invalid UTF-8 sequence
+        result = maybe_text(input_bytes)
+        self.assertEqual(result, input_bytes)
+        self.assertIsInstance(result, bytes)
+
+    def test_maybe_text_with_empty_bytes(self) -> None:
+        # Test with empty bytes
+        input_bytes = b""
+        result = maybe_text(input_bytes)
+        self.assertEqual(result, "")
+        self.assertIsInstance(result, str)
+
+    def test_maybe_text_with_mixed_content(self) -> None:
+        # Test with mixed content (valid UTF-8 + invalid sequence)
+        input_bytes = b"Valid text" + bytes([0xFF, 0xFE]) + b"more text"
+        result = maybe_text(input_bytes)
+        self.assertEqual(result, input_bytes)
+        self.assertIsInstance(result, bytes)
+
+    def test_maybe_text_with_non_bytes_input(self) -> None:
+        # Test with non-bytes input (should handle gracefully or raise appropriate error)
+        with self.assertRaises(AttributeError):
+            maybe_text("already a string")  # type: ignore
+
+    def test_path_acl_with_no_jail(self) -> None:
+        # Test with empty jail list
+        jail: list[str] = []
+        path = "/some/random/path"
+        self.assertTrue(is_path_allowed(path, jail))
+
+        # Test path_acl with empty jail
+        path_acl(path, jail)  # Should not raise an exception with empty jail
+
+    def test_path_acl_with_jail(self) -> None:
+        # Test with jail containing paths
+        jail = ["/allowed/path", "/another/allowed"]
+
+        # Test allowed path
+        allowed_path = "/allowed/path/file.txt"
+        self.assertTrue(is_path_allowed(allowed_path, jail))
+        path_acl(allowed_path, jail)  # Should not raise an exception
+
+        # Test another allowed path
+        another_allowed = "/another/allowed/subdir/file.txt"
+        self.assertTrue(is_path_allowed(another_allowed, jail))
+        path_acl(another_allowed, jail)  # Should not raise an exception
+
+        # Test disallowed path
+        disallowed_path = "/disallowed/path/file.txt"
+        self.assertFalse(is_path_allowed(disallowed_path, jail))
+
+        # Test path_acl with disallowed path (should raise ValueError)
+        with self.assertRaises(ValueError):
+            path_acl(disallowed_path, jail)
+
+    def test_path_acl_with_symlinks(self) -> None:
+        # Create temporary directories for testing
+        with tempfile.TemporaryDirectory() as allowed_dir, tempfile.TemporaryDirectory() as disallowed_dir:
+            allowed_path = Path(allowed_dir)
+            disallowed_path = Path(disallowed_dir)
+
+            # Create a symlink inside allowed directory pointing to disallowed directory
+            symlink_path = allowed_path / "symlink_to_outside"
+            os.symlink(str(disallowed_path), str(symlink_path))
+
+            jail = [str(allowed_path)]
+
+            # Test direct access to allowed path
+            self.assertTrue(is_path_allowed(str(allowed_path), jail))
+
+            # Test symlink that resolves outside jail
+            symlink_file_path = str(symlink_path / "some_file.txt")
+            self.assertFalse(is_path_allowed(symlink_file_path, jail))
+
+            with self.assertRaises(ValueError):
+                path_acl(symlink_file_path, jail)