From 8ba1f064d2bb8187e8ab0c548bd707afaa0ea2e5 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Sun, 24 May 2026 21:25:46 +0800 Subject: [PATCH] feat: add packet filter management tool for GNS3-Copilot fault injection Add comprehensive packet filter management functionality to GNS3-Copilot, enabling AI-powered fault injection scenarios with network simulation capabilities like latency, packet loss, and corruption. ## Changes ### New Features - **GNS3PacketFilterTool**: New LangChain tool for managing packet filters on GNS3 links with support for delay, packet loss, corruption, frequency_drop, and BPF filtering - Actions: get_available, set, get, clear - Integrated into troubleshooting_injection mode for fault scenarios ### API Integration - **Link.available_filters()**: Added method to custom_gns3fy.py Link class - Queries available filter types for specific links - API v3+ only (raises ValueError for v2 connectors) - Returns filter definitions with parameters and constraints ### Tool Integration - Added GNS3PacketFilterTool to TROUBLESHOOTING_INJECTION_MODE_TOOLS - Positioned as 3rd tool in fault injection workflow - Optimized for troubleshooting practice scenarios ## Files Modified - gns3server/agent/gns3_copilot/agent/gns3_copilot.py - gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py - gns3server/agent/gns3_copilot/tools_v2/__init__.py ## Files Added - gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py ## Testing - All validation tests passed - Version checking verified (v3+ only) - Tool integration confirmed in troubleshooting mode --- .../agent/gns3_copilot/agent/gns3_copilot.py | 2 + .../gns3_copilot/gns3_client/custom_gns3fy.py | 60 +++ .../agent/gns3_copilot/tools_v2/__init__.py | 4 + .../tools_v2/gns3_packet_filter.py | 365 ++++++++++++++++++ 4 files changed, 431 insertions(+) create mode 100644 gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py index a675d2074..b5173668c 100644 --- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py +++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py @@ -88,6 +88,7 @@ from gns3server.agent.gns3_copilot.tools_v2 import ( ) from gns3server.agent.gns3_copilot.tools_v2 import GNS3CreateNodeTool from gns3server.agent.gns3_copilot.tools_v2 import GNS3LinkTool +from gns3server.agent.gns3_copilot.tools_v2 import GNS3PacketFilterTool from gns3server.agent.gns3_copilot.tools_v2 import GNS3StartNodeTool from gns3server.agent.gns3_copilot.tools_v2 import GNS3StopNodeTool from gns3server.agent.gns3_copilot.tools_v2 import GNS3SuspendNodeTool @@ -144,6 +145,7 @@ LAB_AUTOMATION_ASSISTANT_MODE_TOOLS = [ TROUBLESHOOTING_INJECTION_MODE_TOOLS = [ ExecuteMultipleDeviceCommands(), # Get device configurations (READ-ONLY) ExecuteMultipleDeviceConfigCommands(), # Inject configuration changes + GNS3PacketFilterTool(), # Manage packet filters on links (delay, loss, corrupt, etc.) InjectionSkillsTool(), # Query injection skills and fault types GNS3TopologyTool(), # Get topology information ] diff --git a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py index cb3770d9f..d565975e7 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py +++ b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py @@ -1042,6 +1042,66 @@ class Link: # Update object self._update(_response.json()) + @verify_connector_and_id + def available_filters(self) -> list[dict[str, Any]]: + """ + Gets the list of available packet filters for this link. + + **NOTE:** This endpoint is only available in GNS3 API v3 or later. + Attempting to call this method with a v2 connector will raise an error. + + **Required Attributes:** + + - `project_id` + - `connector` (must be API v3 or later) + - `link_id` + + **Returns:** + + List of available filter types with their parameters (e.g., frequency_drop, + packet_loss, delay, corrupt, bpf). + + **Example:** + + ```python + >>> link = Link(project_id=, link_id=, connector=) + >>> filters = link.available_filters() + >>> print(filters) + [ + { + "type": "frequency_drop", + "name": "Frequency drop", + "description": "It will drop everything with a -1 frequency...", + "parameters": [...] + }, + ... + ] + ``` + """ + _conn = self.connector + _project_id = self.project_id + + if _conn is None: + raise ValueError("Gns3Connector not assigned under 'connector'") + if _project_id is None: + raise ValueError("Need to submit project_id") + + # Check API version - available_filters endpoint is only available in v3+ + if not hasattr(_conn, "api_version") or _conn.api_version < 3: + raise ValueError( + "The available_filters() method requires GNS3 API v3 or later. " + f"Current connector version: v{getattr(_conn, 'api_version', 2)}. " + "Please use api_version=3 when creating the Gns3Connector." + ) + + _url = ( + f"{_conn.base_url}/projects/{_project_id}/links/{self.link_id}/" + "available_filters" + ) + _response = _conn.http_call("get", _url) + + return cast(list[dict[str, Any]], _response.json()) + @dataclass(config=config) class Node: diff --git a/gns3server/agent/gns3_copilot/tools_v2/__init__.py b/gns3server/agent/gns3_copilot/tools_v2/__init__.py index f99450be4..49957fc79 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/__init__.py +++ b/gns3server/agent/gns3_copilot/tools_v2/__init__.py @@ -32,6 +32,7 @@ This package provides various tools for interacting with GNS3 network simulator: - Multiple device command execution using Nornir - VPCS device configuration using Netmiko - Node and link management +- Packet filter management Main modules: - config_tools_nornir: Multiple device configuration command execution tool using Nornir @@ -42,6 +43,7 @@ Main modules: - gns3_start_node: GNS3 node startup tool - gns3_get_node_temp: GNS3 template retrieval tool - gns3_update_node_name: GNS3 node name update tool +- gns3_packet_filter: GNS3 packet filter management tool Note: GNS3TopologyTool is now available from gns3_client package @@ -54,6 +56,7 @@ from .display_tools_nornir import ExecuteMultipleDeviceCommands from .gns3_create_link import GNS3LinkTool from .gns3_create_node import GNS3CreateNodeTool from .gns3_get_node_temp import GNS3TemplateTool +from .gns3_packet_filter import GNS3PacketFilterTool from .gns3_start_node import GNS3StartNodeQuickTool from .gns3_start_node import GNS3StartNodeTool from .gns3_stop_node import GNS3StopNodeTool @@ -79,6 +82,7 @@ __all__ = [ "ExecuteMultipleDeviceCommands", "GNS3CreateNodeTool", "GNS3LinkTool", + "GNS3PacketFilterTool", "GNS3StartNodeTool", "GNS3StartNodeQuickTool", "GNS3StopNodeTool", diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py new file mode 100644 index 000000000..e2a8dee49 --- /dev/null +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py @@ -0,0 +1,365 @@ +# 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 . +# +# Copyright (C) 2025 Yue Guobin (岳国宾) +# Author: Yue Guobin (岳国宾) +# +# Project Home: https://github.com/yueguobin/gns3-copilot +# + +""" + +GNS3 packet filter management tool for network simulation. + +Provides functionality to manage packet filters on GNS3 links, +including latency, packet loss, corruption, and BPF filtering. +""" + +import json +import logging +from pprint import pprint +from typing import Any + +from langchain.tools import BaseTool +from langchain_core.callbacks import CallbackManagerForToolRun + +from gns3server.agent.gns3_copilot.gns3_client import Link +from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector + +# Configure logging +logger = logging.getLogger(__name__) + + +class GNS3PacketFilterTool(BaseTool): + """ + A LangChain tool to manage packet filters on GNS3 links. + + Supports getting available filters, setting filters, and clearing filters + on network links to simulate various network conditions. + + **Input:** + A JSON object with project_id, link_id, action, and optional filter parameters. + + Example input for getting available filters: + { + "project_id": "uuid-of-project", + "link_id": "uuid-of-link", + "action": "get_available" + } + + Example input for setting filters: + { + "project_id": "uuid-of-project", + "link_id": "uuid-of-link", + "action": "set", + "filters": { + "delay": [100, 10], + "packet_loss": [5] + } + } + + Example input for getting current filters: + { + "project_id": "uuid-of-project", + "link_id": "uuid-of-link", + "action": "get" + } + + Example input for clearing filters: + { + "project_id": "uuid-of-project", + "link_id": "uuid-of-link", + "action": "clear" + } + + **Output:** + A dictionary containing the action result. + + For "get_available": returns list of available filter types + For "set": returns updated link information with applied filters + For "get": returns current filters configured on the link + For "clear": returns confirmation that filters were cleared + """ + + name: str = "manage_gns3_packet_filter" + description: str = """ + Manages packet filters on GNS3 links to inject network faults and simulate network conditions. + + This tool is primarily used for fault injection scenarios to create realistic network problems + for troubleshooting practice, such as latency, packet loss, and corruption. + + Supported actions: + - "get_available": Get list of available filter types for the link + - "set": Set packet filters on the link to inject network faults + - "get": Get current filters configured on the link + - "clear": Clear all filters from the link (remove injected faults) + + Common filter types for fault injection: + - "frequency_drop": Drop every Nth packet (parameter: frequency, -1 to 32767) + - "packet_loss": Packet loss percentage (parameter: chance, 0-100) + - "delay": Delay in ms with optional jitter (parameters: latency 0-32767, jitter 0-32767) + - "corrupt": Packet corruption percentage (parameter: chance, 0-100) + - "bpf": Berkeley Packet Filter (parameter: filter expression text) + + Input is a JSON object with: + - project_id (str): GNS3 project UUID + - link_id (str): GNS3 link UUID + - action (str): One of "get_available", "set", "get", "clear" + - filters (dict, optional): Filter configuration for "set" action + + Example for getting available filters: + { + "project_id": "uuid-of-project", + "link_id": "uuid-of-link", + "action": "get_available" + } + + Example for setting delay and packet loss: + { + "project_id": "uuid-of-project", + "link_id": "uuid-of-link", + "action": "set", + "filters": { + "delay": [100, 10], + "packet_loss": [5] + } + } + + Example for clearing filters: + { + "project_id": "uuid-of-project", + "link_id": "uuid-of-link", + "action": "clear" + } + + Returns a dictionary with action result, filter information, or error message. + """ + + def _run( + self, + tool_input: str, + run_manager: CallbackManagerForToolRun | None = None, + **kwargs: Any, + ) -> dict[str, Any]: + """ + Manages packet filters on a GNS3 link. + + Args: + tool_input: A JSON string with project_id, link_id, action, and optional filters. + run_manager: LangChain run manager (unused). + + Returns: + dict: A dictionary with action result or an error message. + """ + # Log received input + logger.info("Received input: %s", tool_input) + + try: + # Parse input JSON + input_data = json.loads(tool_input) + project_id = input_data.get("project_id") + link_id = input_data.get("link_id") + action = input_data.get("action") + + # Validate required fields + if not project_id: + logger.error("Invalid input: Missing project_id.") + return {"error": "Missing project_id."} + + if not link_id: + logger.error("Invalid input: Missing link_id.") + return {"error": "Missing link_id."} + + if not action: + logger.error("Invalid input: Missing action.") + return {"error": "Missing action."} + + # Validate action + valid_actions = ["get_available", "set", "get", "clear"] + if action not in valid_actions: + logger.error("Invalid action: %s. Must be one of %s", action, valid_actions) + return { + "error": f"Invalid action: {action}. Must be one of {valid_actions}" + } + + # Validate filters for "set" action + if action == "set": + filters = input_data.get("filters") + if not filters or not isinstance(filters, dict): + logger.error("Invalid input: 'set' action requires 'filters' dict.") + return { + "error": "'set' action requires 'filters' dict with filter configuration." + } + + # Initialize Gns3Connector using factory function + logger.info("Connecting to GNS3 server...") + gns3_server = get_gns3_connector() + + if gns3_server is None: + logger.error("Failed to create GNS3 connector") + return { + "error": "Failed to connect to GNS3 server. " + "Please check your configuration." + } + + # Create Link object + logger.info( + "Processing packet filter action '%s' for link %s...", action, link_id + ) + link = Link( + project_id=project_id, link_id=link_id, connector=gns3_server + ) + + # Execute action + if action == "get_available": + result = self._get_available_filters(link) + elif action == "set": + filters = input_data.get("filters", {}) + result = self._set_filters(link, filters) + elif action == "get": + result = self._get_filters(link) + elif action == "clear": + result = self._clear_filters(link) + else: + result = {"error": f"Unknown action: {action}"} + + # Log result + logger.info("Packet filter action '%s' completed successfully.", action) + + return result + + except json.JSONDecodeError as e: + logger.error("Invalid JSON input: %s", e) + return {"error": f"Invalid JSON input: {e}"} + except Exception as e: + logger.error("Failed to process packet filter request: %s", e) + return { + "error": f"Failed to process packet filter request: {str(e)}" + } + + def _get_available_filters(self, link: Link) -> dict[str, Any]: + """Get available filter types for the link.""" + try: + filters = link.available_filters() + logger.info("Retrieved %d available filter types.", len(filters)) + return { + "action": "get_available", + "link_id": link.link_id, + "available_filters": filters, + "count": len(filters), + "status": "success", + } + except Exception as e: + logger.error("Failed to get available filters: %s", e) + return { + "action": "get_available", + "link_id": link.link_id, + "error": f"Failed to get available filters: {str(e)}", + "status": "failed", + } + + def _set_filters(self, link: Link, filters: dict[str, Any]) -> dict[str, Any]: + """Set packet filters on the link.""" + try: + # Update filters + link.update(filters=filters) + + # Get updated link info + link.get() + + logger.info("Successfully set filters on link %s", link.link_id) + return { + "action": "set", + "link_id": link.link_id, + "filters": link.filters, + "status": "success", + "message": "Filters applied successfully", + } + except Exception as e: + logger.error("Failed to set filters: %s", e) + return { + "action": "set", + "link_id": link.link_id, + "error": f"Failed to set filters: {str(e)}", + "status": "failed", + } + + def _get_filters(self, link: Link) -> dict[str, Any]: + """Get current filters configured on the link.""" + try: + # Get link information + link.get() + + logger.info("Retrieved current filters for link %s", link.link_id) + return { + "action": "get", + "link_id": link.link_id, + "filters": link.filters, + "status": "success", + } + except Exception as e: + logger.error("Failed to get filters: %s", e) + return { + "action": "get", + "link_id": link.link_id, + "error": f"Failed to get filters: {str(e)}", + "status": "failed", + } + + def _clear_filters(self, link: Link) -> dict[str, Any]: + """Clear all filters from the link.""" + try: + # Clear filters by setting empty dict + link.update(filters={}) + + # Get updated link info to confirm + link.get() + + logger.info("Successfully cleared filters on link %s", link.link_id) + return { + "action": "clear", + "link_id": link.link_id, + "filters": link.filters, + "status": "success", + "message": "Filters cleared successfully", + } + except Exception as e: + logger.error("Failed to clear filters: %s", e) + return { + "action": "clear", + "link_id": link.link_id, + "error": f"Failed to clear filters: {str(e)}", + "status": "failed", + } + + +if __name__ == "__main__": + # Test the tool locally + # TODO: Replace with actual project and link UUIDs + test_input = json.dumps( + { + "project_id": "d7fc094c-685e-4db1-ac11-5e33a1b2e066", + "link_id": "link-uuid-here", + "action": "get_available", + } + ) + + tool = GNS3PacketFilterTool() + result = tool._run(test_input) + pprint(result)