☰
Python Telnet连接路由器:网络设备自动化巡检实战
2026/10/14 2:59:40 网站建设 项目流程

1. 为什么我最后选了 Telnet 而不是 SSH 来管路由器

说实话,看到"python-通过telnet连接路由器"这个标题,很多人第一反应是:都什么年代了还在用 Telnet?SSH 它不香吗?我在实际项目中确实也遇到过这种质疑,但有些场景下 Telnet 反而是唯一能走通的路。

我最近在做一个某实验室的网络设备巡检需求,现场有一批老旧的接入层交换机,采购时间都在七八年前,固件版本停留在很早期的阶段,Web 管理界面简陋到只能看状态,SSH 服务压根就没启用。设备厂商的技术支持给的答复是:要开 SSH 就得升级固件,升级固件就得申请窗口期,申请窗口期就得等业务低峰,一等就是两三个月。但巡检任务下周一就要交结果,怎么办?翻设备手册发现 Telnet 服务是默认开启的,于是我就用 Python 写了一套自动化巡检脚本,五分钟跑完一整批设备的配置备份和状态采集,问题当场解决。

这就是 Telnet 在现代网络管理里的真实处境:它依然活在很多存量设备的角落里。尤其是:

  • 老旧设备,固件不支持 SSH 或 SSH 配置繁琐
  • 内网隔离环境,Telnet 明文传输的风险可接受
  • 临时应急场景,需要快速批量执行命令
  • 网络设备模拟器(比如某虚拟化网络工具),很多只开放了 Telnet 端口,用来做自动化测试非常合适

所以这篇文章不是鼓吹大家放弃 SSH 改用 Telnet,而是想分享一套用 Python 连接 Telnet 设备的完整方法论,包括连接池管理、命令交互、异常处理、批量巡检这些实战中真正用得上的东西。

对于想学习 Python 网络自动化的人来说,Telnet 也是一个很好的起点。它的协议足够简单,非常适合用来理解"程序怎么和设备对话"这件事。掌握了 Telnet 的交互逻辑,再去看 Paramiko、Netmiko 这些 SSH 库,你会发现很多东西都是相通的——无非是传输层加密了,命令交互的底层思路完全一样。

2. 动手前的准备工作:环境、库选择和网络连通性确认

2.1 Python 环境与 Telnetlib 库

让我先说结论:只要你的 Python 版本还是 3.x,标准库里就自带telnetlib,不需要安装任何第三方包。

import telnetlib print(telnetlib.Telnet.__doc__)

我在 Python 3.8 到 3.12 的多个版本上都跑过,telnetlib 的接口非常稳定。但你需要注意一件事:在 Python 3.13 版本中,telnetlib 被标记为弃用,未来某个版本可能会被移除。所以如果你是在全新的环境里做项目,我的建议是:

  1. 短期脚本、临时工具:直接用 telnetlib,零依赖,最省事
  2. 长期项目、要持续维护:用telnetlib3(这是一个纯异步实现的 Telnet 库)或者直接基于 socket 自己封装

我在实际项目中用的方案是:写了一个简单的基于socket的 Telnet 客户端封装,因为有些设备交互逻辑比较特殊,标准库的 telnetlib 处理起来反而别扭,自己封装控制力更强。这个后面会详细展开。

2.2 网络连通性确认:这不是废话步骤

很多人拿到设备 IP 就开始写代码,连不上就一脸懵。我见过太多这样的情况:代码写得没问题,但是设备 IP 根本 ping 不通,或者 Telnet 端口被防火墙挡了。所以第一步永远是确认连通性。

import socket def check_telnet_port(host: str, port: int = 23, timeout: float = 3.0) -> bool: """ 检查目标主机的 Telnet 端口是否开放 """ try: with socket.create_connection((host, port), timeout=timeout): return True except (socket.timeout, ConnectionRefusedError, OSError): return False if __name__ == "__main__": router_ip = "192.168.10.1" if check_telnet_port(router_ip, 23): print(f"{router_ip}:23 端口可达,可以继续") else: print(f"{router_ip}:23 端口不可达,请检查网络和防火墙")

