feat(security): add forbidden command filtering for device configurations

- Import filter_forbidden_commands utility from command_filter module
- Add _filter_forbidden_commands_from_device_configs method to filter out restricted commands before execution
- Store blocked commands information and log filtered commands for audit purposes
- Update _process_task_results to include blocked commands info in response
- Prevent execution of potentially dangerous commands while maintaining transparency about filtered content

gns3server/agent/gns3_copilot/config/forbidden_commands.txt

@@ -0,0 +1,64 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
+#
+# This file is part of GNS3-Copilot project.
+#
+# GNS3-Copilot is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the
+# Free Software Foundation, either version 3 of the License, or (at your
+# option) any later version.
+#
+# GNS3-Copilot is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
+# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+# for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
+#
+# Copyright (C) 2025 Guobin Yue
+# Author: Guobin Yue
+#
+# Project Home: https://github.com/yueguobin/gns3-copilot
+#
+
+# Forbidden Commands List
+# =======================
+#
+# This file contains commands that are blocked from execution in GNS3-Copilot tools.
+#
+# Why block these commands?
+# ------------------------
+# Commands listed here may:
+# - Run longer than the tool timeout (e.g., traceroute)
+# - Leave the device console in an unusable state for subsequent commands
+# - Generate excessive traffic that could impact lab performance
+#
+# Format
+# ------
+# - One command pattern per line
+# - Simple substring matching (case-insensitive)
+# - Empty lines and lines starting with # are ignored
+# - Match is performed on the beginning of each command
+#
+# Examples
+# --------
+# traceroute          # Blocks "traceroute 8.8.8.8", "traceroute google.com", etc.
+# debug               # Blocks "debug ip routing", "debug ospf events", etc.
+# test                # Blocks "test ..., "test memory", etc.
+
+# Network diagnostic commands that may timeout
+traceroute
+tracepath
+tracert
+
+# Ping flood variants
+ping -f
+ping -f
+
+# Debug commands that may destabilize devices
+debug
+
+# Test commands that may affect device stability
+test

gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py

@@ -45,6 +45,7 @@ from nornir_netmiko.tasks import netmiko_send_config
 
 from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
 from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
+from gns3server.agent.gns3_copilot.utils.command_filter import filter_forbidden_commands
 
 # config log
 logger = logging.getLogger(__name__)
@@ -178,6 +179,11 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
         if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]:
             return device_configs_list
 
+        # Filter forbidden commands and store blocked commands info
+        device_configs_list, blocked_commands_map = self._filter_forbidden_commands_from_device_configs(
+            device_configs_list
+        )
+
         # Create a mapping of device names to their configuration commands
         device_configs_map = self._configs_map(device_configs_list)
 
@@ -205,7 +211,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
             )
 
             # Process results for all devices
-            results = self._process_task_results(device_configs_list, hosts_data, task_result)
+            results = self._process_task_results(device_configs_list, hosts_data, task_result, blocked_commands_map)
 
         except Exception as e:
             # Overall execution failed
@@ -350,6 +356,47 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
         uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
         return bool(re.match(uuid_pattern, project_id, re.IGNORECASE))
 
+    def _filter_forbidden_commands_from_device_configs(
+        self, device_configs_list: list[dict[str, Any]]
+    ) -> tuple[list[dict[str, Any]], dict[str, dict[str, str]]]:
+        """
+        Filter out forbidden commands from device configurations.
+
+        Args:
+            device_configs_list: List of device configurations with config_commands.
+
+        Returns:
+            A tuple of (filtered_device_configs_list, blocked_commands_map):
+            - filtered_device_configs_list: Device configs with forbidden commands removed.
+            - blocked_commands_map: Dict mapping device names to their blocked commands info.
+        """
+        filtered_list = []
+        blocked_commands_map: dict[str, dict[str, str]] = {}
+
+        for device_config in device_configs_list:
+            device_name = device_config["device_name"]
+            commands = device_config["config_commands"]
+
+            # Filter commands
+            allowed_commands, blocked_info = filter_forbidden_commands(commands)
+
+            # Update device config with allowed commands only
+            filtered_config = device_config.copy()
+            filtered_config["config_commands"] = allowed_commands
+            filtered_list.append(filtered_config)
+
+            # Store blocked commands info if any
+            if blocked_info:
+                blocked_commands_map[device_name] = blocked_info
+                logger.info(
+                    "Device %s: %d command(s) blocked: %s",
+                    device_name,
+                    len(blocked_info),
+                    list(blocked_info.keys()),
+                )
+
+        return filtered_list, blocked_commands_map
+
     def _configs_map(self, device_config_list: list[dict[str, Any]]) -> dict[str, list[str]]:
         """Create a mapping of device names to their configuration commands."""
         device_configs_map = {}
