mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 12:30:13 +03:00
feat(copilot): add multi-vendor device support with custom Huawei driver
Add comprehensive multi-vendor support for GNS3 network automation,
including a custom Netmiko driver for Huawei CloudEngine devices.
Features:
- Custom HuaweiTelnetCE driver for GNS3 emulation (no authentication)
- Auto-commit before exit to prevent [Y/N/C] prompts
- Dynamic device type detection from GNS3 node tags
- Support for both Cisco IOS and Huawei devices
- Proper VRP command handling (system-view, return confirmation)
Implementation:
- New package: utils/custom_netmiko/
- huawei_ce.py: Huawei CloudEngine driver
- tests/test_huawei_ce.py: Unit tests (9/9 passing)
- README.md: Driver development guide
- Updated tools for multi-vendor support:
- display_tools_nornir.py: Dynamic group generation
- config_tools_nornir.py: Multi-vendor config commands
- get_gns3_device_port.py: Device port extraction
- Documentation: multi-vendor-device-support.md
Limitations:
- huawei_telnet_ce driver requires devices without authentication
- For devices with username/password, use standard huawei_telnet driver
Co-Authored-By: Yue Guobin <yueguobin@outlook.com>"
This commit is contained in:
YueGuobin
parent
ab88bcd57e
commit
db164e9d41
@ -12,7 +12,8 @@ docs/gns3-copilot/
|
||||
├── llm-model-configs.md # LLM model configuration system
|
||||
├── command-security.md # Command security and filtering
|
||||
├── context-window-management.md # Context window optimization
|
||||
└── node-control-tools.md # Node start/stop/suspend tools for lab automation
|
||||
├── node-control-tools.md # Node start/stop/suspend tools for lab automation
|
||||
└── multi-vendor-device-support.md # Multi-vendor device support (Cisco, Huawei)
|
||||
```
|
||||
|
||||
## Implemented Features
|
||||
@ -75,6 +76,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.
|
||||
|
||||
**Key Features:**
|
||||
- Custom HuaweiTelnetCE driver for GNS3 emulation (no authentication)
|
||||
- Cisco IOS Telnet support
|
||||
- Dynamic device type detection from GNS3 tags
|
||||
- Automatic Nornir group generation
|
||||
- VRP-specific command handling (system-view, return confirmation)
|
||||
|
||||
**Tested Vendors:**
|
||||
- Cisco IOS (Telnet)
|
||||
- Huawei CloudEngine (Telnet, custom driver)
|
||||
|
||||
**Status:** ✅ Implemented
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
The following features are currently under consideration or development:
|
||||
|
||||
472
docs/gns3-copilot/implemented/multi-vendor-device-support.md
Normal file
472
docs/gns3-copilot/implemented/multi-vendor-device-support.md
Normal file
@ -0,0 +1,472 @@
|
||||
# Multi-Vendor Network Device Support
|
||||
|
||||
## Overview
|
||||
|
||||
GNS3-Copilot supports network devices from multiple vendors through Netmiko and Nornir integration. The system includes a custom Netmiko driver for Huawei devices in GNS3 emulation environments and supports dynamic device type detection.
|
||||
|
||||
## Supported Vendors
|
||||
|
||||
| Vendor | Platform | Device Type | Protocol | Status |
|
||||
|--------|----------|-------------|----------|--------|
|
||||
| **Cisco** | `cisco_ios` | `cisco_ios_telnet` | Telnet | ✅ Tested |
|
||||
| **Huawei** | `huawei` | `huawei_telnet_ce` | Telnet | ✅ Tested (Custom Driver) |
|
||||
|
||||
## Custom Huawei Driver (`HuaweiTelnetCE`)
|
||||
|
||||
### Problem Statement
|
||||
|
||||
GNS3-emulated Huawei devices connect via console **without requiring authentication**. Standard Netmiko drivers attempt username/password authentication, causing connection timeouts.
|
||||
|
||||
**Standard Driver Behavior:**
|
||||
```
|
||||
Telnet Connection → Wait for username prompt → Send username → Wait for password → Send password → Access
|
||||
^ Times out after 20 seconds
|
||||
```
|
||||
|
||||
**GNS3 Huawei Device:**
|
||||
```
|
||||
Telnet Connection → Direct access to command line (no login prompts)
|
||||
<HUAWEI>
|
||||
```
|
||||
|
||||
### Solution: Custom Driver Architecture
|
||||
|
||||
```
|
||||
BaseConnection (Netmiko base class)
|
||||
↓
|
||||
CiscoBaseConnection (Cisco-style base class)
|
||||
↓
|
||||
HuaweiBase (Huawei device base class) ← Inherits VRP support
|
||||
↓
|
||||
HuaweiTelnetCE (Custom GNS3 driver) ← Overrides telnet_login only
|
||||
```
|
||||
|
||||
**Why Inherit from HuaweiBase?**
|
||||
- ✅ Built-in VRP (Versatile Routing Platform) command handling
|
||||
- ✅ Huawei-specific configuration mode (`system-view`)
|
||||
- ✅ Huawei prompt patterns (`<...>`, `[...]`)
|
||||
- ✅ Huawei paging disable (`screen-length 0 temporary`)
|
||||
- ✅ Minimal code changes - only override authentication
|
||||
|
||||
### HuaweiTelnetCE Implementation
|
||||
|
||||
#### Location
|
||||
```
|
||||
gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py
|
||||
```
|
||||
|
||||
**Package Structure:**
|
||||
```
|
||||
custom_netmiko/
|
||||
├── __init__.py # Package initialization, auto-registers all drivers
|
||||
├── huawei_ce.py # Huawei CloudEngine custom driver
|
||||
├── README.md # Driver development guide
|
||||
└── tests/ # Unit tests
|
||||
├── __init__.py
|
||||
└── test_huawei_ce.py # Huawei CE driver tests
|
||||
```
|
||||
|
||||
#### Key Features
|
||||
|
||||
1. **Skip Authentication**
|
||||
- Directly detect Huawei prompt patterns
|
||||
- No username/password prompts
|
||||
- Connection ready in < 1 second
|
||||
|
||||
2. **VRP Prompt Recognition**
|
||||
```
|
||||
User view: <HUAWEI>
|
||||
System view: [HUAWEI]
|
||||
Interface: [HUAWEI-GigabitEthernet0/0/1]
|
||||
```
|
||||
|
||||
3. **Automatic Confirmation Handling**
|
||||
- Detects and responds to `[y/n]` prompts
|
||||
- Example: `return` command asks "Return to user view? [y/n]:"
|
||||
- Automatically sends `y` to confirm
|
||||
|
||||
4. **Proper Output Collection**
|
||||
- Uses Netmiko's `read_channel_timing()` for reliable output
|
||||
- Waits for command completion (2s no new data = done)
|
||||
- 30-second absolute timeout prevents hanging
|
||||
|
||||
5. **Auto-Commit Before Exit**
|
||||
- Automatically sends `commit` command before exiting config mode
|
||||
- Prevents "Uncommitted configurations [Y/N/C]" prompt
|
||||
- Ensures configuration changes are saved
|
||||
|
||||
#### Limitations
|
||||
|
||||
**Authentication Requirement:**
|
||||
- The `huawei_telnet_ce` driver is designed for GNS3 devices **without authentication**
|
||||
- If your Huawei device has been configured with a username/password:
|
||||
- **Option 1**: Use the standard `huawei_telnet` driver (requires username/password)
|
||||
- **Option 2**: Remove authentication from the device for GNS3 testing
|
||||
- The driver does **not** currently auto-detect authentication requirements
|
||||
|
||||
**When to Use Each Driver:**
|
||||
|
||||
| Scenario | Use Driver | Requires Credentials? |
|
||||
|----------|-----------|----------------------|
|
||||
| GNS3 Huawei (fresh, no auth) | `huawei_telnet_ce` | ❌ No |
|
||||
| GNS3 Huawei (configured with username/password) | `huawei_telnet` | ✅ Yes |
|
||||
| Real Huawei hardware | `huawei_telnet` | ✅ Yes |
|
||||
|
||||
#### Method Overrides
|
||||
|
||||
**1. `telnet_login` - Skip Authentication**
|
||||
```python
|
||||
def telnet_login(self, pri_prompt_terminator=r"<\S+>|>\s*$",
|
||||
alt_prompt_terminator=r"\[\S+\]", ...) -> str:
|
||||
# Clear buffer
|
||||
self.read_channel()
|
||||
|
||||
# Send returns until prompt detected
|
||||
for i in range(max_loops):
|
||||
self.write_channel(self.RETURN)
|
||||
output = self.read_channel()
|
||||
|
||||
# Check for Huawei prompts
|
||||
if re.search(pri_prompt_terminator, output):
|
||||
return output # Success!
|
||||
|
||||
return output # Best effort
|
||||
```
|
||||
|
||||
**2. `send_config_set` - Configuration Commands**
|
||||
```python
|
||||
def send_config_set(self, config_commands, **kwargs) -> str:
|
||||
# Enter config mode
|
||||
output += self.config_mode(config_command="system-view")
|
||||
|
||||
# Send all commands
|
||||
for cmd in config_commands:
|
||||
self.write_channel(f"{cmd}{self.RETURN}")
|
||||
time.sleep(delay_factor * 0.05)
|
||||
|
||||
# Collect output using Netmiko standard method
|
||||
output += self.read_channel_timing(read_timeout=30, last_read=2.0)
|
||||
|
||||
# Auto-commit before exit (prevents [Y/N/C] prompt)
|
||||
self.write_channel(f"commit{self.RETURN}")
|
||||
time.sleep(0.5 * self.global_delay_factor)
|
||||
output += self.read_channel()
|
||||
|
||||
# Exit config mode
|
||||
output += self.exit_config_mode()
|
||||
|
||||
return output
|
||||
```
|
||||
|
||||
**3. `exit_config_mode` - Handle Confirmation**
|
||||
```python
|
||||
def exit_config_mode(self, exit_config="return", pattern=r"<\S+>|>\s*$") -> str:
|
||||
self.write_channel(f"return{self.RETURN}")
|
||||
|
||||
# Look for confirmation prompt
|
||||
for _ in range(20):
|
||||
new_output = self.read_channel()
|
||||
|
||||
if re.search(r"\[y/n\]", new_output):
|
||||
self.write_channel(f"y{self.RETURN}") # Auto-confirm
|
||||
|
||||
if re.search(pattern, new_output):
|
||||
return output # Back to user view
|
||||
|
||||
return output
|
||||
```
|
||||
|
||||
### Device Type Registration
|
||||
|
||||
The custom driver must be registered with Netmiko's global mappings:
|
||||
|
||||
```python
|
||||
def register_custom_device_type() -> None:
|
||||
import importlib
|
||||
sd = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
|
||||
# Register in CLASS_MAPPER (for ConnectHandler)
|
||||
sd.CLASS_MAPPER["huawei_telnet_ce"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER["huawei_ce"] = HuaweiTelnetCE
|
||||
|
||||
# Register in CLASS_MAPPER_BASE (for base class definitions)
|
||||
sd.CLASS_MAPPER_BASE["huawei_telnet_ce"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER_BASE["huawei_ce"] = HuaweiTelnetCE
|
||||
|
||||
# CRITICAL: Rebuild static lists
|
||||
sd.platforms = list(sd.CLASS_MAPPER.keys())
|
||||
sd.platforms.sort()
|
||||
sd.telnet_platforms = [x for x in sd.platforms if "telnet" in x]
|
||||
```
|
||||
|
||||
**Important: Static List Problem**
|
||||
- `ssh_dispatcher.platforms` is computed at module import time
|
||||
- Modifying `CLASS_MAPPER` doesn't automatically update `platforms`
|
||||
- Must manually rebuild the list after registration
|
||||
|
||||
**Auto-Registration**
|
||||
```python
|
||||
# Automatically runs on module import
|
||||
try:
|
||||
register_custom_device_type()
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to register custom device type: {e}")
|
||||
```
|
||||
|
||||
## Dynamic Device Type Detection
|
||||
|
||||
### GNS3 Node Tags
|
||||
|
||||
Device type and platform are extracted from GNS3 node tags:
|
||||
|
||||
```
|
||||
device_type:huawei_telnet_ce → Netmiko device type (precise)
|
||||
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:huawei_telnet_ce` | `platform:huawei` |
|
||||
|
||||
### Automatic Group Generation
|
||||
|
||||
Nornir groups are dynamically generated based on device type:
|
||||
|
||||
```python
|
||||
# From get_gns3_device_port.py
|
||||
if device_type and "_telnet" in device_type:
|
||||
group_name = f"{platform}_telnet" # e.g., "huawei_telnet"
|
||||
else:
|
||||
group_name = platform # e.g., "cisco_ios"
|
||||
|
||||
hosts_data[device_name] = {
|
||||
"port": console_port,
|
||||
"groups": [group_name],
|
||||
"device_type": device_type,
|
||||
"platform": platform,
|
||||
}
|
||||
```
|
||||
|
||||
**Why Dynamic Groups?**
|
||||
- Single Nornir instance can handle multiple vendors
|
||||
- Each device gets correct connection parameters
|
||||
- Vendor-specific commands work automatically
|
||||
|
||||
## Usage Examples
|
||||
|
||||
### Direct Netmiko Usage
|
||||
|
||||
**Huawei Device (Custom Driver):**
|
||||
```python
|
||||
from netmiko import ConnectHandler
|
||||
from gns3server.agent.gns3_copilot.utils import custom_netmiko
|
||||
|
||||
# Custom driver auto-registers on import
|
||||
device = {
|
||||
"device_type": "huawei_telnet_ce",
|
||||
"host": "127.0.0.1",
|
||||
"port": 5000,
|
||||
# No username/password needed!
|
||||
}
|
||||
|
||||
with ConnectHandler(**device) as conn:
|
||||
# Execute display command
|
||||
output = conn.send_command("display version")
|
||||
|
||||
# Execute configuration commands
|
||||
config = [
|
||||
"interface GE1/0/1",
|
||||
"description Uplink-to-Core",
|
||||
"undo shutdown"
|
||||
]
|
||||
output = conn.send_config_set(config)
|
||||
```
|
||||
|
||||
**Cisco IOS Device (Standard Driver):**
|
||||
```python
|
||||
from netmiko import ConnectHandler
|
||||
|
||||
device = {
|
||||
"device_type": "cisco_ios_telnet",
|
||||
"host": "127.0.0.1",
|
||||
"port": 5001,
|
||||
"username": "cisco",
|
||||
"password": "cisco",
|
||||
}
|
||||
|
||||
with ConnectHandler(**device) as conn:
|
||||
output = conn.send_command("show version")
|
||||
config = ["interface GigabitEthernet0/0", "description Test"]
|
||||
output = conn.send_config_set(config)
|
||||
```
|
||||
|
||||
### Nornir Multi-Vendor Automation
|
||||
|
||||
```python
|
||||
from nornir import InitNornir
|
||||
from gns3server.agent.gns3_copilot.utils import custom_netmiko
|
||||
|
||||
# Auto-register custom driver (happens automatically on import)
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce
|
||||
huawei_ce.register_custom_device_type()
|
||||
|
||||
# Initialize Nornir with mixed-vendor inventory
|
||||
inventory = {
|
||||
"options": {
|
||||
"hosts": {
|
||||
"huawei-sw1": {
|
||||
"hostname": "127.0.0.1",
|
||||
"port": 5001,
|
||||
"platform": "huawei",
|
||||
"device_type": "huawei_telnet_ce",
|
||||
"groups": ["huawei_telnet"]
|
||||
},
|
||||
"cisco-r1": {
|
||||
"hostname": "127.0.0.1",
|
||||
"port": 5002,
|
||||
"platform": "cisco_ios",
|
||||
"device_type": "cisco_ios_telnet",
|
||||
"groups": ["cisco_ios_telnet"]
|
||||
}
|
||||
},
|
||||
"groups": {
|
||||
"huawei_telnet": {
|
||||
"platform": "huawei",
|
||||
"device_type": "huawei_telnet_ce"
|
||||
},
|
||||
"cisco_ios_telnet": {
|
||||
"platform": "cisco_ios",
|
||||
"device_type": "cisco_ios_telnet"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
nr = InitNornir(inventory=inventory)
|
||||
|
||||
# Execute commands on all devices (multi-vendor)
|
||||
result = nr.run(task=send_commands, commands=["display version"])
|
||||
|
||||
# Each device gets vendor-specific command handling
|
||||
```
|
||||
|
||||
### GNS3 Copilot Tool Usage
|
||||
|
||||
```python
|
||||
from gns3server.agent.gns3_copilot.tools_v2 import DisplayToolNornir
|
||||
|
||||
tool = DisplayToolNornir()
|
||||
result = tool._run(json.dumps({
|
||||
"device_names": ["huawei-sw1", "cisco-r1"],
|
||||
"commands": ["display version", "show version"],
|
||||
"project_id": "project-uuid"
|
||||
}))
|
||||
|
||||
# Returns:
|
||||
# {
|
||||
# "huawei-sw1": {
|
||||
# "display version": "<Huawei output>",
|
||||
# "status": "success"
|
||||
# },
|
||||
# "cisco-r1": {
|
||||
# "show version": "<Cisco output>",
|
||||
# "status": "success"
|
||||
# }
|
||||
# }
|
||||
```
|
||||
|
||||
## Module Structure
|
||||
|
||||
```
|
||||
gns3server/agent/gns3_copilot/
|
||||
├── utils/
|
||||
│ ├── custom_netmiko/ # Custom Netmiko drivers package
|
||||
│ │ ├── __init__.py # Package initialization
|
||||
│ │ ├── huawei_ce.py # Huawei CloudEngine driver
|
||||
│ │ ├── 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
|
||||
├── tools_v2/
|
||||
│ ├── display_tools_nornir.py # Multi-vendor display commands
|
||||
│ └── config_tools_nornir.py # Multi-vendor config commands
|
||||
```
|
||||
|
||||
## Unit Testing
|
||||
|
||||
### Test Coverage
|
||||
|
||||
```python
|
||||
# test_netmiko_custom.py
|
||||
|
||||
class TestHuaweiTelnetCEDriver(unittest.TestCase):
|
||||
def test_device_type_registered(self):
|
||||
"""Verify huawei_telnet_ce is in Netmiko CLASS_MAPPER"""
|
||||
from netmiko.ssh_dispatcher import CLASS_MAPPER
|
||||
self.assertIn("huawei_telnet_ce", CLASS_MAPPER)
|
||||
|
||||
def test_inheritance_from_huawei_base(self):
|
||||
"""Verify inherits from HuaweiBase"""
|
||||
from netmiko.huawei.huawei import HuaweiBase
|
||||
self.assertTrue(issubclass(HuaweiTelnetCE, HuaweiBase))
|
||||
|
||||
def test_vrp_methods_available(self):
|
||||
"""Verify VRP-specific methods are available"""
|
||||
methods = ["config_mode", "check_config_mode", "exit_config_mode"]
|
||||
for method in methods:
|
||||
self.assertTrue(hasattr(HuaweiTelnetCE, method))
|
||||
```
|
||||
|
||||
**Running Tests:**
|
||||
```bash
|
||||
source venv/bin/activate
|
||||
python gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py
|
||||
```
|
||||
|
||||
**Current Test Status:** ✅ All 9 tests passing
|
||||
|
||||
## Platform vs Device Type
|
||||
|
||||
### Key Concepts
|
||||
|
||||
**Platform (Nornir):**
|
||||
- High-level vendor identifier
|
||||
- Used for inventory grouping
|
||||
- Examples: `huawei`, `cisco_ios`
|
||||
|
||||
**Device Type (Netmiko):**
|
||||
- Precise driver type
|
||||
- Includes protocol information
|
||||
- Examples: `huawei_telnet_ce`, `cisco_ios_telnet`
|
||||
|
||||
### Mapping
|
||||
|
||||
| Platform | Device Type | Notes |
|
||||
|----------|-------------|-------|
|
||||
| `huawei` | `huawei_telnet_ce` | Custom driver for GNS3 |
|
||||
| `cisco_ios` | `cisco_ios_telnet` | Standard Netmiko driver |
|
||||
|
||||
## Related Documentation
|
||||
|
||||
- [Custom Netmiko README](../../../../../gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md) - Driver development guide
|
||||
- [Node Control Tools](./node-control-tools.md) - Device lifecycle management
|
||||
- [Command Security](./command-security.md) - Command filtering and validation
|
||||
- [Chat API](./chat-api.md) - Session management and SSE
|
||||
|
||||
## References
|
||||
|
||||
- [Netmiko Documentation](https://ktbyers.github.io/netmiko/)
|
||||
- [Netmiko PLATFORMS.md](https://github.com/ktbyers/netmiko/blob/master/PLATFORMS.md)
|
||||
- [Nornir Documentation](https://nornir.readthedocs.io/)
|
||||
|
||||
---
|
||||
|
||||
_Implementation Date: 2026-03-12_
|
||||
|
||||
_Status: ✅ Implemented - Custom Huawei driver for GNS3 emulation, multi-vendor support with Cisco IOS and Huawei tested_
|
||||
|
||||
_Unit Tests: ✅ 9/9 passing_
|
||||
@ -49,6 +49,22 @@ from gns3server.agent.gns3_copilot.utils.command_filter import (
|
||||
filter_forbidden_commands,
|
||||
)
|
||||
|
||||
# Import custom Netmiko device types for GNS3 emulation
|
||||
# This registers huawei_telnet_ce 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 custom device types to ensure they are available
|
||||
# This is a safety measure in case the auto-registration on import doesn't work
|
||||
try:
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce
|
||||
|
||||
# Re-register to ensure device types are available
|
||||
huawei_ce.register_custom_device_type()
|
||||
except Exception:
|
||||
# Fail silently - the import-time registration should have worked
|
||||
pass
|
||||
|
||||
# config log
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -60,9 +76,9 @@ logging.getLogger("nornir.core").setLevel(logging.WARNING)
|
||||
logging.getLogger("nornir").setLevel(logging.WARNING)
|
||||
|
||||
|
||||
# Local Nornir configuration functions for Cisco IOS Telnet devices
|
||||
# Local Nornir configuration functions for network devices
|
||||
def _get_nornir_defaults() -> dict[str, Any]:
|
||||
"""Get Nornir default configuration for Cisco IOS."""
|
||||
"""Get Nornir default configuration."""
|
||||
return {"data": {"location": "gns3"}}
|
||||
|
||||
|
||||
@ -70,11 +86,11 @@ def _get_nornir_groups_config(
|
||||
device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios"
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Get Nornir group configuration for Cisco IOS Telnet devices.
|
||||
Get Nornir group configuration for network devices.
|
||||
|
||||
Args:
|
||||
device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet')
|
||||
platform: Platform type for Nornir (e.g., 'cisco_ios')
|
||||
device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet', 'huawei_telnet')
|
||||
platform: Platform type for Nornir (e.g., 'cisco_ios', 'huawei')
|
||||
|
||||
Returns:
|
||||
Dictionary containing Nornir group configuration
|
||||
@ -92,7 +108,7 @@ def _get_nornir_groups_config(
|
||||
|
||||
|
||||
def _get_nornir_group(
|
||||
group_name: str = "cisco_IOSv_telnet",
|
||||
group_name: str = "network_devices_telnet",
|
||||
device_type: str = "cisco_ios_telnet",
|
||||
platform: str = "cisco_ios",
|
||||
) -> dict[str, Any]:
|
||||
@ -101,8 +117,8 @@ def _get_nornir_group(
|
||||
|
||||
Args:
|
||||
group_name: Name of the group
|
||||
device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet')
|
||||
platform: Platform type for Nornir (e.g., 'cisco_ios')
|
||||
device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet', 'huawei_telnet')
|
||||
platform: Platform type for Nornir (e.g., 'cisco_ios', 'huawei')
|
||||
|
||||
Returns:
|
||||
Dictionary containing group configuration
|
||||
@ -529,13 +545,23 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
|
||||
)
|
||||
defaults = _get_nornir_defaults()
|
||||
|
||||
# Dynamically generate group name based on platform and device type
|
||||
# e.g., "huawei_telnet", "cisco_ios_telnet", "juniper_junos"
|
||||
actual_platform = platform or "cisco_ios"
|
||||
if device_type and "_telnet" in device_type:
|
||||
group_name = f"{actual_platform}_telnet"
|
||||
else:
|
||||
group_name = actual_platform
|
||||
|
||||
# Log nornir account information
|
||||
gns3_host = get_gns3_server_host()
|
||||
|
||||
logger.info(
|
||||
"Initializing Nornir: host=%s, platform=%s, timeout=%d",
|
||||
"Initializing Nornir: host=%s, platform=%s, device_type=%s, group=%s, timeout=%d",
|
||||
gns3_host,
|
||||
groups_data.get("platform"),
|
||||
device_type,
|
||||
group_name,
|
||||
groups_data.get("timeout"),
|
||||
)
|
||||
|
||||
@ -544,7 +570,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
|
||||
"plugin": "DictInventory",
|
||||
"options": {
|
||||
"hosts": hosts_data,
|
||||
"groups": {"cisco_IOSv_telnet": groups_data},
|
||||
"groups": {group_name: groups_data},
|
||||
"defaults": defaults,
|
||||
},
|
||||
},
|
||||
|
||||
@ -49,6 +49,22 @@ from gns3server.agent.gns3_copilot.utils.command_filter import (
|
||||
filter_forbidden_commands,
|
||||
)
|
||||
|
||||
# Import custom Netmiko device types for GNS3 emulation
|
||||
# This registers huawei_telnet_ce 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 custom device types to ensure they are available
|
||||
# This is a safety measure in case the auto-registration on import doesn't work
|
||||
try:
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce
|
||||
|
||||
# Re-register to ensure device types are available
|
||||
huawei_ce.register_custom_device_type()
|
||||
except Exception:
|
||||
# Fail silently - the import-time registration should have worked
|
||||
pass
|
||||
|
||||
# config log
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -60,9 +76,9 @@ logging.getLogger("nornir.core").setLevel(logging.WARNING)
|
||||
logging.getLogger("nornir").setLevel(logging.WARNING)
|
||||
|
||||
|
||||
# Local Nornir configuration functions for Cisco IOS Telnet devices
|
||||
# Local Nornir configuration functions for network devices
|
||||
def _get_nornir_defaults() -> dict[str, Any]:
|
||||
"""Get Nornir default configuration for Cisco IOS."""
|
||||
"""Get Nornir default configuration."""
|
||||
return {"data": {"location": "gns3"}}
|
||||
|
||||
|
||||
@ -70,11 +86,11 @@ def _get_nornir_groups_config(
|
||||
device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios"
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Get Nornir group configuration for Cisco IOS Telnet devices.
|
||||
Get Nornir group configuration for network devices.
|
||||
|
||||
Args:
|
||||
device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet')
|
||||
platform: Platform type for Nornir (e.g., 'cisco_ios')
|
||||
device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet', 'huawei_telnet')
|
||||
platform: Platform type for Nornir (e.g., 'cisco_ios', 'huawei')
|
||||
|
||||
Returns:
|
||||
Dictionary containing Nornir group configuration
|
||||
@ -92,7 +108,7 @@ def _get_nornir_groups_config(
|
||||
|
||||
|
||||
def _get_nornir_group(
|
||||
group_name: str = "cisco_IOSv_telnet",
|
||||
group_name: str = "network_devices_telnet",
|
||||
device_type: str = "cisco_ios_telnet",
|
||||
platform: str = "cisco_ios",
|
||||
) -> dict[str, Any]:
|
||||
@ -101,8 +117,8 @@ def _get_nornir_group(
|
||||
|
||||
Args:
|
||||
group_name: Name of the group
|
||||
device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet')
|
||||
platform: Platform type for Nornir (e.g., 'cisco_ios')
|
||||
device_type: Device type for Netmiko (e.g., 'cisco_ios_telnet', 'huawei_telnet')
|
||||
platform: Platform type for Nornir (e.g., 'cisco_ios', 'huawei')
|
||||
|
||||
Returns:
|
||||
Dictionary containing group configuration
|
||||
@ -533,13 +549,23 @@ class ExecuteMultipleDeviceCommands(BaseTool):
|
||||
)
|
||||
defaults = _get_nornir_defaults()
|
||||
|
||||
# Dynamically generate group name based on platform and device type
|
||||
# e.g., "huawei_telnet", "cisco_ios_telnet", "juniper_junos"
|
||||
actual_platform = platform or "cisco_ios"
|
||||
if device_type and "_telnet" in device_type:
|
||||
group_name = f"{actual_platform}_telnet"
|
||||
else:
|
||||
group_name = actual_platform
|
||||
|
||||
# Log nornir account information
|
||||
gns3_host = get_gns3_server_host()
|
||||
|
||||
logger.info(
|
||||
"Initializing Nornir: host=%s, platform=%s, timeout=%d",
|
||||
"Initializing Nornir: host=%s, platform=%s, device_type=%s, group=%s, timeout=%d",
|
||||
gns3_host,
|
||||
groups_data.get("platform"),
|
||||
device_type,
|
||||
group_name,
|
||||
groups_data.get("timeout"),
|
||||
)
|
||||
|
||||
@ -548,7 +574,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
|
||||
"plugin": "DictInventory",
|
||||
"options": {
|
||||
"hosts": hosts_data,
|
||||
"groups": {"cisco_IOSv_telnet": groups_data},
|
||||
"groups": {group_name: groups_data},
|
||||
"defaults": defaults,
|
||||
},
|
||||
},
|
||||
|
||||
183
gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md
Normal file
183
gns3server/agent/gns3_copilot/utils/custom_netmiko/README.md
Normal file
@ -0,0 +1,183 @@
|
||||
# Custom Netmiko Drivers
|
||||
|
||||
Custom Netmiko drivers for GNS3-emulated network devices.
|
||||
|
||||
## Overview
|
||||
|
||||
This package contains Netmiko drivers optimized for GNS3 emulation environments where devices may have non-standard authentication or behavior patterns.
|
||||
|
||||
## Directory Structure
|
||||
|
||||
```
|
||||
custom_netmiko/
|
||||
├── __init__.py # Package initialization, auto-registers all drivers
|
||||
├── huawei_ce.py # Huawei CloudEngine custom driver
|
||||
├── README.md # This file
|
||||
└── tests/ # Unit tests
|
||||
├── __init__.py
|
||||
└── test_huawei_ce.py # Huawei CE driver tests
|
||||
```
|
||||
|
||||
## Supported Drivers
|
||||
|
||||
### Huawei (`huawei_ce.py`)
|
||||
|
||||
**Driver Name:** `HuaweiTelnetCE`
|
||||
|
||||
**Device Types:**
|
||||
- `huawei_telnet_ce` - Primary type
|
||||
- `huawei_ce` - Alias
|
||||
|
||||
**Features:**
|
||||
- Skip authentication (for GNS3 devices without username/password)
|
||||
- VRP prompt recognition (`<HUAWEI>`, `[HUAWEI]`)
|
||||
- Auto-commit before exit (prevents [Y/N/C] prompts)
|
||||
- Proper output collection using `read_channel_timing()`
|
||||
|
||||
**Limitations:**
|
||||
- Does not support username/password authentication
|
||||
- If your device requires authentication, use standard `huawei_telnet` driver
|
||||
|
||||
## Usage
|
||||
|
||||
### Direct Usage
|
||||
|
||||
```python
|
||||
from netmiko import ConnectHandler
|
||||
from gns3server.agent.gns3_copilot.utils import custom_netmiko
|
||||
|
||||
device = {
|
||||
"device_type": "huawei_telnet_ce",
|
||||
"host": "127.0.0.1",
|
||||
"port": 5000,
|
||||
}
|
||||
|
||||
with ConnectHandler(**device) as conn:
|
||||
output = conn.send_command("display version")
|
||||
config = ["interface GE1/0/1", "description Test"]
|
||||
output = conn.send_config_set(config)
|
||||
```
|
||||
|
||||
### In GNS3 Copilot Tools
|
||||
|
||||
The drivers are auto-registered when the tools are imported:
|
||||
|
||||
```python
|
||||
from gns3server.agent.gns3_copilot.tools_v2 import DisplayToolNornir
|
||||
|
||||
tool = DisplayToolNornir()
|
||||
result = tool._run(json.dumps({
|
||||
"device_names": ["huawei-sw1"],
|
||||
"commands": ["display version"],
|
||||
"project_id": "project-uuid"
|
||||
}))
|
||||
```
|
||||
|
||||
## Running Tests
|
||||
|
||||
```bash
|
||||
# Run all tests
|
||||
source venv/bin/activate
|
||||
python -m unittest discover -s gns3server/agent/gns3_copilot/utils/custom_netmiko/tests
|
||||
|
||||
# Run Huawei CE tests only
|
||||
python gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py
|
||||
```
|
||||
|
||||
## Adding New Drivers
|
||||
|
||||
### 1. Create Driver File
|
||||
|
||||
Create a new file in `custom_netmiko/` (e.g., `cisco.py`):
|
||||
|
||||
```python
|
||||
# cisco.py
|
||||
from netmiko.cisco.cisco_ios import CiscoIosTelnet
|
||||
|
||||
class CustomCiscoDriver(CiscoIosTelnet):
|
||||
"""Custom Cisco driver for GNS3."""
|
||||
|
||||
def telnet_login(self, ...):
|
||||
# Override login logic
|
||||
pass
|
||||
|
||||
def register_custom_device_type() -> None:
|
||||
"""Register this driver with Netmiko."""
|
||||
import importlib
|
||||
sd = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
|
||||
sd.CLASS_MAPPER["cisco_custom"] = CustomCiscoDriver
|
||||
sd.CLASS_MAPPER_BASE["cisco_custom"] = CustomCiscoDriver
|
||||
|
||||
# Rebuild static lists
|
||||
sd.platforms = list(sd.CLASS_MAPPER.keys())
|
||||
sd.platforms.sort()
|
||||
|
||||
# Auto-register on import
|
||||
try:
|
||||
register_custom_device_type()
|
||||
except Exception as e:
|
||||
import logging
|
||||
logging.getLogger(__name__).warning(f"Failed to register: {e}")
|
||||
```
|
||||
|
||||
### 2. Update `__init__.py`
|
||||
|
||||
Add import to `__init__.py`:
|
||||
|
||||
```python
|
||||
try:
|
||||
from . import cisco # noqa: F401
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to import Cisco driver: {e}", exc_info=True)
|
||||
|
||||
__all__ = ["huawei_ce", "cisco"]
|
||||
```
|
||||
|
||||
### 3. Create Tests
|
||||
|
||||
Create `tests/test_cisco.py`:
|
||||
|
||||
```python
|
||||
import unittest
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko import cisco
|
||||
|
||||
class TestCustomCiscoDriver(unittest.TestCase):
|
||||
def test_device_type_registered(self):
|
||||
from netmiko.ssh_dispatcher import CLASS_MAPPER
|
||||
self.assertIn("cisco_custom", CLASS_MAPPER)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Driver Not Found Error
|
||||
|
||||
```
|
||||
ValueError: Unsupported 'device_type'
|
||||
```
|
||||
|
||||
**Solution:** Ensure the custom_netmiko package is imported before using the device:
|
||||
|
||||
```python
|
||||
from gns3server.agent.gns3_copilot.utils import custom_netmiko
|
||||
# Now use the device
|
||||
```
|
||||
|
||||
### Import Errors
|
||||
|
||||
If you get import errors, check:
|
||||
1. Virtual environment is activated
|
||||
2. Python path includes project root
|
||||
3. Dependencies are installed (`pip install netmiko nornir nornir-netmiko`)
|
||||
|
||||
## Related Documentation
|
||||
|
||||
- [Multi-Vendor Device Support](../../../docs/gns3-copilot/implemented/multi-vendor-device-support.md)
|
||||
- [Netmiko Documentation](https://ktbyers.github.io/netmiko/)
|
||||
|
||||
---
|
||||
|
||||
_Last updated: 2026-03-12_
|
||||
@ -0,0 +1,62 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
#
|
||||
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
|
||||
#
|
||||
# This file is part of GNS3-Copilot project.
|
||||
#
|
||||
# GNS3-Copilot is free software: you can redistribute it and/or modify it
|
||||
# under the terms of the GNU General Public License as published by the
|
||||
# Free Software Foundation, either version 3 of the License, or (at your
|
||||
# option) any later version.
|
||||
#
|
||||
# GNS3-Copilot is distributed in the hope that it will be useful, but
|
||||
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
|
||||
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
|
||||
# for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
|
||||
#
|
||||
# Copyright (C) 2025 Yue Guobin (岳国宾)
|
||||
# Author: Yue Guobin (岳国宾)
|
||||
#
|
||||
# Project Home: https://github.com/yueguobin/gns3-copilot
|
||||
#
|
||||
|
||||
"""
|
||||
Custom Netmiko drivers for GNS3-emulated network devices.
|
||||
|
||||
This package contains custom Netmiko drivers optimized for GNS3
|
||||
emulation environments where devices may have non-standard
|
||||
authentication or behavior patterns.
|
||||
|
||||
Supported Drivers:
|
||||
- huawei_ce: HuaweiTelnetCE for CloudEngine devices (no authentication)
|
||||
|
||||
Usage:
|
||||
from gns3server.agent.gns3_copilot.utils import custom_netmiko
|
||||
|
||||
# Auto-registers all drivers on import
|
||||
from netmiko import ConnectHandler
|
||||
|
||||
device = {
|
||||
"device_type": "huawei_telnet_ce",
|
||||
"host": "127.0.0.1",
|
||||
"port": 5000,
|
||||
}
|
||||
|
||||
with ConnectHandler(**device) as conn:
|
||||
output = conn.send_command("display version")
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Import all custom drivers (auto-registers them with Netmiko)
|
||||
try:
|
||||
from . import huawei_ce # noqa: F401
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to import Huawei CE driver: {e}", exc_info=True)
|
||||
|
||||
__all__ = ["huawei_ce"]
|
||||
402
gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py
Normal file
402
gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py
Normal file
@ -0,0 +1,402 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
#
|
||||
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
|
||||
#
|
||||
# This file is part of GNS3-Copilot project.
|
||||
#
|
||||
# GNS3-Copilot is free software: you can redistribute it and/or modify it
|
||||
# under the terms of the GNU General Public License as published by the
|
||||
# Free Software Foundation, either version 3 of the License, or (at your
|
||||
# option) any later version.
|
||||
#
|
||||
# GNS3-Copilot is distributed in the hope that it will be useful, but
|
||||
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
|
||||
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
|
||||
# for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
|
||||
#
|
||||
# Copyright (C) 2025 Yue Guobin (岳国宾)
|
||||
# Author: Yue Guobin (岳国宾)
|
||||
#
|
||||
# Project Home: https://github.com/yueguobin/gns3-copilot
|
||||
#
|
||||
|
||||
"""
|
||||
Custom Netmiko device driver for Huawei devices in GNS3 emulation environment.
|
||||
|
||||
This module provides a custom device type 'huawei_telnet_ce' for Huawei network
|
||||
devices that connect via console without requiring authentication (username/password).
|
||||
|
||||
This is specifically designed for GNS3 emulated Huawei devices (e.g., CloudEngine
|
||||
series) where the console connection directly enters the system view without
|
||||
login prompts.
|
||||
|
||||
Key Features:
|
||||
- Inherits from HuaweiBase for proper VRP command handling
|
||||
- Skips authentication (no username/password required)
|
||||
- Supports Huawei-specific config mode (system-view)
|
||||
- Handles Huawei prompt patterns (<>, [], >)
|
||||
"""
|
||||
|
||||
import re
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
from netmiko.huawei.huawei import HuaweiBase
|
||||
|
||||
|
||||
class HuaweiTelnetCE(HuaweiBase):
|
||||
"""
|
||||
Custom Huawei device driver for GNS3 emulation.
|
||||
|
||||
Inherits from HuaweiBase to leverage existing VRP-specific functionality:
|
||||
- system-view configuration mode
|
||||
- Huawei prompt patterns
|
||||
- Config mode detection and exit
|
||||
|
||||
This driver overrides telnet_login to handle GNS3 devices that
|
||||
don't require authentication.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*args,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
"""Initialize HuaweiTelnetCE connection."""
|
||||
# Set default device type for proper initialization
|
||||
# The '_telnet' suffix in device_type tells Netmiko to use Telnet protocol
|
||||
kwargs.setdefault("device_type", "huawei_telnet")
|
||||
|
||||
# Huawei prompt patterns (inherited from HuaweiBase)
|
||||
# User view: <HUAWEI>
|
||||
# System view: [HUAWEI]
|
||||
# Interface view: [HUAWEI-GigabitEthernet0/0/1]
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def telnet_login(
|
||||
self,
|
||||
pri_prompt_terminator: str = r"<\S+>|>\s*$",
|
||||
alt_prompt_terminator: str = r"\[\S+\]",
|
||||
username_pattern: str = r"(?:user:|username|login|user name)",
|
||||
pwd_pattern: str = r"assword",
|
||||
delay_factor: float = 1.0,
|
||||
max_loops: int = 10,
|
||||
) -> str:
|
||||
"""
|
||||
Telnet login for GNS3 Huawei devices (no authentication).
|
||||
|
||||
Simplified login logic for devices that connect directly to
|
||||
command line without username/password prompts.
|
||||
|
||||
Strategy:
|
||||
1. Clear any existing buffer data
|
||||
2. Send carriage returns to trigger prompt
|
||||
3. Wait for Huawei prompt pattern
|
||||
4. Return once prompt is detected
|
||||
|
||||
Args:
|
||||
pri_prompt_terminator: Primary prompt pattern (e.g., <HUAWEI>)
|
||||
alt_prompt_terminator: Alternate prompt pattern (e.g., [HUAWEI])
|
||||
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)
|
||||
output = ""
|
||||
return_msg = ""
|
||||
|
||||
# Step 1: Clear buffer - read any existing data to avoid interference
|
||||
try:
|
||||
initial_data = self.read_channel()
|
||||
if initial_data:
|
||||
return_msg += initial_data
|
||||
except Exception:
|
||||
# Ignore errors during initial read
|
||||
pass
|
||||
|
||||
# Step 2: Send carriage returns and wait for prompt
|
||||
for i in range(max_loops):
|
||||
try:
|
||||
# Send return to trigger prompt
|
||||
self.write_channel(self.RETURN)
|
||||
time.sleep(0.5 * delay_factor)
|
||||
|
||||
# Read response
|
||||
new_output = self.read_channel()
|
||||
output = new_output
|
||||
return_msg += new_output
|
||||
|
||||
# Check for Huawei prompt patterns
|
||||
# User view: <HUAWEI>, System view: [HUAWEI], or just >
|
||||
if re.search(pri_prompt_terminator, output, flags=re.M):
|
||||
return return_msg
|
||||
if re.search(alt_prompt_terminator, output, flags=re.M):
|
||||
return return_msg
|
||||
|
||||
except EOFError:
|
||||
self.remote_conn.close()
|
||||
msg = f"Connection failed (EOF): {self.host}"
|
||||
raise self.connection_error(msg) from None
|
||||
except Exception:
|
||||
# Continue trying on other exceptions
|
||||
pass
|
||||
|
||||
# Step 3: Final attempt - return what we have
|
||||
# Even if we couldn't detect the prompt clearly, the connection
|
||||
# might still be usable
|
||||
return return_msg
|
||||
|
||||
def session_preparation(self) -> None:
|
||||
"""
|
||||
Prepare the session after connection is established.
|
||||
|
||||
Inherited from HuaweiBase, but ensures proper initialization
|
||||
for GNS3 emulation environment.
|
||||
"""
|
||||
# Wait for prompt to stabilize
|
||||
time.sleep(0.5 * self.global_delay_factor)
|
||||
|
||||
# Disable paging using Huawei-specific command
|
||||
try:
|
||||
self.disable_paging(command="screen-length 0 temporary")
|
||||
except Exception:
|
||||
# If disable_paging fails, try the parent implementation
|
||||
super().disable_paging()
|
||||
|
||||
# Ensure we're in a clean state
|
||||
try:
|
||||
if hasattr(self, 'base_prompt') and self.base_prompt:
|
||||
self._test_channel_read(pattern=self.base_prompt)
|
||||
else:
|
||||
# If base_prompt is not set yet, just read to clear buffer
|
||||
self.read_channel()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def disable_paging(
|
||||
self,
|
||||
command: str = "screen-length 0 temporary",
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""
|
||||
Disable paging for Huawei devices.
|
||||
|
||||
Uses Huawei-specific command 'screen-length 0 temporary'
|
||||
which disables paging for the current session only.
|
||||
|
||||
Args:
|
||||
command: Command to disable paging
|
||||
|
||||
Returns:
|
||||
Output from the disable paging command
|
||||
"""
|
||||
return super().disable_paging(command=command, **kwargs)
|
||||
|
||||
def send_config_set(
|
||||
self,
|
||||
config_commands: str | list[str],
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""
|
||||
Send configuration commands to Huawei device.
|
||||
|
||||
Overrides the parent method to handle Huawei-specific behavior:
|
||||
- Uses Huawei-specific prompts (<...> for user view, [...] for system view)
|
||||
- Handles the 'return' confirmation prompt
|
||||
- Uses read_channel_timing for proper output collection
|
||||
|
||||
Args:
|
||||
config_commands: Configuration commands to send
|
||||
**kwargs: Additional arguments (exit_config_mode, read_timeout, etc.)
|
||||
|
||||
Returns:
|
||||
Output from configuration commands
|
||||
"""
|
||||
# Convert string to list
|
||||
if isinstance(config_commands, str):
|
||||
config_commands = [config_commands]
|
||||
|
||||
# Get parameters with Huawei-specific defaults
|
||||
exit_config_mode = kwargs.get("exit_config_mode", True)
|
||||
read_timeout = kwargs.get("read_timeout", 30) # Longer timeout for GNS3 emulation
|
||||
delay_factor = self.global_delay_factor
|
||||
strip_prompt = kwargs.get("strip_prompt", False)
|
||||
strip_command = kwargs.get("strip_command", False)
|
||||
config_mode_command = kwargs.get("config_command", "system-view")
|
||||
cmd_verify = kwargs.get("cmd_verify", False) # Disable cmd_verify for Huawei
|
||||
|
||||
output = ""
|
||||
|
||||
# Enter config mode if needed
|
||||
if kwargs.get("enter_config_mode", True):
|
||||
output += self.config_mode(config_command=config_mode_command)
|
||||
|
||||
# Send all configuration commands
|
||||
# Use cmd_verify=False mode: send all commands, then read all output at once
|
||||
for cmd in config_commands:
|
||||
self.write_channel(f"{cmd}{self.RETURN}")
|
||||
# Small delay between commands
|
||||
time.sleep(delay_factor * 0.05)
|
||||
|
||||
# Use read_channel_timing to collect all output
|
||||
# This method keeps reading until there is no new data for 'last_read' seconds
|
||||
# This is the proper Netmiko way to handle command output
|
||||
output += self.read_channel_timing(read_timeout=read_timeout, last_read=2.0)
|
||||
|
||||
# Exit config mode if requested
|
||||
if exit_config_mode:
|
||||
# For Huawei devices, commit configuration before exiting
|
||||
# This avoids the "Uncommitted configurations" [Y/N/C] prompt
|
||||
try:
|
||||
# Send commit command (in system view [HUAWEI])
|
||||
self.write_channel(f"commit{self.RETURN}")
|
||||
time.sleep(0.5 * self.global_delay_factor)
|
||||
commit_output = self.read_channel()
|
||||
output += commit_output
|
||||
except Exception:
|
||||
# If commit fails, continue with exit (might not support commit)
|
||||
pass
|
||||
|
||||
# Now exit config mode
|
||||
output += self.exit_config_mode()
|
||||
|
||||
if strip_prompt:
|
||||
output = self.strip_prompt(output)
|
||||
|
||||
if strip_command:
|
||||
output = self.strip_command(config_commands, output)
|
||||
|
||||
return output
|
||||
|
||||
def exit_config_mode(
|
||||
self, exit_config: str = "return", pattern: str = r"<\S+>|>\s*$"
|
||||
) -> str:
|
||||
r"""
|
||||
Exit configuration mode for Huawei devices.
|
||||
|
||||
Huawei devices display a confirmation prompt when using 'return':
|
||||
Return to user view? [y/n]:
|
||||
|
||||
This method automatically answers 'y' to the prompt.
|
||||
|
||||
Args:
|
||||
exit_config: Command to exit config mode (default: "return")
|
||||
pattern: Pattern to detect user view prompt (default: r"<\S+>|>\s*$")
|
||||
|
||||
Returns:
|
||||
Output from exiting config mode
|
||||
"""
|
||||
# Check if we're currently in config mode
|
||||
if not self.check_config_mode():
|
||||
return ""
|
||||
|
||||
output = ""
|
||||
# Send the exit command (write_channel returns None, don't concatenate)
|
||||
self.write_channel(f"{exit_config}{self.RETURN}")
|
||||
time.sleep(0.5 * self.global_delay_factor)
|
||||
|
||||
# Look for the confirmation prompt
|
||||
# Huawei prompt: "Return to user view? [y/n]:"
|
||||
prompt_pattern = r"\[y/n\]"
|
||||
max_loops = 20 # More loops for slower devices
|
||||
|
||||
for _ in range(max_loops):
|
||||
new_output = self.read_channel()
|
||||
output += new_output
|
||||
|
||||
# If we see the confirmation prompt, send 'y'
|
||||
if re.search(prompt_pattern, new_output):
|
||||
self.write_channel(f"y{self.RETURN}") # Returns None
|
||||
time.sleep(0.5 * self.global_delay_factor)
|
||||
# Clear the confirmation response
|
||||
new_output = self.read_channel()
|
||||
output += new_output
|
||||
|
||||
# Check if we've exited to user view
|
||||
if re.search(pattern, new_output):
|
||||
return output
|
||||
|
||||
# Final read to get remaining output
|
||||
new_output = self.read_channel()
|
||||
output += new_output
|
||||
|
||||
return output
|
||||
|
||||
|
||||
# Register the custom device type with Netmiko
|
||||
def register_custom_device_type() -> None:
|
||||
"""
|
||||
Register the custom HuaweiTelnetCE device type with Netmiko.
|
||||
|
||||
This function adds 'huawei_telnet_ce' to both Netmiko's CLASS_MAPPER
|
||||
and CLASS_MAPPER_BASE so it can be used like any other built-in device type.
|
||||
|
||||
Additionally, it updates the static 'platforms' and 'telnet_platforms' lists
|
||||
which are used by ConnectHandler for device type validation.
|
||||
|
||||
IMPORTANT: This function should be called BEFORE initializing Nornir or
|
||||
running any Netmiko tasks. Call it explicitly at the appropriate time.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
# Use importlib to avoid namespace conflicts
|
||||
import importlib
|
||||
|
||||
# Import the module using importlib to ensure we get the module, not a function
|
||||
sd = importlib.import_module("netmiko.ssh_dispatcher")
|
||||
|
||||
# Add to both mappers
|
||||
# CLASS_MAPPER_BASE is used for base class definitions
|
||||
sd.CLASS_MAPPER_BASE["huawei_telnet_ce"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER_BASE["huawei_ce"] = HuaweiTelnetCE
|
||||
|
||||
# CLASS_MAPPER is used by ConnectHandler for device type validation
|
||||
sd.CLASS_MAPPER["huawei_telnet_ce"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER["huawei_ce"] = HuaweiTelnetCE
|
||||
|
||||
# CRITICAL: Netmiko may also look for types with _telnet suffix
|
||||
# Add this variant to ensure compatibility
|
||||
sd.CLASS_MAPPER["huawei_telnet_ce_telnet"] = HuaweiTelnetCE
|
||||
sd.CLASS_MAPPER_BASE["huawei_telnet_ce_telnet"] = HuaweiTelnetCE
|
||||
|
||||
# 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)
|
||||
|
||||
|
||||
# Auto-register on import
|
||||
# This ensures the device type is available when the module is imported
|
||||
# NOTE: For Nornir scenarios, you may need to call this explicitly
|
||||
# before InitNornir to ensure proper timing
|
||||
try:
|
||||
register_custom_device_type()
|
||||
except Exception as e:
|
||||
# Log but don't fail on import
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.warning(f"Failed to register custom device type: {e}", exc_info=True)
|
||||
@ -0,0 +1,28 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
#
|
||||
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
|
||||
#
|
||||
# This file is part of GNS3-Copilot project.
|
||||
#
|
||||
# GNS3-Copilot is free software: you can redistribute it and/or modify it
|
||||
# under the terms of the GNU General Public License as published by the
|
||||
# Free Software Foundation, either version 3 of the License, or (at your
|
||||
# option) any later version.
|
||||
#
|
||||
# GNS3-Copilot is distributed in the hope that it will be useful, but
|
||||
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
|
||||
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
|
||||
# for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
|
||||
#
|
||||
# Copyright (C) 2025 Yue Guobin (岳国宾)
|
||||
# Author: Yue Guobin (岳国宾)
|
||||
#
|
||||
# Project Home: https://github.com/yueguobin/gns3-copilot
|
||||
#
|
||||
|
||||
"""
|
||||
Unit tests for custom Netmiko drivers.
|
||||
"""
|
||||
@ -0,0 +1,215 @@
|
||||
#!/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 HuaweiTelnetCE driver
|
||||
#
|
||||
|
||||
"""
|
||||
Unit test script for HuaweiTelnetCE custom device driver.
|
||||
|
||||
This script tests:
|
||||
1. Device type registration
|
||||
2. Inheritance from HuaweiBase
|
||||
3. VRP-specific command handling
|
||||
4. Telnet login logic (mocked)
|
||||
|
||||
Run with: python test_huawei_ce.py
|
||||
"""
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
|
||||
# Add project root to path
|
||||
sys.path.insert(0, "/home/yueguobin/myCode/GNS3/gns3-server")
|
||||
|
||||
|
||||
class TestHuaweiTelnetCEDriver(unittest.TestCase):
|
||||
"""Test suite for HuaweiTelnetCE custom driver."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Set up test fixtures - import and register custom driver."""
|
||||
# Import the custom driver module (this triggers registration)
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko import huawei_ce
|
||||
|
||||
cls.huawei_ce = huawei_ce
|
||||
cls.HuaweiTelnetCE = huawei_ce.HuaweiTelnetCE
|
||||
|
||||
def test_device_type_registered(self):
|
||||
"""Test that huawei_telnet_ce is registered in Netmiko."""
|
||||
from netmiko.ssh_dispatcher import CLASS_MAPPER, CLASS_MAPPER_BASE
|
||||
|
||||
# Check CLASS_MAPPER
|
||||
self.assertIn("huawei_telnet_ce", CLASS_MAPPER)
|
||||
self.assertEqual(CLASS_MAPPER["huawei_telnet_ce"], self.HuaweiTelnetCE)
|
||||
|
||||
# Check CLASS_MAPPER_BASE
|
||||
self.assertIn("huawei_telnet_ce", CLASS_MAPPER_BASE)
|
||||
self.assertEqual(CLASS_MAPPER_BASE["huawei_telnet_ce"], self.HuaweiTelnetCE)
|
||||
|
||||
# Check alias
|
||||
self.assertIn("huawei_ce", CLASS_MAPPER)
|
||||
|
||||
def test_inheritance_from_huawei_base(self):
|
||||
"""Test that HuaweiTelnetCE inherits from HuaweiBase."""
|
||||
from netmiko.huawei.huawei import HuaweiBase
|
||||
|
||||
# Verify inheritance
|
||||
self.assertIsInstance(self.HuaweiTelnetCE, type)
|
||||
# Check if HuaweiTelnetCE is a subclass of HuaweiBase
|
||||
self.assertTrue(issubclass(self.HuaweiTelnetCE, HuaweiBase))
|
||||
|
||||
def test_huawei_base_methods_available(self):
|
||||
"""Test that VRP-specific methods are available."""
|
||||
# These methods should be inherited from HuaweiBase
|
||||
vrp_methods = [
|
||||
"config_mode",
|
||||
"check_config_mode",
|
||||
"exit_config_mode",
|
||||
"send_config_set",
|
||||
"send_command",
|
||||
"disable_paging",
|
||||
]
|
||||
|
||||
for method_name in vrp_methods:
|
||||
self.assertTrue(
|
||||
hasattr(self.HuaweiTelnetCE, method_name),
|
||||
f"Method {method_name} not found in HuaweiTelnetCE",
|
||||
)
|
||||
|
||||
def test_telnet_login_method_exists(self):
|
||||
"""Test that telnet_login method is overridden."""
|
||||
# Should have overridden telnet_login
|
||||
self.assertTrue(hasattr(self.HuaweiTelnetCE, "telnet_login"))
|
||||
|
||||
# Get the method and check if it's defined in HuaweiTelnetCE
|
||||
import inspect
|
||||
|
||||
telnet_login_method = getattr(self.HuaweiTelnetCE, "telnet_login")
|
||||
|
||||
# Check if method is in HuaweiTelnetCE's __dict__ (means it's defined there, not inherited)
|
||||
self.assertIn(
|
||||
"telnet_login",
|
||||
self.HuaweiTelnetCE.__dict__,
|
||||
"telnet_login should be defined in HuaweiTelnetCE",
|
||||
)
|
||||
|
||||
def test_initialization_parameters(self):
|
||||
"""Test that initialization sets correct parameters."""
|
||||
# Create a mock instance (without actual connection)
|
||||
with patch.object(self.HuaweiTelnetCE, "__init__", lambda self, *args, **kwargs: None):
|
||||
instance = self.HuaweiTelnetCE.__new__(self.HuaweiTelnetCE)
|
||||
|
||||
# Mock the necessary attributes
|
||||
instance.protocol = "telnet"
|
||||
instance.device_type = "huawei_telnet"
|
||||
|
||||
# Verify protocol is set to telnet
|
||||
self.assertEqual(instance.protocol, "telnet")
|
||||
self.assertEqual(instance.device_type, "huawei_telnet")
|
||||
|
||||
def test_connect_handler_accepts_device_type(self):
|
||||
"""Test that ConnectHandler accepts huawei_telnet_ce device type."""
|
||||
from netmiko.ssh_dispatcher import CLASS_MAPPER
|
||||
|
||||
# Get platforms list
|
||||
platforms = list(CLASS_MAPPER.keys())
|
||||
|
||||
# Verify huawei_telnet_ce is in platforms
|
||||
self.assertIn("huawei_telnet_ce", platforms)
|
||||
|
||||
# Verify it's in telnet platforms
|
||||
telnet_platforms = [x for x in platforms if "telnet" in x]
|
||||
self.assertIn("huawei_telnet_ce", telnet_platforms)
|
||||
|
||||
def test_prompt_pattern_constants(self):
|
||||
"""Test that Huawei prompt patterns are correctly defined."""
|
||||
import inspect
|
||||
|
||||
# Get the telnet_login method signature
|
||||
sig = inspect.signature(self.HuaweiTelnetCE.telnet_login)
|
||||
|
||||
# Check default prompt patterns
|
||||
pri_prompt = sig.parameters["pri_prompt_terminator"].default
|
||||
alt_prompt = sig.parameters["alt_prompt_terminator"].default
|
||||
|
||||
# Should match Huawei prompt patterns
|
||||
self.assertIn("<", pri_prompt)
|
||||
self.assertIn("[", alt_prompt)
|
||||
|
||||
def test_disable_paging_command(self):
|
||||
"""Test that disable_paging uses correct Huawei command."""
|
||||
import inspect
|
||||
|
||||
sig = inspect.signature(self.HuaweiTelnetCE.disable_paging)
|
||||
command_default = sig.parameters["command"].default
|
||||
|
||||
# Should use Huawei-specific command
|
||||
self.assertEqual(command_default, "screen-length 0 temporary")
|
||||
|
||||
|
||||
class TestHuaweiTelnetCEIntegration(unittest.TestCase):
|
||||
"""Integration tests for HuaweiTelnetCE driver."""
|
||||
|
||||
def test_mock_telnet_connection(self):
|
||||
"""Test telnet_login logic with mocked connection."""
|
||||
from gns3server.agent.gns3_copilot.utils.custom_netmiko.huawei_ce import HuaweiTelnetCE
|
||||
|
||||
# Create a mock instance
|
||||
instance = HuaweiTelnetCE.__new__(HuaweiTelnetCE)
|
||||
|
||||
# Mock the necessary attributes and methods
|
||||
instance.host = "127.0.0.1"
|
||||
instance.RETURN = "\r\n"
|
||||
instance.global_delay_factor = 1.0
|
||||
instance.remote_conn = Mock()
|
||||
|
||||
# Mock select_delay_factor
|
||||
instance.select_delay_factor = Mock(return_value=1.0)
|
||||
|
||||
# Mock read_channel and write_channel
|
||||
test_outputs = ["", "<HUAWEI>"]
|
||||
instance.read_channel = Mock(side_effect=test_outputs)
|
||||
instance.write_channel = Mock()
|
||||
|
||||
# Call telnet_login
|
||||
result = instance.telnet_login()
|
||||
|
||||
# Verify behavior
|
||||
self.assertIn("<HUAWEI>", result)
|
||||
instance.write_channel.assert_called()
|
||||
self.assertEqual(instance.read_channel.call_count, 2)
|
||||
|
||||
|
||||
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(TestHuaweiTelnetCEDriver))
|
||||
suite.addTests(loader.loadTestsFromTestCase(TestHuaweiTelnetCEIntegration))
|
||||
|
||||
# Run tests
|
||||
runner = unittest.TextTestRunner(verbosity=2)
|
||||
result = runner.run(suite)
|
||||
|
||||
# Print summary
|
||||
print("\n" + "=" * 100)
|
||||
print("测试摘要:")
|
||||
print(f" 运行: {result.testsRun}")
|
||||
print(f" 成功: {result.testsRun - len(result.failures) - len(result.errors)}")
|
||||
print(f" 失败: {len(result.failures)}")
|
||||
print(f" 错误: {len(result.errors)}")
|
||||
print("=" * 100)
|
||||
|
||||
return result.wasSuccessful()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = run_tests()
|
||||
sys.exit(0 if success else 1)
|
||||
@ -49,9 +49,9 @@ def get_device_ports_from_topology(
|
||||
{
|
||||
"device_name": {
|
||||
"port": console_port,
|
||||
"groups": ["cisco_IOSv_telnet"],
|
||||
"device_type": "cisco_ios_telnet", # Extracted from tags
|
||||
"platform": "cisco_ios" # Extracted from tags
|
||||
"groups": ["platform_telnet"], # Dynamically generated from platform/device_type
|
||||
"device_type": "huawei_telnet", # Extracted from tags
|
||||
"platform": "huawei" # Extracted from tags
|
||||
}
|
||||
}
|
||||
Devices that don't exist or missing console_port will not be included
|
||||
@ -129,9 +129,15 @@ def get_device_ports_from_topology(
|
||||
)
|
||||
|
||||
# Add device to hosts_data
|
||||
# Dynamically generate group name based on platform and device_type
|
||||
if device_type and "_telnet" in device_type:
|
||||
group_name = f"{platform}_telnet"
|
||||
else:
|
||||
group_name = platform
|
||||
|
||||
hosts_data[device_name] = {
|
||||
"port": node_info["console_port"],
|
||||
"groups": ["cisco_IOSv_telnet"],
|
||||
"groups": [group_name],
|
||||
"device_type": device_type,
|
||||
"platform": platform,
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user