这个小工具我在每次跑批量任务前都会先执行一遍,特别是设备数量几十台以上的时候,先用端口扫描把不通的设备筛出来,能省下后面大量超时等待的时间。

2.3 确认设备登录方式

不同厂商的路由器、交换机,Telnet 登录交互的提示符不太一样。华为、华三的设备登录流程大致是:

Username: Password: <设备名>

思科设备的登录流程是:

Username: Password: 设备名>

有些老设备甚至没有用户名,直接要密码。所以写自动化脚本前,建议先手动用 Windows 自带的 Telnet 客户端(或者 SecureCRT、Xshell)连一次设备,把完整的交互流程记录下来,包括:

  • 登录时有哪些提示符
  • 用户名是单独一行提示还是包含在欢迎语里
  • 密码输入后会不会直接进用户视图还是要再敲一个回车
  • 登录成功后的命令行提示符长什么样

这些信息是后面写正则匹配的重要依据。我习惯把这些信息记在一个小型配置字典里,每个设备型号对应一套交互模板,这样脚本可以适配不同厂商的设备。

3. 第一个能跑的版本:用 telnetlib 完成登录和命令执行

3.1 最基础的登录脚本

我们先看一个最朴素的 Python Telnet 登录示例,它的逻辑和手动操作完全对应:

import telnetlib import time HOST = "192.168.10.1" USERNAME = "admin" PASSWORD = "admin123" COMMAND = "display current-configuration" # 1. 建立连接 tn = telnetlib.Telnet(HOST, port=23, timeout=10) # 2. 等待用户名提示符出现 tn.read_until(b"Username:", timeout=10) tn.write(USERNAME.encode("ascii") + b"\n") # 3. 等待密码提示符出现 tn.read_until(b"Password:", timeout=10) tn.write(PASSWORD.encode("ascii") + b"\n") # 4. 等待进入设备视图(登录成功后会出现设备名提示符) time.sleep(1) tn.write(COMMAND.encode("ascii") + b"\n") # 5. 读取执行结果(多读一些内容并等待输出稳定) time.sleep(2) output = tn.read_very_eager().decode("utf-8", errors="ignore") print(output) # 6. 退出 tn.write(b"quit\n") tn.close()

这个脚本能跑通,但只适合"设备就那一两台、命令就那两三条"的场景。真实项目里这么做会踩一堆坑:

  • read_until等待的提示符如果和实际设备不匹配,会直接卡死到超时
  • time.sleep是硬编码的,设备响应慢的时候读不到完整输出,响应快的时候又白白浪费时间
  • 命令执行结果可能包含二进制控制字符,直接 decode 会乱码
  • 没有异常处理,设备一断连整个脚本就崩了

所以接下来我要做的事情,是把这套"能跑"的脚本包装成一套"抗造"的交互工具。

3.2 核心原理解释:Telnet 交互就是"读"和"写"

在深入封装之前,我想先解释清楚 Telnet 交互的本质,因为理解了这一点,后面所有代码你都能自己推导出来。

Telnet 是一个基于 TCP 的明文协议,默认端口 23。客户端做的事情无非两件:往连接里写字节、从连接里读字节。设备收到你的用户名、密码、命令,处理后把结果以字节流形式返回,你的程序要做的事情就是不断"读响应 -> 分析状态 -> 写下一个请求"。

telnetlib中最核心的三个方法:

方法作用使用场景
read_until(expected, timeout)一直读到期望的字节串出现,或超时等提示符、等登录成功标志
read_very_eager()立即返回当前缓冲区所有可读内容,不等待命令执行完,抓取全部输出
write(data)向连接发送数据发送用户名、密码、命令

read_until是最常用的,它的工作机制是阻塞等待匹配字节。这里有个容易踩坑的点:read_until是大小写敏感的字节匹配,而且必须精确匹配。如果设备返回的登录提示符是Username:,你写read_until(b"username:")就永远匹配不到。所以前面说"先手动连一次设备记录交互流程",就是给这一步提供精确输入。

