How to read telnet output with telnetlib3 and python?

Question

Using python 3.8.10 and telnetlib I want to connect to a host and just print out the message the host is sending upon connection. I have been trying the following code:

import telnetlib3
import asyncio
async def run_telnet_session():
    total_output = ""
    # Establish a Telnet connection with timeout
    reader, writer = await telnetlib3.open_connection("192.168.200.10", 9000)
    
    try:
        # Read the Telnet output
        while True:
            output = await reader.read(4096)  
            if output:
                print(output)
                total_output += output
            else:
                break  
    
    finally:
        writer.close()
        await writer.wait_closed()
    return total_output

output = asyncio.run(run_telnet_session())
print(output)

and I get the expected output, but the code is blocked! The function never returns. How to change the code so there is a timeout and the function returns the string?

Answer 1

The code you provided is attempting to use the telnetlib3 library, which is not a standard library in Python. It seems that you intended to use the telnetlib library, which is the built-in Telnet client library in Python.

import telnetlib
import time

def run_telnet_session():
    total_output = ""
    # Establish a Telnet connection
    tn = telnetlib.Telnet("192.168.200.10", 9000, timeout=2)
    
    try:
        # Read the Telnet output
        while True:
            output = tn.read_very_eager().decode('utf-8')
            if output:
                total_output += output
                print(output)
                # Reset the timer if new data is received
                start_time = time.time()
            else:
                # Check if no new data received for 2 seconds
                if time.time() - start_time > 2:
                    break
    
    finally:
        tn.close()
    
    return total_output

output = run_telnet_session()
print(output)

In this code, the read_very_eager() method is used to read data from the Telnet connection without blocking. The received data is printed and appended to total_output. The timer (start_time) is reset whenever new data is received, and if no new data is received for 2 seconds, the loop is terminated.

Make sure to replace "192.168.200.10" and 9000 with the correct host IP address and port number you want to connect to.

Answer 2

The solution seems to be that you wrap the reading function into a wait_for like follows:

output = await asyncio.wait_for(reader.read(4096), timeout=2)

That construct now seems to wait for 2 seconds if there is no further output from the read method...

So complete code is

async def telnet_welcome_message(host, port, timeout=2):
    output = ""
    # Establish a Telnet connection with timeout
    reader, writer = await telnetlib3.open_connection(host, port)
        
    try:
        # Read the Telnet output
        while True:
            data = await asyncio.wait_for(reader.read(4096), timeout=timeout)
            if data:
                output += data
    except asyncio.exceptions.TimeoutError:
        # Timeout occurred -> Exiting
        pass
    finally:
        reader.close()
        writer.close()
    return output

message = asyncio.run(telnet_welcome_message(HOST, PORT))
print(message)