text: # 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 base64
  import logging
  import os
  from typing import Iterable, Optional
  from unittest.mock import Mock, patch
  
  import treq
  from netaddr import IPSet
  from service_identity import VerificationError
  from zope.interface import implementer
  
  from twisted.internet import defer
  from twisted.internet._sslverify import ClientTLSOptions, OpenSSLCertificateOptions
  from twisted.internet.interfaces import IProtocolFactory
  from twisted.internet.protocol import Factory
  from twisted.protocols.tls import TLSMemoryBIOFactory, TLSMemoryBIOProtocol
  from twisted.web._newclient import ResponseNeverReceived
  from twisted.web.client import Agent
  from twisted.web.http import HTTPChannel, Request
  from twisted.web.http_headers import Headers
  from twisted.web.iweb import IPolicyForHTTPS
  
  from synapse.config.homeserver import HomeServerConfig
  from synapse.crypto.context_factory import FederationPolicyForHTTPS
  from synapse.http.federation.matrix_federation_agent import MatrixFederationAgent
  from synapse.http.federation.srv_resolver import Server
  from synapse.http.federation.well_known_resolver import (
      WELL_KNOWN_MAX_SIZE,
      WellKnownResolver,
      _cache_period_from_headers,
  )
  from synapse.logging.context import SENTINEL_CONTEXT, LoggingContext, current_context
  from synapse.util.caches.ttlcache import TTLCache
  
  from tests import unittest
  from tests.http import TestServerTLSConnectionFactory, get_test_ca_cert_file
  from tests.server import FakeTransport, ThreadedMemoryReactorClock
  from tests.utils import default_config
  
  logger = logging.getLogger(__name__)
  
  
  # Once Async Mocks or lambdas are supported this can go away.
  def generate_resolve_service(result):
      async def resolve_service(_):
          return result
  
      return resolve_service
  
  
  class MatrixFederationAgentTests(unittest.TestCase):
      def setUp(self):
          self.reactor = ThreadedMemoryReactorClock()
  
          self.mock_resolver = Mock()
  
          config_dict = default_config("test", parse=False)
          config_dict["federation_custom_ca_list"] = [get_test_ca_cert_file()]
  
          self._config = config = HomeServerConfig()
          config.parse_config_dict(config_dict, "", "")
  
          self.tls_factory = FederationPolicyForHTTPS(config)
  
          self.well_known_cache = TTLCache("test_cache", timer=self.reactor.seconds)
          self.had_well_known_cache = TTLCache("test_cache", timer=self.reactor.seconds)
          self.well_known_resolver = WellKnownResolver(
              self.reactor,
              Agent(self.reactor, contextFactory=self.tls_factory),
              b"test-agent",
              well_known_cache=self.well_known_cache,
              had_well_known_cache=self.had_well_known_cache,
          )
  
      def _make_connection(
          self,
          client_factory: IProtocolFactory,
          ssl: bool = True,
          expected_sni: bytes = None,
          tls_sanlist: Optional[Iterable[bytes]] = None,
      ) -> HTTPChannel:
          """Builds a test server, and completes the outgoing client connection
          Args:
              client_factory: the the factory that the
                  application is trying to use to make the outbound connection. We will
                  invoke it to build the client Protocol
  
              ssl: If true, we will expect an ssl connection and wrap
                  server_factory with a TLSMemoryBIOFactory
                  False is set only for when proxy expect http connection.
                  Otherwise federation requests use always https.
  
              expected_sni: the expected SNI value
  
              tls_sanlist: list of SAN entries for the TLS cert presented by the server.
  
          Returns:
              the server Protocol returned by server_factory
          """
  
          # build the test server
          server_factory = _get_test_protocol_factory()
          if ssl:
              server_factory = _wrap_server_factory_for_tls(server_factory, tls_sanlist)
  
          server_protocol = server_factory.buildProtocol(None)
  
          # 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_protocol, self.reactor, client_protocol)
          )
  
          # tell the server protocol to send its stuff back to the client, too
          server_protocol.makeConnection(
              FakeTransport(client_protocol, self.reactor, server_protocol)
          )
  
          if ssl:
              # fish the test server back out of the server-side TLS protocol.
              http_protocol = server_protocol.wrappedProtocol
              # grab a hold of the TLS connection, in case it gets torn down
              tls_connection = server_protocol._tlsConnection
          else:
              http_protocol = server_protocol
              tls_connection = None
  
          # give the reactor a pump to get the TLS juices flowing (if needed)
          self.reactor.advance(0)
  
          # check the SNI
          if expected_sni is not None:
              server_name = tls_connection.get_servername()
              self.assertEqual(
                  server_name,
                  expected_sni,
                  f"Expected SNI {expected_sni!s} but got {server_name!s}",
              )
  
          return http_protocol
  
      @defer.inlineCallbacks
      def _make_get_request(self, uri: bytes):
          """
          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(SENTINEL_CONTEXT)
  
              try:
                  fetch_res = yield fetch_d
                  return fetch_res
              except Exception as e:
                  logger.info("Fetch of %s failed: %s", uri.decode("ascii"), e)
                  raise
              finally:
                  _check_logcontext(context)
  
      def _handle_well_known_connection(
          self,
          client_factory: IProtocolFactory,
          expected_sni: bytes,
          content: bytes,
          response_headers: Optional[dict] = None,
      ) -> HTTPChannel:
          """Handle an outgoing HTTPs connection: wire it up to a server, check that the
          request is for a .well-known, and send the response.
  
          Args:
              client_factory: outgoing connection
              expected_sni: SNI that we expect the outgoing connection to send
              content: content to send back as the .well-known
          Returns:
              server impl
          """
          # make the connection for .well-known
          well_known_server = self._make_connection(
              client_factory, expected_sni=expected_sni
          )
          # check the .well-known request and send a response
          self.assertEqual(len(well_known_server.requests), 1)
          request = well_known_server.requests[0]
          self.assertEqual(
              request.requestHeaders.getRawHeaders(b"user-agent"), [b"test-agent"]
          )
          self._send_well_known_response(request, content, headers=response_headers or {})
          return well_known_server
  
      def _send_well_known_response(
          self,
          request: Request,
          content: bytes,
          headers: Optional[dict] = None,
      ):
          """Check that an incoming request looks like a valid .well-known request, and
          send back the response.
          """
          self.assertEqual(request.method, b"GET")
          self.assertEqual(request.path, b"/.well-known/matrix/server")
          self.assertEqual(request.requestHeaders.getRawHeaders(b"host"), [b"testserv"])
          # send back a response
          for k, v in (headers or {}).items():
              request.setHeader(k, v)
          request.write(content)
          request.finish()
  
          self.reactor.pump((0.1,))
  
      def _make_agent(self) -> MatrixFederationAgent:
          """
          If a proxy server is set, the MatrixFederationAgent must be created again
          because it is created too early during setUp
          """
          return MatrixFederationAgent(
              reactor=self.reactor,
              tls_client_options_factory=self.tls_factory,
              user_agent="test-agent",  # Note that this is unused since _well_known_resolver is provided.
              ip_whitelist=IPSet(),
              ip_blacklist=IPSet(),
              _srv_resolver=self.mock_resolver,
              _well_known_resolver=self.well_known_resolver,
          )
  
      def test_get(self):
          """happy-path test of a GET request with an explicit port"""
          self._do_get()
  
      @patch.dict(
          os.environ,
          {"https_proxy": "proxy.com", "no_proxy": "testserv"},
      )
      def test_get_bypass_proxy(self):
          """test of a GET request with an explicit port and bypass proxy"""
          self._do_get()
  
      def _do_get(self):
          """test of a GET request with an explicit port"""
          self.agent = self._make_agent()
  
          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, expected_sni=b"testserv")
  
          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"]
          )
          self.assertEqual(
              request.requestHeaders.getRawHeaders(b"user-agent"), [b"test-agent"]
          )
          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(b'{ "a": 1 }')
          request.finish()
  
          self.reactor.pump((0.1,))
  
          # check it can be read
          json = self.successResultOf(treq.json_content(response))
          self.assertEqual(json, {"a": 1})
  
      @patch.dict(
          os.environ, {"https_proxy": "http://proxy.com", "no_proxy": "unused.com"}
      )
      def test_get_via_http_proxy(self):
          """test for federation request through a http proxy"""
          self._do_get_via_proxy(expect_proxy_ssl=False, expected_auth_credentials=None)
  
      @patch.dict(
          os.environ,
          {"https_proxy": "http://user:pass@proxy.com", "no_proxy": "unused.com"},
      )
      def test_get_via_http_proxy_with_auth(self):
          """test for federation request through a http proxy with authentication"""
          self._do_get_via_proxy(
              expect_proxy_ssl=False, expected_auth_credentials=b"user:pass"
          )
  
      @patch.dict(
          os.environ, {"https_proxy": "https://proxy.com", "no_proxy": "unused.com"}
      )
      def test_get_via_https_proxy(self):
          """test for federation request through a https proxy"""
          self._do_get_via_proxy(expect_proxy_ssl=True, expected_auth_credentials=None)
  
      @patch.dict(
          os.environ,
          {"https_proxy": "https://user:pass@proxy.com", "no_proxy": "unused.com"},
      )
      def test_get_via_https_proxy_with_auth(self):
          """test for federation request through a https proxy with authentication"""
          self._do_get_via_proxy(
              expect_proxy_ssl=True, expected_auth_credentials=b"user:pass"
          )
  
      def _do_get_via_proxy(
          self,
          expect_proxy_ssl: bool = False,
          expected_auth_credentials: Optional[bytes] = None,
      ):
          """Send a https federation request via an agent and check that it is correctly
              received at the proxy and client. The proxy can use either http or https.
          Args:
              expect_proxy_ssl: True if we expect the request to connect to the proxy via https.
              expected_auth_credentials: credentials we expect to be presented to authenticate at the proxy
          """
          self.agent = self._make_agent()
  
          self.reactor.lookups["testserv"] = "1.2.3.4"
          self.reactor.lookups["proxy.com"] = "9.9.9.9"
          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]
          # make sure we are connecting to the proxy
          self.assertEqual(host, "9.9.9.9")
          self.assertEqual(port, 1080)
  
          # make a test server to act as the proxy, and wire up the client
          proxy_server = self._make_connection(
              client_factory,
              ssl=expect_proxy_ssl,
              tls_sanlist=[b"DNS:proxy.com"] if expect_proxy_ssl else None,
              expected_sni=b"proxy.com" if expect_proxy_ssl else None,
          )
  
          assert isinstance(proxy_server, HTTPChannel)
  
          # now there should be a pending CONNECT request
          self.assertEqual(len(proxy_server.requests), 1)
  
          request = proxy_server.requests[0]
          self.assertEqual(request.method, b"CONNECT")
          self.assertEqual(request.path, b"testserv:8448")
  
          # Check whether auth credentials have been supplied to the proxy
          proxy_auth_header_values = request.requestHeaders.getRawHeaders(
              b"Proxy-Authorization"
          )
  
          if expected_auth_credentials is not None:
              # Compute the correct header value for Proxy-Authorization
              encoded_credentials = base64.b64encode(expected_auth_credentials)
              expected_header_value = b"Basic " + encoded_credentials
  
              # Validate the header's value
              self.assertIn(expected_header_value, proxy_auth_header_values)
          else:
              # Check that the Proxy-Authorization header has not been supplied to the proxy
              self.assertIsNone(proxy_auth_header_values)
  
          # tell the proxy server not to close the connection
          proxy_server.persistent = True
  
          request.finish()
  
          # now we make another test server to act as the upstream HTTP server.
          server_ssl_protocol = _wrap_server_factory_for_tls(
              _get_test_protocol_factory()
          ).buildProtocol(None)
  
          # Tell the HTTP server to send outgoing traffic back via the proxy's transport.
          proxy_server_transport = proxy_server.transport
          server_ssl_protocol.makeConnection(proxy_server_transport)
  
          # ... and replace the protocol on the proxy's transport with the
          # TLSMemoryBIOProtocol for the test server, so that incoming traffic
          # to the proxy gets sent over to the HTTP(s) server.
  
          # See also comment at `_do_https_request_via_proxy`
          # in ../test_proxyagent.py for more details
          if expect_proxy_ssl:
              assert isinstance(proxy_server_transport, TLSMemoryBIOProtocol)
              proxy_server_transport.wrappedProtocol = server_ssl_protocol
          else:
              assert isinstance(proxy_server_transport, FakeTransport)
              client_protocol = proxy_server_transport.other
              c2s_transport = client_protocol.transport
              c2s_transport.other = server_ssl_protocol
  
          self.reactor.advance(0)
  
          server_name = server_ssl_protocol._tlsConnection.get_servername()
          expected_sni = b"testserv"
          self.assertEqual(
              server_name,
              expected_sni,
              f"Expected SNI {expected_sni!s} but got {server_name!s}",
          )
  
          # now there should be a pending request
          http_server = server_ssl_protocol.wrappedProtocol
          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"]
          )
          self.assertEqual(
              request.requestHeaders.getRawHeaders(b"user-agent"), [b"test-agent"]
          )
          # Check that the destination server DID NOT receive proxy credentials
          self.assertIsNone(request.requestHeaders.getRawHeaders(b"Proxy-Authorization"))
          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 test_get_ip_address(self):
          """
          Test the behaviour when the server name contains an explicit IP (with no port)
          """
          self.agent = self._make_agent()
  
          # there will be a getaddrinfo on the IP
          self.reactor.lookups["1.2.3.4"] = "1.2.3.4"
  
          test_d = self._make_get_request(b"matrix://1.2.3.4/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, expected_sni=None)
  
          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"1.2.3.4"])
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_get_ipv6_address(self):
          """
          Test the behaviour when the server name contains an explicit IPv6 address
          (with no port)
          """
          self.agent = self._make_agent()
  
          # there will be a getaddrinfo on the IP
          self.reactor.lookups["::1"] = "::1"
  
          test_d = self._make_get_request(b"matrix://[::1]/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")
          self.assertEqual(port, 8448)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(client_factory, expected_sni=None)
  
          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"[::1]"])
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_get_ipv6_address_with_port(self):
          """
          Test the behaviour when the server name contains an explicit IPv6 address
          (with explicit port)
          """
          self.agent = self._make_agent()
  
          # there will be a getaddrinfo on the IP
          self.reactor.lookups["::1"] = "::1"
  
          test_d = self._make_get_request(b"matrix://[::1]:80/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")
          self.assertEqual(port, 80)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(client_factory, expected_sni=None)
  
          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"[::1]:80"])
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_get_hostname_bad_cert(self):
          """
          Test the behaviour when the certificate on the server doesn't match the hostname
          """
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service([])
          self.reactor.lookups["testserv1"] = "1.2.3.4"
  
          test_d = self._make_get_request(b"matrix://testserv1/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # No SRV record lookup yet
          self.mock_resolver.resolve_service.assert_not_called()
  
          # there should be an attempt to connect on port 443 for the .well-known
          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, 443)
  
          # fonx the connection
          client_factory.clientConnectionFailed(None, Exception("nope"))
  
          # attemptdelay on the hostnameendpoint is 0.3, so takes that long before the
          # .well-known request fails.
          self.reactor.pump((0.4,))
  
          # now there should be a SRV lookup
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.testserv1"
          )
  
          # we should fall back to a direct connection
          self.assertEqual(len(clients), 2)
          (host, port, client_factory, _timeout, _bindAddress) = clients[1]
          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, expected_sni=b"testserv1")
  
          # there should be no requests
          self.assertEqual(len(http_server.requests), 0)
  
          # ... and the request should have failed
          e = self.failureResultOf(test_d, ResponseNeverReceived)
          failure_reason = e.value.reasons[0]
          self.assertIsInstance(failure_reason.value, VerificationError)
  
      def test_get_ip_address_bad_cert(self):
          """
          Test the behaviour when the server name contains an explicit IP, but
          the server cert doesn't cover it
          """
          self.agent = self._make_agent()
  
          # there will be a getaddrinfo on the IP
          self.reactor.lookups["1.2.3.5"] = "1.2.3.5"
  
          test_d = self._make_get_request(b"matrix://1.2.3.5/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.5")
          self.assertEqual(port, 8448)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(client_factory, expected_sni=None)
  
          # there should be no requests
          self.assertEqual(len(http_server.requests), 0)
  
          # ... and the request should have failed
          e = self.failureResultOf(test_d, ResponseNeverReceived)
          failure_reason = e.value.reasons[0]
          self.assertIsInstance(failure_reason.value, VerificationError)
  
      def test_get_no_srv_no_well_known(self):
          """
          Test the behaviour when the server name has no port, no SRV, and no well-known
          """
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service([])
          self.reactor.lookups["testserv"] = "1.2.3.4"
  
          test_d = self._make_get_request(b"matrix://testserv/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # No SRV record lookup yet
          self.mock_resolver.resolve_service.assert_not_called()
  
          # there should be an attempt to connect on port 443 for the .well-known
          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, 443)
  
          # fonx the connection
          client_factory.clientConnectionFailed(None, Exception("nope"))
  
          # attemptdelay on the hostnameendpoint is 0.3, so  takes that long before the
          # .well-known request fails.
          self.reactor.pump((0.4,))
  
          # now there should be a SRV lookup
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.testserv"
          )
  
          # we should fall back to a direct connection
          self.assertEqual(len(clients), 2)
          (host, port, client_factory, _timeout, _bindAddress) = clients[1]
          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, expected_sni=b"testserv")
  
          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"])
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_get_well_known(self):
          """Test the behaviour when the .well-known delegates elsewhere"""
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service([])
          self.reactor.lookups["testserv"] = "1.2.3.4"
          self.reactor.lookups["target-server"] = "1::f"
  
          test_d = self._make_get_request(b"matrix://testserv/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # there should be an attempt to connect on port 443 for the .well-known
          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, 443)
  
          self._handle_well_known_connection(
              client_factory,
              expected_sni=b"testserv",
              content=b'{ "m.server": "target-server" }',
          )
  
          # there should be a SRV lookup
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.target-server"
          )
  
          # now we should get a connection to the target server
          self.assertEqual(len(clients), 2)
          (host, port, client_factory, _timeout, _bindAddress) = clients[1]
          self.assertEqual(host, "1::f")
          self.assertEqual(port, 8448)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(
              client_factory, expected_sni=b"target-server"
          )
  
          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"target-server"]
          )
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
          self.assertEqual(self.well_known_cache[b"testserv"], b"target-server")
  
          # check the cache expires
          self.reactor.pump((48 * 3600,))
          self.well_known_cache.expire()
          self.assertNotIn(b"testserv", self.well_known_cache)
  
      def test_get_well_known_redirect(self):
          """Test the behaviour when the server name has no port and no SRV record, but
          the .well-known has a 300 redirect
          """
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service([])
          self.reactor.lookups["testserv"] = "1.2.3.4"
          self.reactor.lookups["target-server"] = "1::f"
  
          test_d = self._make_get_request(b"matrix://testserv/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # there should be an attempt to connect on port 443 for the .well-known
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop()
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 443)
  
          redirect_server = self._make_connection(
              client_factory, expected_sni=b"testserv"
          )
  
          # send a 302 redirect
          self.assertEqual(len(redirect_server.requests), 1)
          request = redirect_server.requests[0]
          request.redirect(b"https://testserv/even_better_known")
          request.finish()
  
          self.reactor.pump((0.1,))
  
          # now there should be another connection
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop()
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 443)
  
          well_known_server = self._make_connection(
              client_factory, expected_sni=b"testserv"
          )
  
          self.assertEqual(len(well_known_server.requests), 1, "No request after 302")
          request = well_known_server.requests[0]
          self.assertEqual(request.method, b"GET")
          self.assertEqual(request.path, b"/even_better_known")
          request.write(b'{ "m.server": "target-server" }')
          request.finish()
  
          self.reactor.pump((0.1,))
  
          # there should be a SRV lookup
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.target-server"
          )
  
          # now we should get a connection to the target server
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients[0]
          self.assertEqual(host, "1::f")
          self.assertEqual(port, 8448)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(
              client_factory, expected_sni=b"target-server"
          )
  
          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"target-server"]
          )
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
          self.assertEqual(self.well_known_cache[b"testserv"], b"target-server")
  
          # check the cache expires
          self.reactor.pump((48 * 3600,))
          self.well_known_cache.expire()
          self.assertNotIn(b"testserv", self.well_known_cache)
  
      def test_get_invalid_well_known(self):
          """
          Test the behaviour when the server name has an *invalid* well-known (and no SRV)
          """
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service([])
          self.reactor.lookups["testserv"] = "1.2.3.4"
  
          test_d = self._make_get_request(b"matrix://testserv/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # No SRV record lookup yet
          self.mock_resolver.resolve_service.assert_not_called()
  
          # there should be an attempt to connect on port 443 for the .well-known
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop()
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 443)
  
          self._handle_well_known_connection(
              client_factory, expected_sni=b"testserv", content=b"NOT JSON"
          )
  
          # now there should be a SRV lookup
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.testserv"
          )
  
          # we should fall back to a direct connection
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop()
          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, expected_sni=b"testserv")
  
          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"])
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_get_well_known_unsigned_cert(self):
          """Test the behaviour when the .well-known server presents a cert
          not signed by a CA
          """
  
          # we use the same test server as the other tests, but use an agent with
          # the config left to the default, which will not trust it (since the
          # presented cert is signed by a test CA)
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service([])
          self.reactor.lookups["testserv"] = "1.2.3.4"
  
          config = default_config("test", parse=True)
  
          # Build a new agent and WellKnownResolver with a different tls factory
          tls_factory = FederationPolicyForHTTPS(config)
          agent = MatrixFederationAgent(
              reactor=self.reactor,
              tls_client_options_factory=tls_factory,
              user_agent=b"test-agent",  # This is unused since _well_known_resolver is passed below.
              ip_whitelist=IPSet(),
              ip_blacklist=IPSet(),
              _srv_resolver=self.mock_resolver,
              _well_known_resolver=WellKnownResolver(
                  self.reactor,
                  Agent(self.reactor, contextFactory=tls_factory),
                  b"test-agent",
                  well_known_cache=self.well_known_cache,
                  had_well_known_cache=self.had_well_known_cache,
              ),
          )
  
          test_d = agent.request(b"GET", b"matrix://testserv/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # there should be an attempt to connect on port 443 for the .well-known
          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, 443)
  
          http_proto = self._make_connection(client_factory, expected_sni=b"testserv")
  
          # there should be no requests
          self.assertEqual(len(http_proto.requests), 0)
  
          # and there should be a SRV lookup instead
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.testserv"
          )
  
      def test_get_hostname_srv(self):
          """
          Test the behaviour when there is a single SRV record
          """
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service(
              [Server(host=b"srvtarget", port=8443)]
          )
          self.reactor.lookups["srvtarget"] = "1.2.3.4"
  
          test_d = self._make_get_request(b"matrix://testserv/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # the request for a .well-known will have failed with a DNS lookup error.
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.testserv"
          )
  
          # 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, 8443)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(client_factory, expected_sni=b"testserv")
  
          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"])
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_get_well_known_srv(self):
          """Test the behaviour when the .well-known redirects to a place where there
          is a SRV.
          """
          self.agent = self._make_agent()
  
          self.reactor.lookups["testserv"] = "1.2.3.4"
          self.reactor.lookups["srvtarget"] = "5.6.7.8"
  
          test_d = self._make_get_request(b"matrix://testserv/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # there should be an attempt to connect on port 443 for the .well-known
          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, 443)
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service(
              [Server(host=b"srvtarget", port=8443)]
          )
  
          self._handle_well_known_connection(
              client_factory,
              expected_sni=b"testserv",
              content=b'{ "m.server": "target-server" }',
          )
  
          # there should be a SRV lookup
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.target-server"
          )
  
          # now we should get a connection to the target of the SRV record
          self.assertEqual(len(clients), 2)
          (host, port, client_factory, _timeout, _bindAddress) = clients[1]
          self.assertEqual(host, "5.6.7.8")
          self.assertEqual(port, 8443)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(
              client_factory, expected_sni=b"target-server"
          )
  
          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"target-server"]
          )
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_idna_servername(self):
          """test the behaviour when the server name has idna chars in"""
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service([])
  
          # the resolver is always called with the IDNA hostname as a native string.
          self.reactor.lookups["xn--bcher-kva.com"] = "1.2.3.4"
  
          # this is idna for bücher.com
          test_d = self._make_get_request(b"matrix://xn--bcher-kva.com/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          # No SRV record lookup yet
          self.mock_resolver.resolve_service.assert_not_called()
  
          # there should be an attempt to connect on port 443 for the .well-known
          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, 443)
  
          # fonx the connection
          client_factory.clientConnectionFailed(None, Exception("nope"))
  
          # attemptdelay on the hostnameendpoint is 0.3, so  takes that long before the
          # .well-known request fails.
          self.reactor.pump((0.4,))
  
          # now there should have been a SRV lookup
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.xn--bcher-kva.com"
          )
  
          # We should fall back to port 8448
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 2)
          (host, port, client_factory, _timeout, _bindAddress) = clients[1]
          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, expected_sni=b"xn--bcher-kva.com"
          )
  
          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"xn--bcher-kva.com"]
          )
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_idna_srv_target(self):
          """test the behaviour when the target of a SRV record has idna chars"""
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service(
              [Server(host=b"xn--trget-3qa.com", port=8443)]  # târget.com
          )
          self.reactor.lookups["xn--trget-3qa.com"] = "1.2.3.4"
  
          test_d = self._make_get_request(b"matrix://xn--bcher-kva.com/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.xn--bcher-kva.com"
          )
  
          # 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, 8443)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(
              client_factory, expected_sni=b"xn--bcher-kva.com"
          )
  
          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"xn--bcher-kva.com"]
          )
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
      def test_well_known_cache(self):
          self.reactor.lookups["testserv"] = "1.2.3.4"
  
          fetch_d = defer.ensureDeferred(
              self.well_known_resolver.get_well_known(b"testserv")
          )
  
          # there should be an attempt to connect on port 443 for the .well-known
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop(0)
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 443)
  
          well_known_server = self._handle_well_known_connection(
              client_factory,
              expected_sni=b"testserv",
              response_headers={b"Cache-Control": b"max-age=1000"},
              content=b'{ "m.server": "target-server" }',
          )
  
          r = self.successResultOf(fetch_d)
          self.assertEqual(r.delegated_server, b"target-server")
  
          # close the tcp connection
          well_known_server.loseConnection()
  
          # repeat the request: it should hit the cache
          fetch_d = defer.ensureDeferred(
              self.well_known_resolver.get_well_known(b"testserv")
          )
          r = self.successResultOf(fetch_d)
          self.assertEqual(r.delegated_server, b"target-server")
  
          # expire the cache
          self.reactor.pump((1000.0,))
  
          # now it should connect again
          fetch_d = defer.ensureDeferred(
              self.well_known_resolver.get_well_known(b"testserv")
          )
  
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop(0)
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 443)
  
          self._handle_well_known_connection(
              client_factory,
              expected_sni=b"testserv",
              content=b'{ "m.server": "other-server" }',
          )
  
          r = self.successResultOf(fetch_d)
          self.assertEqual(r.delegated_server, b"other-server")
  
      def test_well_known_cache_with_temp_failure(self):
          """Test that we refetch well-known before the cache expires, and that
          it ignores transient errors.
          """
  
          self.reactor.lookups["testserv"] = "1.2.3.4"
  
          fetch_d = defer.ensureDeferred(
              self.well_known_resolver.get_well_known(b"testserv")
          )
  
          # there should be an attempt to connect on port 443 for the .well-known
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop(0)
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 443)
  
          well_known_server = self._handle_well_known_connection(
              client_factory,
              expected_sni=b"testserv",
              response_headers={b"Cache-Control": b"max-age=1000"},
              content=b'{ "m.server": "target-server" }',
          )
  
          r = self.successResultOf(fetch_d)
          self.assertEqual(r.delegated_server, b"target-server")
  
          # close the tcp connection
          well_known_server.loseConnection()
  
          # Get close to the cache expiry, this will cause the resolver to do
          # another lookup.
          self.reactor.pump((900.0,))
  
          fetch_d = defer.ensureDeferred(
              self.well_known_resolver.get_well_known(b"testserv")
          )
  
          # The resolver may retry a few times, so fonx all requests that come along
          attempts = 0
          while self.reactor.tcpClients:
              clients = self.reactor.tcpClients
              (host, port, client_factory, _timeout, _bindAddress) = clients.pop(0)
  
              attempts += 1
  
              # fonx the connection attempt, this will be treated as a temporary
              # failure.
              client_factory.clientConnectionFailed(None, Exception("nope"))
  
              # There's a few sleeps involved, so we have to pump the reactor a
              # bit.
              self.reactor.pump((1.0, 1.0))
  
          # We expect to see more than one attempt as there was previously a valid
          # well known.
          self.assertGreater(attempts, 1)
  
          # Resolver should return cached value, despite the lookup failing.
          r = self.successResultOf(fetch_d)
          self.assertEqual(r.delegated_server, b"target-server")
  
          # Expire both caches and repeat the request
          self.reactor.pump((10000.0,))
  
          # Repeat the request, this time it should fail if the lookup fails.
          fetch_d = defer.ensureDeferred(
              self.well_known_resolver.get_well_known(b"testserv")
          )
  
          clients = self.reactor.tcpClients
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop(0)
          client_factory.clientConnectionFailed(None, Exception("nope"))
          self.reactor.pump((0.4,))
  
          r = self.successResultOf(fetch_d)
          self.assertEqual(r.delegated_server, None)
  
      def test_well_known_too_large(self):
          """A well-known query that returns a result which is too large should be rejected."""
          self.reactor.lookups["testserv"] = "1.2.3.4"
  
          fetch_d = defer.ensureDeferred(
              self.well_known_resolver.get_well_known(b"testserv")
          )
  
          # there should be an attempt to connect on port 443 for the .well-known
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop(0)
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 443)
  
          self._handle_well_known_connection(
              client_factory,
              expected_sni=b"testserv",
              response_headers={b"Cache-Control": b"max-age=1000"},
              content=b'{ "m.server": "' + (b"a" * WELL_KNOWN_MAX_SIZE) + b'" }',
          )
  
          # The result is successful, but disabled delegation.
          r = self.successResultOf(fetch_d)
          self.assertIsNone(r.delegated_server)
  
      def test_srv_fallbacks(self):
          """Test that other SRV results are tried if the first one fails."""
          self.agent = self._make_agent()
  
          self.mock_resolver.resolve_service.side_effect = generate_resolve_service(
              [
                  Server(host=b"target.com", port=8443),
                  Server(host=b"target.com", port=8444),
              ]
          )
          self.reactor.lookups["target.com"] = "1.2.3.4"
  
          test_d = self._make_get_request(b"matrix://testserv/foo/bar")
  
          # Nothing happened yet
          self.assertNoResult(test_d)
  
          self.mock_resolver.resolve_service.assert_called_once_with(
              b"_matrix._tcp.testserv"
          )
  
          # We should see an attempt to connect to the first server
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop(0)
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 8443)
  
          # Fonx the connection
          client_factory.clientConnectionFailed(None, Exception("nope"))
  
          # There's a 300ms delay in HostnameEndpoint
          self.reactor.pump((0.4,))
  
          # Hasn't failed yet
          self.assertNoResult(test_d)
  
          # We shouldnow see an attempt to connect to the second server
          clients = self.reactor.tcpClients
          self.assertEqual(len(clients), 1)
          (host, port, client_factory, _timeout, _bindAddress) = clients.pop(0)
          self.assertEqual(host, "1.2.3.4")
          self.assertEqual(port, 8444)
  
          # make a test server, and wire up the client
          http_server = self._make_connection(client_factory, expected_sni=b"testserv")
  
          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"])
  
          # finish the request
          request.finish()
          self.reactor.pump((0.1,))
          self.successResultOf(test_d)
  
  
  class TestCachePeriodFromHeaders(unittest.TestCase):
      def test_cache_control(self):
          # uppercase
          self.assertEqual(
              _cache_period_from_headers(
                  Headers({b"Cache-Control": [b"foo, Max-Age = 100, bar"]})
              ),
              100,
          )
  
          # missing value
          self.assertIsNone(
              _cache_period_from_headers(Headers({b"Cache-Control": [b"max-age=, bar"]}))
          )
  
          # hackernews: bogus due to semicolon
          self.assertIsNone(
              _cache_period_from_headers(
                  Headers({b"Cache-Control": [b"private; max-age=0"]})
              )
          )
  
          # github
          self.assertEqual(
              _cache_period_from_headers(
                  Headers({b"Cache-Control": [b"max-age=0, private, must-revalidate"]})
              ),
              0,
          )
  
          # google
          self.assertEqual(
              _cache_period_from_headers(
                  Headers({b"cache-control": [b"private, max-age=0"]})
              ),
              0,
          )
  
      def test_expires(self):
          self.assertEqual(
              _cache_period_from_headers(
                  Headers({b"Expires": [b"Wed, 30 Jan 2019 07:35:33 GMT"]}),
                  time_now=lambda: 1548833700,
              ),
              33,
          )
  
          # cache-control overrides expires
          self.assertEqual(
              _cache_period_from_headers(
                  Headers(
                      {
                          b"cache-control": [b"max-age=10"],
                          b"Expires": [b"Wed, 30 Jan 2019 07:35:33 GMT"],
                      }
                  ),
                  time_now=lambda: 1548833700,
              ),
              10,
          )
  
          # invalid expires means immediate expiry
          self.assertEqual(_cache_period_from_headers(Headers({b"Expires": [b"0"]})), 0)
  
  
  def _check_logcontext(context):
      current = current_context()
      if current is not context:
          raise AssertionError("Expected logcontext %s but was %s" % (context, current))
  
  
  def _wrap_server_factory_for_tls(
      factory: IProtocolFactory, sanlist: Iterable[bytes] = None
  ) -> IProtocolFactory:
      """Wrap an existing Protocol Factory with a test TLSMemoryBIOFactory
      The resultant factory will create a TLS server which presents a certificate
      signed by our test CA, valid for the domains in `sanlist`
      Args:
          factory: protocol factory to wrap
          sanlist: list of domains the cert should be valid for
      Returns:
          interfaces.IProtocolFactory
      """
      if sanlist is None:
          sanlist = [
              b"DNS:testserv",
              b"DNS:target-server",
              b"DNS:xn--bcher-kva.com",
              b"IP:1.2.3.4",
              b"IP:::1",
          ]
  
      connection_creator = TestServerTLSConnectionFactory(sanlist=sanlist)
      return TLSMemoryBIOFactory(
          connection_creator, isClient=False, wrappedFactory=factory
      )
  
  
  def _get_test_protocol_factory() -> IProtocolFactory:
      """Get a protocol Factory which will build an HTTPChannel
      Returns:
          interfaces.IProtocolFactory
      """
      server_factory = Factory.forProtocol(HTTPChannel)
  
      # Request.finish expects the factory to have a 'log' method.
      server_factory.log = _log_request
  
      return server_factory
  
  
  def _log_request(request: str):
      """Implements Factory.log, which is expected by Request.finish"""
      logger.info(f"Completed request {request}")
  
  
  @implementer(IPolicyForHTTPS)
  class TrustingTLSPolicyForHTTPS:
      """An IPolicyForHTTPS which checks that the certificate belongs to the
      right server, but doesn't check the certificate chain."""
  
      def creatorForNetloc(self, hostname, port):
          certificateOptions = OpenSSLCertificateOptions()
          return ClientTLSOptions(hostname, certificateOptions.getContext())
  