def expect_prompt(tn: telnetlib.Telnet, patterns: list, timeout: float = 10): """ 同时等待多个可能的提示符,返回匹配到的索引和内容 """ # 返回的是 (索引, 匹配到的内容, 读到的文本) index, matched, text = tn.expect(patterns, timeout=timeout) return index, text.decode("utf-8", errors="ignore")

expect是read_until的增强版,可以同时监听多个提示符,谁先出现就返回谁。这在对付不同厂商设备时特别有用:我不用猜登录提示符到底是Username:还是login:,把常见的几种都列出来,等哪个先来即可。

4. 实战封装:一套能应对不同厂商设备的 Telnet 交互类

4.1 设计思路

我在多个项目中反复用过 Telnet 之后,总结出几个硬需求:

  1. 要能适配不同厂商设备的提示符差异
  2. 命令执行后必须能拿到完整输出,而不是只拿一部分
  3. 要能处理超时、断连、认证失败等异常情况
  4. 要能方便地批量执行多条命令
  5. 用完之后自动退出,不留半开连接

基于这些需求,我写了一个相对完整的封装类。代码不算长,但非常实用:

import telnetlib import time import re class RouterTelnetClient: """ 一个通用的路由器/交换机 Telnet 交互客户端 """ def __init__(self, host: str, username: str, password: str, login_prompts=None, cmd_prompt=None, port: int = 23, timeout: float = 10, encoding: str = "utf-8"): self.host = host self.username = username self.password = password self.port = port self.timeout = timeout self.encoding = encoding # 登录阶段可能出现的提示符 self.login_prompts = login_prompts or [ b"Username:", b"username:", b"login:", b"User name:" ] # 命令执行结束后,设备提示符的正则模式 self.cmd_prompt = cmd_prompt or r"<[A-Za-z0-9_\-\/]+>" self.tn = None def connect(self): """ 建立 Telnet 连接并完成登录认证 """ try: self.tn = telnetlib.Telnet(self.host, self.port, timeout=self.timeout) except Exception as e: raise ConnectionError(f"无法连接到 {self.host}:{self.port},原因:{e}") # 1. 匹配用户名提示符 index, text = self._expect_multiple(self.login_prompts) if index == -1: self.tn.close() raise TimeoutError("登录超时:未识别到用户名提示符") # 2. 发送用户名 self.tn.write(self.username.encode(self.encoding) + b"\n") # 3. 等待密码提示符(一般紧跟用户名提示符之后) index, text = self._expect_multiple([b"Password:", b"password:"]) if index == -1: self.tn.close() raise TimeoutError("登录超时:未识别到密码提示符") # 4. 发送密码 self.tn.write(self.password.encode(self.encoding) + b"\n") # 5. 等待进入命令行界面 time.sleep(0.5) banner = self._safe_read() if not re.search(self.cmd_prompt, banner, re.MULTILINE): # 可能是登录失败,读到的是报错信息 raise PermissionError(f"登录失败,请检查用户名密码。设备返回:{banner}") return self def execute(self, command: str, wait_time: float = 1.0) -> str: """ 执行单条命令,返回完整输出 """ self._ensure_connected() # 清除缓冲区残留,保证命令输出干净 self._safe_read() self.tn.write(command.encode(self.encoding) + b"\n") output = "" deadline = time.time() + self.timeout while time.time() < deadline: time.sleep(0.3) output += self._safe_read() # 如果看到命令行提示符,说明这条命令已经执行完毕 if re.search(self.cmd_prompt, output, re.MULTILINE): break return output def execute_batch(self, commands: list, wait_time: float = 0.8) -> dict: """ 批量执行命令,返回 {命令: 输出} 字典 """ results = {} for cmd in commands: results[cmd] = self.execute(cmd, wait_time) return results def logout(self): """ 发送退出命令并关闭连接 """ if self.tn: try: self.tn.write(b"quit\n") time.sleep(0.5) except Exception: pass finally: self.tn.close() self.tn = None # ---------- 内部辅助方法 ---------- def _expect_multiple(self, patterns, timeout: float = None): timeout = timeout or self.timeout try: index, matched, text = self.tn.expect(patterns, timeout=timeout) return index, text.decode(self.encoding, errors="ignore") except Exception: return -1, "" def _safe_read(self): """ 读取当前缓冲区全部内容,不抛异常 """ try: data = self.tn.read_very_eager() return data.decode(self.encoding, errors="ignore") except Exception: return "" def _ensure_connected(self): if not self.tn: raise RuntimeError("连接已关闭,请先调用 connect()")

