From ad771e29278750ed5d5dcb2699ad80f4e66c2a6e Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 13 Mar 2026 23:33:14 +0800 Subject: [PATCH] feat: add Ruijie telnet driver support to custom Netmiko tools - Import and register Ruijie telnet device type in config_tools_nornir.py - Import and register Ruijie telnet device type in display_tools_nornir.py - Update custom_netmiko __init__.py to include RuijieTelnetEnhanced driver - Add Ruijie telnet driver to __all__ exports for proper module exposure - Extends custom Netmiko support to handle Ruijie devices with interactive prompt handling --- .../tools_v2/config_tools_nornir.py | 4 + .../tools_v2/display_tools_nornir.py | 4 + .../utils/custom_netmiko/__init__.py | 8 +- .../utils/custom_netmiko/ruijie_telnet.py | 181 ++++++++++++++++++ 4 files changed, 196 insertions(+), 1 deletion(-) create mode 100644 gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py diff --git a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py index 0e84a9165..829d6d0fc 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py @@ -58,9 +58,13 @@ from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401 # 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 + from gns3server.agent.gns3_copilot.utils.custom_netmiko.ruijie_telnet import ( + register_custom_device_type as register_ruijie_device_type, + ) # Re-register to ensure device types are available huawei_ce.register_custom_device_type() + register_ruijie_device_type() # CRITICAL: Update netmiko.ssh_dispatcher platforms lists # The platforms variable is calculated at module import time in ssh_dispatcher diff --git a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py index 673635d39..e96d65a81 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py @@ -58,9 +58,13 @@ from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401 # 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 + from gns3server.agent.gns3_copilot.utils.custom_netmiko.ruijie_telnet import ( + register_custom_device_type as register_ruijie_device_type, + ) # Re-register to ensure device types are available huawei_ce.register_custom_device_type() + register_ruijie_device_type() # CRITICAL: Update netmiko.ssh_dispatcher platforms lists # The platforms variable is calculated at module import time in ssh_dispatcher diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py index a81a3ec24..4dc3404c8 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py @@ -32,6 +32,7 @@ authentication or behavior patterns. Supported Drivers: - huawei_ce: HuaweiTelnetCE for CloudEngine devices (no authentication) +- ruijie_telnet: RuijieTelnet for Ruijie devices (interactive prompt handling) Usage: from gns3server.agent.gns3_copilot.utils import custom_netmiko @@ -59,4 +60,9 @@ try: except Exception as e: logger.warning(f"Failed to import Huawei CE driver: {e}", exc_info=True) -__all__ = ["huawei_ce"] +try: + from .ruijie_telnet import RuijieTelnetEnhanced # noqa: F401 +except Exception as e: + logger.warning(f"Failed to import Ruijie driver: {e}", exc_info=True) + +__all__ = ["huawei_ce", "ruijie_telnet"] diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py new file mode 100644 index 000000000..edd46273a --- /dev/null +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py @@ -0,0 +1,181 @@ +# 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 Ruijie (锐捷) devices in GNS3 emulation. + +This module provides enhanced handling for Ruijie network devices, +specifically addressing interactive prompts during configuration. + +Key Features: +- Inherits from RuijieOSBase (Netmiko's native Ruijie support) +- Handles interactive prompts (e.g., OSPF router-id [yes/no] confirmation) +- Maintains compatibility with Netmiko's Ruijie implementation + +Device Type: gns3_ruijie_telnet +""" + +import importlib +import logging +import re +import time + +from netmiko.ruijie.ruijie_os import RuijieOSBase + + +class RuijieTelnetEnhanced(RuijieOSBase): + """ + Enhanced Ruijie device driver with interactive prompt handling. + + Inherits from RuijieOSBase to maintain native Netmiko compatibility, + and adds automatic handling for interactive configuration prompts. + """ + + def __init__( + self, + *args, + **kwargs, + ) -> None: + """Initialize RuijieTelnetEnhanced connection.""" + # Set default_enter for telnet (like Netmiko's RuijieOSTelnet) + default_enter = kwargs.get("default_enter") + kwargs["default_enter"] = "\r\n" if default_enter is None else default_enter + super().__init__(*args, **kwargs) + + def send_config_set( + self, + config_commands: str | list[str], + **kwargs, + ) -> str: + """ + Send configuration commands with interactive prompt handling. + + Strategy: + - Send commands ONE BY ONE (not batch) + - After each command, read output and check for [yes/no] prompt + - If prompt detected, send 'yes' immediately before next command + + 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 defaults + exit_config_mode = kwargs.get("exit_config_mode", True) + enter_config_mode = kwargs.get("enter_config_mode", True) + read_timeout = kwargs.get("read_timeout", 60) + delay_factor = self.global_delay_factor + + output = "" + + # Enter config mode if needed + if enter_config_mode: + output += self.config_mode() + + # Interactive prompt patterns to detect + interactive_patterns = [ + r"\[yes/no\]", + r"\[y/n\]", + r"\[Y/N\]", + ] + + # Send commands ONE BY ONE to detect prompts after each command + for cmd in config_commands: + # Write the command + self.write_channel(f"{cmd}{self.RETURN}") + time.sleep(delay_factor * 0.05) # Reduced from 0.1 + + # Read output after this command + # Use read_channel_timing to handle interactive prompts + new_output = self.read_channel_timing( + read_timeout=10, last_read=0.5 # Reduced from 1.0 + ) + output += new_output + + # Check if interactive prompt appeared after this command + for pattern in interactive_patterns: + if re.search(pattern, new_output, re.IGNORECASE): + logging.info( + "Ruijie device: Detected interactive prompt after '%s', " + "sending 'yes'", + cmd, + ) + # Send 'yes' to confirm BEFORE next command + self.write_channel(f"yes{self.RETURN}") + time.sleep(delay_factor * 0.3) # Reduced from 0.5 + # Read the confirmation response + output += self.read_channel_timing( + read_timeout=30, last_read=0.5 # Reduced from 1.0 + ) + break + + # Exit config mode if requested + if exit_config_mode: + output += self.exit_config_mode() + + return output + + +# Register the custom device type with Netmiko +def register_custom_device_type() -> None: + """ + Register the custom RuijieTelnetEnhanced device type with Netmiko. + + This function adds 'gns3_ruijie_telnet' to Netmiko's CLASS_MAPPER + and updates the platforms lists. + """ + sd = importlib.import_module("netmiko.ssh_dispatcher") + + # Register in both mappers + sd.CLASS_MAPPER_BASE["gns3_ruijie_telnet"] = RuijieTelnetEnhanced + sd.CLASS_MAPPER["gns3_ruijie_telnet"] = RuijieTelnetEnhanced + + # Update platforms lists + 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] + sd.platforms_str = "\n" + "\n".join(sd.platforms_base) + sd.telnet_platforms_str = "\n" + "\n".join(sd.telnet_platforms) + + +# Auto-register on import +try: + register_custom_device_type() +except Exception as e: + logger = logging.getLogger(__name__) + logger.warning( + f"Failed to register Ruijie device type: {e}", + exc_info=True + )