diff options
| -rw-r--r-- | changelog.d/4427.misc | 1 | ||||
| -rw-r--r-- | synapse/http/endpoint.py | 75 | ||||
| -rw-r--r-- | synapse/http/federation/__init__.py | 14 | ||||
| -rw-r--r-- | synapse/http/federation/srv_resolver.py | 124 | ||||
| -rw-r--r-- | tests/http/federation/__init__.py | 14 | ||||
| -rw-r--r-- | tests/http/federation/test_srv_resolver.py (renamed from tests/test_dns.py) | 108 |
6 files changed, 250 insertions, 86 deletions
diff --git a/changelog.d/4427.misc b/changelog.d/4427.misc new file mode 100644 index 0000000000..75500bdbc2 --- /dev/null +++ b/changelog.d/4427.misc @@ -0,0 +1 @@ +Refactor and cleanup for SRV record lookup diff --git a/synapse/http/endpoint.py b/synapse/http/endpoint.py index 1c3b7ea28a..815f8ff2f7 100644 --- a/synapse/http/endpoint.py +++ b/synapse/http/endpoint.py @@ -12,29 +12,17 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -import collections import logging import random import re -import time from twisted.internet import defer from twisted.internet.endpoints import HostnameEndpoint, wrapClientTLS from twisted.internet.error import ConnectError -from twisted.names import client, dns -from twisted.names.error import DNSNameError, DomainError -logger = logging.getLogger(__name__) - -SERVER_CACHE = {} +from synapse.http.federation.srv_resolver import Server, resolve_service -# our record of an individual server which can be tried to reach a destination. -# -# "host" is the hostname acquired from the SRV record. Except when there's -# no SRV record, in which case it is the original hostname. -_Server = collections.namedtuple( - "_Server", "priority weight host port expires" -) +logger = logging.getLogger(__name__) def parse_server_name(server_name): @@ -165,12 +153,9 @@ class SRVClientEndpoint(object): self.service_name = "_%s._%s.%s" % (service, protocol, domain) if default_port is not None: - self.default_server = _Server( + self.default_server = Server( host=domain, port=default_port, - priority=0, - weight=0, - expires=0, ) else: self.default_server = None @@ -240,57 +225,3 @@ class SRVClientEndpoint(object): ) connection = yield endpoint.connect(protocolFactory) defer.returnValue(connection) - - -@defer.inlineCallbacks -def resolve_service(service_name, dns_client=client, cache=SERVER_CACHE, clock=time): - cache_entry = cache.get(service_name, None) - if cache_entry: - if all(s.expires > int(clock.time()) for s in cache_entry): - servers = list(cache_entry) - defer.returnValue(servers) - - servers = [] - - try: - try: - answers, _, _ = yield dns_client.lookupService(service_name) - except DNSNameError: - defer.returnValue([]) - - if (len(answers) == 1 - and answers[0].type == dns.SRV - and answers[0].payload - and answers[0].payload.target == dns.Name(b'.')): - raise ConnectError("Service %s unavailable" % service_name) - - for answer in answers: - if answer.type != dns.SRV or not answer.payload: - continue - - payload = answer.payload - - servers.append(_Server( - host=str(payload.target), - port=int(payload.port), - priority=int(payload.priority), - weight=int(payload.weight), - expires=int(clock.time()) + answer.ttl, - )) - - servers.sort() - cache[service_name] = list(servers) - except DomainError as e: - # We failed to resolve the name (other than a NameError) - # Try something in the cache, else rereaise - cache_entry = cache.get(service_name, None) - if cache_entry: - logger.warn( - "Failed to resolve %r, falling back to cache. %r", - service_name, e - ) - servers = list(cache_entry) - else: - raise e - - defer.returnValue(servers) diff --git a/synapse/http/federation/__init__.py b/synapse/http/federation/__init__.py new file mode 100644 index 0000000000..1453d04571 --- /dev/null +++ b/synapse/http/federation/__init__.py @@ -0,0 +1,14 @@ +# -*- coding: utf-8 -*- +# Copyright 2019 New Vector Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/synapse/http/federation/srv_resolver.py b/synapse/http/federation/srv_resolver.py new file mode 100644 index 0000000000..c49b82c394 --- /dev/null +++ b/synapse/http/federation/srv_resolver.py @@ -0,0 +1,124 @@ +# -*- coding: utf-8 -*- +# Copyright 2014-2016 OpenMarket Ltd +# Copyright 2019 New Vector Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +import time + +import attr + +from twisted.internet import defer +from twisted.internet.error import ConnectError +from twisted.names import client, dns +from twisted.names.error import DNSNameError, DomainError + +from synapse.util.logcontext import make_deferred_yieldable + +logger = logging.getLogger(__name__) + +SERVER_CACHE = {} + + +@attr.s +class Server(object): + """ + Our record of an individual server which can be tried to reach a destination. + + Attributes: + host (bytes): target hostname + port (int): + priority (int): + weight (int): + expires (int): when the cache should expire this record - in *seconds* since + the epoch + """ + host = attr.ib() + port = attr.ib() + priority = attr.ib(default=0) + weight = attr.ib(default=0) + expires = attr.ib(default=0) + + +@defer.inlineCallbacks +def resolve_service(service_name, dns_client=client, cache=SERVER_CACHE, clock=time): + """Look up a SRV record, with caching + + The default resolver in twisted.names doesn't do any caching (it has a CacheResolver, + but the cache never gets populated), so we add our own caching layer here. + + Args: + service_name (unicode|bytes): record to look up + dns_client (twisted.internet.interfaces.IResolver): twisted resolver impl + cache (dict): cache object + clock (object): clock implementation. must provide a time() method. + + Returns: + Deferred[list[Server]]: a list of the SRV records, or an empty list if none found + """ + # TODO: the dns client handles both unicode names (encoding via idna) and pre-encoded + # byteses; however they will obviously end up as separate entries in the cache. We + # should pick one form and stick with it. + cache_entry = cache.get(service_name, None) + if cache_entry: + if all(s.expires > int(clock.time()) for s in cache_entry): + servers = list(cache_entry) + defer.returnValue(servers) + + try: + answers, _, _ = yield make_deferred_yieldable( + dns_client.lookupService(service_name), + ) + except DNSNameError: + # TODO: cache this. We can get the SOA out of the exception, and use + # the negative-TTL value. + defer.returnValue([]) + except DomainError as e: + # We failed to resolve the name (other than a NameError) + # Try something in the cache, else rereaise + cache_entry = cache.get(service_name, None) + if cache_entry: + logger.warn( + "Failed to resolve %r, falling back to cache. %r", + service_name, e + ) + defer.returnValue(list(cache_entry)) + else: + raise e + + if (len(answers) == 1 + and answers[0].type == dns.SRV + and answers[0].payload + and answers[0].payload.target == dns.Name(b'.')): + raise ConnectError("Service %s unavailable" % service_name) + + servers = [] + + for answer in answers: + if answer.type != dns.SRV or not answer.payload: + continue + + payload = answer.payload + + servers.append(Server( + host=payload.target.name, + port=payload.port, + priority=payload.priority, + weight=payload.weight, + expires=int(clock.time()) + answer.ttl, + )) + + servers.sort() # FIXME: get rid of this (it's broken by the attrs change) + cache[service_name] = list(servers) + defer.returnValue(servers) diff --git a/tests/http/federation/__init__.py b/tests/http/federation/__init__.py new file mode 100644 index 0000000000..1453d04571 --- /dev/null +++ b/tests/http/federation/__init__.py @@ -0,0 +1,14 @@ +# -*- coding: utf-8 -*- +# Copyright 2019 New Vector Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/tests/test_dns.py b/tests/http/federation/test_srv_resolver.py index 90bd34be34..1271a495e1 100644 --- a/tests/test_dns.py +++ b/tests/http/federation/test_srv_resolver.py @@ -1,5 +1,6 @@ # -*- coding: utf-8 -*- # Copyright 2014-2016 OpenMarket Ltd +# Copyright 2019 New Vector Ltd # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -16,40 +17,63 @@ from mock import Mock from twisted.internet import defer +from twisted.internet.defer import Deferred +from twisted.internet.error import ConnectError from twisted.names import dns, error -from synapse.http.endpoint import resolve_service +from synapse.http.federation.srv_resolver import resolve_service +from synapse.util.logcontext import LoggingContext +from tests import unittest from tests.utils import MockClock -from . import unittest - -@unittest.DEBUG -class DnsTestCase(unittest.TestCase): - @defer.inlineCallbacks +class SrvResolverTestCase(unittest.TestCase): def test_resolve(self): dns_client_mock = Mock() - service_name = "test_service.example.com" - host_name = "example.com" + service_name = b"test_service.example.com" + host_name = b"example.com" answer_srv = dns.RRHeader( type=dns.SRV, payload=dns.Record_SRV(target=host_name) ) - dns_client_mock.lookupService.return_value = defer.succeed( - ([answer_srv], None, None) - ) + result_deferred = Deferred() + dns_client_mock.lookupService.return_value = result_deferred cache = {} - servers = yield resolve_service( - service_name, dns_client=dns_client_mock, cache=cache - ) + @defer.inlineCallbacks + def do_lookup(): + with LoggingContext("one") as ctx: + resolve_d = resolve_service( + service_name, dns_client=dns_client_mock, cache=cache + ) + + self.assertNoResult(resolve_d) + + # should have reset to the sentinel context + self.assertIs(LoggingContext.current_context(), LoggingContext.sentinel) + + result = yield resolve_d + + # should have restored our context + self.assertIs(LoggingContext.current_context(), ctx) + + defer.returnValue(result) + + test_d = do_lookup() + self.assertNoResult(test_d) dns_client_mock.lookupService.assert_called_once_with(service_name) + result_deferred.callback( + ([answer_srv], None, None) + ) + + servers = self.successResultOf(test_d) + self.assertEquals(len(servers), 1) self.assertEquals(servers, cache[service_name]) self.assertEquals(servers[0].host, host_name) @@ -127,3 +151,59 @@ class DnsTestCase(unittest.TestCase): self.assertEquals(len(servers), 0) self.assertEquals(len(cache), 0) + + def test_disabled_service(self): + """ + test the behaviour when there is a single record which is ".". + """ + service_name = b"test_service.example.com" + + lookup_deferred = Deferred() + dns_client_mock = Mock() + dns_client_mock.lookupService.return_value = lookup_deferred + cache = {} + + resolve_d = resolve_service( + service_name, dns_client=dns_client_mock, cache=cache + ) + self.assertNoResult(resolve_d) + + # returning a single "." should make the lookup fail with a ConenctError + lookup_deferred.callback(( + [dns.RRHeader(type=dns.SRV, payload=dns.Record_SRV(target=b"."))], + None, + None, + )) + + self.failureResultOf(resolve_d, ConnectError) + + def test_non_srv_answer(self): + """ + test the behaviour when the dns server gives us a spurious non-SRV response + """ + service_name = b"test_service.example.com" + + lookup_deferred = Deferred() + dns_client_mock = Mock() + dns_client_mock.lookupService.return_value = lookup_deferred + cache = {} + + resolve_d = resolve_service( + service_name, dns_client=dns_client_mock, cache=cache + ) + self.assertNoResult(resolve_d) + + lookup_deferred.callback(( + [ + dns.RRHeader(type=dns.A, payload=dns.Record_A()), + dns.RRHeader(type=dns.SRV, payload=dns.Record_SRV(target=b"host")), + ], + None, + None, + )) + + servers = self.successResultOf(resolve_d) + + self.assertEquals(len(servers), 1) + self.assertEquals(servers, cache[service_name]) + self.assertEquals(servers[0].host, b"host") |