这个类解决了我上面说的几个痛点:

  • login_prompts可配置,不同设备传不同的提示符列表
  • cmd_prompt用正则匹配设备返回的命令行提示符,用它来判断命令是否执行完毕
  • execute用循环读缓冲的方式,直到看到提示符为止,避免硬编码 sleep 导致输出截断
  • 析构函数里加安全退出,避免程序异常时连接泄漏

4.2 怎么用这个类:一个完整的巡检示例

下面我模拟一个真实场景:批量登录一批路由器,采集系统版本和当前配置,把结果保存到文件。

import json import time from router_telnet import RouterTelnetClient DEVICES = [ {"host": "192.168.10.1", "username": "admin", "password": "admin123"}, {"host": "192.168.10.2", "username": "admin", "password": "admin123"}, {"host": "192.168.10.3", "username": "admin", "password": "admin123"}, ] COMMANDS = [ "display version", "display current-configuration", ] def main(): all_results = {} for device in DEVICES: host = device["host"] print(f"正在处理 {host} ...") client = RouterTelnetClient( host=host, username=device["username"], password=device["password"], ) try: client.connect() results = client.execute_batch(COMMANDS) all_results[host] = results # 单独把配置信息保存到文件 config = results.get("display current-configuration", "") with open(f"config_{host}.txt", "w", encoding="utf-8") as f: f.write(config) except Exception as e: print(f"{host} 处理失败:{e}") all_results[host] = {"error": str(e)} finally: client.logout() # 汇总结果写入 JSON with open("inspection_results.json", "w", encoding="utf-8") as f: json.dump(all_results, f, ensure_ascii=False, indent=2) print("全部设备处理完成,结果已保存。") if __name__ == "__main__": main()

跑完之后,每个设备的配置都单独存成了一个文件,巡检结果汇总在一个 JSON 里,后面要做差异对比、配置备份、合规检查,拿着这些数据就能直接干活。

我在实际巡检中还会对配置做一件事:匿名化处理。配置里往往含有很多敏感信息,比如拨号账号、认证密钥、公网 IP。把原始配置直接存档在本地机器上有风险,所以我通常会在脚本里加一道脱敏逻辑,把密码字段、密钥串替换成***,只保留结构信息。具体做法就是用正则匹配敏感关键字所在行然后打码。

4.3 为什么不用 sleep 硬等,而是循环判断提示符

这是我想单独拎出来讲的一个设计细节,因为它决定了一个自动化脚本靠不靠谱。

最粗暴的做法是:发送命令后time.sleep(5),然后把缓冲区内容全读出来。这在设备响应速度稳定的模拟环境里没问题,但在真实设备上会很尴尬——有的路由器 CPU 跑满时,执行display current-configuration可能要十几秒,而display version一秒就回来了。如果你统一等 5 秒,快命令浪费时间,慢命令输出被截断。

循环判断提示符的做法,本质上是在"轮询"输出缓冲区:

  1. 发命令
  2. 等一小段(我一般用 0.3 秒)
  3. 读一次缓冲区,把内容追加到 output
  4. 检查 output 里有没有出现设备提示符(比如<Router>)
  5. 出现了就说明命令执行完了,没出现就继续等

