__init__.py 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191
  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 incremental import Version
  18. from zope.interface import implementer
  19. import twisted
  20. from OpenSSL import SSL
  21. from OpenSSL.SSL import Connection
  22. from twisted.internet.address import IPv4Address
  23. from twisted.internet.interfaces import (
  24. IOpenSSLServerConnectionCreator,
  25. IProtocolFactory,
  26. IReactorTime,
  27. )
  28. from twisted.internet.ssl import Certificate, trustRootFromCertificates
  29. from twisted.protocols.tls import TLSMemoryBIOFactory, TLSMemoryBIOProtocol
  30. from twisted.web.client import BrowserLikePolicyForHTTPS # noqa: F401
  31. from twisted.web.iweb import IPolicyForHTTPS # noqa: F401
  32. def get_test_https_policy() -> BrowserLikePolicyForHTTPS:
  33. """Get a test IPolicyForHTTPS which trusts the test CA cert
  34. Returns:
  35. IPolicyForHTTPS
  36. """
  37. ca_file = get_test_ca_cert_file()
  38. with open(ca_file) as stream:
  39. content = stream.read()
  40. cert = Certificate.loadPEM(content)
  41. trust_root = trustRootFromCertificates([cert])
  42. return BrowserLikePolicyForHTTPS(trustRoot=trust_root)
  43. def get_test_ca_cert_file() -> str:
  44. """Get the path to the test CA cert
  45. The keypair is generated with:
  46. openssl genrsa -out ca.key 2048
  47. openssl req -new -x509 -key ca.key -days 3650 -out ca.crt \
  48. -subj '/CN=synapse test CA'
  49. """
  50. return os.path.join(os.path.dirname(__file__), "ca.crt")
  51. def get_test_key_file() -> str:
  52. """get the path to the test key
  53. The key file is made with:
  54. openssl genrsa -out server.key 2048
  55. """
  56. return os.path.join(os.path.dirname(__file__), "server.key")
  57. cert_file_count = 0
  58. CONFIG_TEMPLATE = b"""\
  59. [default]
  60. basicConstraints = CA:FALSE
  61. keyUsage=nonRepudiation, digitalSignature, keyEncipherment
  62. subjectAltName = %(sanentries)s
  63. """
  64. def create_test_cert_file(sanlist: List[bytes]) -> str:
  65. """build an x509 certificate file
  66. Args:
  67. sanlist: a list of subjectAltName values for the cert
  68. Returns:
  69. The path to the file
  70. """
  71. global cert_file_count
  72. csr_filename = "server.csr"
  73. cnf_filename = "server.%i.cnf" % (cert_file_count,)
  74. cert_filename = "server.%i.crt" % (cert_file_count,)
  75. cert_file_count += 1
  76. # first build a CSR
  77. subprocess.check_call(
  78. [
  79. "openssl",
  80. "req",
  81. "-new",
  82. "-key",
  83. get_test_key_file(),
  84. "-subj",
  85. "/",
  86. "-out",
  87. csr_filename,
  88. ]
  89. )
  90. # now a config file describing the right SAN entries
  91. sanentries = b",".join(sanlist)
  92. with open(cnf_filename, "wb") as f:
  93. f.write(CONFIG_TEMPLATE % {b"sanentries": sanentries})
  94. # finally the cert
  95. ca_key_filename = os.path.join(os.path.dirname(__file__), "ca.key")
  96. ca_cert_filename = get_test_ca_cert_file()
  97. subprocess.check_call(
  98. [
  99. "openssl",
  100. "x509",
  101. "-req",
  102. "-in",
  103. csr_filename,
  104. "-CA",
  105. ca_cert_filename,
  106. "-CAkey",
  107. ca_key_filename,
  108. "-set_serial",
  109. "1",
  110. "-extfile",
  111. cnf_filename,
  112. "-out",
  113. cert_filename,
  114. ]
  115. )
  116. return cert_filename
  117. @implementer(IOpenSSLServerConnectionCreator)
  118. class TestServerTLSConnectionFactory:
  119. """An SSL connection creator which returns connections which present a certificate
  120. signed by our test CA."""
  121. def __init__(self, sanlist: List[bytes]):
  122. """
  123. Args:
  124. sanlist: a list of subjectAltName values for the cert
  125. """
  126. self._cert_file = create_test_cert_file(sanlist)
  127. def serverConnectionForTLS(self, tlsProtocol: TLSMemoryBIOProtocol) -> Connection:
  128. ctx = SSL.Context(SSL.SSLv23_METHOD)
  129. ctx.use_certificate_file(self._cert_file)
  130. ctx.use_privatekey_file(get_test_key_file())
  131. return Connection(ctx, None)
  132. def wrap_server_factory_for_tls(
  133. factory: IProtocolFactory, clock: IReactorTime, sanlist: List[bytes]
  134. ) -> TLSMemoryBIOFactory:
  135. """Wrap an existing Protocol Factory with a test TLSMemoryBIOFactory
  136. The resultant factory will create a TLS server which presents a certificate
  137. signed by our test CA, valid for the domains in `sanlist`
  138. Args:
  139. factory: protocol factory to wrap
  140. sanlist: list of domains the cert should be valid for
  141. Returns:
  142. interfaces.IProtocolFactory
  143. """
  144. connection_creator = TestServerTLSConnectionFactory(sanlist=sanlist)
  145. # Twisted > 23.8.0 has a different API that accepts a clock.
  146. if twisted.version <= Version("Twisted", 23, 8, 0):
  147. return TLSMemoryBIOFactory(
  148. connection_creator, isClient=False, wrappedFactory=factory
  149. )
  150. else:
  151. return TLSMemoryBIOFactory(
  152. connection_creator, isClient=False, wrappedFactory=factory, clock=clock
  153. )
  154. # A dummy address, useful for tests that use FakeTransport and don't care about where
  155. # packets are going to/coming from.
  156. dummy_address = IPv4Address("TCP", "127.0.0.1", 80)