Security: replace eval() with ast.literal_eval() and add path traversal checks

superagi/agent/output_handler.py

@@ -1,3 +1,4 @@
+import ast
 import json
 from superagi.agent.common_types import TaskExecutorResponse, ToolExecutorResponse
 from superagi.agent.output_parser import AgentSchemaOutputParser
@@ -146,7 +147,7 @@ def __init__(self, agent_execution_id: int, agent_config: dict):
 
     def handle(self, session, assistant_reply):
         assistant_reply = JsonCleaner.extract_json_array_section(assistant_reply)
-        tasks = eval(assistant_reply)
+        tasks = ast.literal_eval(assistant_reply)
         tasks = np.array(tasks).flatten().tolist()
         for task in reversed(tasks):
             self.task_queue.add_task(task)
@@ -177,7 +178,7 @@ def __init__(self, agent_execution_id: int, agent_config: dict):
 
     def handle(self, session, assistant_reply):
         assistant_reply = JsonCleaner.extract_json_array_section(assistant_reply)
-        tasks = eval(assistant_reply)
+        tasks = ast.literal_eval(assistant_reply)
         self.task_queue.clear_tasks()
         for task in reversed(tasks):
             self.task_queue.add_task(task)

superagi/agent/queue_step_handler.py

@@ -1,3 +1,4 @@
+import ast
 import time
 
 import numpy as np
@@ -76,7 +77,7 @@ def _consume_from_queue(self, task_queue: TaskQueue):
     def _process_reply(self, task_queue: TaskQueue, assistant_reply: str):
         assistant_reply = JsonCleaner.extract_json_array_section(assistant_reply)
         print("Queue reply:", assistant_reply)
-        task_array = np.array(eval(assistant_reply)).flatten().tolist()
+        task_array = np.array(ast.literal_eval(assistant_reply)).flatten().tolist()
         for task in task_array:
             task_queue.add_task(str(task))
             logger.info("RAMRAM: Added task to queue: ", task)

superagi/controllers/knowledges.py

@@ -1,3 +1,4 @@
+import ast
 from fastapi_sqlalchemy import db
 from fastapi import HTTPException, Depends, Query, status
 from fastapi import APIRouter
