conftest.py 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127
  1. import collections
  2. import contextlib
  3. import threading
  4. import platform
  5. import sys
  6. import pytest
  7. import trustme
  8. from tornado import web, ioloop
  9. from dummyserver.handlers import TestingApp
  10. from dummyserver.server import run_tornado_app
  11. from dummyserver.server import (
  12. DEFAULT_CA,
  13. DEFAULT_CA_KEY,
  14. CLIENT_INTERMEDIATE_PEM,
  15. CLIENT_NO_INTERMEDIATE_PEM,
  16. CLIENT_INTERMEDIATE_KEY,
  17. HAS_IPV6,
  18. )
  19. # The Python 3.8+ default loop on Windows breaks Tornado
  20. @pytest.fixture(scope="session", autouse=True)
  21. def configure_windows_event_loop():
  22. if sys.version_info >= (3, 8) and platform.system() == "Windows":
  23. import asyncio
  24. asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
  25. @pytest.fixture(scope="session")
  26. def certs_dir(tmp_path_factory):
  27. tmpdir = tmp_path_factory.mktemp("certs")
  28. # Start from existing root CA as we don't want to change the server certificate yet
  29. with open(DEFAULT_CA, "rb") as crt, open(DEFAULT_CA_KEY, "rb") as key:
  30. root_ca = trustme.CA.from_pem(crt.read(), key.read())
  31. # client cert chain
  32. intermediate_ca = root_ca.create_child_ca()
  33. cert = intermediate_ca.issue_cert(u"example.com")
  34. cert.private_key_pem.write_to_path(str(tmpdir / CLIENT_INTERMEDIATE_KEY))
  35. # Write the client cert and the intermediate CA
  36. client_cert = str(tmpdir / CLIENT_INTERMEDIATE_PEM)
  37. cert.cert_chain_pems[0].write_to_path(client_cert)
  38. cert.cert_chain_pems[1].write_to_path(client_cert, append=True)
  39. # Write only the client cert
  40. cert.cert_chain_pems[0].write_to_path(str(tmpdir / CLIENT_NO_INTERMEDIATE_PEM))
  41. yield tmpdir
  42. ServerConfig = collections.namedtuple("ServerConfig", ["host", "port", "ca_certs"])
  43. @contextlib.contextmanager
  44. def run_server_in_thread(scheme, host, tmpdir, ca, server_cert):
  45. ca_cert_path = str(tmpdir / "ca.pem")
  46. server_cert_path = str(tmpdir / "server.pem")
  47. server_key_path = str(tmpdir / "server.key")
  48. ca.cert_pem.write_to_path(ca_cert_path)
  49. server_cert.private_key_pem.write_to_path(server_key_path)
  50. server_cert.cert_chain_pems[0].write_to_path(server_cert_path)
  51. server_certs = {"keyfile": server_key_path, "certfile": server_cert_path}
  52. io_loop = ioloop.IOLoop.current()
  53. app = web.Application([(r".*", TestingApp)])
  54. server, port = run_tornado_app(app, io_loop, server_certs, scheme, host)
  55. server_thread = threading.Thread(target=io_loop.start)
  56. server_thread.start()
  57. yield ServerConfig(host, port, ca_cert_path)
  58. io_loop.add_callback(server.stop)
  59. io_loop.add_callback(io_loop.stop)
  60. server_thread.join()
  61. @pytest.fixture
  62. def no_san_server(tmp_path_factory):
  63. tmpdir = tmp_path_factory.mktemp("certs")
  64. ca = trustme.CA()
  65. # only common name, no subject alternative names
  66. server_cert = ca.issue_cert(common_name=u"localhost")
  67. with run_server_in_thread("https", "localhost", tmpdir, ca, server_cert) as cfg:
  68. yield cfg
  69. @pytest.fixture
  70. def ip_san_server(tmp_path_factory):
  71. tmpdir = tmp_path_factory.mktemp("certs")
  72. ca = trustme.CA()
  73. # IP address in Subject Alternative Name
  74. server_cert = ca.issue_cert(u"127.0.0.1")
  75. with run_server_in_thread("https", "127.0.0.1", tmpdir, ca, server_cert) as cfg:
  76. yield cfg
  77. @pytest.fixture
  78. def ipv6_addr_server(tmp_path_factory):
  79. if not HAS_IPV6:
  80. pytest.skip("Only runs on IPv6 systems")
  81. tmpdir = tmp_path_factory.mktemp("certs")
  82. ca = trustme.CA()
  83. # IP address in Common Name
  84. server_cert = ca.issue_cert(common_name=u"::1")
  85. with run_server_in_thread("https", "::1", tmpdir, ca, server_cert) as cfg:
  86. yield cfg
  87. @pytest.fixture
  88. def ipv6_san_server(tmp_path_factory):
  89. if not HAS_IPV6:
  90. pytest.skip("Only runs on IPv6 systems")
  91. tmpdir = tmp_path_factory.mktemp("certs")
  92. ca = trustme.CA()
  93. # IP address in Subject Alternative Name
  94. server_cert = ca.issue_cert(u"::1")
  95. with run_server_in_thread("https", "::1", tmpdir, ca, server_cert) as cfg:
  96. yield cfg