server.py 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226
  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 socket
  9. import ssl
  10. import sys
  11. import threading
  12. import warnings
  13. from datetime import datetime
  14. import tornado.httpserver
  15. import tornado.ioloop
  16. import tornado.netutil
  17. import tornado.web
  18. import trustme
  19. from cryptography.hazmat.backends import default_backend
  20. from cryptography.hazmat.primitives import serialization
  21. from urllib3.exceptions import HTTPWarning
  22. from urllib3.util import ALPN_PROTOCOLS, resolve_cert_reqs, resolve_ssl_version
  23. log = logging.getLogger(__name__)
  24. CERTS_PATH = os.path.join(os.path.dirname(__file__), "certs")
  25. DEFAULT_CERTS = {
  26. "certfile": os.path.join(CERTS_PATH, "server.crt"),
  27. "keyfile": os.path.join(CERTS_PATH, "server.key"),
  28. "cert_reqs": ssl.CERT_OPTIONAL,
  29. "ca_certs": os.path.join(CERTS_PATH, "cacert.pem"),
  30. "alpn_protocols": ALPN_PROTOCOLS,
  31. }
  32. DEFAULT_CA = os.path.join(CERTS_PATH, "cacert.pem")
  33. DEFAULT_CA_KEY = os.path.join(CERTS_PATH, "cacert.key")
  34. def _resolves_to_ipv6(host):
  35. """Returns True if the system resolves host to an IPv6 address by default."""
  36. resolves_to_ipv6 = False
  37. try:
  38. for res in socket.getaddrinfo(host, None, socket.AF_UNSPEC):
  39. af, _, _, _, _ = res
  40. if af == socket.AF_INET6:
  41. resolves_to_ipv6 = True
  42. except socket.gaierror:
  43. pass
  44. return resolves_to_ipv6
  45. def _has_ipv6(host):
  46. """Returns True if the system can bind an IPv6 address."""
  47. sock = None
  48. has_ipv6 = False
  49. if socket.has_ipv6:
  50. # has_ipv6 returns true if cPython was compiled with IPv6 support.
  51. # It does not tell us if the system has IPv6 support enabled. To
  52. # determine that we must bind to an IPv6 address.
  53. # https://github.com/urllib3/urllib3/pull/611
  54. # https://bugs.python.org/issue658327
  55. try:
  56. sock = socket.socket(socket.AF_INET6)
  57. sock.bind((host, 0))
  58. has_ipv6 = _resolves_to_ipv6("localhost")
  59. except Exception:
  60. pass
  61. if sock:
  62. sock.close()
  63. return has_ipv6
  64. # Some systems may have IPv6 support but DNS may not be configured
  65. # properly. We can not count that localhost will resolve to ::1 on all
  66. # systems. See https://github.com/urllib3/urllib3/pull/611 and
  67. # https://bugs.python.org/issue18792
  68. HAS_IPV6_AND_DNS = _has_ipv6("localhost")
  69. HAS_IPV6 = _has_ipv6("::1")
  70. # Different types of servers we have:
  71. class NoIPv6Warning(HTTPWarning):
  72. "IPv6 is not available"
  73. pass
  74. class SocketServerThread(threading.Thread):
  75. """
  76. :param socket_handler: Callable which receives a socket argument for one
  77. request.
  78. :param ready_event: Event which gets set when the socket handler is
  79. ready to receive requests.
  80. """
  81. USE_IPV6 = HAS_IPV6_AND_DNS
  82. def __init__(self, socket_handler, host="localhost", port=8081, ready_event=None):
  83. threading.Thread.__init__(self)
  84. self.daemon = True
  85. self.socket_handler = socket_handler
  86. self.host = host
  87. self.ready_event = ready_event
  88. def _start_server(self):
  89. if self.USE_IPV6:
  90. sock = socket.socket(socket.AF_INET6)
  91. else:
  92. warnings.warn("No IPv6 support. Falling back to IPv4.", NoIPv6Warning)
  93. sock = socket.socket(socket.AF_INET)
  94. if sys.platform != "win32":
  95. sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  96. sock.bind((self.host, 0))
  97. self.port = sock.getsockname()[1]
  98. # Once listen() returns, the server socket is ready
  99. sock.listen(1)
  100. if self.ready_event:
  101. self.ready_event.set()
  102. self.socket_handler(sock)
  103. sock.close()
  104. def run(self):
  105. self.server = self._start_server()
  106. def ssl_options_to_context(
  107. keyfile=None,
  108. certfile=None,
  109. server_side=None,
  110. cert_reqs=None,
  111. ssl_version=None,
  112. ca_certs=None,
  113. do_handshake_on_connect=None,
  114. suppress_ragged_eofs=None,
  115. ciphers=None,
  116. alpn_protocols=None,
  117. ):
  118. """Return an equivalent SSLContext based on ssl.wrap_socket args."""
  119. ssl_version = resolve_ssl_version(ssl_version)
  120. cert_none = resolve_cert_reqs("CERT_NONE")
  121. if cert_reqs is None:
  122. cert_reqs = cert_none
  123. else:
  124. cert_reqs = resolve_cert_reqs(cert_reqs)
  125. ctx = ssl.SSLContext(ssl_version)
  126. ctx.load_cert_chain(certfile, keyfile)
  127. ctx.verify_mode = cert_reqs
  128. if ctx.verify_mode != cert_none:
  129. ctx.load_verify_locations(cafile=ca_certs)
  130. if alpn_protocols and hasattr(ctx, "set_alpn_protocols"):
  131. try:
  132. ctx.set_alpn_protocols(alpn_protocols)
  133. except NotImplementedError:
  134. pass
  135. return ctx
  136. def run_tornado_app(app, io_loop, certs, scheme, host):
  137. assert io_loop == tornado.ioloop.IOLoop.current()
  138. # We can't use fromtimestamp(0) because of CPython issue 29097, so we'll
  139. # just construct the datetime object directly.
  140. app.last_req = datetime(1970, 1, 1)
  141. if scheme == "https":
  142. if sys.version_info < (2, 7, 9):
  143. ssl_opts = certs
  144. else:
  145. ssl_opts = ssl_options_to_context(**certs)
  146. http_server = tornado.httpserver.HTTPServer(app, ssl_options=ssl_opts)
  147. else:
  148. http_server = tornado.httpserver.HTTPServer(app)
  149. sockets = tornado.netutil.bind_sockets(None, address=host)
  150. port = sockets[0].getsockname()[1]
  151. http_server.add_sockets(sockets)
  152. return http_server, port
  153. def run_loop_in_thread(io_loop):
  154. t = threading.Thread(target=io_loop.start)
  155. t.start()
  156. return t
  157. def get_unreachable_address():
  158. # reserved as per rfc2606
  159. return ("something.invalid", 54321)
  160. if __name__ == "__main__":
  161. # For debugging dummyserver itself - python -m dummyserver.server
  162. from .testcase import TestingApp
  163. host = "127.0.0.1"
  164. io_loop = tornado.ioloop.IOLoop.current()
  165. app = tornado.web.Application([(r".*", TestingApp)])
  166. server, port = run_tornado_app(app, io_loop, None, "http", host)
  167. server_thread = run_loop_in_thread(io_loop)
  168. print("Listening on http://{host}:{port}".format(host=host, port=port))
  169. def encrypt_key_pem(private_key_pem, password):
  170. private_key = serialization.load_pem_private_key(
  171. private_key_pem.bytes(), password=None, backend=default_backend()
  172. )
  173. encrypted_key = private_key.private_bytes(
  174. serialization.Encoding.PEM,
  175. serialization.PrivateFormat.TraditionalOpenSSL,
  176. serialization.BestAvailableEncryption(password),
  177. )
  178. return trustme.Blob(encrypted_key)