@@ -154,7 +155,7 @@ def install_selected_knowledge(knowledge_name: str, vector_db_index_id: int, org
 def uninstall_selected_knowledge(knowledge_name: str, organisation = Depends(get_user_organisation)):
     knowledge = db.session.query(Knowledges).filter(Knowledges.name == knowledge_name, Knowledges.organisation_id == organisation.id).first()
     knowledge_config = KnowledgeConfigs.get_knowledge_config_from_knowledge_id(db.session, knowledge.id)
-    vector_ids = eval(knowledge_config["vector_ids"])
+    vector_ids = ast.literal_eval(knowledge_config["vector_ids"])
     vector_db_index = VectordbIndices.get_vector_index_from_id(db.session, knowledge.vector_db_index_id)
     vector = Vectordbs.get_vector_db_from_id(db.session, vector_db_index.vector_db_id)
     db_creds = VectordbConfigs.get_vector_db_config_from_db_id(db.session, vector.id)

superagi/helper/resource_helper.py

@@ -96,7 +96,7 @@ def get_resource_path(cls, file_name: str):
         Args:
             file_name (str): The name of the file.
         """
-        return ResourceHelper.get_root_output_dir() + file_name
+        return ResourceHelper._validate_path_containment(ResourceHelper.get_root_output_dir(), file_name)
 
     @classmethod
     def get_root_output_dir(cls):
@@ -124,6 +124,14 @@ def get_root_input_dir(cls):
             root_dir = os.getcwd() + "/"
         return root_dir
 
+    @staticmethod
+    def _validate_path_containment(root_dir: str, file_name: str) -> str:
+        final_path = os.path.abspath(root_dir + file_name)
+        root_dir_abs = os.path.abspath(root_dir)
+        if not final_path.startswith(root_dir_abs + os.sep) and final_path != root_dir_abs:
+            raise ValueError(f"Path traversal detected: {file_name}")
+        return final_path
+
     @classmethod
     def get_agent_write_resource_path(cls, file_name: str, agent: Agent, agent_execution: AgentExecution):
         """Get agent resource path to write files
@@ -140,7 +148,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._validate_path_containment(root_dir, file_name)
         return final_path
 
     @staticmethod
@@ -162,15 +170,16 @@ def get_agent_read_resource_path(cls, file_name, agent: Agent, agent_execution:
             agent (Agent): The agent corresponding to resource.
             agent_execution (AgentExecution): The agent execution corresponding to the resource.
         """
-        final_path = ResourceHelper.get_root_input_dir() + file_name
+        input_root_dir = ResourceHelper.get_root_input_dir()
+        final_path = ResourceHelper._validate_path_containment(input_root_dir, file_name)
         if "{agent_id}" in final_path:
             final_path = ResourceHelper.get_formatted_agent_level_path(
                 agent=agent,
                 path=final_path)
         output_root_dir = ResourceHelper.get_root_output_dir()
         if final_path is None or cls.__check_file_path_exists(final_path):
             if output_root_dir is not None:
-                final_path = ResourceHelper.get_root_output_dir() + file_name
+                final_path = ResourceHelper._validate_path_containment(output_root_dir, file_name)
                 if "{agent_id}" in final_path:
                     final_path = ResourceHelper.get_formatted_agent_level_path(
                         agent=agent,

superagi/models/agent.py

@@ -113,7 +113,7 @@ def eval_agent_config(cls, key, value):
         elif key in ["project_id", "memory_window", "max_iterations", "iteration_interval"]:
             return int(value)
         elif key in ["goal", "constraints", "instruction", "is_deleted"]:
-            return eval(value)
+            return ast.literal_eval(value)
         elif key == "tools":
             return list(ast.literal_eval(value))
 

superagi/models/agent_execution_config.py

@@ -107,7 +107,7 @@ def eval_agent_config(cls, key, value):
         """
 
         if key == "goal" or key == "instruction" or key == "tools":
-            return eval(value)
+            return ast.literal_eval(value)
 
     @classmethod
     def build_agent_execution_config(cls, session, agent, results_agent, results_agent_execution, total_calls, total_tokens):
@@ -120,7 +120,7 @@ def build_agent_execution_config(cls, session, agent, results_agent, results_age
 
         # Construct the response
         if 'goal' in results_agent_dict:
-            results_agent_dict['goal'] = eval(results_agent_dict['goal'])
+            results_agent_dict['goal'] = ast.literal_eval(results_agent_dict['goal'])
 
         if "toolkits" in results_agent_dict:
             results_agent_dict["toolkits"] = list(ast.literal_eval(results_agent_dict["toolkits"]))
@@ -130,10 +130,10 @@ def build_agent_execution_config(cls, session, agent, results_agent, results_age
             tools = session.query(Tool).filter(Tool.id.in_(results_agent_dict["tools"])).all()
             results_agent_dict["tools"] = tools
         if 'instruction' in results_agent_dict:
-            results_agent_dict['instruction'] = eval(results_agent_dict['instruction'])
+            results_agent_dict['instruction'] = ast.literal_eval(results_agent_dict['instruction'])
 
         if 'constraints' in results_agent_dict:
-            results_agent_dict['constraints'] = eval(results_agent_dict['constraints'])
+            results_agent_dict['constraints'] = ast.literal_eval(results_agent_dict['constraints'])
 
         results_agent_dict["name"] = agent.name
         agent_workflow = AgentWorkflow.find_by_id(session, agent.agent_workflow_id)
@@ -158,7 +158,7 @@ def build_scheduled_agent_execution_config(cls, session, agent, results_agent, t
 
         # Construct the response
         if 'goal' in results_agent_dict:
-            results_agent_dict['goal'] = eval(results_agent_dict['goal'])
+            results_agent_dict['goal'] = ast.literal_eval(results_agent_dict['goal'])
 
         if "toolkits" in results_agent_dict:
             results_agent_dict["toolkits"] = list(ast.literal_eval(results_agent_dict["toolkits"]))
@@ -168,10 +168,10 @@ def build_scheduled_agent_execution_config(cls, session, agent, results_agent, t
             tools = session.query(Tool).filter(Tool.id.in_(results_agent_dict["tools"])).all()
             results_agent_dict["tools"] = tools
         if 'instruction' in results_agent_dict:
-            results_agent_dict['instruction'] = eval(results_agent_dict['instruction'])
+            results_agent_dict['instruction'] = ast.literal_eval(results_agent_dict['instruction'])
 
         if 'constraints' in results_agent_dict:
-            results_agent_dict['constraints'] = eval(results_agent_dict['constraints'])
+            results_agent_dict['constraints'] = ast.literal_eval(results_agent_dict['constraints'])
 
         results_agent_dict["name"] = agent.name
         agent_workflow = AgentWorkflow.find_by_id(session, agent.agent_workflow_id)

superagi/models/agent_template.py

@@ -1,3 +1,4 @@
+import ast
 import json
 
 import requests
@@ -221,6 +222,6 @@ def eval_agent_config(cls, key, value):
             else:
                 return None
         elif key == "goal" or key == "constraints" or key == "instruction":
-            return eval(value)
+            return ast.literal_eval(value)
         elif key == "tools":
-            return [str(x) for x in eval(value)]
+            return [str(x) for x in ast.literal_eval(value)]