Fix: Prevent arbitrary file write via path traversal in ResourceHelper

superagi/helper/resource_helper.py

@@ -10,6 +10,18 @@
 
 
 class ResourceHelper:
+    @staticmethod
+    def _secure_path_join(base_dir: str, file_name: str):
+        """
+        Safely join paths and prevent path traversal attacks.
+        """
+        base_dir = os.path.abspath(base_dir)
+        final_path = os.path.abspath(os.path.join(base_dir, file_name))
+
+        if os.path.commonpath([base_dir, final_path]) != base_dir:
+            raise Exception("SecurityError: Path traversal detected")
+
+        return final_path
     @classmethod
     def make_written_file_resource(cls, file_name: str, agent: Agent, agent_execution: AgentExecution, session):
         """
@@ -91,13 +103,15 @@ def get_formatted_agent_execution_level_path(cls, agent_execution: AgentExecutio
 
     @classmethod
     def get_resource_path(cls, file_name: str):
-        """Get final path of the resource.
-
+        """
+        Get the full path for a resource file in the root output directory.
         Args:
             file_name (str): The name of the file.
+        Returns:
+            str: The absolute path where the resource file should be stored.
         """
-        return ResourceHelper.get_root_output_dir() + file_name
-
+        root_dir = ResourceHelper.get_root_output_dir()
+        return ResourceHelper._secure_path_join(root_dir, file_name)
     @classmethod
     def get_root_output_dir(cls):
         """Get root dir of the resource.
@@ -126,12 +140,14 @@ def get_root_input_dir(cls):
 
     @classmethod
     def get_agent_write_resource_path(cls, file_name: str, agent: Agent, agent_execution: AgentExecution):
-        """Get agent resource path to write files
-
+        """
+        Get the full path for writing a resource file for a specific agent execution.
         Args:
             file_name (str): The name of the file.
-            agent (Agent): The unique identifier of the agent.
-            agent_execution (AgentExecution): The unique identifier of the agent.
+            agent (Agent): The agent associated with the file.
+            agent_execution (AgentExecution): The agent execution context.
+        Returns:
+            str: The absolute path where the file should be written.
         """
         root_dir = ResourceHelper.get_root_output_dir()
         if agent is not None and "{agent_id}" in root_dir:
@@ -140,7 +156,7 @@ def get_agent_write_resource_path(cls, file_name: str, agent: Agent, agent_execu
                 root_dir = ResourceHelper.get_formatted_agent_execution_level_path(agent_execution, root_dir)
             directory = os.path.dirname(root_dir)
             os.makedirs(directory, exist_ok=True)
-        final_path = root_dir + file_name
+        final_path = ResourceHelper._secure_path_join(root_dir, file_name)
         return final_path
 
     @staticmethod

superagi/resource_manager/file_manager.py

@@ -13,6 +13,18 @@ def __init__(self, session: Session, agent_id: int = None, agent_execution_id: i
         self.session = session
         self.agent_id = agent_id
         self.agent_execution_id = agent_execution_id
+
+    def _validate_safe_path(self, final_path: str):
+        """
+        Ensure the resolved file path stays inside the allowed resource directory.
+        Prevents path traversal vulnerabilities.
+        """
+        base_dir = os.path.abspath(os.path.dirname(final_path))
+        target_path = os.path.abspath(final_path)
+
+        if os.path.commonpath([base_dir, target_path]) != base_dir:
+            raise Exception("SecurityError: Attempted path traversal outside resource directory")
+
     def write_binary_file(self, file_name: str, data):
         if self.agent_id is not None:
             final_path = ResourceHelper.get_agent_write_resource_path(file_name,
@@ -24,6 +36,7 @@ def write_binary_file(self, file_name: str, data):
         else:
             final_path = ResourceHelper.get_resource_path(file_name)
         try:
+            self._validate_safe_path(final_path)
             with open(final_path, mode="wb") as img:
                 img.write(data)
                 img.close()
@@ -56,6 +69,7 @@ def write_file(self, file_name: str, content):
         else:
             final_path = ResourceHelper.get_resource_path(file_name)
         try:
+            self._validate_safe_path(final_path)
             with open(final_path, mode="w") as file:
                 file.write(content)
                 file.close()
@@ -75,6 +89,7 @@ def write_csv_file(self, file_name: str, csv_data):
         else:
             final_path = ResourceHelper.get_resource_path(file_name)
         try:
+            self._validate_safe_path(final_path)
             with open(final_path, mode="w", newline="") as file:
                 writer = csv.writer(file, lineterminator="\n")
                 writer.writerows(csv_data)

superagi/tools/code/write_code.py

@@ -1,4 +1,5 @@
 import re
+import os 
 from typing import Type, Optional, List
 
 from pydantic import BaseModel, Field
@@ -95,6 +96,31 @@ def _execute(self, code_description: str) -> str:
             file_name = re.sub(r'[<>"|?*]', "", match.group(1))
             if not file_name[0].isalnum():
                 file_name = file_name[1:-1]
+
+            # Block path traversal attempts
+            if ".." in file_name:
+                logger.error(f"Blocked path traversal attempt: {file_name}")
+                return "Error: Invalid filename (path traversal detected)"
+
+            # Block absolute paths
+            if file_name.startswith("/") or file_name.startswith("\\"):
+                logger.error(f"Blocked absolute path attempt: {file_name}")
+                return "Error: Invalid filename (absolute paths not allowed)"
+
+            # Allow only safe filename characters
+            if not re.match(r'^[a-zA-Z0-9_.\-/]+$', file_name):
+                logger.error(f"Blocked unsafe filename: {file_name}")
+                return "Error: Filename contains unsafe characters"
+
+            # Normalize path to remove tricks like a/../b
+            normalized_path = os.path.normpath(file_name)
+
+            # Ensure normalized path still doesn't escape directory
+            if normalized_path.startswith(".."):
+                logger.error(f"Blocked normalized path traversal: {file_name}")
+                return "Error: Invalid filename after normalization"
+
+            file_name = normalized_path
 
             # Get the code
             code = match.group(2)