使用 netmiko 的 is_alive() 方法实现 ssh 保活机制

发表于 分类于 阅读次数： 本文字数： 1.9k 阅读时长 ≈ 2 分钟

在使用 Netmiko 的过程中，出于某些原因（SSH 新建连接慢、VTY 数量限制等），需要把 Netmiko 已经连接到设备上的 SSH 会话保持住，防止超时。

网络设备的超时时间是按照最后一次接收到命令的时间开始算的（以华三设备为例），所以一般发送一个回车、空字符等等都可以实现保活。

Netmiko 本身提供了一个 is_alive() 方法来对会话进行验证，底层实现是通过发送 null 来做的。

简单记录一下验证过程：

网络设备配置：

1
2
3
4
line vty 0 63
 ...
 idle-timeout 0 10
#

测试代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from netmiko import ConnectHandler as ch
import time

host = {
    'device_type': 'hp_comware',
    'host': '192.168.56.20',
    'username': 'netdevops',
    'password': 'NetDevops@01',
}

conn = ch(**host)
# 设备配置 10s 超时，每隔 3s 发送 is_alive()，15s 后仍然可以执行命令
# 说明 is_alive() 这个方法发送的空字符可以保活
for i in range(0,5):
    print(conn.is_alive())
    time.sleep(3)
result = conn.send_command("dis arp")
print(result)
# 打印正常
# True
# True
# True
# True
# True
#   Type: S-Static   D-Dynamic   O-Openflow   R-Rule   M-Multiport  I-Invalid
# IP address       MAC address    SVLAN/VSI Interface/Link ID        Aging Type  
# 192.168.56.1     0a00-2700-0032 --        XGE1/0/49                16    D   

# 设备配置 10s 超时，sleep 11s 后，执行命令提示超时
time.sleep(11)
result = conn.send_command("dis arp")
print(result)
# 报错
#     raise socket.error("Socket is closed")
# OSError: Socket is closed

不同的设备实现机制可能会不一样，需要根据自己实际的情况进行验证。

Netmiko 的源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def is_alive(self):
    """Returns a boolean flag with the state of the connection."""
    null = chr(0)
    if self.remote_conn is None:
        log.error("Connection is not initialised, is_alive returns False")
        return False
    if self.protocol == "telnet":
        ...
    else:
        # SSH
        try:
            # Try sending ASCII null byte to maintain the connection alive
            log.debug("Sending the NULL byte")
            # 向 channel 发送一个空字符串，如果能正常发送成功，则说明会话还在
            # 网络设备的超时时间是按照最后一次接收到命令的时间开始算的
            # 所以每次进行 is_alive() 的判断，都会刷新 vty 的超时时间
            self.write_channel(null)
            return self.remote_conn.transport.is_active()
        except (socket.error, EOFError):
            log.error("Unable to send", exc_info=True)
            # If unable to send, we can tell for sure that the connection is unusable
            return False
    return False