security(api): add path traversal and CRLF injection protection

- Add is_safe_path_target() and get_safe_file_path() utilities
- Validate history id and snapshot target parameters in API endpoints
- Sanitize config string values to prevent CRLF injection

comfyui_manager/legacy/manager_core.py

@@ -42,7 +42,7 @@ from ..common.enums import NetworkMode, SecurityLevel, DBMode
 from ..common import context
 
 
-version_code = [4, 0, 4]
+version_code = [4, 0, 5]
 version_str = f"V{version_code[0]}.{version_code[1]}" + (f'.{version_code[2]}' if len(version_code) > 2 else '')
 
 
@@ -1612,6 +1612,11 @@ def write_config():
         'db_mode': get_config()['db_mode'],
     }
 
+    # Sanitize all string values to prevent CRLF injection attacks
+    for key, value in config['default'].items():
+        if isinstance(value, str):
+            config['default'][key] = value.replace('\r', '').replace('\n', '').replace('\x00', '')
+
     directory = os.path.dirname(context.manager_config_path)
     if not os.path.exists(directory):
         os.makedirs(directory)

comfyui_manager/legacy/manager_server.py

@@ -814,8 +814,13 @@ async def get_history_list(request):
 @routes.get("/v2/manager/queue/history")
 async def get_history(request):
     try:
-        json_name = request.rel_url.query["id"]+'.json'
-        batch_path = os.path.join(context.manager_batch_history_path, json_name)
+        history_id = request.rel_url.query["id"]
+
+        # Prevent path traversal attacks
+        batch_path = manager_security.get_safe_file_path(history_id, context.manager_batch_history_path)
+        if batch_path is None:
+            logging.warning(f"[Security] Invalid history id rejected: {history_id}")
+            return web.Response(text="Invalid history id", status=400)
 
         with open(batch_path, 'r', encoding='utf-8') as file:
             json_str = file.read()
@@ -1159,7 +1164,12 @@ async def remove_snapshot(request):
     try:
         target = request.rel_url.query["target"]
 
-        path = os.path.join(context.manager_snapshot_path, f"{target}.json")
+        # Prevent path traversal attacks
+        path = manager_security.get_safe_file_path(target, context.manager_snapshot_path)
+        if path is None:
+            logging.warning(f"[Security] Invalid snapshot target rejected: {target}")
+            return web.Response(text="Invalid target", status=400)
+
         if os.path.exists(path):
             os.remove(path)
 
@@ -1177,7 +1187,12 @@ async def restore_snapshot(request):
     try:
         target = request.rel_url.query["target"]
 
-        path = os.path.join(context.manager_snapshot_path, f"{target}.json")
+        # Prevent path traversal attacks
+        path = manager_security.get_safe_file_path(target, context.manager_snapshot_path)
+        if path is None:
+            logging.warning(f"[Security] Invalid snapshot target rejected: {target}")
+            return web.Response(text="Invalid target", status=400)
+
         if os.path.exists(path):
             if not os.path.exists(context.manager_startup_script_path):
                 os.makedirs(context.manager_startup_script_path)