#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
""" A telnet server which negotiates"""
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import argparse
import os
import sys
import logging
try: # Python 2
import SocketServer as socketserver
except ImportError: # Python 3
import socketserver
log = logging.getLogger(__name__)
HOST = "localhost"
IDENT = "NTEL"
# The strings that indicate the test framework is checking our aliveness
VERIFIED_REQ = "verifiedserver"
VERIFIED_RSP = "WE ROOLZ: {pid}"
def telnetserver(options):
"""
Starts up a TCP server with a telnet handler and serves DICT requests
forever.
"""
if options.pidfile:
pid = os.getpid()
with open(options.pidfile, "w") as f:
f.write(str(pid))
local_bind = (HOST, options.port)
log.info("Listening on %s", local_bind)
# Need to set the allow_reuse on the class, not on the instance.
socketserver.TCPServer.allow_reuse_address = True
server = socketserver.TCPServer(local_bind, NegotiatingTelnetHandler)
server.serve_forever()
return ScriptRC.SUCCESS
class NegotiatingTelnetHandler(socketserver.BaseRequestHandler):
"""Handler class for Telnet connections.
"""
def handle(self):
"""
Negotiates options before reading data.
"""
neg = Negotiator(self.request)
try:
# Send some initial negotiations.
neg.send_do("NEW_ENVIRON")
neg.send_will("NEW_ENVIRON")
neg.send_dont("NAWS")
neg.send_wont("NAWS")
# Get the data passed through the negotiator
data = neg.recv(1024)
log.debug("Incoming data: %r", data)
if VERIFIED_REQ.encode('ascii') in data:
log.debug("Received verification request from test framework")
response = VERIFIED_RSP.format(pid=os.getpid())
response_data = response.encode('ascii')
else:
log.debug("Received normal request - echoing back")
response_data = data.strip()
if response_data:
log.debug("Sending %r", response_data)
self.request.sendall(response_data)
except IOError:
log.exception("IOError hit during request")
class Negotiator(object):
NO_NEG = 0
START_NEG = 1
WILL = 2
WONT = 3
DO = 4
DONT = 5
def __init__(self, tcp):
self.tcp = tcp
self.state = self.NO_NEG
def recv(self, bytes):
"""
Read bytes from TCP, handling negotiation sequences
:param bytes: Number of bytes to read
:return: a buffer of bytes
"""
buffer = bytearray()
# If we keep receiving negotiation sequences, we won't fill the buffer.
# Keep looping while we can, and until we have something to give back
# to the caller.
while len(buffer) == 0:
data = self.tcp.recv(bytes)
if not data:
# TCP failed to give us any data. Break out.
break
for byte_int in bytearray(data):
if self.state == self.NO_NEG:
self.no_neg(byte_int, buffer)
elif self.state == self.START_NEG:
self.start_neg(byte_int)
elif self.state in [self.WILL, self.WONT, self.DO, self.DONT]:
self.handle_option(byte_int)
else:
# Received an unexpected byte. Stop negotiations
log.error("Unexpected byte %s in state %s",
byte_int,
self.state)
self.state = self.NO_NEG
return buffer
def no_neg(self, byte_int, buffer):
# Not negotiating anything thus far. Check to see if we
# should.
if byte_int == NegTokens.IAC:
# Start negotiation
log.debug("Starting negotiation (IAC)")
self.state = self.START_NEG
else:
# Just append the incoming byte to the buffer
buffer.append(byte_int)
def start_neg(self, byte_int):
# In a negotiation.
log.debug("In negotiation (%s)",
NegTokens.from_val(byte_int))
if byte_int == NegTokens.WILL:
# Client is confirming they are willing to do an option
log.debug("Client is willing")
self.state = self.WILL
elif byte_int == NegTokens.WONT:
# Client is confirming they are unwilling to do an
# option
log.debug("Client is unwilling")
self.state = self.WONT
elif byte_int == NegTokens.DO:
# Client is indicating they can do an option
log.debug("Client can do")
self.state = self.DO
elif byte_int == NegTokens.DONT:
# Client is indicating they can't do an option
log.debug("Client can't do")
self.state = self.DONT
else:
# Received an unexpected byte. Stop negotiations
log.error("Unexpected byte %s in state %s",
byte_int,
self.state)
self.state = self.NO_NEG
def handle_option(self, byte_int):
if byte_int in [NegOptions.BINARY,
NegOptions.CHARSET,
NegOptions.SUPPRESS_GO_AHEAD,
NegOptions.NAWS,
NegOptions.NEW_ENVIRON]:
log.debug("Option: %s", NegOptions.from_val(byte_int))
# No further negotiation of this option needed. Reset the state.
self.state = self.NO_NEG
else:
# Received an unexpected byte. Stop negotiations
log.error("Unexpected byte %s in state %s",
byte_int,
self.state)
self.state = self.NO_NEG
def send_message(self, message_ints):
self.tcp.sendall(bytearray(message_ints))
def send_iac(self, arr):
message = [NegTokens.IAC]
message.extend(arr)
self.send_message(message)
def send_do(self, option_str):
log.debug("Sending DO %s", option_str)
self.send_iac([NegTokens.DO, NegOptions.to_val(option_str)])
def send_dont(self, option_str):
log.debug("Sending DONT %s", option_str)
self.send_iac([NegTokens.DONT, NegOptions.to_val(option_str)])
def send_will(self, option_str):
log.debug("Sending WILL %s", option_str)
self.send_iac([NegTokens.WILL, NegOptions.to_val(option_str)])
def send_wont(self, option_str):
log.debug("Sending WONT %s", option_str)
self.send_iac([NegTokens.WONT, NegOptions.to_val(option_str)])
class NegBase(object):
@classmethod
def to_val(cls, name):
return getattr(cls, name)
@classmethod
def from_val(cls, val):
for k in cls.__dict__.keys():
if getattr(cls, k) == val:
return k
return "<unknown>"
class NegTokens(NegBase):
# The start of a negotiation sequence
IAC = 255
# Confirm willingness to negotiate
WILL = 251
# Confirm unwillingness to negotiate
WONT = 252
# Indicate willingness to negotiate
DO = 253
# Indicate unwillingness to negotiate
DONT = 254
# The start of sub-negotiation options.
SB = 250
# The end of sub-negotiation options.
SE = 240
class NegOptions(NegBase):
# Binary Transmission
BINARY = 0
# Suppress Go Ahead
SUPPRESS_GO_AHEAD = 3
# NAWS - width and height of client
NAWS = 31
# NEW-ENVIRON - environment variables on client
NEW_ENVIRON = 39
# Charset option
CHARSET = 42
def get_options():
parser = argparse.ArgumentParser()
parser.add_argument("--port", action="store", default=9019,
type=int, help="port to listen on")
parser.add_argument("--verbose", action="store", type=int, default=0,
help="verbose output")
parser.add_argument("--pidfile", action="store",
help="file name for the PID")
parser.add_argument("--logfile", action="store",
help="file name for the log")
parser.add_argument("--srcdir", action="store", help="test directory")
parser.add_argument("--id", action="store", help="server ID")
parser.add_argument("--ipv4", action="store_true", default=0,
help="IPv4 flag")
return parser.parse_args()
def setup_logging(options):
"""
Set up logging from the command line options
"""
root_logger = logging.getLogger()
add_stdout = False
formatter = logging.Formatter("%(asctime)s %(levelname)-5.5s "
"[{ident}] %(message)s"
.format(ident=IDENT))
# Write out to a logfile
if options.logfile:
handler = logging.FileHandler(options.logfile, mode="w")
handler.setFormatter(formatter)
handler.setLevel(logging.DEBUG)
root_logger.addHandler(handler)
else:
# The logfile wasn't specified. Add a stdout logger.
add_stdout = True
if options.verbose:
# Add a stdout logger as well in verbose mode
root_logger.setLevel(logging.DEBUG)
add_stdout = True
else:
root_logger.setLevel(logging.INFO)
if add_stdout:
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setFormatter(formatter)
stdout_handler.setLevel(logging.DEBUG)
root_logger.addHandler(stdout_handler)
class ScriptRC(object):
"""Enum for script return codes"""
SUCCESS = 0
FAILURE = 1
EXCEPTION = 2
class ScriptException(Exception):
pass
if __name__ == '__main__':
# Get the options from the user.
options = get_options()
# Setup logging using the user options
setup_logging(options)
# Run main script.
try:
rc = telnetserver(options)
except Exception as e:
log.exception(e)
rc = ScriptRC.EXCEPTION
log.info("Returning %d", rc)
sys.exit(rc)