Annotation of embedaddon/curl/tests/negtelnetserver.py, revision 1.1.1.1
1.1 misho 1: #!/usr/bin/env python
2: # -*- coding: utf-8 -*-
3: #
4: # Project ___| | | | _ \| |
5: # / __| | | | |_) | |
6: # | (__| |_| | _ <| |___
7: # \___|\___/|_| \_\_____|
8: #
9: # Copyright (C) 2017 - 2020, Daniel Stenberg, <daniel@haxx.se>, et al.
10: #
11: # This software is licensed as described in the file COPYING, which
12: # you should have received as part of this distribution. The terms
13: # are also available at https://curl.haxx.se/docs/copyright.html.
14: #
15: # You may opt to use, copy, modify, merge, publish, distribute and/or sell
16: # copies of the Software, and permit persons to whom the Software is
17: # furnished to do so, under the terms of the COPYING file.
18: #
19: # This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
20: # KIND, either express or implied.
21: #
22: """ A telnet server which negotiates"""
23:
24: from __future__ import (absolute_import, division, print_function,
25: unicode_literals)
26: import argparse
27: import os
28: import sys
29: import logging
30: if sys.version_info.major >= 3:
31: import socketserver
32: else:
33: import SocketServer as socketserver
34:
35: log = logging.getLogger(__name__)
36: HOST = "localhost"
37: IDENT = "NTEL"
38:
39:
40: # The strings that indicate the test framework is checking our aliveness
41: VERIFIED_REQ = "verifiedserver"
42: VERIFIED_RSP = "WE ROOLZ: {pid}"
43:
44:
45: def telnetserver(options):
46: """
47: Starts up a TCP server with a telnet handler and serves DICT requests
48: forever.
49: """
50: if options.pidfile:
51: pid = os.getpid()
52: # see tests/server/util.c function write_pidfile
53: if os.name == "nt":
54: pid += 65536
55: with open(options.pidfile, "w") as f:
56: f.write(str(pid))
57:
58: local_bind = (HOST, options.port)
59: log.info("Listening on %s", local_bind)
60:
61: # Need to set the allow_reuse on the class, not on the instance.
62: socketserver.TCPServer.allow_reuse_address = True
63: server = socketserver.TCPServer(local_bind, NegotiatingTelnetHandler)
64: server.serve_forever()
65:
66: return ScriptRC.SUCCESS
67:
68:
69: class NegotiatingTelnetHandler(socketserver.BaseRequestHandler):
70: """Handler class for Telnet connections.
71:
72: """
73: def handle(self):
74: """
75: Negotiates options before reading data.
76: """
77: neg = Negotiator(self.request)
78:
79: try:
80: # Send some initial negotiations.
81: neg.send_do("NEW_ENVIRON")
82: neg.send_will("NEW_ENVIRON")
83: neg.send_dont("NAWS")
84: neg.send_wont("NAWS")
85:
86: # Get the data passed through the negotiator
87: data = neg.recv(1024)
88: log.debug("Incoming data: %r", data)
89:
90: if VERIFIED_REQ.encode('utf-8') in data:
91: log.debug("Received verification request from test framework")
92: pid = os.getpid()
93: # see tests/server/util.c function write_pidfile
94: if os.name == "nt":
95: pid += 65536
96: response = VERIFIED_RSP.format(pid=pid)
97: response_data = response.encode('utf-8')
98: else:
99: log.debug("Received normal request - echoing back")
100: response_data = data.decode('utf-8').strip().encode('utf-8')
101:
102: if response_data:
103: log.debug("Sending %r", response_data)
104: self.request.sendall(response_data)
105:
106: except IOError:
107: log.exception("IOError hit during request")
108:
109:
110: class Negotiator(object):
111: NO_NEG = 0
112: START_NEG = 1
113: WILL = 2
114: WONT = 3
115: DO = 4
116: DONT = 5
117:
118: def __init__(self, tcp):
119: self.tcp = tcp
120: self.state = self.NO_NEG
121:
122: def recv(self, bytes):
123: """
124: Read bytes from TCP, handling negotiation sequences
125:
126: :param bytes: Number of bytes to read
127: :return: a buffer of bytes
128: """
129: buffer = bytearray()
130:
131: # If we keep receiving negotiation sequences, we won't fill the buffer.
132: # Keep looping while we can, and until we have something to give back
133: # to the caller.
134: while len(buffer) == 0:
135: data = self.tcp.recv(bytes)
136: if not data:
137: # TCP failed to give us any data. Break out.
138: break
139:
140: for byte_int in bytearray(data):
141: if self.state == self.NO_NEG:
142: self.no_neg(byte_int, buffer)
143: elif self.state == self.START_NEG:
144: self.start_neg(byte_int)
145: elif self.state in [self.WILL, self.WONT, self.DO, self.DONT]:
146: self.handle_option(byte_int)
147: else:
148: # Received an unexpected byte. Stop negotiations
149: log.error("Unexpected byte %s in state %s",
150: byte_int,
151: self.state)
152: self.state = self.NO_NEG
153:
154: return buffer
155:
156: def no_neg(self, byte_int, buffer):
157: # Not negotiating anything thus far. Check to see if we
158: # should.
159: if byte_int == NegTokens.IAC:
160: # Start negotiation
161: log.debug("Starting negotiation (IAC)")
162: self.state = self.START_NEG
163: else:
164: # Just append the incoming byte to the buffer
165: buffer.append(byte_int)
166:
167: def start_neg(self, byte_int):
168: # In a negotiation.
169: log.debug("In negotiation (%s)",
170: NegTokens.from_val(byte_int))
171:
172: if byte_int == NegTokens.WILL:
173: # Client is confirming they are willing to do an option
174: log.debug("Client is willing")
175: self.state = self.WILL
176: elif byte_int == NegTokens.WONT:
177: # Client is confirming they are unwilling to do an
178: # option
179: log.debug("Client is unwilling")
180: self.state = self.WONT
181: elif byte_int == NegTokens.DO:
182: # Client is indicating they can do an option
183: log.debug("Client can do")
184: self.state = self.DO
185: elif byte_int == NegTokens.DONT:
186: # Client is indicating they can't do an option
187: log.debug("Client can't do")
188: self.state = self.DONT
189: else:
190: # Received an unexpected byte. Stop negotiations
191: log.error("Unexpected byte %s in state %s",
192: byte_int,
193: self.state)
194: self.state = self.NO_NEG
195:
196: def handle_option(self, byte_int):
197: if byte_int in [NegOptions.BINARY,
198: NegOptions.CHARSET,
199: NegOptions.SUPPRESS_GO_AHEAD,
200: NegOptions.NAWS,
201: NegOptions.NEW_ENVIRON]:
202: log.debug("Option: %s", NegOptions.from_val(byte_int))
203:
204: # No further negotiation of this option needed. Reset the state.
205: self.state = self.NO_NEG
206:
207: else:
208: # Received an unexpected byte. Stop negotiations
209: log.error("Unexpected byte %s in state %s",
210: byte_int,
211: self.state)
212: self.state = self.NO_NEG
213:
214: def send_message(self, message_ints):
215: self.tcp.sendall(bytearray(message_ints))
216:
217: def send_iac(self, arr):
218: message = [NegTokens.IAC]
219: message.extend(arr)
220: self.send_message(message)
221:
222: def send_do(self, option_str):
223: log.debug("Sending DO %s", option_str)
224: self.send_iac([NegTokens.DO, NegOptions.to_val(option_str)])
225:
226: def send_dont(self, option_str):
227: log.debug("Sending DONT %s", option_str)
228: self.send_iac([NegTokens.DONT, NegOptions.to_val(option_str)])
229:
230: def send_will(self, option_str):
231: log.debug("Sending WILL %s", option_str)
232: self.send_iac([NegTokens.WILL, NegOptions.to_val(option_str)])
233:
234: def send_wont(self, option_str):
235: log.debug("Sending WONT %s", option_str)
236: self.send_iac([NegTokens.WONT, NegOptions.to_val(option_str)])
237:
238:
239: class NegBase(object):
240: @classmethod
241: def to_val(cls, name):
242: return getattr(cls, name)
243:
244: @classmethod
245: def from_val(cls, val):
246: for k in cls.__dict__.keys():
247: if getattr(cls, k) == val:
248: return k
249:
250: return "<unknown>"
251:
252:
253: class NegTokens(NegBase):
254: # The start of a negotiation sequence
255: IAC = 255
256: # Confirm willingness to negotiate
257: WILL = 251
258: # Confirm unwillingness to negotiate
259: WONT = 252
260: # Indicate willingness to negotiate
261: DO = 253
262: # Indicate unwillingness to negotiate
263: DONT = 254
264:
265: # The start of sub-negotiation options.
266: SB = 250
267: # The end of sub-negotiation options.
268: SE = 240
269:
270:
271: class NegOptions(NegBase):
272: # Binary Transmission
273: BINARY = 0
274: # Suppress Go Ahead
275: SUPPRESS_GO_AHEAD = 3
276: # NAWS - width and height of client
277: NAWS = 31
278: # NEW-ENVIRON - environment variables on client
279: NEW_ENVIRON = 39
280: # Charset option
281: CHARSET = 42
282:
283:
284: def get_options():
285: parser = argparse.ArgumentParser()
286:
287: parser.add_argument("--port", action="store", default=9019,
288: type=int, help="port to listen on")
289: parser.add_argument("--verbose", action="store", type=int, default=0,
290: help="verbose output")
291: parser.add_argument("--pidfile", action="store",
292: help="file name for the PID")
293: parser.add_argument("--logfile", action="store",
294: help="file name for the log")
295: parser.add_argument("--srcdir", action="store", help="test directory")
296: parser.add_argument("--id", action="store", help="server ID")
297: parser.add_argument("--ipv4", action="store_true", default=0,
298: help="IPv4 flag")
299:
300: return parser.parse_args()
301:
302:
303: def setup_logging(options):
304: """
305: Set up logging from the command line options
306: """
307: root_logger = logging.getLogger()
308: add_stdout = False
309:
310: formatter = logging.Formatter("%(asctime)s %(levelname)-5.5s "
311: "[{ident}] %(message)s"
312: .format(ident=IDENT))
313:
314: # Write out to a logfile
315: if options.logfile:
316: handler = logging.FileHandler(options.logfile, mode="w")
317: handler.setFormatter(formatter)
318: handler.setLevel(logging.DEBUG)
319: root_logger.addHandler(handler)
320: else:
321: # The logfile wasn't specified. Add a stdout logger.
322: add_stdout = True
323:
324: if options.verbose:
325: # Add a stdout logger as well in verbose mode
326: root_logger.setLevel(logging.DEBUG)
327: add_stdout = True
328: else:
329: root_logger.setLevel(logging.INFO)
330:
331: if add_stdout:
332: stdout_handler = logging.StreamHandler(sys.stdout)
333: stdout_handler.setFormatter(formatter)
334: stdout_handler.setLevel(logging.DEBUG)
335: root_logger.addHandler(stdout_handler)
336:
337:
338: class ScriptRC(object):
339: """Enum for script return codes"""
340: SUCCESS = 0
341: FAILURE = 1
342: EXCEPTION = 2
343:
344:
345: class ScriptException(Exception):
346: pass
347:
348:
349: if __name__ == '__main__':
350: # Get the options from the user.
351: options = get_options()
352:
353: # Setup logging using the user options
354: setup_logging(options)
355:
356: # Run main script.
357: try:
358: rc = telnetserver(options)
359: except Exception as e:
360: log.exception(e)
361: rc = ScriptRC.EXCEPTION
362:
363: log.info("Returning %d", rc)
364: sys.exit(rc)
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>
![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to negtelnetserver.py CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / curl / tests