diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py
index 4efbd5824..341f4281f 100644
--- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py
+++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py
@@ -95,6 +95,7 @@ from gns3server.agent.gns3_copilot.tools_v2 import GNS3TemplateTool
from gns3server.agent.gns3_copilot.tools_v2 import GNS3UpdateNodeNameTool
from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands
from gns3server.agent.gns3_copilot.tools_v2 import PacketCaptureTool
+from gns3server.agent.gns3_copilot.tools_v2 import PacketAnalysisTool
from gns3server.agent.gns3_copilot.skills import DeviceSkillsTool
from gns3server.agent.gns3_copilot.skills import InjectionSkillsTool
@@ -116,6 +117,7 @@ TEACHING_ASSISTANT_MODE_TOOLS = [
ExecuteMultipleDeviceCommands(), # Execute show/display/debug commands
# (READ-ONLY)
PacketCaptureTool(), # Analyze packets from active capture
+ PacketAnalysisTool(), # Protocol-oriented packet analysis with tshark
DeviceSkillsTool(), # Get device-specific skills and command knowledge
]
@@ -133,6 +135,7 @@ LAB_AUTOMATION_ASSISTANT_MODE_TOOLS = [
ExecuteMultipleDeviceConfigCommands(), # Execute configuration commands
VPCSCommands(), # Execute VPCS commands using Netmiko
PacketCaptureTool(), # Analyze packets from active capture
+ PacketAnalysisTool(), # Protocol-oriented packet analysis with tshark
DeviceSkillsTool(), # Get device-specific skills and command knowledge
]
diff --git a/gns3server/agent/gns3_copilot/skills/loader.py b/gns3server/agent/gns3_copilot/skills/loader.py
index 88b825ef8..fde4abd7e 100644
--- a/gns3server/agent/gns3_copilot/skills/loader.py
+++ b/gns3server/agent/gns3_copilot/skills/loader.py
@@ -188,6 +188,46 @@ class SkillsLoader:
logger.error(f"Failed to load forbidden commands from {config_file}: {e}")
return []
+ def load_packet_analysis_protocols(self) -> Dict[str, Dict[str, Any]]:
+ """
+ Load packet analysis protocol definitions from YAML files.
+
+ Returns:
+ Dictionary mapping protocol keys to protocol definitions.
+ Each protocol contains tshark_field, display_filter, and filter_examples.
+ """
+ if yaml is None:
+ logger.error("PyYAML is not installed. Cannot load packet analysis protocols.")
+ return {}
+
+ protocols = {}
+ packet_analysis_dir = self.skills_dir / "packet_analysis"
+
+ if not packet_analysis_dir.exists():
+ logger.debug(f"Packet analysis directory not found: {packet_analysis_dir}")
+ return {}
+
+ for yaml_file in packet_analysis_dir.glob("*.yaml"):
+ try:
+ protocol_data = self._load_yaml(yaml_file)
+ if not protocol_data:
+ logger.warning(f"Skipping empty YAML file: {yaml_file}")
+ continue
+
+ # Use protocol_key field from YAML as the key
+ protocol_key = protocol_data.get("protocol_key")
+ if not protocol_key:
+ logger.warning(f"No 'protocol_key' field in {yaml_file}, skipping")
+ continue
+
+ protocols[protocol_key] = protocol_data
+ logger.debug(f"Loaded packet analysis protocol: {protocol_key} from {yaml_file}")
+ except Exception as e:
+ logger.error(f"Failed to load protocol from {yaml_file}: {e}")
+
+ logger.info(f"Loaded {len(protocols)} packet analysis protocols from {packet_analysis_dir}")
+ return protocols
+
def _load_yaml(self, file_path: Path) -> Dict[str, Any]:
"""
Load a YAML file and return its content.
diff --git a/gns3server/agent/gns3_copilot/skills/manager.py b/gns3server/agent/gns3_copilot/skills/manager.py
index c5ca14c47..cbbfc9f22 100644
--- a/gns3server/agent/gns3_copilot/skills/manager.py
+++ b/gns3server/agent/gns3_copilot/skills/manager.py
@@ -247,6 +247,34 @@ class SkillsManager:
logger.error(f"Failed to reload skills: {e}")
return False
+ def reload_packet_analysis_protocols(self) -> bool:
+ """
+ Hot reload packet analysis protocol definitions from YAML files.
+
+ Loads the latest protocol definitions from YAML files and updates
+ the PACKET_ANALYSIS_REGISTRY.
+
+ Returns:
+ True if successful, False otherwise
+ """
+ try:
+ from .registry import PACKET_ANALYSIS_REGISTRY
+
+ # Load new packet analysis protocols from YAML files
+ new_protocols = self.loader.load_packet_analysis_protocols()
+
+ # Update registry (safe replace - never leaves dict empty)
+ for k in list(PACKET_ANALYSIS_REGISTRY):
+ if k not in new_protocols:
+ del PACKET_ANALYSIS_REGISTRY[k]
+ PACKET_ANALYSIS_REGISTRY.update(new_protocols)
+
+ logger.info(f"Successfully reloaded {len(new_protocols)} packet analysis protocols")
+ return True
+ except Exception as e:
+ logger.error(f"Failed to reload packet analysis protocols: {e}")
+ return False
+
def reload_prompts(self) -> bool:
"""
Hot reload prompts from Markdown files.
diff --git a/gns3server/agent/gns3_copilot/skills/registry.py b/gns3server/agent/gns3_copilot/skills/registry.py
index eed12608b..40e04f4a6 100644
--- a/gns3server/agent/gns3_copilot/skills/registry.py
+++ b/gns3server/agent/gns3_copilot/skills/registry.py
@@ -54,6 +54,10 @@ SKILLS_REGISTRY: dict[str, dict[str, Any]] = {}
# This registry can be hot-reloaded via SkillsManager
INJECTION_SKILLS_REGISTRY: dict[str, dict[str, Any]] = {}
+# Packet analysis protocols registry - loaded from external repository
+# Contains protocol definitions for tshark-based packet analysis
+PACKET_ANALYSIS_REGISTRY: dict[str, dict[str, Any]] = {}
+
# Global skills manager instance for hot reload
_skills_manager = None
_init_in_progress = False
@@ -171,6 +175,7 @@ def reload_skills_repository() -> dict[str, Any]:
# Reload everything from local files
skills_ok = manager.reload_skills()
prompts_ok = manager.reload_prompts()
+ protocols_ok = manager.reload_packet_analysis_protocols()
# Reload forbidden commands (local import to avoid circular dependency)
from gns3server.agent.gns3_copilot.utils.command_filter import reload_forbidden_commands as _reload_fc
@@ -180,11 +185,13 @@ def reload_skills_repository() -> dict[str, Any]:
forbidden_commands = get_forbidden_commands()
return {
- "success": skills_ok or prompts_ok,
+ "success": skills_ok or prompts_ok or protocols_ok,
"skills": skills_ok,
"skill_count": manager.get_skill_count(),
"prompts": prompts_ok,
"prompt_count": manager.get_prompt_count(),
+ "protocols": protocols_ok,
+ "protocol_count": len(PACKET_ANALYSIS_REGISTRY),
"forbidden_commands": len(forbidden_commands),
"version": manager.get_current_version(),
}
@@ -568,6 +575,53 @@ def list_available_injection_skills(context: list[str] | None = None) -> list[di
return skills
+def get_packet_analysis_protocol(protocol: str) -> dict[str, Any]:
+ """
+ Get a packet analysis protocol definition.
+
+ Args:
+ protocol: The protocol key (e.g., "ospf", "bgp", "icmp")
+
+ Returns:
+ Protocol definition dictionary with available_fields, base_filter, etc.
+ Returns error dict if protocol not found.
+ """
+ protocol_data = PACKET_ANALYSIS_REGISTRY.get(protocol)
+
+ if not protocol_data:
+ # Try case-insensitive match
+ for key, data in PACKET_ANALYSIS_REGISTRY.items():
+ if key.lower() == protocol.lower():
+ protocol_data = data
+ protocol = key
+ break
+
+ if not protocol_data:
+ return {
+ "error": f"Unknown protocol: {protocol}",
+ "available_protocols": list(PACKET_ANALYSIS_REGISTRY.keys()),
+ }
+
+ return protocol_data
+
+
+def list_available_packet_analysis_protocols() -> list[dict[str, str]]:
+ """
+ List all available packet analysis protocols.
+
+ Returns:
+ List of protocol info dicts with protocol, name, and description.
+ """
+ protocols = []
+ for key, data in PACKET_ANALYSIS_REGISTRY.items():
+ protocols.append({
+ "protocol": key,
+ "name": data.get("name", key),
+ "description": data.get("description", ""),
+ })
+ return protocols
+
+
class DeviceSkillsTool(BaseTool):
"""
LangChain tool for querying device/feature skills.
diff --git a/gns3server/agent/gns3_copilot/tools_v2/__init__.py b/gns3server/agent/gns3_copilot/tools_v2/__init__.py
index 8cb9d744a..a4a0734a1 100644
--- a/gns3server/agent/gns3_copilot/tools_v2/__init__.py
+++ b/gns3server/agent/gns3_copilot/tools_v2/__init__.py
@@ -60,6 +60,7 @@ from .gns3_stop_node import GNS3StopNodeTool
from .gns3_suspend_node import GNS3SuspendNodeTool
from .gns3_update_node_name import GNS3UpdateNodeNameTool
from .packet_capture_tools import PacketCaptureTool
+from .packet_analysis_tool import PacketAnalysisTool
# Dynamic version management
try:
@@ -86,6 +87,7 @@ __all__ = [
"GNS3UpdateNodeNameTool",
"GNS3TemplateTool",
"PacketCaptureTool",
+ "PacketAnalysisTool",
]
# Package initialization message
diff --git a/gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py b/gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py
new file mode 100644
index 000000000..c4b0bc010
--- /dev/null
+++ b/gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py
@@ -0,0 +1,308 @@
+# 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
+#
+
+"""
+Protocol-Oriented Packet Analysis Tool
+
+Analyzes packets from an active GNS3 capture using tshark with user-provided arguments.
+The LLM constructs tshark commands based on protocol knowledge from packet analysis skills.
+"""
+
+import logging
+import os
+import subprocess
+import tempfile
+
+from langchain.tools import BaseTool
+from langchain_core.callbacks import CallbackManagerForToolRun
+
+from gns3server.agent.gns3_copilot.gns3_client.context_helpers import (
+ get_current_jwt_token,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class PacketAnalysisTool(BaseTool):
+ """
+ LangChain tool for analyzing packets using tshark with user-provided arguments.
+
+ The LLM constructs tshark command arguments based on protocol knowledge
+ from packet analysis skills (available_fields, base_filter, etc.).
+
+ Input:
+ project_id (str, required): UUID of the GNS3 project
+ link_id (str, required): UUID of the link to analyze
+ tshark_args (str, required): tshark command arguments (e.g., '-Y "ospf" -T fields -e ip.src')
+
+ Output:
+ str: tshark output in text format (tab-separated fields or JSON)
+ """
+
+ name: str = "packet_analysis"
+ description: str = """
+ Analyze packets from an active GNS3 capture using tshark.
+
+ Use this tool to analyze network packets for troubleshooting. Construct tshark
+ arguments based on the protocol's available fields from packet_analysis skills.
+
+ Input (JSON format):
+ - project_id (str, required): UUID of the GNS3 project
+ - link_id (str, required): UUID of the link to analyze
+ - tshark_args (str, required): tshark command arguments (everything after 'tshark -r ')
+
+ Common tshark arguments:
+ -Y "": Display filter (e.g., 'ospf', 'bgp', 'icmp')
+ -T fields: Output as tab-separated fields
+ -e : Extract specific field (can use multiple -e)
+ -T json: Output as JSON
+ -c : Limit packet count
+
+ Examples:
+ # OSPF Hello packets with specific fields
+ {"project_id": "xxx", "link_id": "yyy",
+ "tshark_args": "-Y 'ospf.msg == 1' -T fields -e ip.src -e ospf.hello.interval -e ospf.dead.interval"}
+
+ # BGP messages
+ {"project_id": "xxx", "link_id": "yyy",
+ "tshark_args": "-Y 'bgp' -T fields -e ip.src -e bgp.type"}
+
+ # First 50 ICMP packets
+ {"project_id": "xxx", "link_id": "yyy",
+ "tshark_args": "-Y 'icmp' -T fields -e icmp.type -e icmp.code -c 50"}
+ """
+
+ def _run(
+ self,
+ project_id: str,
+ link_id: str,
+ tshark_args: str,
+ run_manager: CallbackManagerForToolRun | None = None,
+ ) -> str:
+ """
+ Analyze packets using tshark with user-provided arguments.
+
+ Args:
+ project_id: UUID of the GNS3 project
+ link_id: UUID of the link to analyze
+ tshark_args: tshark command arguments (after '-r ')
+ run_manager: LangChain run manager (unused)
+
+ Returns:
+ str: tshark output
+ """
+ logger.info(
+ f"PacketAnalysisTool invoked: project_id={project_id}, "
+ f"link_id={link_id}, tshark_args={tshark_args}"
+ )
+
+ # Validate inputs
+ if not project_id:
+ return '{"error": "project_id is required"}'
+ if not link_id:
+ return '{"error": "link_id is required"}'
+ if not tshark_args or not tshark_args.strip():
+ return '{"error": "tshark_args is required"}'
+
+ temp_file = None
+ try:
+ # Download capture file
+ temp_file = self._download_capture(project_id, link_id)
+ if not temp_file:
+ return '{"error": "Failed to download capture file"}'
+
+ # Check if file exists and has content
+ if not os.path.exists(temp_file):
+ return '{"error": "Capture file not found"}'
+
+ file_size = os.path.getsize(temp_file)
+ if file_size == 0:
+ return '{"error": "Capture file is empty, no packets captured yet"}'
+
+ logger.info(f"Capture file downloaded: {temp_file}, size={file_size} bytes")
+
+ # Run tshark with user-provided arguments
+ result = self._run_tshark(temp_file, tshark_args)
+ return result
+
+ except Exception as e:
+ logger.error(f"PacketAnalysisTool error: {e}", exc_info=True)
+ return f'{{"error": "Analysis failed: {str(e)}"}}'
+
+ finally:
+ # Clean up temp file
+ if temp_file and os.path.exists(temp_file):
+ try:
+ os.remove(temp_file)
+ logger.debug(f"Temporary file removed: {temp_file}")
+ except Exception as e:
+ logger.warning(f"Failed to remove temp file: {e}")
+
+ def _download_capture(self, project_id: str, link_id: str) -> str | None:
+ """
+ Download capture file from GNS3 server.
+
+ Args:
+ project_id: Project UUID
+ link_id: Link UUID
+
+ Returns:
+ str: Path to temporary capture file, or None on failure
+ """
+ jwt_token = get_current_jwt_token()
+ if not jwt_token:
+ logger.error("JWT token not found in context")
+ return None
+
+ # Detect GNS3 server URL
+ url = self._detect_gns3_url()
+ if not url:
+ return None
+
+ capture_url = f"{url}/v3/projects/{project_id}/links/{link_id}/capture/file"
+ logger.info(f"Downloading capture from: {capture_url}")
+
+ # Create temp file
+ temp_fd, temp_file = tempfile.mkstemp(suffix=".pcap", prefix="gns3_capture_")
+ os.close(temp_fd)
+
+ try:
+ import requests
+
+ headers = {"Authorization": f"Bearer {jwt_token}"}
+ # Use verify=False for HTTPS to skip certificate validation (for self-signed certs)
+ verify_cert = not capture_url.startswith("https://")
+ response = requests.get(
+ capture_url,
+ headers=headers,
+ stream=True,
+ timeout=30,
+ verify=verify_cert,
+ )
+
+ if response.status_code != 200:
+ logger.error(f"Failed to download capture: HTTP {response.status_code}")
+ os.remove(temp_file)
+ return None
+
+ # Write to temp file
+ with open(temp_file, "wb") as f:
+ for chunk in response.iter_content(chunk_size=8192):
+ f.write(chunk)
+
+ logger.info(f"Capture file saved: {temp_file}, size={os.path.getsize(temp_file)} bytes")
+ return temp_file
+
+ except Exception as e:
+ logger.error(f"Failed to download capture: {e}", exc_info=True)
+ if os.path.exists(temp_file):
+ os.remove(temp_file)
+ return None
+
+ def _detect_gns3_url(self) -> str | None:
+ """
+ Detect GNS3 server URL from Controller or Config.
+
+ Returns:
+ str: GNS3 server URL, or None on failure
+ """
+ try:
+ from gns3server.controller import Controller
+
+ controller = Controller.instance()
+ local_compute = controller.get_compute("local")
+ url = f"{local_compute.protocol}://{local_compute.host}:{local_compute.port}"
+ logger.debug(f"Detected GNS3 URL from Controller: {url}")
+ return url
+ except Exception as e:
+ logger.debug(f"Cannot get URL from Controller: {e}")
+
+ try:
+ from gns3server.config import Config
+
+ server_config = Config.instance().settings.Server
+ url = f"{server_config.protocol.value}://{server_config.host}:{server_config.port}"
+ logger.debug(f"Detected GNS3 URL from Config: {url}")
+ return url
+ except Exception as e:
+ logger.debug(f"Cannot get URL from Config: {e}")
+
+ # Fallback default
+ default_url = "http://127.0.0.1:3080"
+ logger.warning(f"Using fallback default URL: {default_url}")
+ return default_url
+
+ def _run_tshark(self, pcap_file: str, tshark_args: str) -> str:
+ """
+ Run tshark with user-provided arguments.
+
+ Args:
+ pcap_file: Path to the capture file
+ tshark_args: tshark command arguments (after '-r ')
+
+ Returns:
+ str: tshark output
+ """
+ # Build command: tshark -r
+ cmd = ["tshark", "-r", pcap_file] + tshark_args.split()
+
+ logger.info(f"Running tshark: {' '.join(cmd)}")
+
+ try:
+ result = subprocess.run(
+ cmd,
+ capture_output=True,
+ text=True,
+ timeout=30,
+ )
+
+ output = result.stdout
+ if result.stderr:
+ if "tshark:" in result.stderr.lower():
+ logger.warning(f"tshark stderr: {result.stderr}")
+
+ if not output.strip():
+ return "No matching packets found"
+
+ logger.info(f"tshark output: {len(output)} characters")
+ return output
+
+ except subprocess.TimeoutExpired:
+ logger.error("tshark timeout after 30 seconds")
+ return '{"error": "tshark timeout after 30 seconds"}'
+ except FileNotFoundError:
+ logger.error("tshark not found. Please install tshark: apt install tshark")
+ return '{"error": "tshark not installed. Please install tshark: apt install tshark"}'
+ except Exception as e:
+ logger.error(f"tshark execution error: {e}", exc_info=True)
+ return f'{{"error": "tshark failed: {str(e)}"}}'
+
+
+if __name__ == "__main__":
+ # Test the tool
+ tool = PacketAnalysisTool()
+ print("Testing PacketAnalysisTool...")
+ print("Note: Set project_id, link_id and tshark_args to test with actual GNS3 capture")
+ print(tool.description)