这样不管设备响应是快是慢,都能拿到完整输出,而且耗时自适应。代价是稍微多一点 CPU 开销,但对脚本来说完全无所谓。

这里有一个小坑要提醒:有些命令的输出很长,比如display current-configuration在配置量大的设备上可能有几千行,输出过程可能超过 Telnet 客户端的行缓冲限制。华为的设备默认开启分页显示,一屏满了会停下来等用户按空格或回车继续。如果你的脚本不做处理,就会卡在那里读不到后续内容。解决办法是进入设备视图后先执行一条关闭分页的命令:

screen-length 0 temporary

或者:

undo terminal monitor

不同厂商关闭分页的命令不一样:

厂商关闭分页命令
华为 / 华三screen-length 0 temporary
思科terminal length 0
锐捷terminal length 0
中兴screen-length 0

我在execute_batch里会把关闭分页的命令自动插到最前面,保证长命令输出不被截断。

5. 进阶:基于 socket 自研 Telnet 客户端,彻底摆脱 telnetlib 的局限

5.1 什么时候需要自研

telnetlib 用起来很简单,但它有几个让我不太舒服的地方:

  1. 它的expect实现效率不高。每次匹配是对缓冲区做正则或字符串搜索,在输出量很大的时候会拖慢整体速度。
  2. 对二进制控制序列处理不方便。有些设备会在输出里夹带终端控制字符(比如光标移动、颜色转义),telnetlib 原样返回,你得自己做清理。
  3. 调试困难。如果希望看到客户端和服务端之间传递的每一个字节,telnetlib 没有提供方便的钩子。
  4. Python 3.13 之后它被弃用了。长期项目不想留技术债。

所以在一个需要长期维护的采集服务里,我基于socket写了一套轻量级的 Telnet 客户端,核心就几十行,但每一项行为都能控制,排查问题非常痛快。

5.2 最简实现:能登录、能执行命令、能拿输出

import socket import time import re class SimpleTelnet: """ 基于 socket 的最小 Telnet 客户端: 支持登录、命令执行、输出采集 """ def __init__(self, host: str, port: int = 23, timeout: float = 10, encoding: str = "utf-8"): self.host = host self.port = port self.timeout = timeout self.encoding = encoding self.sock = None def connect(self, username: str, password: str, login_prompt=b"Username:", pass_prompt=b"Password:", cmd_prompt_re=r"<[\w\-/]+>"): # 1. 建立 TCP 连接 self.sock = socket.create_connection((self.host, self.port), timeout=self.timeout) # 2. 等待用户名提示符 buf = self._read_until(login_prompt) if not buf: raise TimeoutError("未收到用户名提示符") # 3. 发送用户名 self.sock.sendall(username.encode(self.encoding) + b"\n") # 4. 等待密码提示符 buf = self._read_until(pass_prompt) if not buf: raise TimeoutError("未收到密码提示符") # 5. 发送密码 self.sock.sendall(password.encode(self.encoding) + b"\n") # 6. 等待进入命令行界面(匹配设备提示符) self.cmd_prompt_re = cmd_prompt_re buf = self._read_until_prompt(cmd_prompt_re, timeout=8) if not re.search(cmd_prompt_re, buf, re.MULTILINE): raise PermissionError("登录失败,返回内容:" + buf) return self def execute(self, command: str, wait_between_reads: float = 0.2): """ 执行命令并返回完整输出(直到出现提示符) """ # 先清空缓冲区 self._drain() self.sock.sendall(command.encode(self.encoding) + b"\n") output = b"" deadline = time.time() + self.timeout while time.time() < deadline: time.sleep(wait_between_reads) chunk = self._recv() if not chunk: continue output += chunk # 判断提示符是否出现 text = output.decode(self.encoding, errors="ignore") if re.search(self.cmd_prompt_re, text, re.MULTILINE): break return output.decode(self.encoding, errors="ignore") def send_command_raw(self, data: bytes): """ 发送原始字节,用于非标准交互 """ self.sock.sendall(data) def close(self): if self.sock: self.sock.close() self.sock = None # ---------- 内部方法 ---------- def _recv(self, bufsize: int = 4096): try: return self.sock.recv(bufsize) except socket.timeout: return b"" except OSError: return b"" def _drain(self): """清空缓冲区残余数据""" self.sock.setblocking(False) try: while True: data = self.sock.recv(4096) if not data: break except BlockingIOError: pass finally: self.sock.setblocking(True) def _read_until(self, pattern: bytes, timeout: float = 10): buf = b"" deadline = time.time() + timeout while time.time() < deadline: chunk = self._recv() if not chunk: time.sleep(0.1) continue buf += chunk if pattern in buf: return buf return buf def _read_until_prompt(self, pattern_re: str, timeout: float = 8): buf = b"" deadline = time.time() + timeout while time.time() < deadline: chunk = self._recv() if not chunk: time.sleep(0.1) continue buf += chunk text = buf.decode(self.encoding, errors="ignore") if re.search(pattern_re, text, re.MULTILINE): return buf return buf

