device_connector.py•34.1 kB
from typing import Dict, List, Optional, Any, Union, Tuple
import logging
import sys
import asyncio
import re
import os
from abc import ABC, abstractmethod
from datetime import datetime
# 创建日志目录
os.makedirs("logs", exist_ok=True)
# 引入相同的日志格式化处理
class JsonFormatter(logging.Formatter):
def __init__(self, fmt=None, datefmt=None, style='%'):
super().__init__(fmt, datefmt, style)
def format(self, record):
log_record = super().format(record)
return log_record.encode('utf-8', errors='replace').decode('utf-8')
# 配置日志格式
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(stream=sys.stdout)
]
)
# 设置所有处理器使用UTF-8编码格式化
for handler in logging.root.handlers:
if isinstance(handler, logging.StreamHandler):
handler.setFormatter(JsonFormatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
logger = logging.getLogger("device_connector")
# 全局变量
HAS_HUAWEI_SUPPORT = False
SCRAPLI_IMPORT_SUCCESS = False
# 导入scrapli并记录版本信息
try:
# 导入同步版本的Scrapli
from scrapli import Scrapli
# 导入异步版本的Scrapli
from scrapli.driver.core import AsyncIOSXEDriver, AsyncNXOSDriver, AsyncJunosDriver
# 导入更多平台支持
try:
from scrapli_community import huawei_vrp
HAS_HUAWEI_SUPPORT = True
logger.info("成功加载华为设备驱动")
except ImportError:
HAS_HUAWEI_SUPPORT = False
logger.warning("未找到华为设备驱动，将使用通用驱动")
logger.info(f"成功导入scrapli, 版本: {getattr(Scrapli, '__version__', '未知')}")
# 打印可用的异常类，帮助调试
import scrapli.exceptions
logger.debug("Available scrapli exceptions:")
for attr_name in dir(scrapli.exceptions):
if 'Error' in attr_name:
logger.debug(f" - {attr_name}")
SCRAPLI_IMPORT_SUCCESS = True
except ImportError as e:
logger.error(f"导入scrapli失败: {str(e)}，请确保已经安装: pip install scrapli>=2025.1.30")
# 这里不再抛出异常，而是允许程序继续运行
SCRAPLI_IMPORT_SUCCESS = False
from network_devices import (
NetworkDevice,
DeviceCredential,
ConnectionProtocol,
DeviceStatus
)
class CommandResult:
"""命令执行结果类"""
def __init__(self,
command: str,
output: str,
success: bool = True,
error_message: Optional[str] = None):
self.command = command
self.output = output
self.success = success
self.error_message = error_message
self.execution_time = datetime.now()
def __str__(self) -> str:
status = "成功" if self.success else "失败"
return f"命令 '{self.command}' 执行{status}: {len(self.output)}字节"
def to_dict(self) -> Dict[str, Any]:
"""转换为字典表示"""
return {
"command": self.command,
"output": self.output,
"success": self.success,
"error_message": self.error_message,
"execution_time": self.execution_time.isoformat()
}
class DeviceConnector(ABC):
"""设备连接器抽象基类"""
def __init__(self, device: NetworkDevice, credential: DeviceCredential):
self.device = device
self.credential = credential
self.connection = None
self.connected = False
self.last_activity = None
@abstractmethod
async def connect(self) -> bool:
"""
连接到设备
Returns:
是否连接成功
"""
pass
@abstractmethod
async def disconnect(self) -> bool:
"""
断开与设备的连接
Returns:
是否断开成功
"""
pass
@abstractmethod
async def send_command(self, command: str, timeout: int = 30) -> CommandResult:
"""
发送命令到设备
Args:
command: 要执行的命令
timeout: 命令超时时间（秒）
Returns:
命令执行结果
"""
pass
@abstractmethod
async def send_commands(self, commands: List[str], timeout: int = 30) -> List[CommandResult]:
"""
发送多个命令到设备
Args:
commands: 要执行的命令列表
timeout: 每个命令的超时时间（秒）
Returns:
命令执行结果列表
"""
pass
class ScrapliConnector(DeviceConnector):
"""Scrapli连接器实现"""
async def connect(self) -> bool:
"""
通过Scrapli连接到设备
Returns:
是否连接成功
"""
if not SCRAPLI_IMPORT_SUCCESS:
logger.error(f"无法连接到设备 {self.device.name}，Scrapli库导入失败")
return False
logger.info(f"正在通过Scrapli连接到设备: {self.device.name} ({self.device.ip_address})")
try:
# 优先使用设备的platform字段，如果不存在则从vendor推断
if hasattr(self.device, 'platform') and self.device.platform:
platform = self.device.platform
logger.debug(f"使用设备指定的平台类型: {platform}")
else:
# 映射设备类型和厂商到Scrapli platform参数
vendor = self.device.vendor.value.lower() if self.device.vendor else ""
device_type = self.device.device_type.value.lower() if self.device.device_type else ""
# 根据设备厂商和类型确定platform
if "cisco" in vendor:
if "switch" in device_type:
platform = "cisco_iosxe" # 思科交换机
elif "router" in device_type:
platform = "cisco_iosxe" # 思科路由器
elif "nexus" in device_type or "nxos" in vendor:
platform = "cisco_nxos" # 思科Nexus交换机
else:
platform = "cisco_iosxe" # 默认思科平台
elif "huawei" in vendor:
platform = "huawei_vrp" # 华为平台
elif "juniper" in vendor:
platform = "juniper_junos" # Juniper平台
elif "arista" in vendor:
platform = "arista_eos" # Arista平台
else:
platform = "cisco_iosxe" # 默认使用思科IOS XE平台
logger.debug(f"根据vendor推断平台类型: {platform}")
logger.info(f"使用平台类型: {platform} 连接设备")
# 准备连接参数
device_params = {
"host": self.device.ip_address,
"auth_username": self.credential.username,
"auth_strict_key": False,
"timeout_socket": 15, # 连接超时时间
"timeout_transport": 30, # 传输超时时间
"platform": platform,
"port": self.credential.port or 22,
"transport": "paramiko" # 在Windows上使用paramiko传输方式
}
# 设置认证方式：优先使用SSH密钥文件，如果没有则使用密码
if self.credential.ssh_key_file and self.credential.protocol == ConnectionProtocol.SSH:
logger.info(f"使用SSH密钥文件认证: {self.credential.ssh_key_file}")
device_params["auth_private_key"] = self.credential.ssh_key_file
elif self.credential.password:
logger.info("使用密码认证")
device_params["auth_password"] = self.credential.password
else:
logger.error("没有提供认证方式（密码或SSH密钥文件）")
return False
# 如果是SSH，添加SSH特有参数
if self.credential.protocol == ConnectionProtocol.SSH:
# 保留为异步方式连接
if platform == "cisco_iosxe":
driver = AsyncIOSXEDriver
elif platform == "cisco_nxos":
driver = AsyncNXOSDriver
elif platform == "juniper_junos":
driver = AsyncJunosDriver
else:
# 对于其他平台，尝试使用同步方式连接
logger.info(f"没有{platform}的专用异步驱动，使用同步Scrapli")
# 使用同步方式，但在异步方法中运行
# 使用设备参数中的platform自动选择合适的驱动
conn = Scrapli(**device_params)
# 用asyncio运行同步打开连接操作
await asyncio.to_thread(conn.open)
self.connection = conn
self.connected = True
self.last_activity = datetime.now()
logger.info(f"已成功连接到设备: {self.device.name} ({self.device.ip_address})")
# 获取并记录设备提示符
try:
prompt = await asyncio.to_thread(conn.get_prompt)
logger.info(f"设备提示符: {prompt}")
except Exception as e:
logger.warning(f"获取设备提示符失败: {str(e)}")
return True
# 这里使用异步方式连接
logger.debug(f"使用异步驱动: {driver.__name__}")
conn = driver(**device_params)
logger.debug(f"正在打开连接到 {self.device.ip_address}...")
await conn.open()
self.connection = conn
# 如果是Telnet，处理Telnet连接
elif self.credential.protocol == ConnectionProtocol.TELNET:
# 修改传输选项为telnet
device_params["transport"] = "telnet"
device_params["port"] = self.credential.port or 23
# 使用同步方式，但在异步方法中运行
conn = Scrapli(**device_params)
# 用asyncio运行同步打开连接操作
await asyncio.to_thread(conn.open)
self.connection = conn
self.connected = True
self.last_activity = datetime.now()
logger.info(f"已成功连接到设备: {self.device.name} ({self.device.ip_address})")
# 获取并记录设备提示符
try:
if hasattr(self.connection, "get_prompt"):
if asyncio.iscoroutinefunction(self.connection.get_prompt):
prompt = await self.connection.get_prompt()
else:
prompt = await asyncio.to_thread(self.connection.get_prompt)
logger.info(f"设备提示符: {prompt}")
except Exception as e:
logger.warning(f"获取设备提示符失败: {str(e)}")
return True
except Exception as e:
logger.error(f"连接设备失败: {self.device.name} ({self.device.ip_address}) - {type(e).__name__}: {str(e)}")
self.connected = False
return False
async def disconnect(self) -> bool:
"""
断开Scrapli连接
Returns:
是否断开成功
"""
if not self.connected:
logger.warning(f"设备未连接: {self.device.name} ({self.device.ip_address})")
return True
try:
if self.connection:
await self.connection.close()
self.connected = False
logger.info(f"已断开与设备的连接: {self.device.name} ({self.device.ip_address})")
return True
except Exception as e:
logger.error(f"断开连接失败: {self.device.name} ({self.device.ip_address}) - {str(e)}")
return False
async def send_command(self, command: str, timeout: int = 30) -> CommandResult:
"""
通过Scrapli发送命令
Args:
command: 要执行的命令
timeout: 命令超时时间（秒）
Returns:
命令执行结果
"""
if not self.connected:
logger.warning(f"设备未连接: {self.device.name} ({self.device.ip_address})")
return CommandResult(
command=command,
output="",
success=False,
error_message="设备未连接"
)
logger.info(f"向设备 {self.device.name} 发送命令: {command}")
try:
# 根据连接对象是同步还是异步类型，采用不同的处理方式
if asyncio.iscoroutinefunction(self.connection.send_command):
# 异步方法直接调用
logger.debug("使用异步方式发送命令")
response = await self.connection.send_command(
command=command,
timeout_ops=timeout
)
else:
# 同步方法需要在线程中运行
logger.debug("使用同步方式在线程中发送命令")
response = await asyncio.to_thread(
self.connection.send_command,
command=command,
timeout_ops=timeout
)
self.last_activity = datetime.now()
# 处理响应对象，提取结果
result = ""
is_failed = False
# 尝试提取结果和失败状态
if hasattr(response, "result"):
result = response.result
if hasattr(response, "failed"):
is_failed = response.failed
else:
# 如果没有result属性，尝试将整个响应转为字符串
result = str(response)
# 规范化输出格式，确保换行符一致
if isinstance(result, str):
# 对齐输出格式，确保每行开头无多余空格
lines = result.splitlines()
# 修复：不再移除每行开头的空格，保留原始输出格式
# 重新组合输出，使用统一的换行符
result = '\n'.join(lines)
# 命令执行完毕后，获取当前提示符
try:
# 获取最新的提示符
if hasattr(self.connection, "get_prompt"):
if asyncio.iscoroutinefunction(self.connection.get_prompt):
prompt = await self.connection.get_prompt()
else:
prompt = await asyncio.to_thread(self.connection.get_prompt)
# 确保输出末尾有提示符
if prompt and not result.endswith(prompt):
result = result + "\n" + prompt
logger.debug(f"添加提示符到输出: '{prompt}'")
except Exception as e:
logger.warning(f"获取提示符失败: {str(e)}")
if is_failed:
logger.error(f"命令 '{command}' 在设备 {self.device.name} 上执行失败: {result}")
return CommandResult(
command=command,
output=result,
success=False,
error_message="命令执行失败"
)
logger.info(f"命令 '{command}' 在设备 {self.device.name} 上执行成功")
return CommandResult(
command=command,
output=result,
success=True
)
except asyncio.TimeoutError:
logger.error(f"命令 '{command}' 在设备 {self.device.name} 上执行超时")
return CommandResult(
command=command,
output="",
success=False,
error_message="命令执行超时"
)
except Exception as e:
# 检查是否为连接断开错误
error_name = type(e).__name__
error_msg = str(e)
# 检测连接相关错误
if "ScrapliConnectionNotOpened" in error_name or "connection not opened" in error_msg:
logger.warning(f"连接已断开，尝试重新连接: {self.device.name} ({self.device.ip_address})")
try:
# 首先尝试关闭当前连接
try:
if self.connection:
await self.connection.close()
except Exception as close_error:
logger.warning(f"关闭旧连接失败: {str(close_error)}")
# 标记为未连接状态
self.connected = False
# 尝试重新连接
reconnect_success = await self.connect()
if reconnect_success:
logger.info(f"重新连接成功，重试命令: {command}")
# 递归调用自身重试命令
return await self.send_command(command, timeout)
else:
logger.error(f"重新连接失败: {self.device.name} ({self.device.ip_address})")
return CommandResult(
command=command,
output="",
success=False,
error_message="连接断开且重连失败"
)
except Exception as reconnect_error:
logger.error(f"重新连接过程中发生错误: {str(reconnect_error)}")
return CommandResult(
command=command,
output="",
success=False,
error_message=f"重连过程错误: {str(reconnect_error)}"
)
# 其他错误正常处理
logger.error(f"命令 '{command}' 在设备 {self.device.name} 上执行失败: {error_name}: {error_msg}")
return CommandResult(
command=command,
output="",
success=False,
error_message=f"{error_name}: {error_msg}"
)
async def send_commands(self, commands: List[str], timeout: int = 30) -> List[CommandResult]:
"""
通过Scrapli发送多个命令
Args:
commands: 要执行的命令列表
timeout: 每个命令的超时时间（秒）
Returns:
命令执行结果列表
"""
logger.info(f"向设备 {self.device.name} 发送多个命令，共 {len(commands)} 条")
results = []
for i, command in enumerate(commands):
logger.debug(f"执行第 {i+1}/{len(commands)} 条命令: {command}")
result = await self.send_command(command, timeout)
results.append(result)
# 如果命令执行失败，记录并中断执行
if not result.success:
logger.warning(f"命令 '{command}' 执行失败，中断后续命令执行")
break
# 命令之间添加短暂延迟，避免设备负载过高
if i < len(commands) - 1: # 如果不是最后一条命令
await asyncio.sleep(0.5)
return results
async def send_control_command(self, control_code: str) -> CommandResult:
"""
发送控制字符到设备
Args:
control_code: 控制字符代码，例如'C'表示Ctrl+C
Returns:
命令执行结果
"""
if not self.connection or not self.connected:
logger.warning(f"未连接到设备，无法发送控制命令: {control_code}")
return CommandResult(
command=f"CTRL+{control_code}",
output="",
success=False,
error_message="设备未连接"
)
try:
logger.info(f"发送控制字符 CTRL+{control_code} 到设备: {self.device.name}")
# 不同控制字符的处理
if control_code == 'C': # CTRL+C
special_char = '\x03'
description = "中断"
elif control_code == 'D': # CTRL+D
special_char = '\x04'
description = "EOF"
elif control_code == 'Z': # CTRL+Z
special_char = '\x1A'
description = "挂起"
else:
logger.warning(f"不支持的控制字符: CTRL+{control_code}")
return CommandResult(
command=f"CTRL+{control_code}",
output="",
success=False,
error_message=f"不支持的控制字符: CTRL+{control_code}"
)
# 发送特殊字符
if hasattr(self.connection, 'channel'):
# 直接通过channel发送控制字符
await self.connection.channel.write(special_char)
logger.debug(f"已发送{description}信号到设备: {self.device.name}")
# 等待输出
await asyncio.sleep(1)
output = await self.connection.channel.read()
else:
# 尝试通过send_command发送
logger.debug(f"尝试通过send_command发送控制命令")
resp = await self.connection.send_command(special_char)
output = resp.result
self.last_activity = datetime.now()
return CommandResult(
command=f"CTRL+{control_code}",
output=output,
success=True
)
except Exception as e:
error_name = type(e).__name__
error_msg = str(e)
# 检测连接相关错误
if "ScrapliConnectionNotOpened" in error_name or "connection not opened" in error_msg:
logger.warning(f"发送控制字符时检测到连接已断开，尝试重新连接: {self.device.name}")
try:
# 尝试关闭当前连接
try:
if self.connection:
await self.connection.close()
except Exception as close_error:
logger.warning(f"关闭旧连接失败: {str(close_error)}")
# 标记为未连接状态
self.connected = False
# 尝试重新连接
reconnect_success = await self.connect()
if reconnect_success:
logger.info(f"重新连接成功，重试发送控制字符: CTRL+{control_code}")
# 递归调用自身重试发送
return await self.send_control_command(control_code)
else:
logger.error(f"重新连接失败: {self.device.name}")
return CommandResult(
command=f"CTRL+{control_code}",
output="",
success=False,
error_message="连接断开且重连失败"
)
except Exception as reconnect_error:
logger.error(f"重新连接过程中发生错误: {str(reconnect_error)}")
return CommandResult(
command=f"CTRL+{control_code}",
output="",
success=False,
error_message=f"重连过程错误: {str(reconnect_error)}"
)
# 其他错误正常处理
error_msg = f"发送控制字符失败: {error_name}: {error_msg}"
logger.error(error_msg)
return CommandResult(
command=f"CTRL+{control_code}",
output="",
success=False,
error_message=error_msg
)
class ConnectorFactory:
"""连接器工厂，根据设备和凭据创建适当的连接器"""
@staticmethod
def create_connector(device: NetworkDevice, credential: DeviceCredential) -> DeviceConnector:
"""
创建设备连接器
Args:
device: 网络设备
credential: 设备凭据
Returns:
设备连接器
Raises:
ValueError: 如果不支持的协议
"""
# 无论是SSH还是TELNET，都使用ScrapliConnector
# Scrapli将根据配置自动选择合适的传输机制
if credential.protocol in [ConnectionProtocol.SSH, ConnectionProtocol.TELNET]:
return ScrapliConnector(device, credential)
else:
raise ValueError(f"不支持的协议: {credential.protocol.value}")
# 连接管理器单例
class ConnectionManager:
"""连接管理器，管理设备连接和会话"""
def __init__(self):
self.active_connections: Dict[str, DeviceConnector] = {}
logger.info("连接管理器初始化完成")
async def connect_device(self,
device: NetworkDevice,
credential: DeviceCredential) -> Tuple[bool, Optional[str]]:
"""
连接到设备
Args:
device: 网络设备
credential: 设备凭据
Returns:
(是否成功, 错误消息)
"""
# 检查是否已经连接
connection_key = f"{device.id}_{credential.id}"
if connection_key in self.active_connections:
logger.info(f"设备已连接: {device.name} ({device.ip_address})")
return True, None
try:
# 记录设备和凭据信息
logger.info(f"尝试连接设备: {device.name} ({device.ip_address}), 凭据: {credential.username}@{credential.protocol.value}:{credential.port}")
# 创建连接器
connector = ConnectorFactory.create_connector(device, credential)
# 连接到设备
success = await connector.connect()
if success:
# 保存连接
self.active_connections[connection_key] = connector
return True, None
else:
return False, "连接失败"
except Exception as e:
logger.error(f"连接设备时发生错误: {str(e)}")
return False, str(e)
async def disconnect_device(self, device_id: str, credential_id: str) -> Tuple[bool, Optional[str]]:
"""
断开与设备的连接
Args:
device_id: 设备ID
credential_id: 凭据ID
Returns:
(是否成功, 错误消息)
"""
connection_key = f"{device_id}_{credential_id}"
connector = self.active_connections.get(connection_key)
if not connector:
logger.warning(f"设备未连接: {device_id}")
return True, None
try:
# 断开连接
success = await connector.disconnect()
if success:
# 移除连接
del self.active_connections[connection_key]
return success, None
except Exception as e:
logger.error(f"断开连接时发生错误: {str(e)}")
return False, str(e)
async def send_command(self,
device_id: str,
credential_id: str,
command: str,
timeout: int = 30) -> Tuple[Optional[CommandResult], Optional[str]]:
"""
向设备发送命令
Args:
device_id: 设备ID
credential_id: 凭据ID
command: 要执行的命令
timeout: 命令超时时间（秒）
Returns:
(命令结果, 错误消息)
"""
connection_key = f"{device_id}_{credential_id}"
connector = self.active_connections.get(connection_key)
if not connector:
logger.warning(f"设备未连接: {device_id}，请先调用connect_device连接")
return None, "设备未连接，请先连接设备"
try:
# 发送命令
logger.info(f"通过连接管理器向设备 {device_id} 发送命令: {command}")
result = await connector.send_command(command, timeout)
# 如果成功执行，但输出为空，添加提示
if result.success and not result.output.strip():
logger.info("命令执行成功，但返回内容为空")
result.output = "[命令执行成功，但没有返回内容]"
return result, None
except Exception as e:
error_msg = f"发送命令时发生错误: {type(e).__name__}: {str(e)}"
logger.error(error_msg)
return None, error_msg
async def send_commands(self,
device_id: str,
credential_id: str,
commands: List[str],
timeout: int = 30) -> Tuple[Optional[List[CommandResult]], Optional[str]]:
"""
向设备发送多个命令
Args:
device_id: 设备ID
credential_id: 凭据ID
commands: 要执行的命令列表
timeout: 每个命令的超时时间（秒）
Returns:
(命令结果列表, 错误消息)
"""
connection_key = f"{device_id}_{credential_id}"
connector = self.active_connections.get(connection_key)
if not connector:
logger.warning(f"设备未连接: {device_id}，请先调用connect_device连接")
return None, "设备未连接，请先连接设备"
try:
# 发送命令
logger.info(f"通过连接管理器向设备 {device_id} 发送 {len(commands)} 条命令")
results = await connector.send_commands(commands, timeout)
return results, None
except Exception as e:
error_msg = f"发送命令时发生错误: {type(e).__name__}: {str(e)}"
logger.error(error_msg)
return None, error_msg
def get_active_connections(self) -> List[Dict[str, Any]]:
"""
获取活动连接列表
Returns:
活动连接信息列表
"""
result = []
for connection_key, connector in self.active_connections.items():
device_id, credential_id = connection_key.split('_')
result.append({
"device_id": device_id,
"device_name": connector.device.name,
"device_ip": connector.device.ip_address,
"credential_id": credential_id,
"protocol": connector.credential.protocol.value,
"last_activity": connector.last_activity.isoformat() if connector.last_activity else None
})
return result
async def send_control_command(self,
device_id: str,
credential_id: str,
control_code: str) -> Tuple[Optional[CommandResult], Optional[str]]:
"""
向设备发送控制字符
Args:
device_id: 设备ID
credential_id: 凭证ID
control_code: 控制字符代码，例如'C'表示Ctrl+C
Returns:
(命令结果, 错误信息)
"""
# 获取连接
connection_key = f"{device_id}_{credential_id}"
connector = self.active_connections.get(connection_key)
if not connector:
error_msg = f"未找到活跃连接: {connection_key}"
logger.error(error_msg)
return None, error_msg
try:
result = await connector.send_control_command(control_code)
return result, None
except Exception as e:
error_msg = f"发送控制字符失败: {str(e)}"
logger.error(error_msg)
return None, error_msg
# 创建全局连接管理器实例
connection_manager = ConnectionManager()