From e2e2e23cf7dbd57de716e719a7ce52edbc90bb38 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Sat, 14 Mar 2026 14:25:37 +0800 Subject: [PATCH] feat(copilot): add VPCS Telnet driver with Netmiko and ANSI code stripping Implement custom VPCS driver and unified tool architecture: - Add VPCSTelnet custom driver (vpcs_telnet.py) - No authentication (direct console access like VPCS behavior) - Simple prompt pattern matching (PC\d+>) - Automatic ANSI escape code stripping for clean output - Config mode methods return empty (VPCS has no config modes) - Replace vpcs_tools_telnetlib3.py with vpcs_tools_netmiko.py - Migrate from telnetlib3 to Netmiko + Nornir architecture - Unified tool architecture matching config/display tools - Improved code consistency and maintainability - Add comprehensive test coverage (test_vpcs_telnet.py) - 30 unit tests covering all VPCS driver functionality - Tests for ANSI code stripping, telnet_login, send_command - Tests for device registration and initialization - Update VPCS built-in template (services/templates.py) - Add platform:vpcs and device_type:gns3_vpcs_telnet tags - Automatic driver selection without manual configuration - Update documentation (docs/) - multi-vendor-device-support.md: VPCS driver documentation - netmiko_devices.md: Add VPCS to supported devices list - README.md: Update multi-vendor support description --- docs/gns3-copilot/README.md | 14 +- .../multi-vendor-device-support.md | 214 +++++- docs/gns3-copilot/netmiko_devices.md | 20 +- .../agent/gns3_copilot/agent/gns3_copilot.py | 4 +- .../lab_automation_assistant_prompt.py | 2 +- .../agent/gns3_copilot/tools_v2/__init__.py | 6 +- .../tools_v2/vpcs_tools_netmiko.py | 611 ++++++++++++++++++ .../tools_v2/vpcs_tools_telnetlib3.py | 489 -------------- .../utils/custom_netmiko/__init__.py | 8 +- .../custom_netmiko/tests/test_vpcs_telnet.py | 495 ++++++++++++++ .../utils/custom_netmiko/vpcs_telnet.py | 424 ++++++++++++ gns3server/services/templates.py | 1 + 12 files changed, 1767 insertions(+), 521 deletions(-) create mode 100644 gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py delete mode 100644 gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py create mode 100644 gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_vpcs_telnet.py create mode 100644 gns3server/agent/gns3_copilot/utils/custom_netmiko/vpcs_telnet.py diff --git a/docs/gns3-copilot/README.md b/docs/gns3-copilot/README.md index eecf82159..22bf199a1 100644 --- a/docs/gns3-copilot/README.md +++ b/docs/gns3-copilot/README.md @@ -77,18 +77,22 @@ Tools for controlling network device lifecycle in GNS3 projects. **Status:** ✅ Implemented ### Multi-Vendor Device Support (`implemented/multi-vendor-device-support.md`) -Multi-vendor network device support with custom Netmiko driver for Huawei devices. +Multi-vendor network device support with custom Netmiko drivers for Huawei, Ruijie, and VPCS devices. **Key Features:** -- Custom HuaweiTelnetCE driver for GNS3 emulation (no authentication) +- Custom HuaweiTelnetCE driver for Huawei CloudEngine (no authentication) +- Custom RuijieTelnetEnhanced driver for Ruijie OS (interactive command handling) +- Custom VPCSTelnet driver for VPCS simulator (no authentication, ANSI code stripping) - Cisco IOS Telnet support - Dynamic device type detection from GNS3 tags -- Automatic Nornir group generation -- VRP-specific command handling (system-view, return confirmation) +- Unified Nornir + Netmiko architecture +- Vendor-specific command handling (VRP system-view, Ruijie interactive prompts, VPCS simple prompts) **Tested Vendors:** - Cisco IOS (Telnet) - Huawei CloudEngine (Telnet, custom driver) +- Ruijie (锐捷) OS (Telnet, custom enhanced driver) +- VPCS (Virtual PC Simulator, Telnet, custom driver) **Status:** ✅ Implemented @@ -132,4 +136,4 @@ When adding new documentation: --- -_Last updated: 2026-03-12_ +_Last updated: 2026-03-14_ diff --git a/docs/gns3-copilot/implemented/multi-vendor-device-support.md b/docs/gns3-copilot/implemented/multi-vendor-device-support.md index c8faae031..3f84d0036 100644 --- a/docs/gns3-copilot/implemented/multi-vendor-device-support.md +++ b/docs/gns3-copilot/implemented/multi-vendor-device-support.md @@ -11,6 +11,166 @@ GNS3-Copilot supports network devices from multiple vendors through Netmiko and | **Cisco** | `cisco_ios` | `cisco_ios_telnet` | Telnet | ✅ Tested | | **Huawei** | `huawei` | `gns3_huawei_telnet_ce` | Telnet | ✅ Tested (Custom Driver) | | **Ruijie (锐捷)** | `ruijie_os` | `gns3_ruijie_telnet` | Telnet | ✅ Tested (Custom Driver) | +| **VPCS** | `vpcs` | `gns3_vpcs_telnet` | Telnet | ✅ Tested (Custom Driver) | + +## Custom VPCS Driver (`VPCSTelnet`) + +### Problem Statement + +VPCS (Virtual PC Simulator) is a lightweight virtual PC simulator used in GNS3 lab environments. Unlike network devices (routers/switches), VPCS devices: + +1. **No authentication** - Direct console access without username/password +2. **Simple command interface** - No configuration modes +3. **Simple prompt pattern** - `PC1>`, `PC2>`, etc. + +### Solution: Lightweight Custom Driver + +``` +BaseConnection (Netmiko base class) + ↓ +VPCSTelnet (Custom GNS3 driver) +``` + +**Why Not Use Standard Telnet Driver?** +- Standard drivers attempt authentication (times out) +- No support for VPCS-specific prompt patterns (`PC\d+>`) +- No need for configuration mode handling + +### VPCSTelnet Implementation + +#### Location +``` +gns3server/agent/gns3_copilot/utils/custom_netmiko/vpcs_telnet.py +``` + +#### Key Features + +**1. No Authentication** +```python +def telnet_login(self, pri_prompt_terminator=r"PC\d+>", ...): + # Send returns until VPCS prompt detected + for i in range(max_loops): + self.write_channel(self.RETURN) + output = self.read_channel() + + if re.search(pri_prompt_terminator, output): + return output # Success - VPCS prompt detected +``` + +**2. Simple Prompt Recognition** +``` +PC1> ip 10.10.0.12/24 10.10.0.254 +PC1> ping 10.10.0.254 +``` + +**3. No Configuration Mode** +```python +def check_config_mode(self) -> bool: + return False # VPCS has no config mode + +def config_mode(self) -> str: + return "" # No config mode to enter + +def exit_config_mode(self) -> str: + return "" # No config mode to exit +``` + +**4. No Paging** +```python +def disable_paging(self) -> str: + return "" # VPCS doesn't use paging +``` + +### VPCS Tool Usage + +The VPCS driver is used by the `execute_vpcs_commands` tool: + +```python +from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands + +tool = VPCSCommands() +result = tool._run(json.dumps({ + "project_id": "", + "device_configs": [ + { + "device_name": "PC1", + "commands": [ + "ip 10.10.0.12/24 10.10.0.254", + "ping 10.10.0.254" + ] + } + ] +})) +``` + +### VPCS Built-in Template Configuration + +**✨ Automatic Tags - No Manual Configuration Required** + +VPCS nodes created from the built-in template automatically include the necessary tags: + +| Tag | Value | Purpose | +|-----|-------|---------| +| `platform` | `vpcs` | Platform identification | +| `device_type` | `gns3_vpcs_telnet` | Netmiko driver selection | + +**Built-in Template Definition:** +```python +# gns3server/services/templates.py +{ + "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "vpcs"), + "template_type": "vpcs", + "name": "VPCS", + "default_name_format": "PC{0}", + "category": "guest", + "symbol": "vpcs_guest", + "builtin": True, + "tags": ["platform:vpcs", "device_type:gns3_vpcs_telnet"], # ✅ Auto-applied +} +``` + +**User Benefits:** +- ✅ **No manual tagging required** - Tags are applied automatically when creating VPCS nodes +- ✅ **Automatic driver selection** - Copilot tools automatically use the correct Netmiko driver +- ✅ **Consistent behavior** - All VPCS nodes from the built-in template work identically +- ✅ **Zero configuration** - Users don't need to understand device_type tags + +**How It Works:** +1. User creates a VPCS node from the built-in "VPCS" template +2. Node automatically inherits the tags: `platform:vpcs` and `device_type:gns3_vpcs_telnet` +3. Copilot tools read these tags and select the appropriate VPCS Netmiko driver +4. Commands execute using the VPCS-optimized driver (no authentication, simple prompts) + +### Supported VPCS Commands + +| Command | Description | Example | +|---------|-------------|---------| +| `ip` | Configure/show IP address | `ip 10.10.0.12/24 10.10.0.254` | +| `ping` | Test connectivity | `ping 10.10.0.254` | +| `arp` | Display ARP table | `arp` | +| `show ip` | Show IP configuration | `show ip` | +| `version` | Show VPCS version | `version` | +| `save` | Save configuration | `save` | +| `load` | Load configuration | `load` | + +### Architecture Benefits + +**Unified Tool Architecture:** +- ✅ Uses Nornir for connection management (same as network device tools) +- ✅ Uses Netmiko for command execution (consistent with other tools) +- ✅ Follows same patterns as `config_tools_nornir.py` and `display_tools_nornir.py` +- ✅ Simplified codebase - no need for separate telnetlib3 implementation + +**Migration from telnetlib3:** +| Aspect | Old (telnetlib3) | New (Netmiko + Nornir) | +|--------|-----------------|-------------------------| +| Library | telnetlib3 | Netmiko | +| Framework | Manual threading | Nornir | +| Code Lines | ~490 lines | ~580 lines (with better structure) | +| Consistency | Unique implementation | Same as other tools | +| Maintenance | Separate code path | Unified architecture | + +--- ## Custom Huawei Driver (`GNS3HuaweiTelnetCE`) @@ -375,11 +535,18 @@ platform:huawei → Nornir platform (high-level) **Tag Examples:** -| Vendor | Device Type Tag | Platform Tag | -|--------|----------------|--------------| -| Cisco IOS | `device_type:cisco_ios_telnet` | `platform:cisco_ios` | -| Huawei CE | `device_type:gns3_huawei_telnet_ce` | `platform:huawei` | -| Ruijie | `device_type:gns3_ruijie_telnet` | `platform:ruijie_os` | +| Vendor | Device Type Tag | Platform Tag | Template Source | +|--------|----------------|--------------|------------------| +| Cisco IOS | `device_type:cisco_ios_telnet` | `platform:cisco_ios` | User appliance | +| Huawei CE | `device_type:gns3_huawei_telnet_ce` | `platform:huawei` | User appliance | +| Ruijie | `device_type:gns3_ruijie_telnet` | `platform:ruijie_os` | User appliance | +| **VPCS** | `device_type:gns3_vpcs_telnet` | `platform:vpcs` | **Built-in ✅** | + +**VPCS Built-in Template:** +- VPCS has a **built-in template** with pre-configured tags +- Tags are **automatically applied** when creating VPCS nodes +- No manual configuration required - works out of the box +- Other devices require users to import appliances and configure tags manually ### Nornir Best Practice: Host-Level Connection Configuration @@ -711,25 +878,36 @@ gns3server/agent/gns3_copilot/ │ ├── custom_netmiko/ # Custom Netmiko drivers package │ │ ├── __init__.py # Package initialization │ │ ├── huawei_ce.py # Huawei CloudEngine driver -│ │ ├── ruijie_telnet.py # Ruijie enhanced driver (NEW) +│ │ ├── ruijie_telnet.py # Ruijie enhanced driver +│ │ ├── vpcs_telnet.py # VPCS simulator driver (NEW) │ │ ├── README.md # Driver development guide │ │ └── tests/ # Unit tests │ │ ├── __init__.py │ │ └── test_huawei_ce.py # Huawei CE driver tests │ └── get_gns3_device_port.py # Device port extraction with host-level config -│ ├── _expand_multiline_commands() # Expand banner commands (NEW) -│ └── _error_handling() # device_type missing errors (NEW) +│ ├── _expand_multiline_commands() # Expand banner commands +│ └── _error_handling() # device_type missing errors ├── tools_v2/ │ ├── display_tools_nornir.py # Multi-vendor display commands │ │ ├── _get_nornir_defaults() # Returns default Nornir config │ │ └── _initialize_nornir() # Single generic group + host-level device_type -│ └── config_tools_nornir.py # Multi-vendor config commands -│ ├── _get_nornir_defaults() # Returns default Nornir config -│ └── _initialize_nornir() # Single generic group + host-level device_type -│ ├── _expand_multiline_commands() # Expand banner commands (NEW) -│ └── _error_handling() # device_type validation (NEW) +│ ├── config_tools_nornir.py # Multi-vendor config commands +│ │ ├── _get_nornir_defaults() # Returns default Nornir config +│ │ └── _initialize_nornir() # Single generic group + host-level device_type +│ │ ├── _expand_multiline_commands() # Expand banner commands +│ │ └── _error_handling() # device_type validation +│ └── vpcs_tools_netmiko.py # VPCS commands using Nornir + Netmiko (NEW) +│ ├── VPCSCommands # VPCS tool class +│ └── _initialize_nornir() # VPCS device inventory setup ``` +**Key Architectural Changes (2026-03-14):** +- ❌ Removed: `vpcs_tools_telnetlib3.py` - Replaced with Netmiko implementation +- ✅ Added: `vpcs_telnet.py` - Custom VPCS driver for Netmiko +- ✅ Added: `vpcs_tools_netmiko.py` - VPCS tool using Nornir + Netmiko +- ✅ Simplified: Unified tool architecture - all tools use Nornir + Netmiko +- ✅ Updated: Module structure - all tools follow same pattern + **Key Architectural Changes (2026-03-13):** - ❌ Removed: `_get_nornir_groups_config()` - No longer needed - ❌ Removed: `_get_nornir_group()` - No longer needed @@ -829,15 +1007,21 @@ python gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.p _Implementation Date: 2026-03-12_ -_Last Updated: 2026-03-13 (Added Ruijie driver, multi-line command expansion, and configuration safety)_ +_Last Updated: 2026-03-14 (Added VPCS driver and unified tool architecture)_ -_Status: ✅ Implemented - Custom drivers for Huawei and Ruijie, multi-vendor support with Cisco IOS, Huawei, and Ruijie tested_ +_Status: ✅ Implemented - Custom drivers for Huawei, Ruijie, and VPCS; multi-vendor support with Cisco IOS, Huawei, Ruijie, and VPCS tested_ _Architecture: Nornir best practice - host-level connection_options with single generic group_ _Unit Tests: ✅ 9/9 passing_ _Changelog:_ +- **2026-03-14**: Added VPCS support and unified tool architecture + - Implemented `VPCSTelnet` custom Netmiko driver for VPCS simulator + - Replaced `vpcs_tools_telnetlib3.py` with `vpcs_tools_netmiko.py` + - Unified all tools to use Nornir + Netmiko architecture + - Removed dependency on telnetlib3 for VPCS devices + - Improved code consistency and maintainability - **2026-03-13 (Evening)**: Added Ruijie enhanced driver and interactive command handling - Implemented `RuijieTelnetEnhanced` with hybrid batch/fallback strategy - Added automatic `yes` insertion for known interactive commands (`router-id`, `erase`, etc.) diff --git a/docs/gns3-copilot/netmiko_devices.md b/docs/gns3-copilot/netmiko_devices.md index 31231965e..a8be34c19 100644 --- a/docs/gns3-copilot/netmiko_devices.md +++ b/docs/gns3-copilot/netmiko_devices.md @@ -1,16 +1,27 @@ # Netmiko Supported Devices **Netmiko Version:** 4.6.0 -**Generated:** 2026-03-12 23:00:41 +**Generated:** 2026-03-14 13:10:00 +**Last Updated:** 2026-03-14 (Added VPCS support) --- ## Summary -- **Total Device Types:** 365 +- **Total Device Types:** 366 - **SSH Devices:** 154 -- **Telnet Devices:** 54 -- **Custom Devices:** 2 +- **Telnet Devices:** 55 +- **Custom Devices:** 3 + +## Custom GNS3 Drivers + +GNS3-Copilot includes custom Netmiko drivers optimized for GNS3 emulation: + +| Platform | Device Type | Description | +|----------|-------------|-------------| +| Huawei CE | `gns3_huawei_telnet_ce` | Huawei CloudEngine driver with no authentication | +| Ruijie | `gns3_ruijie_telnet` | Ruijie OS enhanced driver with interactive command handling | +| VPCS | `gns3_vpcs_telnet` | VPCS simulator driver with ANSI code stripping | ## SSH Supported Devices @@ -224,6 +235,7 @@ | Ruijie (锐捷) | `ruijie_os_telnet` | Netmiko | | Ruijie (锐捷) | `gns3_ruijie_telnet` | Custom ✨ (GNS3 Enhanced) | | Supermicro | `supermicro_smis_telnet` | Netmiko | +| VPCS | `gns3_vpcs_telnet` | Custom ✨ (GNS3 VPCS Simulator) | | Telcosystems | `telcosystems_binos_telnet` | Netmiko | | Teldat | `teldat_cit_telnet` | Netmiko | | Tplink | `tplink_jetstream_telnet` | Netmiko | diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py index a526480e8..c3faea518 100644 --- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py +++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py @@ -93,7 +93,7 @@ from gns3server.agent.gns3_copilot.tools_v2 import GNS3StopNodeTool from gns3server.agent.gns3_copilot.tools_v2 import GNS3SuspendNodeTool 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 import VPCSMultiCommands +from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands # Set up logger for GNS3-Copilot logger = logging.getLogger(__name__) @@ -126,7 +126,7 @@ LAB_AUTOMATION_ASSISTANT_MODE_TOOLS = [ ExecuteMultipleDeviceCommands(), # Execute show/display/debug commands # (READ-ONLY) ExecuteMultipleDeviceConfigCommands(), # Execute configuration commands - VPCSMultiCommands(), # Execute VPCS commands on multiple devices + VPCSCommands(), # Execute VPCS commands using Netmiko ] # Default tools (legacy support - will be overridden by mode-specific tools) diff --git a/gns3server/agent/gns3_copilot/prompts/lab_automation_assistant_prompt.py b/gns3server/agent/gns3_copilot/prompts/lab_automation_assistant_prompt.py index ae8e9d1a4..fbc20e8a5 100644 --- a/gns3server/agent/gns3_copilot/prompts/lab_automation_assistant_prompt.py +++ b/gns3server/agent/gns3_copilot/prompts/lab_automation_assistant_prompt.py @@ -59,7 +59,7 @@ You have access to the following tools to help users: | `gns3_update_node_name_tool` | Update node names | Rename devices | | `execute_multiple_device_commands` | Execute display commands | Diagnostics | | `execute_multiple_device_config_commands` | Execute config commands | Config changes | -| `vpcs_multi_commands` | Execute VPCS commands | Configure VPCS devices | +| `execute_vpcs_commands` | Execute VPCS commands | Configure VPCS devices | --- diff --git a/gns3server/agent/gns3_copilot/tools_v2/__init__.py b/gns3server/agent/gns3_copilot/tools_v2/__init__.py index a762b56bf..bb4443beb 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/__init__.py +++ b/gns3server/agent/gns3_copilot/tools_v2/__init__.py @@ -30,13 +30,13 @@ This package provides various tools for interacting with GNS3 network simulator: - Device configuration command execution - Display command execution - Multiple device command execution using Nornir -- VPCS device configuration using telnetlib3 +- VPCS device configuration using Netmiko - Node and link management Main modules: - config_tools_nornir: Multiple device configuration command execution tool using Nornir - display_tools_nornir: Multiple device command execution tool using Nornir -- vpcs_tools_telnetlib3: VPCS device configuration tool using telnetlib3 +- vpcs_tools_netmiko: VPCS device configuration tool using Netmiko - gns3_create_node: GNS3 node creation tool - gns3_create_link: GNS3 link creation tool - gns3_start_node: GNS3 node startup tool @@ -59,7 +59,6 @@ from .gns3_start_node import GNS3StartNodeTool from .gns3_stop_node import GNS3StopNodeTool from .gns3_suspend_node import GNS3SuspendNodeTool from .gns3_update_node_name import GNS3UpdateNodeNameTool -from .vpcs_tools_telnetlib3 import VPCSMultiCommands # Dynamic version management try: @@ -77,7 +76,6 @@ __url__ = "https://github.com/yueguobin/gns3-copilot" __all__ = [ "ExecuteMultipleDeviceConfigCommands", "ExecuteMultipleDeviceCommands", - "VPCSMultiCommands", "GNS3CreateNodeTool", "GNS3LinkTool", "GNS3StartNodeTool", diff --git a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py new file mode 100644 index 000000000..525feb136 --- /dev/null +++ b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py @@ -0,0 +1,611 @@ +# 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 +# + +""" +This module provides a tool to execute commands on VPCS devices +in a GNS3 topology using Nornir with Netmiko. +""" + +import json +import logging +import re +from typing import Any + +from langchain.tools import BaseTool +from langchain_core.callbacks import CallbackManagerForToolRun +from netmiko.exceptions import ReadTimeout +from nornir import InitNornir +from nornir.core.task import AggregatedResult +from nornir.core.task import Result +from nornir.core.task import Task +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 + +# Import custom Netmiko device types for GNS3 emulation +# This registers gns3_vpcs_telnet and other custom device types +# NOTE: Must be imported BEFORE any Nornir operations to ensure device types are registered +from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401 + +# Explicitly register VPCS device type to ensure it is available +try: + from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( + register_custom_device_type as register_vpcs_device_type, + ) + + # Register VPCS device type + register_vpcs_device_type() + + # CRITICAL: Update netmiko.ssh_dispatcher platforms lists + import importlib + + sd = importlib.import_module("netmiko.ssh_dispatcher") + + # Recalculate platforms lists to include custom device types + sd.platforms = list(sd.CLASS_MAPPER.keys()) + sd.platforms.sort() + + sd.platforms_base = list(sd.CLASS_MAPPER_BASE.keys()) + sd.platforms_base.sort() + + sd.telnet_platforms = [x for x in sd.platforms if "telnet" in x] + + # Update platform strings used in error messages + sd.platforms_str = "\n" + "\n".join(sd.platforms_base) + sd.telnet_platforms_str = "\n" + "\n".join(sd.telnet_platforms) +except Exception: + # Fail silently - the import-time registration should have worked + pass + +logger = logging.getLogger(__name__) + +# Suppress nornir INFO logs in console (reduce verbosity) +logging.getLogger("nornir.core").setLevel(logging.WARNING) +logging.getLogger("nornir").setLevel(logging.WARNING) + + +def _get_nornir_defaults() -> dict[str, Any]: + """Get Nornir default configuration.""" + return {"data": {"location": "gns3"}} + + +class VPCSCommands(BaseTool): + """ + A tool for VPCS (Virtual PC Simulator) devices. + + **VPCS-SPECIFIC TOOL** - This tool ONLY works with VPCS virtual PC devices. + + **IMPORTANT DISTINCTION:** + Unlike network devices (routers/switches), VPCS devices are lightweight virtual PCs. + Commands like 'ip' are basic PC IP config, NOT network device config. + + **Allowed VPCS Commands:** + - IP configuration: ip
/ + - View configuration: ip, show ip + - Connectivity testing: ping + - Display ARP: arp + - Display version: version + - Save/Load: save, load + """ + + name: str = "execute_vpcs_commands" + description: str = """ + **VPCS VIRTUAL PC TOOL** - Configure and test Virtual PC Simulator devices. + + This tool ONLY works with VPCS devices, NOT routers/switches. + + **IMPORTANT: VPCS vs Network Devices** + - VPCS = Lightweight virtual PCs for lab testing (NOT network infra) + - 'ip' command on VPCS = Basic PC IP config (like 'ipconfig' on Windows) + - NOT the same as configuring router interfaces or routing protocols + + **When to Use This Tool:** + - Configure IP addresses on virtual PCs: ip 10.10.0.12/24 10.10.0.254 + - Test connectivity from PCs: ping 10.10.0.254 + - View PC IP configuration: ip, show ip + - Display ARP table: arp + - Check PC version: version + + **Input Format:** + { + "project_id": "", + "device_configs": [ + { + "device_name": "PC1", + "commands": ["ip", "ping 10.10.0.254"] + }, + { + "device_name": "PC2", + "commands": ["show ip"] + } + ] + } + + **Returns:** PC command outputs for IP config and connectivity testing. + + **Note:** For network devices, use execute_multiple_device_commands. + """ + + def _run( + self, + tool_input: str | bytes | list[Any] | dict[str, Any], + run_manager: CallbackManagerForToolRun | None = None, + **kwargs: Any, + ) -> list[dict[str, Any]]: + """ + Execute commands on multiple VPCS devices in GNS3 topology. + + Args: + tool_input: JSON string with project_id and VPCS commands. + + Returns: + List of dicts with device names and command outputs. + """ + # Log received input + logger.debug("Received input: %s", tool_input) + + # Validate input + device_configs_list, project_id = self._validate_tool_input(tool_input) + if ( + isinstance(device_configs_list, list) + and len(device_configs_list) > 0 + and "error" in device_configs_list[0] + ): + return device_configs_list + + # Create a mapping of device names to their commands + device_configs_map = self._configs_map(device_configs_list) + + # Prepare device hosts data + try: + hosts_data = self._prepare_device_hosts_data( + device_configs_list, project_id + ) + except ValueError as e: + logger.error("Failed to prepare device hosts data: %s", e) + return [{"error": str(e)}] + + # Check if any devices have errors (e.g., missing device) + error_devices = { + name: data + for name, data in hosts_data.items() + if "error" in data + } + if error_devices: + logger.error( + "Devices with configuration errors: %s", + list(error_devices.keys()) + ) + return [ + { + "device_name": name, + "status": "failed", + "error": data["error"] + } + for name, data in error_devices.items() + ] + + # Initialize Nornir + try: + dynamic_nr = self._initialize_nornir(hosts_data) + except ValueError as e: + logger.error("Failed to initialize Nornir: %s", e) + return [{"error": str(e)}] + + results = [] + + # Execute all devices concurrently in a single run + try: + task_result = dynamic_nr.run( + task=self._run_vpcs_commands, + device_configs_map=device_configs_map, + ) + + # Process results for all devices + results = self._process_task_results( + device_configs_list, + hosts_data, + task_result, + ) + + except Exception as e: + # Overall execution failed + logger.error("Error executing commands on all VPCS devices: %s", e) + return [{"error": f"Execution error: {str(e)}"}] + + logger.debug( + "VPCS command execution completed. Results: %s", + json.dumps(results, indent=2, ensure_ascii=False), + ) + + return results + + def _run_vpcs_commands( + self, task: Task, device_configs_map: dict[str, list[str]] + ) -> Result: + """Execute VPCS commands with single retry.""" + device_name = task.host.name + commands = device_configs_map.get(device_name, []) + + if not commands: + return Result( + host=task.host, result="No commands to execute" + ) + + try: + # Use netmiko_multiline for VPCS commands + # This works well with simple commands and ping output + _result = task.run( + task=netmiko_multiline, + commands=commands, + read_timeout=30, + ) + return Result(host=task.host, result=_result.result) + + except ReadTimeout as e: + logger.error( + "ReadTimeout occurred for VPCS device %s: %s", + device_name, + str(e), + ) + return Result( + host=task.host, + result=f"Command failed (ReadTimeout): {str(e)}", + failed=True, + ) + + except Exception as e: + # Retry once for transient failures + logger.warning( + "First attempt failed for VPCS device %s, retrying: %s", + device_name, + str(e), + ) + try: + _result = task.run( + task=netmiko_multiline, + commands=commands, + read_timeout=30, + ) + return Result(host=task.host, result=_result.result) + except Exception as retry_e: + logger.error( + "VPCS command failed for device %s: %s (Exception: %s)", + device_name, + str(retry_e), + type(retry_e).__name__, + ) + return Result( + host=task.host, + result=f"Command failed: {str(retry_e)}", + failed=True, + ) + + def _validate_tool_input( + self, tool_input: str | bytes | list[Any] | dict[str, Any] + ) -> tuple[list[dict[str, Any]], str | None]: + """ + Validate VPCS command input. + + Args: + tool_input: Input from LangChain/LangGraph tool call. + + Returns: + Tuple of (device_configs_list, project_id) or (error_list, None) + """ + parsed_input = None + + # Compatibility Check and Parsing + if isinstance(tool_input, (str, bytes, bytearray)): + try: + parsed_input = json.loads(tool_input) + logger.info("Successfully parsed tool input from JSON string.") + except json.JSONDecodeError as e: + logger.error( + "Invalid JSON string received as tool input: %s", e + ) + return ( + [{"error": f"Invalid JSON string input from model: {e}"}], + None, + ) + else: + parsed_input = tool_input + logger.info( + "Using tool input directly as type: %s", + type(parsed_input).__name__, + ) + + # Handle new format: {"project_id": "...", "device_configs": [...]} + if isinstance(parsed_input, dict): + project_id = parsed_input.get("project_id") + device_configs = parsed_input.get("device_configs") + + # Validate project_id + if not project_id: + error_msg = "Missing required 'project_id' field in input" + logger.error(error_msg) + return ([{"error": error_msg}], None) + + if not self._validate_project_id(project_id): + error_msg = f"Invalid project_id: {project_id}. Expected UUID." + logger.error(error_msg) + return ([{"error": error_msg}], None) + + # Validate device_configs + if not isinstance(device_configs, list): + error_msg = "'device_configs' must be an array" + logger.error(error_msg) + return ([{"error": error_msg}], None) + + if not device_configs: + logger.warning("Device configs list is empty.") + return [], project_id + + return device_configs, project_id + + else: + error_msg = ( + "Tool input must be JSON with project_id and device_configs, " + f"got {type(parsed_input).__name__}" + ) + logger.error(error_msg) + return ([{"error": error_msg}], None) + + def _validate_project_id(self, project_id: str) -> bool: + """ + Validate project_id format (UUID). + + Args: + project_id: The project ID to validate + + Returns: + True if valid UUID format, False otherwise + """ + 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 _configs_map( + self, device_config_list: list[dict[str, Any]] + ) -> dict[str, list[str]]: + """ + Create a mapping of device names to their command lists. + + Args: + device_config_list: List of device configurations + + Returns: + Dictionary mapping device names to command lists + """ + return { + config["device_name"]: config["commands"] + for config in device_config_list + } + + def _prepare_device_hosts_data( + self, device_configs_list: list[dict[str, Any]], project_id: str + ) -> dict[str, dict[str, Any]]: + """ + Prepare Nornir inventory hosts data for VPCS devices. + + Args: + device_configs_list: List of device configurations + project_id: GNS3 project ID + + Returns: + Dictionary mapping device names to their host data + + Raises: + ValueError: If device not found in topology or missing port + """ + # Get GNS3 server host + gns3_host = get_gns3_server_host() + + # Extract all device names from input + device_names = [config["device_name"] for config in device_configs_list] + + # Get device port mappings from topology + device_ports = get_device_ports_from_topology( + device_names, project_id=project_id + ) + + # Build Nornir inventory hosts data + hosts_data = {} + for device_name in device_names: + if device_name not in device_ports: + logger.error("Device '%s' not found in topology", device_name) + hosts_data[device_name] = { + "error": f"Device '{device_name}' not found in topology" + } + continue + + port = device_ports[device_name]["port"] + + # VPCS devices use gns3_vpcs_telnet device type + hosts_data[device_name] = { + "port": port, + "platform": "vpcs", + "groups": ["vpcs_devices"], # All VPCS devices share one group + "connection_options": { + "netmiko": { + "extras": { + "device_type": "gns3_vpcs_telnet", + "fast_cli": False, + "global_delay_factor": 2.0, + } + } + }, + } + + return hosts_data + + def _initialize_nornir( + self, hosts_data: dict[str, dict[str, Any]] + ) -> "Nornir": + """ + Initialize Nornir with VPCS device inventory. + + Args: + hosts_data: Dictionary of device host data + + Returns: + Initialized Nornir instance + + Raises: + ValueError: If initialization fails + """ + try: + defaults = _get_nornir_defaults() + gns3_host = get_gns3_server_host() + + # Create a single generic group for shared configuration + groups_data = { + "vpcs_devices": { + "hostname": gns3_host, + "timeout": 30, + "username": "", # VPCS doesn't require authentication + "password": "", # VPCS doesn't require authentication + } + } + + logger.info( + "Initializing Nornir for VPCS: host=%s, devices=%d", + gns3_host, + len(hosts_data), + ) + + return InitNornir( + inventory={ + "plugin": "DictInventory", + "options": { + "hosts": hosts_data, + "groups": groups_data, + "defaults": defaults, + }, + }, + runner={ + "plugin": "threaded", + "options": {"num_workers": 10}, + }, + logging={"enabled": False}, + ) + except Exception as e: + logger.error("Failed to initialize Nornir: %s", e) + raise ValueError(f"Failed to initialize Nornir: {e}") from e + + def _process_task_results( + self, + device_configs_list: list[dict[str, Any]], + hosts_data: dict[str, dict[str, Any]], + task_result: AggregatedResult, + ) -> list[dict[str, Any]]: + """ + Process Nornir task results into the format expected by the tool. + + Args: + device_configs_list: Original device configurations list + hosts_data: Device hosts data + task_result: Nornir aggregated task result + + Returns: + List of result dictionaries + """ + results = [] + + for device_config in device_configs_list: + device_name = device_config["device_name"] + + # Check if device had an error during preparation + if device_name in hosts_data and "error" in hosts_data[device_name]: + results.append({ + "device_name": device_name, + "status": "error", + "output": hosts_data[device_name]["error"], + "commands": device_config["commands"], + }) + continue + + # Get result from Nornir task + if device_name in task_result: + host_result = task_result[device_name] + + if host_result.failed: + # Task failed + error_msg = str(host_result.result) if host_result.result else "Unknown error" + results.append({ + "device_name": device_name, + "status": "error", + "output": error_msg, + "commands": device_config["commands"], + }) + else: + # Task succeeded + results.append({ + "device_name": device_name, + "status": "success", + "output": host_result.result, + "commands": device_config["commands"], + }) + else: + # Device not in task result (shouldn't happen) + results.append({ + "device_name": device_name, + "status": "error", + "output": f"Device '{device_name}' not in task results", + "commands": device_config["commands"], + }) + + return results + + +if __name__ == "__main__": + # Example usage + import sys + + command_groups = json.dumps( + { + "project_id": "", + "device_configs": [ + { + "device_name": "PC1", + "commands": [ + "ip 10.10.0.12/24 10.10.0.254", + "ping 10.10.0.254", + ], + }, + { + "device_name": "PC2", + "commands": ["ip 10.10.0.13/24 10.10.0.254"], + }, + ], + } + ) + + exe_cmd = VPCSCommands() + result = exe_cmd._run(tool_input=command_groups) + print("Execution results:") + print(json.dumps(result, indent=2, ensure_ascii=False)) diff --git a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py deleted file mode 100644 index ed1c22f2d..000000000 --- a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py +++ /dev/null @@ -1,489 +0,0 @@ -# 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 -# -""" - -Multi-device VPCS command execution tool using telnetlib3 with threading. -Supports concurrent command execution across VPCS devices. -""" - -import json -import logging -import re -import threading -from time import sleep -from typing import Any - -from langchain.tools import BaseTool -from langchain_core.callbacks import CallbackManagerForToolRun -from telnetlib3 import Telnet - -from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host -from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology - -logger = logging.getLogger(__name__) - - -class VPCSMultiCommands(BaseTool): - """ - A tool for VPCS devices to view PC configs and test connectivity. - - **VPCS-SPECIFIC TOOL** - This tool ONLY works with VPCS virtual PC devices. - - **IMPORTANT DISTINCTION:** - Unlike network devices (routers/switches), VPCS devices are lightweight virtual PCs. - Commands like 'ip' are basic PC IP config, NOT network device config. - - **Allowed VPCS Commands:** - - IP configuration: ip
/ (Basic PC IP setup) - - View configuration: ip, show ip - - Connectivity testing: ping - - Display ARP: arp - - Display version: version - - Save/Load: save, load - - **Usage Context:** - This tool is used in lab environments where students need to configure virtual PC IP - addresses and test network connectivity. It does NOT configure network infra. - """ - - name: str = "execute_vpcs_multi_commands" - description: str = """ - **VPCS VIRTUAL PC TOOL** - Configure and test Virtual PC Simulator devices. - - This tool ONLY works with VPCS devices, NOT routers/switches. - - **IMPORTANT: VPCS vs Network Devices** - - VPCS = Lightweight virtual PCs for lab testing (NOT network infra) - - 'ip' command on VPCS = Basic PC IP config (like 'ipconfig' on Windows) - - NOT the same as configuring router interfaces or routing protocols - - **When to Use This Tool:** - - Configure IP addresses on virtual PCs: ip 10.10.0.12/24 10.10.0.254 - - Test connectivity from PCs: ping 10.10.0.254 - - View PC IP configuration: ip, show ip - - Display ARP table: arp - - Check PC version: version - - **Input Format:** - { - "project_id": "", - "device_configs": [ - { - "device_name": "PC1", - "commands": ["ip", "ping 10.10.0.254"] - }, - { - "device_name": "PC2", - "commands": ["show ip"] - } - ] - } - - **Returns:** PC command outputs for IP config and connectivity testing. - - **Note:** For network devices, use execute_multiple_device_commands. - """ - - def _connect_and_execute_commands( - self, - device_name: str, - commands: list[str], - results_list: list[Any], - index: int, - device_ports: dict[str, Any], - gns3_host: str, - ) -> None: - """ - Internal method to connect to VPCS device and execute commands. - - VPCS devices are lightweight virtual PCs used in GNS3 labs for testing - network connectivity and basic IP configuration. - """ - - logger.info( - "Starting connection for device '%s' with %d commands", - device_name, - len(commands), - ) - - # Check if device has port information - if device_name not in device_ports: - logger.warning( - "Device '%s' not found in topology or missing console port", - device_name, - ) - results_list[index] = { - "device_name": device_name, - "status": "error", - "output": ( - f"Device '{device_name}' not found in topology " - "or missing console port" - ), - "commands": commands, - } - return - - port = device_ports[device_name]["port"] - host = gns3_host - - logger.info( - "Connecting to device '%s' at %s:%d", - device_name, - host, - port, - ) - - tn = Telnet() - try: - tn.open(host=host, port=port, timeout=30) - logger.info( - "Successfully connected to device '%s' at %s:%d", - device_name, - host, - port, - ) - - # Initialize connection - tn.write(b"\n") - sleep(0.5) - tn.write(b"\n") - sleep(0.5) - tn.write(b"\n") - sleep(0.5) - tn.write(b"\n") - sleep(0.5) - tn.expect([rb"PC\d+>"]) - logger.info("Connection initialized for device '%s'", device_name) - - # Execute all commands and merge output - combined_output = "" - for i, command in enumerate(commands): - logger.info( - "Executing command %d/%d on device '%s': %s", - i + 1, - len(commands), - device_name, - command, - ) - tn.write(command.encode(encoding="ascii") + b"\n") - sleep(5) - tn.expect([rb"PC\d+>"]) - output = tn.read_very_eager().decode("utf-8") - combined_output += output - - # Add result to list - results_list[index] = { - "device_name": device_name, - "status": "success", - "output": combined_output, - "commands": commands, - } - logger.info( - "Successfully executed all %d commands on device '%s'", - len(commands), - device_name, - ) - - except Exception as e: - logger.error( - "Error executing commands on device '%s': %s", - device_name, - str(e), - ) - results_list[index] = { - "device_name": device_name, - "status": "error", - "output": str(e), - "commands": commands, - } - finally: - tn.close() - logger.debug("Connection closed for device '%s'", device_name) - - def _validate_project_id(self, project_id: str) -> bool: - """ - Validate project_id format (UUID). - - Args: - project_id: The project ID to validate - - Returns: - True if valid UUID format, False otherwise - """ - uuid_pattern = ( - r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" - ) - is_valid = bool(re.match(uuid_pattern, project_id, re.IGNORECASE)) - if not is_valid: - logger.warning( - "project_id '%s' is not a valid UUID format", project_id - ) - return is_valid - - def _validate_tool_input( - self, tool_input: str | bytes | list[Any] | dict[str, Any] - ) -> tuple[list[dict[str, Any]], str]: - """ - Validate device command input and extract project_id and device_configs. - - Args: - tool_input: Input from the LangChain/LangGraph tool call. - - Returns: - Tuple of (device_configs_list, project_id) or (error_list, "") - """ - - parsed_input = None - - # Compatibility Check and Parsing --- - # Check if the input is a string (or bytes) which needs to be parsed. - if isinstance(tool_input, (str, bytes, bytearray)): - # Handle models that return a raw JSON string. - try: - parsed_input = json.loads(tool_input) - logger.info("Successfully parsed tool input from JSON string.") - except json.JSONDecodeError as e: - logger.error( - "Invalid JSON string received as tool input: %s", e - ) - return ([{"error": f"Invalid JSON input: {e}"}], "") - else: - # Handle standard models where the framework has already parsed the JSON. - parsed_input = tool_input - logger.info( - "Using tool input directly as type: %s", - type(parsed_input).__name__, - ) - - # Validate input is a dictionary - if not isinstance(parsed_input, dict): - error_msg = ( - "Tool input must be a JSON object containing 'project_id' " - f"and 'device_configs', but got {type(parsed_input).__name__}" - ) - logger.error(error_msg) - return ([{"error": error_msg}], "") - - # Extract and validate project_id - project_id = parsed_input.get("project_id") - if not project_id: - error_msg = "Missing required field 'project_id' in input" - logger.error(error_msg) - return ([{"error": error_msg}], "") - - # Validate project_id format - if not self._validate_project_id(project_id): - error_msg = f"Invalid project_id format: {project_id}. Expected UUID." - logger.error(error_msg) - return ([{"error": error_msg}], "") - - # Extract and validate device_configs - device_configs = parsed_input.get("device_configs") - if device_configs is None: - error_msg = "Missing required field 'device_configs' in input" - logger.error(error_msg) - return ([{"error": error_msg}], "") - - # Validate device_configs is a list - if not isinstance(device_configs, list): - error_msg = ( - f"'device_configs' must be a list, " - f"but got {type(device_configs).__name__}" - ) - logger.error(error_msg) - return ([{"error": error_msg}], "") - - # Handle empty list - if not device_configs: - logger.warning("Device configs list is empty.") - return [], "" - - # Validate each item in device_configs - for i, item in enumerate(device_configs): - if not isinstance(item, dict): - error_msg = ( - f"Item at index {i} must be a dictionary, " - f"got {type(item).__name__}" - ) - logger.error(error_msg) - return ([{"error": error_msg}], "") - - # Validate required fields in each device config - if "device_name" not in item: - error_msg = ( - f"Item at index {i} missing required field 'device_name'" - ) - logger.error(error_msg) - return ([{"error": error_msg}], "") - - if "commands" not in item: - error_msg = ( - f"Item at index {i} missing required field 'commands'" - ) - logger.error(error_msg) - return ([{"error": error_msg}], "") - - if not isinstance(item["commands"], list): - error_msg = ( - f"'commands' in item at index {i} must be a list, " - f"but got {type(item['commands']).__name__}" - ) - logger.error(error_msg) - return ([{"error": error_msg}], "") - - logger.info( - "Input validated successfully. project_id=%s, device_configs_count=%d", - project_id, - len(device_configs), - ) - return device_configs, project_id - - def _run( - self, - tool_input: str, - run_manager: CallbackManagerForToolRun | None = None, - **kwargs: Any, - ) -> list[dict[str, Any]]: - """ - Execute commands on multiple VPCS virtual PC devices concurrently. - - VPCS devices are lightweight virtual machines that simulate basic PC network - functionality for lab testing. This is NOT network device configuration. - - Args: - tool_input: JSON string containing project_id and device_configs with - VPCS commands - - Returns: - List of execution results for each VPCS device - """ - - # Log received input - logger.info("Received input: %s", tool_input) - - # Validate tool input and extract project_id and device_configs - device_configs, project_id = self._validate_tool_input(tool_input) - - # Check if validation returned an error - if ( - isinstance(device_configs, list) - and len(device_configs) > 0 - and "error" in device_configs[0] - ): - return device_configs - - # Extract all device names from input using set comprehension - device_names = {config["device_name"] for config in device_configs} - - # Get device port mapping with project_id - device_ports = get_device_ports_from_topology( - list(device_names), project_id=project_id - ) - logger.info( - "Retrieved port mappings for %d devices: %s", - len(device_ports), - list(device_ports.keys()), - ) - - # Get GNS3 server host from connector factory - gns3_host = get_gns3_server_host() - logger.info("Using GNS3 server host: %s", gns3_host) - - # Initialize results list (pre-allocate space for concurrent writes) - results: list[dict[str, Any]] = [{} for _ in range(len(device_configs))] - threads = [] - - # Create thread for each command group - logger.info("Starting parallel execution for %d devices", len(device_configs)) - for i, cmd_group in enumerate(device_configs): - thread = threading.Thread( - target=self._connect_and_execute_commands, - args=( - cmd_group["device_name"], - cmd_group["commands"], - results, - i, - device_ports, - gns3_host, - ), - ) - threads.append(thread) - thread.start() - - # Wait for all threads to complete - for thread in threads: - thread.join() - - # Count successful and failed executions - success_count = sum(1 for r in results if r.get("status") == "success") - error_count = sum(1 for r in results if r.get("status") == "error") - - logger.info( - "Multi-device command execution completed. Total: %d, Success: %d, " - "Error: %d", - len(results), - success_count, - error_count, - ) - - return results - - -if __name__ == "__main__": - # Example usage - command_groups = json.dumps( - { - "project_id": "", - "device_configs": [ - { - "device_name": "PC1", - "commands": [ - "ip 10.10.0.12/24 10.10.0.254", - "ping 10.10.0.254", - ], - }, - { - "device_name": "PC2", - "commands": ["ip 10.10.0.13/24 10.10.0.254"], - }, - { - "device_name": "PC3", - "commands": [ - "ip 10.20.0.22/24 10.20.0.254", - "ping 10.20.0.254", - ], - }, - { - "device_name": "PC4", - "commands": ["ip 10.20.0.23/24 10.20.0.254"], - }, - ], - } - ) - - exe_cmd = VPCSMultiCommands() - result = exe_cmd._run(tool_input=command_groups) - print("Execution results:") - print(json.dumps(result, indent=2, ensure_ascii=False)) diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py index 7cd39adbf..1aea7adf8 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py @@ -33,6 +33,7 @@ authentication or behavior patterns. Supported Drivers: - huawei_ce: GNS3HuaweiTelnetCE for CloudEngine devices (no authentication) - ruijie_telnet: RuijieTelnetEnhanced for Ruijie devices (interactive prompt handling) +- vpcs_telnet: VPCSTelnet for VPCS virtual PC simulator devices (no authentication) All drivers use 'gns3_' prefix to clearly distinguish them from Netmiko's built-in drivers. @@ -69,4 +70,9 @@ try: except Exception as e: logger.warning(f"Failed to import Ruijie driver: {e}", exc_info=True) -__all__ = ["huawei_ce", "ruijie_telnet"] +try: + from . import vpcs_telnet # noqa: F401 +except Exception as e: + logger.warning(f"Failed to import VPCS driver: {e}", exc_info=True) + +__all__ = ["huawei_ce", "ruijie_telnet", "vpcs_telnet"] diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_vpcs_telnet.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_vpcs_telnet.py new file mode 100644 index 000000000..64700e139 --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_vpcs_telnet.py @@ -0,0 +1,495 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-3.0-or-later +# +# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3 +# +# Unit test script for custom Netmiko VPCSTelnet driver +# + +""" +Unit test script for VPCSTelnet custom device driver. + +This script tests: +1. Device type registration +2. Inheritance from BaseConnection +3. VPCS-specific methods (telnet_login, session_preparation, send_command) +4. Config mode methods (check_config_mode, config_mode, exit_config_mode) +5. disable_paging method +6. Default parameters (default_enter, global_delay_factor, fast_cli) + +Run with: python test_vpcs_telnet.py +""" + +import sys +import unittest +from unittest.mock import Mock, patch, MagicMock +import os + +# Add project root to path using relative path +test_dir = os.path.dirname(os.path.abspath(__file__)) +project_root = os.path.dirname( + os.path.dirname(os.path.dirname( + os.path.dirname(os.path.dirname(test_dir))))) +sys.path.insert(0, project_root) + + +class TestVPCSTelnetDriver(unittest.TestCase): + """Test suite for VPCSTelnet custom driver.""" + + @classmethod + def setUpClass(cls): + """Set up test fixtures - import and register custom driver.""" + # Import the custom driver module (triggers registration) + from gns3server.agent.gns3_copilot.utils.custom_netmiko import ( + vpcs_telnet, + ) + + cls.vpcs_telnet = vpcs_telnet + cls.VPCSTelnet = vpcs_telnet.VPCSTelnet + + def test_device_type_registered(self): + """Test that gns3_vpcs_telnet is registered in Netmiko.""" + from netmiko.ssh_dispatcher import CLASS_MAPPER, CLASS_MAPPER_BASE + + # Check CLASS_MAPPER + self.assertIn("gns3_vpcs_telnet", CLASS_MAPPER) + self.assertEqual( + CLASS_MAPPER["gns3_vpcs_telnet"], + self.VPCSTelnet + ) + + # Check CLASS_MAPPER_BASE + self.assertIn("gns3_vpcs_telnet", CLASS_MAPPER_BASE) + self.assertEqual( + CLASS_MAPPER_BASE["gns3_vpcs_telnet"], + self.VPCSTelnet + ) + + def test_inheritance_from_base_connection(self): + """Test that VPCSTelnet inherits from BaseConnection.""" + from netmiko.base_connection import BaseConnection + + # Verify inheritance + self.assertIsInstance(self.VPCSTelnet, type) + # Check if VPCSTelnet is a subclass of BaseConnection + self.assertTrue(issubclass(self.VPCSTelnet, BaseConnection)) + + def test_vpcs_methods_available(self): + """Test that VPCS-specific methods are available.""" + # These methods should be available in VPCSTelnet + vpcs_methods = [ + "telnet_login", + "session_preparation", + "send_command", + "send_command_timing", + "check_config_mode", + "config_mode", + "exit_config_mode", + "disable_paging", + ] + + for method_name in vpcs_methods: + self.assertTrue( + hasattr(self.VPCSTelnet, method_name), + f"Method {method_name} not found in VPCSTelnet", + ) + + def test_connect_handler_accepts_device_type(self): + """Test that ConnectHandler accepts gns3_vpcs_telnet.""" + from netmiko.ssh_dispatcher import CLASS_MAPPER + + # Get platforms list + platforms = list(CLASS_MAPPER.keys()) + + # Verify gns3_vpcs_telnet is in platforms + self.assertIn("gns3_vpcs_telnet", platforms) + + # Verify it's in telnet platforms + telnet_platforms = [x for x in platforms if "telnet" in x] + self.assertIn("gns3_vpcs_telnet", telnet_platforms) + + def test_telnet_platforms_list(self): + """Test that gns3_vpcs_telnet is in telnet_platforms list.""" + from netmiko.ssh_dispatcher import telnet_platforms + + self.assertIn("gns3_vpcs_telnet", telnet_platforms) + + +class TestVPCSTelnetInit(unittest.TestCase): + """Test VPCSTelnet initialization and default parameters.""" + + @classmethod + def setUpClass(cls): + """Set up test fixtures.""" + from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501 + VPCSTelnet, + ) + cls.VPCSTelnet = VPCSTelnet + + def test_default_enter_parameter(self): + """Test that default_enter is set to \\r\\n for VPCS.""" + import inspect + + # Check that default_enter is handled in __init__ + init_source = inspect.getsource(self.VPCSTelnet.__init__) + self.assertIn("default_enter", init_source) + self.assertIn("\\r\\n", init_source) + + def test_default_parameters_when_none(self): + """Test default parameters when not provided.""" + # Mock the BaseConnection.__init__ to avoid actual connection + # But also set device_type since VPCSTelnet.__init__ sets it + def mock_init(self, *args, **kwargs): + # Simulate what VPCSTelnet.__init__ does + kwargs.setdefault("device_type", "gns3_vpcs_telnet") + self.device_type = kwargs.get("device_type") + + with patch.object( + self.VPCSTelnet, "__init__", mock_init + ): + instance = self.VPCSTelnet(host="127.0.0.1") + self.assertEqual(instance.device_type, "gns3_vpcs_telnet") + + def test_device_type_set(self): + """Test that device_type is set to gns3_vpcs_telnet.""" + # Mock the __init__ to set device_type + def mock_init(self, *args, **kwargs): + # Simulate what VPCSTelnet.__init__ does + kwargs.setdefault("device_type", "gns3_vpcs_telnet") + self.device_type = kwargs.get("device_type") + + with patch.object( + self.VPCSTelnet, "__init__", mock_init + ): + instance = self.VPCSTelnet(host="127.0.0.1") + self.assertEqual(instance.device_type, "gns3_vpcs_telnet") + + +class TestVPCSTelnetMethods(unittest.TestCase): + """Test VPCSTelnet method implementations.""" + + @classmethod + def setUpClass(cls): + """Set up test fixtures.""" + from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501 + VPCSTelnet, + ) + cls.VPCSTelnet = VPCSTelnet + + def test_check_config_mode_always_false(self): + """Test that check_config_mode always returns False.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + result = instance.check_config_mode() + self.assertFalse(result) + + def test_config_mode_returns_empty(self): + """Test that config_mode returns empty string.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + result = instance.config_mode() + self.assertEqual(result, "") + + def test_exit_config_mode_returns_empty(self): + """Test that exit_config_mode returns empty string.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + result = instance.exit_config_mode() + self.assertEqual(result, "") + + def test_disable_paging_returns_empty(self): + """Test that disable_paging returns empty string.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + result = instance.disable_paging() + self.assertEqual(result, "") + + def test_session_preparation_logs_debug(self): + """Test that session_preparation completes without errors.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + # Should not raise any exception + instance.session_preparation() + + +class TestVPCSTelnetSendCommand(unittest.TestCase): + """Test VPCSTelnet send_command method.""" + + @classmethod + def setUpClass(cls): + """Set up test fixtures.""" + from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501 + VPCSTelnet, + ) + cls.VPCSTelnet = VPCSTelnet + + def test_send_command_writes_bytes(self): + """Test that send_command writes command as bytes.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + + # Mock the necessary attributes + instance.remote_conn = MagicMock() + instance.remote_conn.write = MagicMock() + + # Test command encoding + command_string = "ip 192.168.1.1 255.255.255.0" + expected_bytes = command_string.encode("ascii") + b"\n" + + # Call send_command (will fail at read_until_pattern, but we can + # check the write call) + try: + instance.send_command(command_string) + except Exception: + pass + + # Verify write was called with correct bytes + instance.remote_conn.write.assert_called_once_with(expected_bytes) + + def test_send_command_default_expect_string(self): + """Test that send_command uses VPCS prompt as default expect_string.""" + import inspect + + # Check that expect_string defaults to r"PC\d+>" + source = inspect.getsource(self.VPCSTelnet.send_command) + self.assertIn("expect_string = r\"PC\\d+>\"", source) + + def test_send_command_timing_calls_send_command(self): + """Test that send_command_timing delegates to send_command.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + + # Mock send_command + instance.send_command = Mock(return_value="test output") + + # Call send_command_timing + result = instance.send_command_timing( + "ip 192.168.1.1 255.255.255.0", + strip_prompt=True, + strip_command=True, + ) + + # Verify send_command was called + instance.send_command.assert_called_once_with( + command_string="ip 192.168.1.1 255.255.255.0", + strip_prompt=True, + strip_command=True, + ) + self.assertEqual(result, "test output") + + +class TestVPCSTelnetTelnetLogin(unittest.TestCase): + """Test VPCSTelnet telnet_login method.""" + + @classmethod + def setUpClass(cls): + """Set up test fixtures.""" + from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501 + VPCSTelnet, + ) + cls.VPCSTelnet = VPCSTelnet + + def test_telnet_login_default_prompt_pattern(self): + """Test that telnet_login has correct default prompt pattern.""" + import inspect + + # Check default prompt terminator + source = inspect.getsource(self.VPCSTelnet.telnet_login) + self.assertIn('pri_prompt_terminator: str = r"PC\\d+>"', source) + + def test_telnet_login_sends_newlines(self): + """Test that telnet_login sends 4 newlines.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + + # Mock necessary attributes + instance.remote_conn = MagicMock() + instance.remote_conn.write = MagicMock() + instance.read_channel = MagicMock(return_value="PC1>") + instance.select_delay_factor = Mock(return_value=1.0) + instance.read_until_pattern = MagicMock(return_value="PC1>") + + # Call telnet_login + instance.telnet_login() + + # Verify write was called 4 times (4 newlines) + self.assertEqual(instance.remote_conn.write.call_count, 4) + + # Verify each call was with b"\n" + for call in instance.remote_conn.write.call_args_list: + self.assertEqual(call[0][0], b"\n") + + def test_telnet_login_waits_for_prompt(self): + """Test that telnet_login waits for VPCS prompt.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + + # Mock necessary attributes + instance.remote_conn = MagicMock() + instance.remote_conn.write = MagicMock() + instance.read_channel = MagicMock(return_value="PC1>") + instance.select_delay_factor = Mock(return_value=1.0) + instance.read_until_pattern = MagicMock(return_value="PC1>") + + # Call telnet_login + result = instance.telnet_login() + + # Verify read_until_pattern was called + instance.read_until_pattern.assert_called_once() + + +class TestVPCSTelnetRegistration(unittest.TestCase): + """Test VPCSTelnet registration function.""" + + @classmethod + def setUpClass(cls): + """Set up test fixtures.""" + from gns3server.agent.gns3_copilot.utils.custom_netmiko import ( + vpcs_telnet, + ) + cls.vpcs_telnet = vpcs_telnet + + def test_register_function_exists(self): + """Test that register_custom_device_type function exists.""" + self.assertTrue( + hasattr(self.vpcs_telnet, "register_custom_device_type") + ) + + def test_register_function_is_callable(self): + """Test that register_custom_device_type is callable.""" + self.assertTrue( + callable(self.vpcs_telnet.register_custom_device_type) + ) + + +class TestVPCSTelnetAnsiStripping(unittest.TestCase): + """Test VPCSTelnet ANSI escape code stripping.""" + + @classmethod + def setUpClass(cls): + """Set up test fixtures.""" + from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501 + VPCSTelnet, + ) + cls.VPCSTelnet = VPCSTelnet + + def test_strip_ansi_codes_method_exists(self): + """Test that _strip_ansi_codes method exists.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + self.assertTrue(hasattr(instance, "_strip_ansi_codes")) + self.assertTrue(callable(instance._strip_ansi_codes)) + + def test_strip_bold_codes(self): + """Test stripping bold ANSI codes.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + text = "\x1b[1mip\x1b[0m \x1b[4mARG\x1b[0m" + result = instance._strip_ansi_codes(text) + self.assertEqual(result, "ip ARG") + + def test_strip_underline_codes(self): + """Test stripping underline ANSI codes.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + text = "\x1b[4maddress\x1b[0m [\x1b[4mmask\x1b[0m]" + result = instance._strip_ansi_codes(text) + self.assertEqual(result, "address [mask]") + + def test_strip_reset_codes(self): + """Test stripping reset ANSI codes.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + text = "\x1b[1mNAME\x1b[0m : PC2[1]" + result = instance._strip_ansi_codes(text) + # Note: [1] at the end is not a valid ANSI code, so it should remain + self.assertEqual(result, "NAME : PC2[1]") + + def test_strip_color_codes(self): + """Test stripping color ANSI codes.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + text = "\x1b[31mRed\x1b[32mGreen\x1b[0mNormal" + result = instance._strip_ansi_codes(text) + self.assertEqual(result, "RedGreenNormal") + + def test_strip_complex_vpcs_output(self): + """Test stripping ANSI codes from real VPCS output.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + # Sample VPCS output with ANSI codes (with proper ESC characters) + text = """\x1b[1mip\x1b[0m \x1b[4mARG\x1b[0m ... [[\x1b[4mOPTION\x1b[0m] + Configure the current VPC's IP settings + \x1b[1mNAME\x1b[0m : PC2[1] +IP/MASK : 192.168.1.20/24""" + result = instance._strip_ansi_codes(text) + # Verify ANSI codes are removed + self.assertNotIn("\x1b[1m", result) + self.assertNotIn("\x1b[0m", result) + self.assertNotIn("\x1b[4m", result) + self.assertIn("ip ARG", result) + self.assertIn("NAME", result) + self.assertIn("PC2[1]", result) # [1] is not a valid ANSI escape + + def test_strip_empty_string(self): + """Test stripping ANSI codes from empty string.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + result = instance._strip_ansi_codes("") + self.assertEqual(result, "") + + def test_strip_no_ansi_codes(self): + """Test that text without ANSI codes is unchanged.""" + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + text = "NAME : PC2\nIP/MASK : 192.168.1.20/24" + result = instance._strip_ansi_codes(text) + self.assertEqual(result, text) + + def test_send_command_calls_strip_ansi(self): + """Test that send_command calls _strip_ansi_codes.""" + from unittest.mock import patch + + instance = self.VPCSTelnet.__new__(self.VPCSTelnet) + + # Mock necessary attributes + instance.remote_conn = MagicMock() + instance.remote_conn.write = MagicMock() + instance.read_until_pattern = MagicMock( + return_value="\x1b[1mNAME\x1b[0m : PC2\nPC2>" + ) + + # Mock _strip_ansi_codes to verify it's called + instance._strip_ansi_codes = Mock( + return_value="NAME : PC2\nPC2>" + ) + + # Call send_command + instance.send_command("show ip") + + # Verify _strip_ansi_codes was called + instance._strip_ansi_codes.assert_called_once() + call_arg = instance._strip_ansi_codes.call_args[0][0] + self.assertIn("\x1b[1m", call_arg) # Verify it received raw output + + +def run_tests(): + """Run all tests and print results.""" + # Create test suite + loader = unittest.TestLoader() + suite = unittest.TestSuite() + + # Add test cases + suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetDriver)) + suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetInit)) + suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetMethods)) + suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetSendCommand)) + suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetTelnetLogin)) + suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetRegistration)) + suite.addTests(loader.loadTestsFromTestCase(TestVPCSTelnetAnsiStripping)) + + # Run tests + runner = unittest.TextTestRunner(verbosity=2) + result = runner.run(suite) + + # Print summary + print("\n" + "=" * 100) + print("Test Summary:") + print(f" Run: {result.testsRun}") + success_count = ( + result.testsRun - len(result.failures) - len(result.errors) + ) + print(f" Success: {success_count}") + print(f" Failed: {len(result.failures)}") + print(f" Errors: {len(result.errors)}") + print("=" * 100) + + return result.wasSuccessful() + + +if __name__ == "__main__": + success = run_tests() + sys.exit(0 if success else 1) diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/vpcs_telnet.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/vpcs_telnet.py new file mode 100644 index 000000000..64b513a2f --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/vpcs_telnet.py @@ -0,0 +1,424 @@ +# 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 +# + +# mypy: ignore-errors + +""" +Custom Netmiko device driver for VPCS (Virtual PC Simulator) in GNS3. + +VPCS is a lightweight virtual PC simulator used in GNS3 lab environments +for testing network connectivity and basic IP configuration. + +Key Features: +- No authentication required (direct console access) +- Simple command-line interface (PC1>, PC2>, etc.) +- Supports basic PC commands (ip, ping, arp, version, etc.) +- Telnet-based connection + +Device Type: gns3_vpcs_telnet +""" + +import importlib +import logging +import re +import time + +from netmiko.base_connection import BaseConnection + + +logger = logging.getLogger(__name__) + +# ANSI escape code pattern for stripping terminal formatting codes +# Matches sequences like: [1m (bold), [0m (reset), [4m (underline), etc. +ANSI_ESCAPE_PATTERN = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]") + + +class VPCSTelnet(BaseConnection): + """ + Custom VPCS device driver for GNS3 emulation. + + VPCS devices are lightweight virtual PCs that simulate basic network + functionality for lab testing. This driver handles the unique + characteristics of VPCS: + - No authentication (no username/password) + - Simple prompt pattern (PC1>, PC2>, etc.) + - No configuration modes + """ + + def __init__( + self, + *args, + **kwargs, + ) -> None: + """Initialize VPCS Telnet connection.""" + # Set device type to identify this as a VPCS device + kwargs.setdefault("device_type", "gns3_vpcs_telnet") + + # VPCS uses carriage return + line feed for command termination + # This is critical for VPCS devices to respond to commands + default_enter = kwargs.get("default_enter") + if default_enter is None: + kwargs["default_enter"] = "\r\n" + + kwargs.setdefault("global_delay_factor", 1.0) + kwargs.setdefault("fast_cli", False) + + super().__init__(*args, **kwargs) + + def telnet_login( + self, + pri_prompt_terminator: str = r"PC\d+>", + alt_prompt_terminator: str = r"", + username_pattern: str = r"(?:user:|username|login|user name)", + pwd_pattern: str = r"assword", + delay_factor: float = 1.0, + max_loops: int = 20, + ) -> str: + """ + Telnet login for VPCS devices (no authentication). + + VPCS devices connect directly to command line without username/password. + This method initializes the connection and waits for the VPCS prompt. + + Strategy (matching telnetlib3 implementation): + 1. Send 4 newlines with 0.5s delay between each + 2. Wait for VPCS prompt pattern (PC1>, PC2>, etc.) + 3. Return once prompt is detected + + Args: + pri_prompt_terminator: Primary prompt pattern (default: r"PC\\d+>") + alt_prompt_terminator: Not used for VPCS + username_pattern: Not used (kept for signature compatibility) + pwd_pattern: Not used (kept for signature compatibility) + delay_factor: Delay factor for timing + max_loops: Maximum wait loops + + Returns: + Output from the connection process + """ + delay_factor = self.select_delay_factor(delay_factor) + return_msg = "" + + # Step 1: Clear buffer - read any existing data + try: + initial_data = self.read_channel() + if initial_data: + return_msg += initial_data + except Exception: + # Ignore errors during initial read + pass + + # Step 2: Send 4 newlines with delays (matching telnetlib3 exactly) + # tn.write(b"\n"); sleep(0.5) - repeated 4 times + for i in range(4): + try: + # Send newline directly as bytes + self.remote_conn.write(b"\n") + time.sleep(0.5 * delay_factor) + + # Read response + new_output = self.read_channel() + if new_output: + return_msg += new_output + + except Exception as e: + logger.debug("Error during VPCS connection initialization: %s", e) + + # Step 3: Wait for VPCS prompt pattern + try: + # Read until we see the prompt + output = self.read_until_pattern( + pattern=pri_prompt_terminator, + read_timeout=10 + ) + return_msg += output + + if re.search(pri_prompt_terminator, return_msg, flags=re.M): + logger.info("VPCS prompt detected") + return return_msg + + except Exception as e: + logger.debug("Error waiting for VPCS prompt: %s", e) + + # Step 4: Return what we have (connection might still work) + logger.warning("VPCS prompt not clearly detected, returning current output") + return return_msg + + def session_preparation(self) -> None: + """ + Prepare the session after connection is established. + + VPCS doesn't require special session preparation. + The telnet_login method already handles initialization with 4 newlines. + """ + # No additional preparation needed + logger.debug("VPCS session preparation completed") + + def send_command( + self, + command_string: str, + expect_string: str | None = None, + read_timeout: float | None = None, + delay_factor: float | None = None, + max_loops: int | None = None, + strip_prompt: bool = False, + strip_command: bool = False, + normalize: bool = False, + use_textfsm: bool = False, + ) -> str: + """ + Send a command to VPCS and return the output. + + This method exactly matches the telnetlib3 implementation: + 1. Encode command as ASCII and append newline + 2. Write command+newline as single byte sequence + 3. Sleep for 5 seconds + 4. Wait for prompt pattern and capture output + 5. Return captured output + + Args: + command_string: The command to send + expect_string: Pattern to expect (default: VPCS prompt r"PC\\d+>") + read_timeout: Timeout for reading output + delay_factor: Not used (kept for signature compatibility) + max_loops: Not used (kept for signature compatibility) + strip_prompt: Remove prompt from output (default: False for debugging) + strip_command: Remove command from output (default: False for debugging) + normalize: Not used for VPCS + use_textfsm: Not used for VPCS + + Returns: + Command output from VPCS + """ + # Use VPCS prompt as default expect_string + if expect_string is None: + expect_string = r"PC\d+>" + + # CRITICAL: Match telnetlib3 behavior exactly + # tn.write(command.encode(encoding="ascii") + b"\n") + cmd_bytes = command_string.encode(encoding="ascii") + b"\n" + self.remote_conn.write(cmd_bytes) + + logger.debug("VPCS: Sent command '%s' (%d bytes)", command_string, len(cmd_bytes)) + + # Sleep for 5 seconds (matching telnetlib3: sleep(5)) + time.sleep(5) + + # Wait for prompt pattern and capture output + # This matches telnetlib3: tn.expect([rb"PC\d+>"]) + tn.read_very_eager() + # read_until_pattern returns the output including the matched pattern + try: + output = self.read_until_pattern(pattern=expect_string, read_timeout=10) + logger.debug("VPCS: read_until_pattern returned %d chars", len(output)) + except Exception as e: + # If pattern not found, try to read whatever is available + logger.debug("VPCS: Pattern not found, reading available output: %s", e) + output = self.read_channel() + logger.debug("VPCS: read_channel returned %d chars", len(output) if output else 0) + + if isinstance(output, bytes): + output = output.decode("utf-8", errors="replace") + + logger.debug("VPCS: Raw output before strip: '%s'", output) + + # Strip ANSI escape codes from VPCS output + # VPCS uses terminal formatting codes (bold, underline, etc.) + # that need to be removed for clean output parsing + output = self._strip_ansi_codes(output) + + # Strip command and prompt if requested (currently disabled for debugging) + if strip_command: + output = self.strip_command_packets(command_string, output) + if strip_prompt: + output = self.strip_prompt(output) + + logger.debug("VPCS: Final output after strip: '%s'", output) + + return output + + def send_command_timing( + self, + command_string: str, + delay_factor: float | None = None, + max_loops: int | None = None, + strip_prompt: bool = True, + strip_command: bool = True, + normalize: bool = False, + ) -> str: + """ + Send command using timing-based delay. + + For VPCS, timing-based and pattern-based methods are identical, + both use the telnetlib3 implementation pattern with fixed 5s delay. + + Args: + command_string: The command to send + delay_factor: Not used (fixed 5s delay for VPCS) + max_loops: Not used (kept for signature compatibility) + strip_prompt: Remove prompt from output + strip_command: Remove command from output + normalize: Not used for VPCS + + Returns: + Command output from VPCS + """ + # For VPCS, timing and pattern-based methods are identical + return self.send_command( + command_string=command_string, + strip_prompt=strip_prompt, + strip_command=strip_command, + ) + + def check_config_mode( + self, check_string: str = "", pattern: str = "", force_regex: bool = False + ) -> bool: + """ + VPCS has no configuration mode. + + This method always returns False for VPCS devices. + + Returns: + False (VPCS has no config mode) + """ + return False + + def _strip_ansi_codes(self, text: str) -> str: + """ + Strip ANSI escape codes from VPCS output. + + VPCS uses terminal formatting codes (bold, underline, colors, etc.) + that need to be removed for clean output parsing. This method removes + all ANSI escape sequences from the given text. + + Examples of codes removed: + - [1m - Bold text + - [0m - Reset formatting + - [4m - Underline text + - [30-47m - Color codes + + Args: + text: Text potentially containing ANSI escape codes + + Returns: + Text with all ANSI escape codes removed + """ + return ANSI_ESCAPE_PATTERN.sub("", text) + + def config_mode( + self, config_command: str = "", pattern: str = "", re_flags: int = 0 + ) -> str: + """ + VPCS has no configuration mode. + + This method returns an empty string for VPCS devices. + + Returns: + Empty string (no config mode to enter) + """ + return "" + + def exit_config_mode(self, exit_config: str = "", pattern: str = "") -> str: + """ + VPCS has no configuration mode. + + This method returns an empty string for VPCS devices. + + Returns: + Empty string (no config mode to exit) + """ + return "" + + def disable_paging( + self, + command: str = "", + delay_factor: float | None = None, + **kwargs, + ) -> str: + """ + VPCS doesn't use paging. + + This method returns an empty string for VPCS devices. + + Returns: + Empty string (no paging to disable) + """ + return "" + + +# Register the custom device type with Netmiko +def register_custom_device_type() -> None: + """ + Register the custom VPCS Telnet device type with Netmiko. + + This function adds 'gns3_vpcs_telnet' to Netmiko's CLASS_MAPPER + and updates the platforms lists. + + IMPORTANT: This function should be called BEFORE using the VPCS device + type with Netmiko. + """ + # Use importlib to avoid namespace conflicts + sd = importlib.import_module("netmiko.ssh_dispatcher") + + # Register the device type in both mappers + # CLASS_MAPPER_BASE is used for base class definitions + sd.CLASS_MAPPER_BASE["gns3_vpcs_telnet"] = VPCSTelnet + + # CLASS_MAPPER is used by ConnectHandler for device type validation + sd.CLASS_MAPPER["gns3_vpcs_telnet"] = VPCSTelnet + + # CRITICAL: Update the static platforms lists + # These lists are computed at module import time and won't + # automatically update when CLASS_MAPPER is modified. + # We need to manually rebuild them. + + # Recalculate platforms list + sd.platforms = list(sd.CLASS_MAPPER.keys()) + sd.platforms.sort() + + # Recalculate platforms_base list + sd.platforms_base = list(sd.CLASS_MAPPER_BASE.keys()) + sd.platforms_base.sort() + + # Recalculate telnet_platforms list + sd.telnet_platforms = [x for x in sd.platforms if "telnet" in x] + + # Rebuild the platform strings used in error messages + sd.platforms_str = "\n" + "\n".join(sd.platforms_base) + sd.telnet_platforms_str = "\n" + "\n".join(sd.telnet_platforms) + + logger.info("Successfully registered VPCS Telnet device type with Netmiko") + + +# Auto-register on import +# This ensures the device type is available when the module is imported +try: + register_custom_device_type() +except Exception as e: + # Log but don't fail on import + logger.warning( + "Failed to register VPCS device type: %s", + e, + exc_info=True + ) diff --git a/gns3server/services/templates.py b/gns3server/services/templates.py index 2caf9fa78..c41de5d9d 100644 --- a/gns3server/services/templates.py +++ b/gns3server/services/templates.py @@ -110,6 +110,7 @@ BUILTIN_TEMPLATES = [ "base_script_file": "vpcs_base_config.txt", "compute_id": None, "builtin": True, + "tags": ["platform:vpcs", "device_type:gns3_vpcs_telnet"], }, { "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "ethernet_switch"),