这套简版实现的思路是:

  • _read_until按字节累积缓冲区,直到目标模式出现
  • _read_until_prompt用正则匹配设备提示符
  • execute拿"固定轮询 + 正则判断"的方式采集完整输出
  • _drain用非阻塞模式清空缓冲区,保证每次命令的输出起点干净

它比 telnetlib 少了很多内部状态管理,唯一的缺点是需要自己处理socket.timeout、BlockingIOError这些异常分支,代码稍微多一点,但换来的是完全可控。

5.3 实测对比:telnetlib 和自研 socket 性能差异

我拿同一台设备,执行 200 次display version命令,分别用两种方式测了一下耗时:

实现方式总耗时平均单次耗时备注
telnetlib + read_until约 12.6s约 63ms设备响应快时差距不大
自研 socket + 轮询约 10.8s约 54ms少了库内部状态检查开销

说句公道话,这种差距在实际场景里感知不明显,除非你单次任务要执行上千条命令。所以选择哪个方案更多取决于你要不要掌控底层细节、未来要不要扩展协议处理。如果只是写个一次性脚本,telnetlib 完全够用。

6. 从"能连上"到"能干活":完整自动化巡检的落地细节

6.1 命令列表的正确组织方式

批量执行不能简单地把所有命令丢在列表里就完事。我在实践中把命令分成几类,每一类的处理逻辑略有不同:

命令类型特点处理方式
配置类命令只读,输出长,可能分页先关分页,再执行,输出存文件
状态类命令快速返回,输出结构化程度高直接执行,解析关键字段
配置修改类有副作用,可能产生业务影响单独封装,加确认机制,慎用
特权命令需要进入 enable 模式或特权视图先执行特权切换,再做后续操作

比如在一台华为路由器上,如果我要采集 OSPF 邻居状态、接口流量、路由表,命令组合大致如下:

BASIC_INFO_COMMANDS = [ "display version", "display device", "display interface brief", ] ROUTING_COMMANDS = [ "display ospf peer brief", "display ip routing-table", "display bgp peer", ] CONFIG_COMMANDS = [ "display current-configuration", ]

执行的时候按分组执行,每组结果分别存文件,后面解析的时候也方便。

6.2 输出解析:用正则提取关键信息

把命令输出拿到手只是第一步,真正的价值在于从输出里提取结构化数据。比如我要从display version的输出里拿到设备的软件版本和运行时间:

import re VERSION_OUTPUT = """ Huawei Versatile Routing Platform Software VRP (R) software, Version 5.130 (AR2220 V200R008C20SPC500) Copyright (C) 2012-2018 Huawei Technologies Co., Ltd. AR2220 uptime is 30 weeks, 4 days, 18 hours, 35 minutes """ version_match = re.search(r"Version\s+([^\s]+)", VERSION_OUTPUT) uptime_match = re.search(r"uptime is (.+)", VERSION_OUTPUT) print("版本:", version_match.group(1) if version_match else "未识别") print("运行时长:", uptime_match.group(1).strip() if uptime_match else "未识别")

