server.py 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178
  1. #!/usr/bin/env python
  2. """
  3. Dummy server used for unit testing.
  4. """
  5. from __future__ import print_function
  6. import logging
  7. import os
  8. import random
  9. import string
  10. import sys
  11. import threading
  12. import socket
  13. import warnings
  14. import ssl
  15. from datetime import datetime
  16. from urllib3.exceptions import HTTPWarning
  17. import tornado.httpserver
  18. import tornado.ioloop
  19. import tornado.netutil
  20. import tornado.web
  21. log = logging.getLogger(__name__)
  22. CERTS_PATH = os.path.join(os.path.dirname(__file__), "certs")
  23. DEFAULT_CERTS = {
  24. "certfile": os.path.join(CERTS_PATH, "server.crt"),
  25. "keyfile": os.path.join(CERTS_PATH, "server.key"),
  26. "cert_reqs": ssl.CERT_OPTIONAL,
  27. "ca_certs": os.path.join(CERTS_PATH, "cacert.pem"),
  28. }
  29. CLIENT_INTERMEDIATE_PEM = "client_intermediate.pem"
  30. CLIENT_NO_INTERMEDIATE_PEM = "client_no_intermediate.pem"
  31. CLIENT_INTERMEDIATE_KEY = "client_intermediate.key"
  32. CLIENT_CERT = os.path.join(CERTS_PATH, CLIENT_INTERMEDIATE_PEM)
  33. PASSWORD_KEYFILE = os.path.join(CERTS_PATH, "server_password.key")
  34. PASSWORD_CLIENT_KEYFILE = os.path.join(CERTS_PATH, "client_password.key")
  35. DEFAULT_CA = os.path.join(CERTS_PATH, "cacert.pem")
  36. DEFAULT_CA_KEY = os.path.join(CERTS_PATH, "cacert.key")
  37. DEFAULT_CA_BAD = os.path.join(CERTS_PATH, "client_bad.pem")
  38. COMBINED_CERT_AND_KEY = os.path.join(CERTS_PATH, "server.combined.pem")
  39. def _has_ipv6(host):
  40. """ Returns True if the system can bind an IPv6 address. """
  41. sock = None
  42. has_ipv6 = False
  43. if socket.has_ipv6:
  44. # has_ipv6 returns true if cPython was compiled with IPv6 support.
  45. # It does not tell us if the system has IPv6 support enabled. To
  46. # determine that we must bind to an IPv6 address.
  47. # https://github.com/urllib3/urllib3/pull/611
  48. # https://bugs.python.org/issue658327
  49. try:
  50. sock = socket.socket(socket.AF_INET6)
  51. sock.bind((host, 0))
  52. has_ipv6 = True
  53. except Exception:
  54. pass
  55. if sock:
  56. sock.close()
  57. return has_ipv6
  58. # Some systems may have IPv6 support but DNS may not be configured
  59. # properly. We can not count that localhost will resolve to ::1 on all
  60. # systems. See https://github.com/urllib3/urllib3/pull/611 and
  61. # https://bugs.python.org/issue18792
  62. HAS_IPV6_AND_DNS = _has_ipv6("localhost")
  63. HAS_IPV6 = _has_ipv6("::1")
  64. # Different types of servers we have:
  65. class NoIPv6Warning(HTTPWarning):
  66. "IPv6 is not available"
  67. pass
  68. class SocketServerThread(threading.Thread):
  69. """
  70. :param socket_handler: Callable which receives a socket argument for one
  71. request.
  72. :param ready_event: Event which gets set when the socket handler is
  73. ready to receive requests.
  74. """
  75. USE_IPV6 = HAS_IPV6_AND_DNS
  76. def __init__(self, socket_handler, host="localhost", port=8081, ready_event=None):
  77. threading.Thread.__init__(self)
  78. self.daemon = True
  79. self.socket_handler = socket_handler
  80. self.host = host
  81. self.ready_event = ready_event
  82. def _start_server(self):
  83. if self.USE_IPV6:
  84. sock = socket.socket(socket.AF_INET6)
  85. else:
  86. warnings.warn("No IPv6 support. Falling back to IPv4.", NoIPv6Warning)
  87. sock = socket.socket(socket.AF_INET)
  88. if sys.platform != "win32":
  89. sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  90. sock.bind((self.host, 0))
  91. self.port = sock.getsockname()[1]
  92. # Once listen() returns, the server socket is ready
  93. sock.listen(1)
  94. if self.ready_event:
  95. self.ready_event.set()
  96. self.socket_handler(sock)
  97. sock.close()
  98. def run(self):
  99. self.server = self._start_server()
  100. def run_tornado_app(app, io_loop, certs, scheme, host):
  101. assert io_loop == tornado.ioloop.IOLoop.current()
  102. # We can't use fromtimestamp(0) because of CPython issue 29097, so we'll
  103. # just construct the datetime object directly.
  104. app.last_req = datetime(1970, 1, 1)
  105. if scheme == "https":
  106. http_server = tornado.httpserver.HTTPServer(app, ssl_options=certs)
  107. else:
  108. http_server = tornado.httpserver.HTTPServer(app)
  109. sockets = tornado.netutil.bind_sockets(None, address=host)
  110. port = sockets[0].getsockname()[1]
  111. http_server.add_sockets(sockets)
  112. return http_server, port
  113. def run_loop_in_thread(io_loop):
  114. t = threading.Thread(target=io_loop.start)
  115. t.start()
  116. return t
  117. def get_unreachable_address():
  118. while True:
  119. host = "".join(random.choice(string.ascii_lowercase) for _ in range(60))
  120. sockaddr = (host, 54321)
  121. # check if we are really "lucky" and hit an actual server
  122. try:
  123. s = socket.create_connection(sockaddr)
  124. except socket.error:
  125. return sockaddr
  126. else:
  127. s.close()
  128. if __name__ == "__main__":
  129. # For debugging dummyserver itself - python -m dummyserver.server
  130. from .testcase import TestingApp
  131. host = "127.0.0.1"
  132. io_loop = tornado.ioloop.IOLoop.current()
  133. app = tornado.web.Application([(r".*", TestingApp)])
  134. server, port = run_tornado_app(app, io_loop, None, "http", host)
  135. server_thread = run_loop_in_thread(io_loop)
  136. print("Listening on http://{host}:{port}".format(host=host, port=port))