Merge pull request #4428 from matrix-org/rav/matrix_federation_agent

Move SRV magic into an Agent-like thing

3 files changed, 301 insertions, 24 deletions

diff --git a/tests/http/federation/test_matrix_federation_agent.py b/tests/http/federation/test_matrix_federation_agent.py
new file mode 100644
index 0000000000..eb963d80fb
--- /dev/null
+++ b/tests/http/federation/test_matrix_federation_agent.py
@@ -0,0 +1,183 @@
+# -*- 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.
+import logging
+
+from mock import Mock
+
+import treq
+
+from twisted.internet import defer
+from twisted.internet.protocol import Factory
+from twisted.protocols.tls import TLSMemoryBIOFactory
+from twisted.test.ssl_helpers import ServerTLSContext
+from twisted.web.http import HTTPChannel
+
+from synapse.crypto.context_factory import ClientTLSOptionsFactory
+from synapse.http.federation.matrix_federation_agent import MatrixFederationAgent
+from synapse.util.logcontext import LoggingContext
+
+from tests.server import FakeTransport, ThreadedMemoryReactorClock
+from tests.unittest import TestCase
+
+logger = logging.getLogger(__name__)
+
+
+class MatrixFederationAgentTests(TestCase):
+    def setUp(self):
+        self.reactor = ThreadedMemoryReactorClock()
+
+        self.mock_resolver = Mock()
+
+        self.agent = MatrixFederationAgent(
+            reactor=self.reactor,
+            tls_client_options_factory=ClientTLSOptionsFactory(None),
+            _srv_resolver=self.mock_resolver,
+        )
+
+    def _make_connection(self, client_factory):
+        """Builds a test server, and completes the outgoing client connection
+
+        Returns:
+            HTTPChannel: the test server
+        """
+
+        # build the test server
+        server_tls_protocol = _build_test_server()
+
+        # now, tell the client protocol factory to build the client protocol (it will be a
+        # _WrappingProtocol, around a TLSMemoryBIOProtocol, around an
+        # HTTP11ClientProtocol) and wire the output of said protocol up to the server via
+        # a FakeTransport.
+        #
+        # Normally this would be done by the TCP socket code in Twisted, but we are
+        # stubbing that out here.
+        client_protocol = client_factory.buildProtocol(None)
+        client_protocol.makeConnection(FakeTransport(server_tls_protocol, self.reactor))
+
+        # tell the server tls protocol to send its stuff back to the client, too
+        server_tls_protocol.makeConnection(FakeTransport(client_protocol, self.reactor))
+
+        # finally, give the reactor a pump to get the TLS juices flowing.
+        self.reactor.pump((0.1,))
+
+        # fish the test server back out of the server-side TLS protocol.
+        return server_tls_protocol.wrappedProtocol
+
+    @defer.inlineCallbacks
+    def _make_get_request(self, uri):
+        """
+        Sends a simple GET request via the agent, and checks its logcontext management
+        """
+        with LoggingContext("one") as context:
+            fetch_d = self.agent.request(b'GET', uri)
+
+            # Nothing happened yet
+            self.assertNoResult(fetch_d)
+
+            # should have reset logcontext to the sentinel
+            _check_logcontext(LoggingContext.sentinel)
+
+            try:
+                fetch_res = yield fetch_d
+                defer.returnValue(fetch_res)
+            finally:
+                _check_logcontext(context)
+
+    def test_get(self):
+        """
+        happy-path test of a GET request
+        """
+        self.reactor.lookups["testserv"] = "1.2.3.4"
+        test_d = self._make_get_request(b"matrix://testserv:8448/foo/bar")
+
+        # Nothing happened yet
+        self.assertNoResult(test_d)
+
+        # Make sure treq is trying to connect
+        clients = self.reactor.tcpClients
+        self.assertEqual(len(clients), 1)
+        (host, port, client_factory, _timeout, _bindAddress) = clients[0]
+        self.assertEqual(host, '1.2.3.4')
+        self.assertEqual(port, 8448)
+
+        # make a test server, and wire up the client
+        http_server = self._make_connection(client_factory)
+
+        self.assertEqual(len(http_server.requests), 1)
+        request = http_server.requests[0]
+        self.assertEqual(request.method, b'GET')
+        self.assertEqual(request.path, b'/foo/bar')
+        self.assertEqual(
+            request.requestHeaders.getRawHeaders(b'host'),
+            [b'testserv:8448']
+        )
+        content = request.content.read()
+        self.assertEqual(content, b'')
+
+        # Deferred is still without a result
+        self.assertNoResult(test_d)
+
+        # send the headers
+        request.responseHeaders.setRawHeaders(b'Content-Type', [b'application/json'])
+        request.write('')
+
+        self.reactor.pump((0.1,))
+
+        response = self.successResultOf(test_d)
+
+        # that should give us a Response object
+        self.assertEqual(response.code, 200)
+
+        # Send the body
+        request.write('{ "a": 1 }'.encode('ascii'))
+        request.finish()
+
+        self.reactor.pump((0.1,))
+
+        # check it can be read
+        json = self.successResultOf(treq.json_content(response))
+        self.assertEqual(json, {"a": 1})
+
+
+def _check_logcontext(context):
+    current = LoggingContext.current_context()
+    if current is not context:
+        raise AssertionError(
+            "Expected logcontext %s but was %s" % (context, current),
+        )
+
+
+def _build_test_server():
+    """Construct a test server
+
+    This builds an HTTP channel, wrapped with a TLSMemoryBIOProtocol
+
+    Returns:
+        TLSMemoryBIOProtocol
+    """
+    server_factory = Factory.forProtocol(HTTPChannel)
+    # Request.finish expects the factory to have a 'log' method.
+    server_factory.log = _log_request
+
+    server_tls_factory = TLSMemoryBIOFactory(
+        ServerTLSContext(), isClient=False, wrappedFactory=server_factory,
+    )
+
+    return server_tls_factory.buildProtocol(None)
+
+
+def _log_request(request):
+    """Implements Factory.log, which is expected by Request.finish"""
+    logger.info("Completed request %s", request)
diff --git a/tests/http/federation/test_srv_resolver.py b/tests/http/federation/test_srv_resolver.py
index 1271a495e1..a872e2441e 100644
--- a/tests/http/federation/test_srv_resolver.py
+++ b/tests/http/federation/test_srv_resolver.py
@@ -21,7 +21,7 @@ from twisted.internet.defer import Deferred
 from twisted.internet.error import ConnectError
 from twisted.names import dns, error
 
