endpoint.py 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2014-2016 OpenMarket Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. import collections
  16. import logging
  17. import random
  18. import re
  19. import time
  20. from twisted.internet import defer
  21. from twisted.internet.endpoints import HostnameEndpoint, wrapClientTLS
  22. from twisted.internet.error import ConnectError
  23. from twisted.names import client, dns
  24. from twisted.names.error import DNSNameError, DomainError
  25. logger = logging.getLogger(__name__)
  26. SERVER_CACHE = {}
  27. # our record of an individual server which can be tried to reach a destination.
  28. #
  29. # "host" is the hostname acquired from the SRV record. Except when there's
  30. # no SRV record, in which case it is the original hostname.
  31. _Server = collections.namedtuple(
  32. "_Server", "priority weight host port expires"
  33. )
  34. def parse_server_name(server_name):
  35. """Split a server name into host/port parts.
  36. Args:
  37. server_name (str): server name to parse
  38. Returns:
  39. Tuple[str, int|None]: host/port parts.
  40. Raises:
  41. ValueError if the server name could not be parsed.
  42. """
  43. try:
  44. if server_name[-1] == ']':
  45. # ipv6 literal, hopefully
  46. return server_name, None
  47. domain_port = server_name.rsplit(":", 1)
  48. domain = domain_port[0]
  49. port = int(domain_port[1]) if domain_port[1:] else None
  50. return domain, port
  51. except Exception:
  52. raise ValueError("Invalid server name '%s'" % server_name)
  53. VALID_HOST_REGEX = re.compile(
  54. "\\A[0-9a-zA-Z.-]+\\Z",
  55. )
  56. def parse_and_validate_server_name(server_name):
  57. """Split a server name into host/port parts and do some basic validation.
  58. Args:
  59. server_name (str): server name to parse
  60. Returns:
  61. Tuple[str, int|None]: host/port parts.
  62. Raises:
  63. ValueError if the server name could not be parsed.
  64. """
  65. host, port = parse_server_name(server_name)
  66. # these tests don't need to be bulletproof as we'll find out soon enough
  67. # if somebody is giving us invalid data. What we *do* need is to be sure
  68. # that nobody is sneaking IP literals in that look like hostnames, etc.
  69. # look for ipv6 literals
  70. if host[0] == '[':
  71. if host[-1] != ']':
  72. raise ValueError("Mismatched [...] in server name '%s'" % (
  73. server_name,
  74. ))
  75. return host, port
  76. # otherwise it should only be alphanumerics.
  77. if not VALID_HOST_REGEX.match(host):
  78. raise ValueError("Server name '%s' contains invalid characters" % (
  79. server_name,
  80. ))
  81. return host, port
  82. def matrix_federation_endpoint(reactor, destination, tls_client_options_factory=None,
  83. timeout=None):
  84. """Construct an endpoint for the given matrix destination.
  85. Args:
  86. reactor: Twisted reactor.
  87. destination (unicode): The name of the server to connect to.
  88. tls_client_options_factory
  89. (synapse.crypto.context_factory.ClientTLSOptionsFactory):
  90. Factory which generates TLS options for client connections.
  91. timeout (int): connection timeout in seconds
  92. """
  93. domain, port = parse_server_name(destination)
  94. endpoint_kw_args = {}
  95. if timeout is not None:
  96. endpoint_kw_args.update(timeout=timeout)
  97. if tls_client_options_factory is None:
  98. transport_endpoint = HostnameEndpoint
  99. default_port = 8008
  100. else:
  101. # the SNI string should be the same as the Host header, minus the port.
  102. # as per https://github.com/matrix-org/synapse/issues/2525#issuecomment-336896777,
  103. # the Host header and SNI should therefore be the server_name of the remote
  104. # server.
  105. tls_options = tls_client_options_factory.get_options(domain)
  106. def transport_endpoint(reactor, host, port, timeout):
  107. return wrapClientTLS(
  108. tls_options,
  109. HostnameEndpoint(reactor, host, port, timeout=timeout),
  110. )
  111. default_port = 8448
  112. if port is None:
  113. return SRVClientEndpoint(
  114. reactor, "matrix", domain, protocol="tcp",
  115. default_port=default_port, endpoint=transport_endpoint,
  116. endpoint_kw_args=endpoint_kw_args
  117. )
  118. else:
  119. return transport_endpoint(
  120. reactor, domain, port, **endpoint_kw_args
  121. )
  122. class SRVClientEndpoint(object):
  123. """An endpoint which looks up SRV records for a service.
  124. Cycles through the list of servers starting with each call to connect
  125. picking the next server.
  126. Implements twisted.internet.interfaces.IStreamClientEndpoint.
  127. """
  128. def __init__(self, reactor, service, domain, protocol="tcp",
  129. default_port=None, endpoint=HostnameEndpoint,
  130. endpoint_kw_args={}):
  131. self.reactor = reactor
  132. self.service_name = "_%s._%s.%s" % (service, protocol, domain)
  133. if default_port is not None:
  134. self.default_server = _Server(
  135. host=domain,
  136. port=default_port,
  137. priority=0,
  138. weight=0,
  139. expires=0,
  140. )
  141. else:
  142. self.default_server = None
  143. self.endpoint = endpoint
  144. self.endpoint_kw_args = endpoint_kw_args
  145. self.servers = None
  146. self.used_servers = None
  147. @defer.inlineCallbacks
  148. def fetch_servers(self):
  149. self.used_servers = []
  150. self.servers = yield resolve_service(self.service_name)
  151. def pick_server(self):
  152. if not self.servers:
  153. if self.used_servers:
  154. self.servers = self.used_servers
  155. self.used_servers = []
  156. self.servers.sort()
  157. elif self.default_server:
  158. return self.default_server
  159. else:
  160. raise ConnectError(
  161. "No server available for %s" % self.service_name
  162. )
  163. # look for all servers with the same priority
  164. min_priority = self.servers[0].priority
  165. weight_indexes = list(
  166. (index, server.weight + 1)
  167. for index, server in enumerate(self.servers)
  168. if server.priority == min_priority
  169. )
  170. total_weight = sum(weight for index, weight in weight_indexes)
  171. target_weight = random.randint(0, total_weight)
  172. for index, weight in weight_indexes:
  173. target_weight -= weight
  174. if target_weight <= 0:
  175. server = self.servers[index]
  176. # XXX: this looks totally dubious:
  177. #
  178. # (a) we never reuse a server until we have been through
  179. # all of the servers at the same priority, so if the
  180. # weights are A: 100, B:1, we always do ABABAB instead of
  181. # AAAA...AAAB (approximately).
  182. #
  183. # (b) After using all the servers at the lowest priority,
  184. # we move onto the next priority. We should only use the
  185. # second priority if servers at the top priority are
  186. # unreachable.
  187. #
  188. del self.servers[index]
  189. self.used_servers.append(server)
  190. return server
  191. @defer.inlineCallbacks
  192. def connect(self, protocolFactory):
  193. if self.servers is None:
  194. yield self.fetch_servers()
  195. server = self.pick_server()
  196. logger.info("Connecting to %s:%s", server.host, server.port)
  197. endpoint = self.endpoint(
  198. self.reactor, server.host, server.port, **self.endpoint_kw_args
  199. )
  200. connection = yield endpoint.connect(protocolFactory)
  201. defer.returnValue(connection)
  202. @defer.inlineCallbacks
  203. def resolve_service(service_name, dns_client=client, cache=SERVER_CACHE, clock=time):
  204. cache_entry = cache.get(service_name, None)
  205. if cache_entry:
  206. if all(s.expires > int(clock.time()) for s in cache_entry):
  207. servers = list(cache_entry)
  208. defer.returnValue(servers)
  209. servers = []
  210. try:
  211. try:
  212. answers, _, _ = yield dns_client.lookupService(service_name)
  213. except DNSNameError:
  214. defer.returnValue([])
  215. if (len(answers) == 1
  216. and answers[0].type == dns.SRV
  217. and answers[0].payload
  218. and answers[0].payload.target == dns.Name(b'.')):
  219. raise ConnectError("Service %s unavailable" % service_name)
  220. for answer in answers:
  221. if answer.type != dns.SRV or not answer.payload:
  222. continue
  223. payload = answer.payload
  224. servers.append(_Server(
  225. host=str(payload.target),
  226. port=int(payload.port),
  227. priority=int(payload.priority),
  228. weight=int(payload.weight),
  229. expires=int(clock.time()) + answer.ttl,
  230. ))
  231. servers.sort()
  232. cache[service_name] = list(servers)
  233. except DomainError as e:
  234. # We failed to resolve the name (other than a NameError)
  235. # Try something in the cache, else rereaise
  236. cache_entry = cache.get(service_name, None)
  237. if cache_entry:
  238. logger.warn(
  239. "Failed to resolve %r, falling back to cache. %r",
  240. service_name, e
  241. )
  242. servers = list(cache_entry)
  243. else:
  244. raise e
  245. defer.returnValue(servers)