@@ -443,6 +490,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
         device_configs_list: list[dict[str, Any]],
         hosts_data: dict[str, dict[str, Any]],
         task_result: AggregatedResult,
+        blocked_commands_map: dict[str, dict[str, str]],
     ) -> list[dict[str, Any]]:
         """Process the task results and format them for return."""
         results = []
@@ -450,6 +498,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
         for device_config in device_configs_list:
             device_name = device_config["device_name"]
             config_commands = device_config["config_commands"]
+            blocked_commands_info = blocked_commands_map.get(device_name, {})
 
             # Check if device is in topology
             if device_name not in hosts_data:
@@ -458,6 +507,10 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
                     "status": "failed",
                     "error": (f"Device '{device_name}' not found in topology or missing console_port"),
                 }
+                # Add blocked commands info if any
+                if blocked_commands_info:
+                    device_result["blocked_commands"] = list(blocked_commands_info.keys())
+                    device_result["blocked_info"] = blocked_commands_info
                 results.append(device_result)
                 continue
 
@@ -468,6 +521,10 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
                     "status": "failed",
                     "error": (f"Device '{device_name}' not found in task results"),
                 }
+                # Add blocked commands info if any
+                if blocked_commands_info:
+                    device_result["blocked_commands"] = list(blocked_commands_info.keys())
+                    device_result["blocked_info"] = blocked_commands_info
                 results.append(device_result)
                 continue
 
@@ -486,6 +543,14 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
                 device_result["output"] = multi_result[0].result
                 device_result["config_commands"] = config_commands
 
+            # Add blocked commands info if any
+            if blocked_commands_info:
+                device_result["blocked_commands"] = list(blocked_commands_info.keys())
+                device_result["blocked_info"] = blocked_commands_info
+                # Update status if some commands were blocked but execution succeeded
+                if device_result["status"] == "success":
+                    device_result["status"] = "partial_success"
+
             results.append(device_result)
 
         return results

gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py

@@ -45,6 +45,7 @@ from nornir_netmiko.tasks import netmiko_multiline
 
 from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
 from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
+from gns3server.agent.gns3_copilot.utils.command_filter import filter_forbidden_commands
 
 # config log
 logger = logging.getLogger(__name__)
@@ -185,6 +186,11 @@ class ExecuteMultipleDeviceCommands(BaseTool):
         if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]:
             return device_configs_list
 
+        # Filter forbidden commands and store blocked commands info
+        device_configs_list, blocked_commands_map = self._filter_forbidden_commands_from_device_configs(
+            device_configs_list
+        )
+
         # Create a mapping of device names to their display commands
         device_configs_map = self._configs_map(device_configs_list)
 
@@ -212,7 +218,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
             )
 
             # Process results for all devices
-            results = self._process_task_results(device_configs_list, hosts_data, task_result)
+            results = self._process_task_results(device_configs_list, hosts_data, task_result, blocked_commands_map)
 
         except Exception as e:
             # Overall execution failed
@@ -369,6 +375,47 @@ class ExecuteMultipleDeviceCommands(BaseTool):
         uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
         return bool(re.match(uuid_pattern, project_id, re.IGNORECASE))
 
+    def _filter_forbidden_commands_from_device_configs(
+        self, device_configs_list: list[dict[str, Any]]
+    ) -> tuple[list[dict[str, Any]], dict[str, dict[str, str]]]:
+        """
+        Filter out forbidden commands from device configurations.
+
+        Args:
+            device_configs_list: List of device configurations with commands.
+
+        Returns:
+            A tuple of (filtered_device_configs_list, blocked_commands_map):
+            - filtered_device_configs_list: Device configs with forbidden commands removed.
+            - blocked_commands_map: Dict mapping device names to their blocked commands info.
+        """
+        filtered_list = []
+        blocked_commands_map: dict[str, dict[str, str]] = {}
+
+        for device_config in device_configs_list:
+            device_name = device_config["device_name"]
+            commands = device_config["commands"]
+
+            # Filter commands
+            allowed_commands, blocked_info = filter_forbidden_commands(commands)
+
+            # Update device config with allowed commands only
+            filtered_config = device_config.copy()
+            filtered_config["commands"] = allowed_commands
+            filtered_list.append(filtered_config)
+
+            # Store blocked commands info if any
+            if blocked_info:
+                blocked_commands_map[device_name] = blocked_info
+                logger.info(
+                    "Device %s: %d command(s) blocked: %s",
+                    device_name,
+                    len(blocked_info),
+                    list(blocked_info.keys()),
+                )
+
+        return filtered_list, blocked_commands_map
+
     def _configs_map(self, device_config_list: list[dict[str, Any]]) -> dict[str, list[str]]:
         """Create a mapping of device names to their diagnostic commands."""
         device_diagnostic_map = {}
