__init__.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151
  1. # Copyright 2019 New Vector Ltd
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. import os.path
  15. import subprocess
  16. from typing import List
  17. from zope.interface import implementer
  18. from OpenSSL import SSL
  19. from OpenSSL.SSL import Connection
  20. from twisted.internet.interfaces import IOpenSSLServerConnectionCreator
  21. from twisted.internet.ssl import Certificate, trustRootFromCertificates
  22. from twisted.web.client import BrowserLikePolicyForHTTPS # noqa: F401
  23. from twisted.web.iweb import IPolicyForHTTPS # noqa: F401
  24. def get_test_https_policy():
  25. """Get a test IPolicyForHTTPS which trusts the test CA cert
  26. Returns:
  27. IPolicyForHTTPS
  28. """
  29. ca_file = get_test_ca_cert_file()
  30. with open(ca_file) as stream:
  31. content = stream.read()
  32. cert = Certificate.loadPEM(content)
  33. trust_root = trustRootFromCertificates([cert])
  34. return BrowserLikePolicyForHTTPS(trustRoot=trust_root)
  35. def get_test_ca_cert_file():
  36. """Get the path to the test CA cert
  37. The keypair is generated with:
  38. openssl genrsa -out ca.key 2048
  39. openssl req -new -x509 -key ca.key -days 3650 -out ca.crt \
  40. -subj '/CN=synapse test CA'
  41. """
  42. return os.path.join(os.path.dirname(__file__), "ca.crt")
  43. def get_test_key_file():
  44. """get the path to the test key
  45. The key file is made with:
  46. openssl genrsa -out server.key 2048
  47. """
  48. return os.path.join(os.path.dirname(__file__), "server.key")
  49. cert_file_count = 0
  50. CONFIG_TEMPLATE = b"""\
  51. [default]
  52. basicConstraints = CA:FALSE
  53. keyUsage=nonRepudiation, digitalSignature, keyEncipherment
  54. subjectAltName = %(sanentries)s
  55. """
  56. def create_test_cert_file(sanlist: List[bytes]) -> str:
  57. """build an x509 certificate file
  58. Args:
  59. sanlist: a list of subjectAltName values for the cert
  60. Returns:
  61. The path to the file
  62. """
  63. global cert_file_count
  64. csr_filename = "server.csr"
  65. cnf_filename = "server.%i.cnf" % (cert_file_count,)
  66. cert_filename = "server.%i.crt" % (cert_file_count,)
  67. cert_file_count += 1
  68. # first build a CSR
  69. subprocess.check_call(
  70. [
  71. "openssl",
  72. "req",
  73. "-new",
  74. "-key",
  75. get_test_key_file(),
  76. "-subj",
  77. "/",
  78. "-out",
  79. csr_filename,
  80. ]
  81. )
  82. # now a config file describing the right SAN entries
  83. sanentries = b",".join(sanlist)
  84. with open(cnf_filename, "wb") as f:
  85. f.write(CONFIG_TEMPLATE % {b"sanentries": sanentries})
  86. # finally the cert
  87. ca_key_filename = os.path.join(os.path.dirname(__file__), "ca.key")
  88. ca_cert_filename = get_test_ca_cert_file()
  89. subprocess.check_call(
  90. [
  91. "openssl",
  92. "x509",
  93. "-req",
  94. "-in",
  95. csr_filename,
  96. "-CA",
  97. ca_cert_filename,
  98. "-CAkey",
  99. ca_key_filename,
  100. "-set_serial",
  101. "1",
  102. "-extfile",
  103. cnf_filename,
  104. "-out",
  105. cert_filename,
  106. ]
  107. )
  108. return cert_filename
  109. @implementer(IOpenSSLServerConnectionCreator)
  110. class TestServerTLSConnectionFactory:
  111. """An SSL connection creator which returns connections which present a certificate
  112. signed by our test CA."""
  113. def __init__(self, sanlist):
  114. """
  115. Args:
  116. sanlist: list[bytes]: a list of subjectAltName values for the cert
  117. """
  118. self._cert_file = create_test_cert_file(sanlist)
  119. def serverConnectionForTLS(self, tlsProtocol):
  120. ctx = SSL.Context(SSL.SSLv23_METHOD)
  121. ctx.use_certificate_file(self._cert_file)
  122. ctx.use_privatekey_file(get_test_key_file())
  123. return Connection(ctx, None)