基于Python与Netmiko通过SSH读取思科路由器接口及流量的技术咨询

前置准备 #

先安装Netmiko依赖：

pip install netmiko

步骤1：建立SSH连接 #

用Netmiko的ConnectHandler快速建立与思科IOS设备的SSH连接，替换成你的设备实际参数即可：

from netmiko import ConnectHandler
from netmiko.exceptions import NetMikoAuthenticationException, NetMikoTimeoutException

def connect_to_device(device_info):
    try:
        connection = ConnectHandler(**device_info)
        print("SSH连接建立成功")
        return connection
    except NetMikoTimeoutException:
        print("连接超时，请检查设备IP或网络连通性")
    except NetMikoAuthenticationException:
        print("认证失败，请检查用户名/密码")
    except Exception as e:
        print(f"连接出错: {str(e)}")
    return None

# 设备配置信息
device = {
    "device_type": "cisco_ios",
    "ip": "192.168.1.1",  # 替换为你的路由器IP
    "username": "admin",  # 替换为你的登录用户名
    "password": "your_password",  # 替换为你的登录密码
    # 若需要特权模式，添加下面一行
    # "secret": "enable_password",
}

# 初始化连接
conn = connect_to_device(device)
if not conn:
    exit()

# 进入特权模式（如果需要）
# conn.enable()

步骤2：读取接口列表 #

发送show ip interface brief命令获取接口列表，开启use_textfsm=True可直接得到结构化数据，不用手动解析纯文本：

def get_interface_list(connection):
    output = connection.send_command("show ip interface brief", use_textfsm=True)
    # 提取所有接口名称
    interfaces = [entry["intf"] for entry in output]
    print(f"获取到的接口列表: {interfaces}")
    return interfaces

# 获取接口列表
interface_list = get_interface_list(conn)
if not interface_list:
    print("未获取到接口信息")
    conn.disconnect()
    exit()

步骤3：读取并监控接口实时流量 #

通过两次读取接口流量统计（间隔指定时间），计算出实时流量速率：

import time

def get_interface_traffic(connection, interface):
    # 发送命令并解析结构化流量数据
    output = connection.send_command(f"show interfaces {interface}", use_textfsm=True)[0]
    return {
        "in_bytes": int(output["input_bytes"]),
        "out_bytes": int(output["output_bytes"])
    }

def monitor_traffic(connection, interfaces, interval=5):
    print(f"\n开始监控接口流量（间隔{interval}秒），按Ctrl+C停止...")
    try:
        # 读取初始流量值
        initial_traffic = {intf: get_interface_traffic(connection, intf) for intf in interfaces}
        while True:
            time.sleep(interval)
            # 读取当前流量值
            current_traffic = {intf: get_interface_traffic(connection, intf) for intf in interfaces}
            # 计算并打印速率
            print("\n" + "-"*60)
            for intf in interfaces:
                in_diff = current_traffic[intf]["in_bytes"] - initial_traffic[intf]["in_bytes"]
                out_diff = current_traffic[intf]["out_bytes"] - initial_traffic[intf]["out_bytes"]
                # 转换为KB/s单位
                in_speed_kb = (in_diff / interval) / 1024
                out_speed_kb = (out_diff / interval) / 1024
                print(f"接口 {intf}: 入站 {in_speed_kb:.2f} KB/s | 出站 {out_speed_kb:.2f} KB/s")
            # 更新初始值为当前值，用于下一轮计算
            initial_traffic = current_traffic.copy()
    except KeyboardInterrupt:
        print("\n监控已停止")
    except Exception as e:
        print(f"监控出错: {str(e)}")

# 启动流量监控
monitor_traffic(conn, interface_list)

# 断开SSH连接
conn.disconnect()

注意事项 #

• 确保路由器已开启SSH服务，且你的账号有权限执行show ip interface brief和show interfaces命令
• 如果设备需要进入特权模式才能查看详细流量，记得在设备配置中添加secret参数，并调用conn.enable()
• TextFSM解析模板Netmiko已内置，无需额外配置，结构化数据比纯文本处理更可靠

内容的提问来源于stack exchange，提问作者salman ansari