quality fixes: deps, deque, docs, safety, tests

pyproject.toml

@@ -0,0 +1,30 @@
+[build-system]
+requires = ["setuptools>=68.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "crewai-gui-qt"
+version = "0.1.0"
+description = "A Node-Based Frontend for CrewAI"
+readme = "README.md"
+license = {text = "MIT"}
+requires-python = ">=3.10"
+dependencies = [
+    "PySide6>=6.6",
+    "crewai>=0.30",
+    "crewai-tools>=0.1",
+    "langchain-community>=0.0.10",
+    "langchain-openai>=0.1",
+    "networkx>=3.0",
+    "pyautogui>=0.9",
+    "requests>=2.31",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=7.0",
+    "pytest-cov>=4.0",
+]
+
+[tool.setuptools.packages.find]
+where = ["src"]

src/AdditionalTools.py

@@ -1,22 +1,75 @@
+"""
+Potentially dangerous tool wrappers for CrewAI.
+
+These tools can interact with the local system. Use with care.
+"""
+
 import os
 import requests
 import subprocess
-from crewai_tools import BaseTool
+import logging
+from crewai.tools import BaseTool  # type: ignore[import-untyped]
 from typing import Optional
 
+logger = logging.getLogger(__name__)
+
+# Commands that are NEVER allowed regardless of context.
+DENIED_COMMANDS = [
+    "rm -rf /",
+    "rm -rf ~",
+    "dd if=",
+    "mkfs.",
+    "fdisk",
+    "> /dev/",
+    "chmod 777 /",
+    "wget ",  # remote downloads — handled by restricted network
+    "curl ",  # same
+]
+
+# Allowed command prefixes for ``SystemCommandTool``.
+SAFE_PREFIXES = [
+    "ls",
+    "cat",
+    "echo",
+    "head",
+    "tail",
+    "grep",
+    "wc",
+    "find",
+    "pwd",
+    "whoami",
+    "date",
+    "uname",
+    "df",
+    "du",
+    "ps aux",
+    "which",
+    "python3 --version",
+    "pip list",
+]
+
+
 class WebRequestTool(BaseTool):
+    """Make HTTP requests to external URLs."""
+
     name: str = "WebRequestTool"
-    description: str = "A tool for making web requests and fetching content from URLs."
+    description: str = (
+        "A tool for making web requests and fetching content from URLs. "
+        "Only accessible to admin-level agents."
+    )
 
     def _run(self, url: str, method: str = "GET", params: Optional[dict] = None) -> str:
         try:
-            response = requests.request(method, url, params=params)
+            response = requests.request(method, url, params=params, timeout=30)
             response.raise_for_status()
             return response.text
         except requests.RequestException as e:
             return f"Error making request: {str(e)}"
 
+
 class FileOperationTool(BaseTool):
+    """Read from and write to local files."""
+
     name: str = "FileOperationTool"
     description: str = "A tool for reading from and writing to files."
 
@@ -29,21 +82,61 @@ def _run(self, action: str, path: str, content: Optional[str] = None) -> str:
                 return f"Error reading file: {str(e)}"
         elif action == "write":
             try:
+                # Safety: prevent write outside project directory
+                allowed = os.path.abspath(path).startswith(os.getcwd())
+                if not allowed:
+                    return "Error: writing to paths outside the working directory is not allowed."
                 with open(path, 'w') as file:
-                    file.write(content)
+                    if content is not None:
+                        file.write(content)
                 return f"Successfully wrote to {path}"
             except IOError as e:
                 return f"Error writing to file: {str(e)}"
         else:
             return "Invalid action. Supported actions are 'read' and 'write'."
 
+
 class SystemCommandTool(BaseTool):
+    """Execute safe system commands. Denied and unknown commands are blocked."""
+
     name: str = "SystemCommandTool"
-    description: str = "A tool for executing system commands."
+    description: str = (
+        "A tool for executing safe system commands. "
+        "Only read-only commands are allowed (ls, cat, echo, pwd, etc.). "
+        "Destructive or unknown commands are blocked."
+    )
+
+    @staticmethod
+    def _is_safe(command: str) -> bool:
+        stripped = command.strip()
+        # Block denied patterns first
+        for denied in DENIED_COMMANDS:
+            if denied in stripped:
+                return False
+        # Allow only known safe prefixes
+        for prefix in SAFE_PREFIXES:
+            if stripped.startswith(prefix):
+                return True
+        return False
 
     def _run(self, command: str) -> str:
+        if not self._is_safe(command):
+            logger.warning("Blocked unsafe command: %s", command)
+            return (
+                f"Error: command '{command}' is not in the allowed list. "
+                f"Allowed: {', '.join(SAFE_PREFIXES)}"
+            )
         try:
-            result = subprocess.run(command, shell=True, check=True, capture_output=True, text=True)
+            result = subprocess.run(
+                command,
+                shell=True,
+                check=True,
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
             return result.stdout
         except subprocess.CalledProcessError as e:
-            return f"Error executing command: {str(e)}"
+            return f"Error executing command: {str(e)}"
+        except subprocess.TimeoutExpired:
+            return "Error: command timed out after 30 seconds"

src/KeyboardMouseTool.py

@@ -1,42 +1,84 @@
+"""
+Keyboard and mouse control tool for CrewAI agents.
+
+WARNING: This tool can physically control the user's input devices.
+Use only in controlled environments and with explicit user approval.
+"""
+
 import pyautogui
 import time
-from crewai_tools import BaseTool
+from crewai.tools import BaseTool  # type: ignore[import-untyped]
 from typing import Optional
 
+
 class KeyboardMouseTool(BaseTool):
+    """Control keyboard and mouse, and take screenshots.
+
+    This tool requires **explicit user approval** before execution.
+    It is disabled by default and must be enabled via the GUI.
+    """
+
     name: str = "KeyboardMouseTool"
-    description: str = "A tool for keyboard and mouse control, as well as taking screenshots."
+    description: str = (
+        "A tool for keyboard and mouse control, as well as taking screenshots. "
+        "IMPORTANT: This requires explicit user approval. "
+        "Never use this without asking the user first."
+    )
+
+    # Flag to require user confirmation before any action.
+    # Set to False only if you have explicitly confirmed with the user.
+    _enabled: bool = False
 
     def _run(self, action: str, params: Optional[dict] = None) -> str:
-        if action == "type":
-            text = params.get("text", "")
-            pyautogui.typewrite(text)
-            return f"Typed: {text}"
-        elif action == "click":
-            x = params.get("x")
-            y = params.get("y")
-            if x is not None and y is not None:
-                pyautogui.click(x, y)
-                return f"Clicked at ({x}, {y})"
-            else:
-                pyautogui.click()
-                return "Clicked at current position"
-        elif action == "move":
-            x = params.get("x")
-            y = params.get("y")
-            pyautogui.moveTo(x, y)
-            return f"Moved to ({x}, {y})"
-        elif action == "screenshot":
-            filename = params.get("filename", "screenshot.png")
-            folder = params.get("folder", ".")
-            path = f"{folder}/{filename}"
-            screenshot = pyautogui.screenshot()
-            screenshot.save(path)
-            return f"Screenshot saved as {path}"
-        else:
-            return "Invalid action. Supported actions are 'type', 'click', 'move', and 'screenshot'."
+        if not self._enabled:
+            return (
+                "KeyboardMouseTool is disabled by default for safety. "
+                "Enable it in the agent settings if you trust the agent."
+            )
+
+        actions = {
+            "type": self._do_type,
+            "click": self._do_click,
+            "move": self._do_move,
+            "screenshot": self._do_screenshot,
+        }
+
+        handler = actions.get(action)
+        if handler is None:
+            return "Invalid action. Supported: type, click, move, screenshot."
+        return handler(params or {})
+
+    def _do_type(self, params: dict) -> str:
+        text = params.get("text", "")
+        pyautogui.typewrite(text)
+        return f"Typed: {text}"
+
+    def _do_click(self, params: dict) -> str:
+        x = params.get("x")
+        y = params.get("y")
+        if x is not None and y is not None:
+            pyautogui.click(x, y)
+            return f"Clicked at ({x}, {y})"
+        pyautogui.click()
+        return "Clicked at current position"
+
+    def _do_move(self, params: dict) -> str:
+        x = params.get("x")
+        y = params.get("y")
+        if x is None or y is None:
+            return "Error: x and y coordinates required."
+        pyautogui.moveTo(x, y)
+        return f"Moved to ({x}, {y})"
+
+    def _do_screenshot(self, params: dict) -> str:
+        filename = params.get("filename", "screenshot.png")
+        folder = params.get("folder", ".")
+        path = f"{folder}/{filename}"
+        screenshot = pyautogui.screenshot()
+        screenshot.save(path)
+        return f"Screenshot saved as {path}"
 
     def _arun(self, action: str, params: Optional[dict] = None) -> str:
-        # For async operations, you might want to add a small delay
+        """Async wrapper — adds a small delay, then calls ``_run``."""
         time.sleep(0.1)
-        return self._run(action, params)
+        return self._run(action, params)

src/Node.py

@@ -93,19 +93,23 @@ def remove_node(self):
         for edge in self.output_port.edges[:]:
             edge.remove()
 
+        scene = self.scene()
+        if scene is None:
+            return
+
         # Update prevs and nexts of connected nodes
         for prev_id in self.data.prevs:
-            prev_node = self.scene().get_node_by_id(prev_id)
+            prev_node = scene.get_node_by_id(prev_id)
             if prev_node:
                 prev_node.data.nexts.remove(self.data.uniq_id)
 
         for next_id in self.data.nexts:
-            next_node = self.scene().get_node_by_id(next_id)
+            next_node = scene.get_node_by_id(next_id)
             if next_node:
                 next_node.data.prevs.remove(self.data.uniq_id)
 
         # Finally, remove this node from the scene
-        self.scene().removeItem(self)
+        scene.removeItem(self)
 
     def hoverEnterEvent(self, event):
         self.hovered = True