如何使用Python与telnetlib3实现命令发送及交互式会话

先安装依赖：

pip install telnetlib3

以下是完整实现代码，满足你三个核心需求：初始化命令、交互式会话、结束后执行额外命令：

import asyncio
import sys
import telnetlib3

async def telnet_workflow(host, port, init_cmds, post_cmds, prompt="# "):
    # 建立Telnet连接
    reader, writer = await telnetlib3.open_connection(host, port)

    # 1. 执行初始化命令，等待每个命令完成
    print("=== 执行初始化命令 ===")
    for cmd in init_cmds:
        writer.write(f"{cmd}\r\n")
        await writer.drain()
        
        # 读取输出直到出现提示符，确保命令执行完成
        while True:
            output = await reader.read(4096)
            if not output:
                break
            sys.stdout.write(output)
            sys.stdout.flush()
            if prompt in output:
                break

    # 2. 启动交互式会话，绑定用户输入输出与Telnet连接
    print("\n=== 进入交互式会话，按Ctrl+D退出 ===")
    
    async def handle_user_input(writer):
        while True:
            try:
                # 读取用户输入（同步调用避免异步stdin的兼容性问题）
                user_input = await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)
                if not user_input:
                    break  # 用户输入EOF（Ctrl+D）
                writer.write(user_input)
                await writer.drain()
            except EOFError:
                break

    async def handle_server_output(reader):
        while True:
            output = await reader.read(4096)
            if not output:
                break
            sys.stdout.write(output)
            sys.stdout.flush()

    # 启动输入输出处理任务
    input_task = asyncio.create_task(handle_user_input(writer))
    output_task = asyncio.create_task(handle_server_output(reader))

    # 等待任一任务结束（用户退出或服务器断开）
    done, pending = await asyncio.wait([input_task, output_task], return_when=asyncio.FIRST_COMPLETED)
    # 清理未完成的任务
    for task in pending:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

    # 3. 执行结束后的额外命令
    print("\n=== 执行后续命令 ===")
    for cmd in post_cmds:
        writer.write(f"{cmd}\r\n")
        await writer.drain()
        
        # 读取输出直到提示符
        while True:
            output = await reader.read(4096)
            if not output:
                break
            sys.stdout.write(output)
            sys.stdout.flush()
            if prompt in output:
                break

    # 关闭连接
    writer.close()
    await writer.wait_closed()

if __name__ == "__main__":
    # 配置参数，根据实际情况修改
    TELNET_HOST = "192.168.1.1"
    TELNET_PORT = 23
    INIT_COMMANDS = ["enable", "configure terminal", "interface GigabitEthernet0/0"]
    POST_COMMANDS = ["exit", "write memory", "exit"]
    CLI_PROMPT = "# "  # 匹配你的服务器CLI提示符

    asyncio.run(telnet_workflow(TELNET_HOST, TELNET_PORT, INIT_COMMANDS, POST_COMMANDS, CLI_PROMPT))

关键细节说明： #

• 初始化命令处理：通过等待服务器返回的提示符来确保每个命令执行完成，避免命令发送过快导致的混乱，你需要根据目标设备的实际提示符修改CLI_PROMPT参数（比如普通用户模式用> ）。
• 交互式会话：用两个异步任务分别处理用户输入和服务器输出，实现和原生Telnet一致的交互体验；用户可以通过按Ctrl+D退出交互模式。
• 后续命令执行：交互结束后自动执行预设的收尾命令，同样通过提示符判断命令执行状态。
• 兼容性：使用run_in_executor处理标准输入，避免异步环境下stdin的兼容性问题，比直接用异步stdin更稳定。

扩展建议： #

• 如果需要处理登录验证，把用户名、密码加入INIT_COMMANDS即可，比如["admin", "yourpassword"]。
• 可以添加超时处理，比如用asyncio.wait_for包裹读取操作，避免无限等待。

内容的提问来源于stack exchange，提问作者Giles Knap