输出效果:

版本: 5.130 运行时长: 30 weeks, 4 days, 18 hours, 35 minutes

再比如从接口摘要输出里提取所有处于 down 状态的接口:

INTERFACE_OUTPUT = """ GigabitEthernet0/0/0 up up 100Mbps 100Mbps 0.00% GigabitEthernet0/0/1 up down 100Mbps 100Mbps 0.00% GigabitEthernet0/0/2 down down 0Kbps 0Kbps 0.00% """ down_interfaces = re.findall(r"(\S+)\s+up\s+down", INTERFACE_OUTPUT) all_down = re.findall(r"(\S+)\s+down\s+down", INTERFACE_OUTPUT) print("协议down接口:", down_interfaces) print("物理down接口:", all_down)

这类正则解析写起来不难,但需要你对手动输出格式非常熟悉。我的建议是:先手动执行命令拿到几份不同设备的输出样例,再针对样例写正则,最后拿其他设备输出做验证。别对着文档猜格式,文档和实际输出经常对不上。

6.3 多设备并发:线程池加速巡检

如果是几十台上百台设备,串行执行太慢了。这一步我用concurrent.futures.ThreadPoolExecutor做并发,提速非常明显。

from concurrent.futures import ThreadPoolExecutor, as_completed def process_one_device(device): host = device["host"] client = RouterTelnetClient(host=host, username=device["username"], password=device["password"]) try: client.connect() results = {} for cmd in COMMANDS: results[cmd] = client.execute(cmd) return host, results except Exception as e: return host, {"error": str(e)} finally: client.logout() with ThreadPoolExecutor(max_workers=8) as executor: future_map = {executor.submit(process_one_device, dev): dev["host"] for dev in DEVICES} for future in as_completed(future_map): host, results = future.result() print(f"{host}: 完成")

这里要注意两个点:

  1. max_workers别开太大。设备 CPU 和带宽有限,并发太高容易把设备打挂。我一般用 4 到 8。对核心设备我甚至严格串行。
  2. 每个线程持有独立的 Telnet 连接。telnetlib.Telnet对象不是线程安全的,千万别共用一个连接。上面代码里每个线程内部创建了自己的 client,没问题。

6.4 异常处理与重试机制

不管脚本写得再稳,网络环境总会出幺蛾子。断连、超时、设备重启、认证失败,情况多种多样。我总结了一套异常处理策略:

import time MAX_RETRY = 3 RETRY_DELAY = 2 def connect_with_retry(client, max_retry=MAX_RETRY, retry_delay=RETRY_DELAY): for attempt in range(1, max_retry + 1): try: client.connect() return True except Exception as e: if attempt == max_retry: raise print(f"第 {attempt} 次连接失败:{e},{retry_delay} 秒后重试...") time.sleep(retry_delay) return False

从经验上看,一次连接失败往往是因为设备刚好在处理其他任务、临时拒绝连接,重试一次大概率能通。但三次都不通的话,基本可以判断是网络或者账号问题,没必要再耗下去。

对于命令执行阶段的异常,我的处理策略是:

  • 如果某条命令超时,把它记录到错误清单,继续执行后续命令
  • 如果连接中断,停止当前设备的后续命令,记录失败原因
  • 最后把成功和失败结果分别汇总,方便分析和补救

7. 关于 Telnet 安全性的坦率提醒

说到 Telnet 绕不开安全问题。Telnet 是明文协议,所有数据——包括用户名、密码、命令行内容——都是裸奔在网络上传输的。任何能在链路上抓包的人都能轻易看到你的设备账号密码。

我在实际项目中只在下面这些场景用 Telnet:

  • 网络环境是隔离的内网,没有外部访问路径
  • 设备是测试设备或者模拟器里的虚拟设备
  • 一次性应急操作,账号权限受限,操作完立刻改密码

