NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-08-27 20:40:13 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat: add packet filter management tool for GNS3-Copilot fault injection

Browse Source 
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
This commit is contained in:
YueGuobin 2026-05-24 21:25:46 +08:00
parent 279ea9eed3
commit 8ba1f064d2
No known key found for this signature in database
4 changed files with 431 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
gns3server/agent/gns3_copilot/agent/gns3_copilot.py
View File

@ -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
]

60
gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py
View File

@ -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=<pr_id>, link_id=<link_id>, connector=<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:

4
gns3server/agent/gns3_copilot/tools_v2/__init__.py
View File

@ -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",

365
gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py Normal file
View File

@ -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 <https://www.gnu.org/licenses/>.
#
# 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)