@@ -462,6 +509,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
         device_configs_list: list[dict[str, Any]],
         hosts_data: dict[str, dict[str, Any]],
         task_result: AggregatedResult,
+        blocked_commands_map: dict[str, dict[str, str]],
     ) -> list[dict[str, Any]]:
         """Process the task results and format them for return."""
         results = []
@@ -469,6 +517,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
         for device_config in device_configs_list:
             device_name = device_config["device_name"]
             diagnostic_commands = device_config["commands"]
+            blocked_commands_info = blocked_commands_map.get(device_name, {})
 
             # Check if device is in topology
             if device_name not in hosts_data:
@@ -477,6 +526,10 @@ class ExecuteMultipleDeviceCommands(BaseTool):
                     "status": "failed",
                     "error": (f"Device '{device_name}' not found in topology or missing console_port"),
                 }
+                # Add blocked commands info if any
+                if blocked_commands_info:
+                    device_result["blocked_commands"] = list(blocked_commands_info.keys())
+                    device_result["blocked_info"] = blocked_commands_info
                 results.append(device_result)
                 continue
 
@@ -487,6 +540,10 @@ class ExecuteMultipleDeviceCommands(BaseTool):
                     "status": "failed",
                     "error": (f"Device '{device_name}' not found in task results"),
                 }
+                # Add blocked commands info if any
+                if blocked_commands_info:
+                    device_result["blocked_commands"] = list(blocked_commands_info.keys())
+                    device_result["blocked_info"] = blocked_commands_info
                 results.append(device_result)
                 continue
 
@@ -505,6 +562,14 @@ class ExecuteMultipleDeviceCommands(BaseTool):
                 device_result["output"] = multi_result[0].result
                 device_result["diagnostic_commands"] = diagnostic_commands
 
+            # Add blocked commands info if any
+            if blocked_commands_info:
+                device_result["blocked_commands"] = list(blocked_commands_info.keys())
+                device_result["blocked_info"] = blocked_commands_info
+                # Update status if some commands were blocked but execution succeeded
+                if device_result["status"] == "success":
+                    device_result["status"] = "partial_success"
+
             results.append(device_result)
 
         return results

gns3server/agent/gns3_copilot/utils/command_filter.py