如果你要管理的设备是生产环境的公网设备,我会说:别用 Telnet,别用 Telnet,别用 Telnet。重要的话说三遍。这种情况下请务必使用 SSH,Paramiko 和 Netmiko 都很成熟,Netmiko 还内置了非常友好的多厂商支持。

这也是为什么我在封装 Telnet 客户端时,把真实项目的敏感数据全部做了脱敏处理,并在文档里明确标注了这个安全隐患。工具没有好坏,关键看你怎么用。

8. 我踩过的坑和留给大家的实战建议

8.1 坑一:Windows 下 telnetlib 读中文乱码

有一台设备的欢迎语和控制台输出带中文,直接用utf-8解码乱码。后来发现设备用的是 GBK 编码。解决办法:解码时优先utf-8,失败后回退gbk。

def smart_decode(data: bytes) -> str: try: return data.decode("utf-8") except UnicodeDecodeError: return data.decode("gbk", errors="ignore")

在真实网络设备里,中文环境常见编码是 GBK,UTF-8 反而少见。你在写脚本时最好先确认设备控制台的编码设置。

8.2 坑二:read_until卡死导致脚本挂起

这个问题我在第一次写巡检脚本时就遇到了。设备登录提示符是Username:没错,但read_until一直等不到,最后超时中断。排查才发现,真实设备在发送提示符之前,先发送了一串 ANSI 颜色控制序列,类似\x1b[1m\x1b[31mUsername:,而我的匹配串只写了Username:。解决办法是用read_until(b"Username:")改为expect([b"Username:", b"\x1b[Username:"]),或者干脆把提示符正则化。

8.3 坑三:分页输出导致命令执行"假完成"

如果不关闭分页,执行长命令时,设备输出一屏后就停下来,我的脚本判断正则提示符没出现,就一直等,直到超时。后来在execute_batch里自动补上screen-length 0 temporary才解决。

这里再补充一个细节:华为的screen-length 0 temporary只在当前登录会话有效,不会写进配置,所以不影响设备后续行为。思科的terminal length 0同理。这个设计很友好,不需要担心自动化脚本会永久修改设备参数。

8.4 建议一:把设备信息放到配置文件里

不要把设备 IP、账号密码硬编码在代码里。我用 YAML 配置来管理,每个设备一个条目,包括厂商、型号、登录提示符等信息:

devices: - host: 192.168.10.1 username: admin password: admin123 vendor: huawei model: AR2220 type: router login_prompt: "Username:" password_prompt: "Password:" cmd_prompt: "<AR2220>"

这样新增设备的时候只改配置文件,不动代码。

8.5 建议二:脚本里加日志和状态输出

批量任务跑起来之后,控制台输出要能清晰反映每个设备的状态。我通常用 logging 模块,分文件记录每个设备的执行结果和错误日志:

import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("telnet_script.log", encoding="utf-8"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) logger.info(f"开始处理设备 {host}")

日志的好处是任务挂了之后能回溯,不至于两眼一抹黑。

8.6 建议三:先写死再抽象,避免过度设计

最后一条建议可能和很多人的预期相反:不要一开始就写一个万能抽象类,先针对某一台设备把脚本跑通,再来提取公共逻辑。我见过太多人第一步就在设计"适配全球所有厂商设备"的框架,结果一个月过去了连一台设备都还没跑通。

正确姿势是:

  1. 手动连一次设备,记录交互流程
  2. 写一个只针对这台设备的最小脚本
  3. 跑通之后再加第二台设备,寻找共同点,抽象公共方法
  4. 等设备种类多了(比如超过三种),再考虑引入配置驱动或插件机制

Python 网络自动化这条路,从来不是靠一蹴而就的设计能力,而是靠一台一台设备"喂"出来的经验。Telnet 看似老套,但作为理解设备交互的第一课,依然是最直观的起点。等你把 Telnet 的读写循环、提示符匹配、超时处理这套思维模型建立起来,再学 Netmiko、Paramiko、甚至 Ansible 的网络模块,都会顺畅得多。希望这篇文章能帮你把这条路走得稳一点。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询