-from synapse.http.federation.srv_resolver import resolve_service
+from synapse.http.federation.srv_resolver import SrvResolver
 from synapse.util.logcontext import LoggingContext
 
 from tests import unittest
@@ -43,13 +43,13 @@ class SrvResolverTestCase(unittest.TestCase):
         dns_client_mock.lookupService.return_value = result_deferred
 
         cache = {}
+        resolver = SrvResolver(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
-                )
+                resolve_d = resolver.resolve_service(service_name)
 
                 self.assertNoResult(resolve_d)
 
@@ -83,16 +83,15 @@ class SrvResolverTestCase(unittest.TestCase):
         dns_client_mock = Mock()
         dns_client_mock.lookupService.return_value = defer.fail(error.DNSServerError())
 
-        service_name = "test_service.example.com"
+        service_name = b"test_service.example.com"
 
         entry = Mock(spec_set=["expires"])
         entry.expires = 0
 
         cache = {service_name: [entry]}
+        resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
 
-        servers = yield resolve_service(
-            service_name, dns_client=dns_client_mock, cache=cache
-        )
+        servers = yield resolver.resolve_service(service_name)
 
         dns_client_mock.lookupService.assert_called_once_with(service_name)
 
@@ -106,17 +105,18 @@ class SrvResolverTestCase(unittest.TestCase):
         dns_client_mock = Mock(spec_set=['lookupService'])
         dns_client_mock.lookupService = Mock(spec_set=[])
 
-        service_name = "test_service.example.com"
+        service_name = b"test_service.example.com"
 
         entry = Mock(spec_set=["expires"])
         entry.expires = 999999999
 
         cache = {service_name: [entry]}
