tests/test-l7.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2015, 2016 Nicira, Inc.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import argparse
  17 import socket
  18
  19 try:  # Python 2.7
  20     from BaseHTTPServer import HTTPServer
  21     from SimpleHTTPServer import SimpleHTTPRequestHandler
  22     from SocketServer import TCPServer
  23 except:
  24     from http.server import HTTPServer, SimpleHTTPRequestHandler
  25     from socketserver import TCPServer
  26
  27
  28 class TCPServerV6(HTTPServer):
  29     address_family = socket.AF_INET6
  30
  31
  32 def get_ftpd():
  33     try:
  34         from pyftpdlib.authorizers import DummyAuthorizer
  35         from pyftpdlib.handlers import FTPHandler
  36         from pyftpdlib.servers import FTPServer
  37
  38         import logging
  39         import pyftpdlib.log
  40         pyftpdlib.log.LEVEL = logging.DEBUG
  41
  42         class OVSFTPHandler(FTPHandler):
  43             authorizer = DummyAuthorizer()
  44             authorizer.add_anonymous("/tmp")
  45             # Hack around a bug in pyftpdlib, which rejects EPRT
  46             # connection due to mismatching textual representation of
  47             # the IPv6 address.
  48             permit_foreign_addresses = True
  49         server = [FTPServer, OVSFTPHandler, 21]
  50     except ImportError:
  51         server = None
  52         pass
  53     return server
  54
  55
  56 def get_tftpd():
  57     try:
  58         from tftpy import TftpServer, TftpShared
  59
  60         class OVSTFTPServer(TftpServer):
  61             def __init__(self, listen, handler=None):
  62                 (ip, port) = listen
  63                 self.ip = ip
  64                 self.port = port
  65                 TftpServer.__init__(self, tftproot='./')
  66
  67             def serve_forever(self):
  68                 self.listen(self.ip, self.port)
  69         server = [OVSTFTPServer, None, TftpShared.DEF_TFTP_PORT]
  70     except (ImportError, SyntaxError):
  71         server = None
  72         pass
  73     return server
  74
  75
  76 def main():
  77     SERVERS = {
  78         'http': [TCPServer, SimpleHTTPRequestHandler, 80],
  79         'http6': [TCPServerV6, SimpleHTTPRequestHandler, 80],
  80         'ftp': get_ftpd(),
  81         'tftp': get_tftpd(),
  82     }
  83
  84     protocols = [srv for srv in SERVERS if SERVERS[srv] is not None]
  85     parser = argparse.ArgumentParser(
  86         description='Run basic application servers.')
  87     parser.add_argument('proto', default='http', nargs='?',
  88                         help='protocol to serve (%s)' % protocols)
  89     args = parser.parse_args()
  90
  91     if args.proto not in protocols:
  92         parser.print_help()
  93         exit(1)
  94
  95     constructor = SERVERS[args.proto][0]
  96     handler = SERVERS[args.proto][1]
  97     port = SERVERS[args.proto][2]
  98     srv = constructor(('', port), handler)
  99     srv.serve_forever()
 100
 101
 102 if __name__ == '__main__':
 103     main()