@@ -0,0 +1,217 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
+#
+# This file is part of GNS3-Copilot project.
+#
+# GNS3-Copilot is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the
+# Free Software Foundation, either version 3 of the License, or (at your
+# option) any later version.
+#
+# GNS3-Copilot is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
+# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+# for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
+#
+# Copyright (C) 2025 Guobin Yue
+# Author: Guobin Yue
+#
+# Project Home: https://github.com/yueguobin/gns3-copilot
+#
+
+"""
+Command filter module for GNS3-Copilot.
+
+This module provides functionality to filter out dangerous or long-running
+commands that may cause issues with tool execution timeouts or device console
+availability.
+"""
+
+import logging
+import os
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+
+def _get_gns3_copilot_root() -> Path:
+    """Get the root directory of the GNS3-Copilot project."""
+    # Get the directory containing this file
+    current_file = Path(__file__).resolve()
+    # Go up to the gns3_copilot directory (utils parent)
+    return current_file.parent.parent
+
+# Default forbidden commands (fallback if file not found)
+DEFAULT_FORBIDDEN_COMMANDS = [
+    "traceroute",
+    "tracepath",
+    "tracert",
+    "ping -f",
+    "debug",
+    "test",
+]
+
+# Cache for forbidden commands to avoid repeated file reads
+_forbidden_commands_cache: list[str] | None = None
+
+
+def _get_forbidden_commands_file_path() -> Path:
+    """Get the path to the forbidden commands configuration file."""
+    return _get_gns3_copilot_root() / "config" / "forbidden_commands.txt"
+
+
+def _load_forbidden_commands() -> list[str]:
+    """
+    Load forbidden commands from the configuration file.
+
+    Returns:
+        List of forbidden command patterns. If the file cannot be read,
+        returns the default list.
+    """
+    global _forbidden_commands_cache
+
+    # Return cached value if available
+    if _forbidden_commands_cache is not None:
+        return _forbidden_commands_cache
+
+    file_path = _get_forbidden_commands_file_path()
+
+    try:
+        if not file_path.exists():
+            logger.warning(
+                "Forbidden commands file not found: %s. Using default list.",
+                file_path,
+            )
+            _forbidden_commands_cache = DEFAULT_FORBIDDEN_COMMANDS.copy()
+            return _forbidden_commands_cache
+
+        with open(file_path, "r", encoding="utf-8") as f:
+            forbidden_commands = []
+            for line in f:
+                line = line.strip()
+
+                # Skip empty lines and comments
+                if not line or line.startswith("#"):
+                    continue
+
+                forbidden_commands.append(line.lower())
+
+        if not forbidden_commands:
+            logger.warning("No forbidden commands found in %s. Using default list.", file_path)
+            _forbidden_commands_cache = DEFAULT_FORBIDDEN_COMMANDS.copy()
+        else:
+            logger.info(
+                "Loaded %d forbidden command patterns from %s",
+                len(forbidden_commands),
+                file_path,
+            )
+            _forbidden_commands_cache = forbidden_commands
+
+        return _forbidden_commands_cache
+
+    except Exception as e:
+        logger.error(
+            "Error reading forbidden commands file %s: %s. Using default list.",
+            file_path,
+            e,
+        )
+        _forbidden_commands_cache = DEFAULT_FORBIDDEN_COMMANDS.copy()
+        return _forbidden_commands_cache
+
+
+def reload_forbidden_commands() -> None:
+    """
+    Reload the forbidden commands list from the configuration file.
+
+    This clears the cache and forces a reload from the file on the next
+    call to filter_forbidden_commands() or get_forbidden_commands().
+
+    Use this after modifying the forbidden_commands.txt file to apply
+    changes without restarting the GNS3 server.
+    """
+    global _forbidden_commands_cache
+    _forbidden_commands_cache = None
+    logger.info("Forbidden commands cache cleared. Will reload on next access.")
+
+
+def get_forbidden_commands() -> list[str]:
+    """
+    Get the current list of forbidden command patterns.
+
+    Returns:
+        List of forbidden command patterns (lowercase).
+    """
+    return _load_forbidden_commands()
+
+
+def is_command_forbidden(command: str) -> bool:
+    """
+    Check if a single command is forbidden.
+
+    Args:
+        command: The command string to check.
+
+    Returns:
+        True if the command matches a forbidden pattern, False otherwise.
+    """
+    forbidden_commands = _load_forbidden_commands()
+    command_lower = command.strip().lower()
+
+    for forbidden_pattern in forbidden_commands:
+        if command_lower.startswith(forbidden_pattern):
+            return True
+
+    return False
+
+
+def filter_forbidden_commands(
+    commands: list[str],
+) -> tuple[list[str], dict[str, str]]:
+    """
+    Filter out forbidden commands from a list of commands.
+
+    Args:
+        commands: List of command strings to filter.
+
+    Returns:
+        A tuple of (allowed_commands, blocked_commands_info):
+        - allowed_commands: List of commands that are not forbidden.
+        - blocked_commands_info: Dict mapping blocked commands to their reasons.
+    """
+    allowed_commands: list[str] = []
+    blocked_commands_info: dict[str, str] = {}
+
+    for command in commands:
+        if is_command_forbidden(command):
+            # Find which pattern matched
+            forbidden_commands = _load_forbidden_commands()
+            command_lower = command.strip().lower()
+            matched_pattern = None
+
+            for pattern in forbidden_commands:
+                if command_lower.startswith(pattern):
+                    matched_pattern = pattern
+                    break
+
+            reason = (
+                f"Command '{command}' is not allowed because it matches the "
+                f"forbidden pattern '{matched_pattern}'. "
+                f"This command may run longer than the tool timeout or leave "
+                f"the device console unavailable for subsequent commands."
+            )
+            blocked_commands_info[command] = reason
+        else:
+            allowed_commands.append(command)
+
+    if blocked_commands_info:
+        logger.info(
+            "Filtered %d command(s): %s",
+            len(blocked_commands_info),
+            list(blocked_commands_info.keys()),
+        )
+
+    return allowed_commands, blocked_commands_info