-
-        servers = yield resolve_service(
-            service_name, dns_client=dns_client_mock, cache=cache, clock=clock
+        resolver = SrvResolver(
+            dns_client=dns_client_mock, cache=cache, get_time=clock.time,
         )
 
+        servers = yield resolver.resolve_service(service_name)
+
         self.assertFalse(dns_client_mock.lookupService.called)
 
         self.assertEquals(len(servers), 1)
@@ -128,12 +128,13 @@ class SrvResolverTestCase(unittest.TestCase):
 
         dns_client_mock.lookupService.return_value = defer.fail(error.DNSServerError())
 
-        service_name = "test_service.example.com"
+        service_name = b"test_service.example.com"
 
         cache = {}
+        resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
 
         with self.assertRaises(error.DNSServerError):
-            yield resolve_service(service_name, dns_client=dns_client_mock, cache=cache)
+            yield resolver.resolve_service(service_name)
 
     @defer.inlineCallbacks
     def test_name_error(self):
@@ -141,13 +142,12 @@ class SrvResolverTestCase(unittest.TestCase):
 
         dns_client_mock.lookupService.return_value = defer.fail(error.DNSNameError())
 
-        service_name = "test_service.example.com"
+        service_name = b"test_service.example.com"
 
         cache = {}
+        resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
 
-        servers = yield resolve_service(
-            service_name, dns_client=dns_client_mock, cache=cache
-        )
+        servers = yield resolver.resolve_service(service_name)
 
         self.assertEquals(len(servers), 0)
         self.assertEquals(len(cache), 0)
@@ -162,10 +162,9 @@ class SrvResolverTestCase(unittest.TestCase):
         dns_client_mock = Mock()
         dns_client_mock.lookupService.return_value = lookup_deferred
         cache = {}
+        resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
 
-        resolve_d = resolve_service(
-            service_name, dns_client=dns_client_mock, cache=cache
-        )
+        resolve_d = resolver.resolve_service(service_name)
         self.assertNoResult(resolve_d)
 
         # returning a single "." should make the lookup fail with a ConenctError
@@ -187,10 +186,9 @@ class SrvResolverTestCase(unittest.TestCase):
         dns_client_mock = Mock()
         dns_client_mock.lookupService.return_value = lookup_deferred
         cache = {}
+        resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
 
-        resolve_d = resolve_service(
-            service_name, dns_client=dns_client_mock, cache=cache
-        )
+        resolve_d = resolver.resolve_service(service_name)
         self.assertNoResult(resolve_d)
 
         lookup_deferred.callback((
diff --git a/tests/http/test_fedclient.py b/tests/http/test_fedclient.py
index 8426eee400..d37f8f9981 100644
--- a/tests/http/test_fedclient.py
+++ b/tests/http/test_fedclient.py
@@ -15,6 +15,7 @@
 
 from mock import Mock
 
+from twisted.internet import defer
 from twisted.internet.defer import TimeoutError
 from twisted.internet.error import ConnectingCancelledError, DNSLookupError
 from twisted.test.proto_helpers import StringTransport
@@ -26,11 +27,20 @@ from synapse.http.matrixfederationclient import (
     MatrixFederationHttpClient,
     MatrixFederationRequest,
 )
+from synapse.util.logcontext import LoggingContext
 
 from tests.server import FakeTransport
 from tests.unittest import HomeserverTestCase
 
 
+def check_logcontext(context):
+    current = LoggingContext.current_context()
+    if current is not context:
+        raise AssertionError(
+            "Expected logcontext %s but was %s" % (context, current),
+        )
+
+
 class FederationClientTests(HomeserverTestCase):
     def make_homeserver(self, reactor, clock):
 
@@ -43,6 +53,70 @@ class FederationClientTests(HomeserverTestCase):
         self.cl = MatrixFederationHttpClient(self.hs)
         self.reactor.lookups["testserv"] = "1.2.3.4"
 
+    def test_client_get(self):
+        """
+        happy-path test of a GET request
+        """
+        @defer.inlineCallbacks
+        def do_request():
+            with LoggingContext("one") as context:
+                fetch_d = self.cl.get_json("testserv:8008", "foo/bar")
+
+                # Nothing happened yet
+                self.assertNoResult(fetch_d)
+
+                # should have reset logcontext to the sentinel
+                check_logcontext(LoggingContext.sentinel)
+
+                try:
+                    fetch_res = yield fetch_d
+                    defer.returnValue(fetch_res)
+                finally:
+                    check_logcontext(context)
+
+        test_d = do_request()
+
+        self.pump()
+
+        # Nothing happened yet
+        self.assertNoResult(test_d)
+
+        # Make sure treq is trying to connect
+        clients = self.reactor.tcpClients
+        self.assertEqual(len(clients), 1)
+        (host, port, factory, _timeout, _bindAddress) = clients[0]
+        self.assertEqual(host, '1.2.3.4')
+        self.assertEqual(port, 8008)
+
+        # complete the connection and wire it up to a fake transport
+        protocol = factory.buildProtocol(None)
+        transport = StringTransport()
+        protocol.makeConnection(transport)
+
+        # that should have made it send the request to the transport
+        self.assertRegex(transport.value(), b"^GET /foo/bar")
+
+        # Deferred is still without a result
+        self.assertNoResult(test_d)
+
+        # Send it the HTTP response
+        res_json = '{ "a": 1 }'.encode('ascii')
+        protocol.dataReceived(
+            b"HTTP/1.1 200 OK\r\n"
+            b"Server: Fake\r\n"
+            b"Content-Type: application/json\r\n"
+            b"Content-Length: %i\r\n"
+            b"\r\n"
+            b"%s" % (len(res_json), res_json)
+        )
+
+        self.pump()
+
+        res = self.successResultOf(test_d)
+
+        # check the response is as expected
+        self.assertEqual(res, {"a": 1})
+
     def test_dns_error(self):
         """
         If the DNS lookup returns an error, it will bubble up.
@@ -54,6 +128,28 @@ class FederationClientTests(HomeserverTestCase):
         self.assertIsInstance(f.value, RequestSendFailed)
         self.assertIsInstance(f.value.inner_exception, DNSLookupError)
 
+    def test_client_connection_refused(self):
+        d = self.cl.get_json("testserv:8008", "foo/bar", timeout=10000)
+
+        self.pump()
+
+        # Nothing happened yet
+        self.assertNoResult(d)
+
+        clients = self.reactor.tcpClients
+        self.assertEqual(len(clients), 1)
+        (host, port, factory, _timeout, _bindAddress) = clients[0]
+        self.assertEqual(host, '1.2.3.4')
+        self.assertEqual(port, 8008)
+        e = Exception("go away")
+        factory.clientConnectionFailed(None, e)
+        self.pump(0.5)
+
+        f = self.failureResultOf(d)
+
+        self.assertIsInstance(f.value, RequestSendFailed)
+        self.assertIs(f.value.inner_exception, e)
+
     def test_client_never_connect(self):
         """
         If the HTTP request is not connected and is timed out, it'll give a