Python telnetlib3 示例

3
我想了解如何在简单场景下使用telnetlib3。
长期以来,telnetlib(不是3)在https://docs.python.org/3/library/telnetlib.html有一个简单的示例,其中python程序连接到telnet服务器,然后查找提示并提供响应。人们可以很容易地看到如何将此示例扩展到不同的提示,添加超时并添加更多的提示-响应步骤。
import getpass
import telnetlib

HOST = "localhost"
user = input("Enter your remote account: ")
password = getpass.getpass()

tn = telnetlib.Telnet(HOST)

tn.read_until(b"login: ")
tn.write(user.encode('ascii') + b"\n")
if password:
    tn.read_until(b"Password: ")
    tn.write(password.encode('ascii') + b"\n")

tn.write(b"ls\n")
tn.write(b"exit\n")

print(tn.read_all().decode('ascii'))

然而,telnetlib(不是3)已被弃用。

替代方案telnetlib3(https://telnetlib3.readthedocs.io/en/latest/intro.html#quick-example)基于asyncio提供了一个示例,并且异步的“shell”函数(与服务器交互)会阻塞等待提示符(异步的理由),并始终使用“y”响应服务器。

import asyncio, telnetlib3

async def shell(reader, writer):
    while True:
        # read stream until '?' mark is found
        outp = await reader.read(1024)
        if not outp:
            # End of File
            break
        elif '?' in outp:
            # reply all questions with 'y'.
            writer.write('y')

        # display all server output
        print(outp, flush=True)

    # EOF
    print()

loop = asyncio.get_event_loop()
coro = telnetlib3.open_connection('localhost', 6023, shell=shell)
reader, writer = loop.run_until_complete(coro)
loop.run_until_complete(writer.protocol.waiter_closed)

我在如何使结构化的代码执行主流任务方面缺少一些线索,这在(字面上!)直接的telnetlib(非3)示例中得到了展示,其中服务器提供一系列不同的提示,而Python程序则提供相应的响应。我怀疑这部分是由于我对asyncio的陌生和应该使用哪些代码模式来使异步函数执行一系列步骤。

因此,看到以这种方式实现的telnetlib(非3)示例将是一个很大的帮助。

2个回答

5

我认为把 telnetlib3 称为 telnetlib 的“替代品”有点牵强。虽然它允许你编写 Telnet 客户端(或服务器),但它确实是一个完全不同的东西。对于在最初的 telnetlib 示例中所做的类似工作,我通常会选择 pexpect(或者只是普通的 expect)。

看起来 PEP 594 也将 Exscript 视为一种解决方案,它看起来比 telnetlib3 更接近 telnetlib


这里是我编写的一个示例,使用telnetlib3连接到本地交换机,登录,发送enable命令,然后退出:

import asyncio
import telnetlib3

async def shell(reader, writer):
    rules = [
            ('User:', 'admin'),
            ('Password:', 'secret'),
            (') >', 'enable'),
            (') #', 'exit'),
            (') >', 'logout'),
            ]

    ruleiter = iter(rules)
    expect, send = next(ruleiter)
    while True:
        outp = await reader.read(1024)
        if not outp:
            break

        if expect in outp:
            writer.write(send)
            writer.write('\r\n')
            try:
                expect, send = next(ruleiter)
            except StopIteration:
                break

        # display all server output
        print(outp, flush=True)

    # EOF
    print()

async def main():
    reader, writer = await telnetlib3.open_connection('office-sw-0', 23, shell=shell)
    await writer.protocol.waiter_closed


if __name__ == '__main__':
    asyncio.run(main())

我不是很喜欢它。我更愿意做这样的事情:

import sys
import pexpect

rules = [
    ("User:", "admin"),
    ("Password:", "secret"),
    (r"\) >", "enable"),
    (r"\) #", "exit"),
    (r"\) >", "logout"),
    (pexpect.EOF, None),
]

client = pexpect.spawn("telnet office-sw-0")

# This is so we can see what's happening.
client.logfile = sys.stdout.buffer

for expect, send in rules:
    client.expect(expect)
    if send is None:
        break
    client.sendline(send)

2
感谢larsks的回答。我之前没有仔细看Exscript,因为它的文档显示需要Python 2.7(我使用3.x),而且它看起来对我的需求过于复杂。此外,虽然源代码中有注释,但没有任何API的文档。 但现在我发现Exscript也可能支持Python 3.x。值得注意的是,它包括一个名为telnetlib.py的模块,该模块看起来与旧的telnetlib.py密切相关,并包含一些相关的注释。不确定如何处理这个问题,以及是否可以仅使用该模块。 - gwideman
1
感谢您提供的asyncio和pexpect示例。您的asyncio示例朝着我认为可能需要的方向前进,但我不相信它会将与服务器的交互从主程序中“漂移”出去。我将不得不学习asyncio以更全面地了解多个任务如何相互通信和协调。 同时,我看到您的pexpect示例完成了工作,但代价是调用外部程序,并且不是跨平台的。但还是很好知道! - gwideman
exscript似乎没有得到维护,而pexpect在telnetlib3方面无法使用。 - Alex
Exscript似乎没有得到维护,而pexpect在telnetlib3上无法正常工作。 - undefined

2

对于简单的操作和响应应用程序,您可以使用telnetlib3编写代码,与telnetlib(没有3)和pexpect几乎没有区别,只是存在一些异步关键字。以下是通过telnet发送电子邮件的完整工作示例。希望对您有所帮助,不客气。

import asyncio
import re
import telnetlib3

rules = [
    ("220 .* SMTP", "HELO localhost"),
    ("250 Hello and welcome", "MAIL FROM:<from_email_address"),
    ("250 Ok", "RCPT TO:<to_email_address>"),
    ("250 Ok", "DATA"),
    ("354 Proceed", "<email header and body>"),
    ("250 Ok", "QUIT")
]

async def main():
    reader, writer = await telnetlib3.open_connection(host, port)

    for prompt, action in rules:
        response = await reader.readline()
        print(response.strip())

        if re.match(prompt, response):
            writer.write(action + "\n")
            print(action)

asyncio.run(main())

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接