Support using SSL on worker endpoints.

changelog.d/14128.misc

@@ -0,0 +1 @@
+Add TLS support for generic worker endpoints.

synapse/app/_base.py

@@ -47,6 +47,7 @@
 from twisted.logger import LoggingFile, LogLevel
 from twisted.protocols.tls import TLSMemoryBIOFactory
 from twisted.python.threadpool import ThreadPool
+from twisted.web.resource import Resource
 
 import synapse.util.caches
 from synapse.api.constants import MAX_PDU_SIZE
@@ -55,12 +56,13 @@
 from synapse.config import ConfigError
 from synapse.config._base import format_config_error
 from synapse.config.homeserver import HomeServerConfig
-from synapse.config.server import ManholeConfig
+from synapse.config.server import ListenerConfig, ManholeConfig
 from synapse.crypto import context_factory
 from synapse.events.presence_router import load_legacy_presence_router
 from synapse.events.spamcheck import load_legacy_spam_checkers
 from synapse.events.third_party_rules import load_legacy_third_party_event_rules
 from synapse.handlers.auth import load_legacy_password_auth_providers
+from synapse.http.site import SynapseSite
 from synapse.logging.context import PreserveLoggingContext
 from synapse.logging.opentracing import init_tracer
 from synapse.metrics import install_gc_manager, register_threadpool
@@ -357,6 +359,55 @@ def listen_tcp(
     return r  # type: ignore[return-value]
 
 
+def listen_http(
+    listener_config: ListenerConfig,
+    root_resource: Resource,
+    version_string: str,
+    max_request_body_size: int,
+    context_factory: IOpenSSLContextFactory,
+    reactor: IReactorSSL = reactor,
+) -> List[Port]:
+    port = listener_config.port
+    bind_addresses = listener_config.bind_addresses
+    tls = listener_config.tls
+
+    assert listener_config.http_options is not None
+
+    site_tag = listener_config.http_options.tag
+    if site_tag is None:
+        site_tag = str(port)
+
+    site = SynapseSite(
+        "synapse.access.%s.%s" % ("https" if tls else "http", site_tag),
+        site_tag,
+        listener_config,
+        root_resource,
+        version_string,
+        max_request_body_size=max_request_body_size,
+        reactor=reactor,
+    )
+    if tls:
+        # refresh_certificate should have been called before this.
+        assert context_factory is not None
+        ports = listen_ssl(
+            bind_addresses,
+            port,
+            site,
+            context_factory,
+            reactor=reactor,
+        )
+        logger.info("Synapse now listening on TCP port %d (TLS)", port)
+    else:
+        ports = listen_tcp(
+            bind_addresses,
+            port,
+            site,
+            reactor=reactor,
+        )
+        logger.info("Synapse now listening on TCP port %d", port)
+    return ports
+
+
 def listen_ssl(
     bind_addresses: Collection[str],
     port: int,

synapse/app/generic_worker.py

@@ -44,7 +44,7 @@
 from synapse.federation.transport.server import TransportLayerServer
 from synapse.http.server import JsonResource, OptionsResource
 from synapse.http.servlet import RestServlet, parse_json_object_from_request
-from synapse.http.site import SynapseRequest, SynapseSite
+from synapse.http.site import SynapseRequest
 from synapse.logging.context import LoggingContext
 from synapse.metrics import METRICS_PREFIX, MetricsResource, RegistryProxy
 from synapse.replication.http import REPLICATION_PREFIX, ReplicationRestResource
@@ -288,15 +288,9 @@ class GenericWorkerServer(HomeServer):
     DATASTORE_CLASS = GenericWorkerSlavedStore  # type: ignore
 
     def _listen_http(self, listener_config: ListenerConfig) -> None:
-        port = listener_config.port
-        bind_addresses = listener_config.bind_addresses
 
         assert listener_config.http_options is not None
 
-        site_tag = listener_config.http_options.tag
-        if site_tag is None:
-            site_tag = str(port)
-
         # We always include a health resource.
         resources: Dict[str, Resource] = {"/health": HealthResource()}
 
@@ -395,23 +389,15 @@ def _listen_http(self, listener_config: ListenerConfig) -> None:
 
         root_resource = create_resource_tree(resources, OptionsResource())
 
-        _base.listen_tcp(
-            bind_addresses,
-            port,
-            SynapseSite(
-                "synapse.access.http.%s" % (site_tag,),
-                site_tag,
-                listener_config,
-                root_resource,
-                self.version_string,
-                max_request_body_size=max_request_body_size(self.config),
-                reactor=self.get_reactor(),
-            ),
+        _base.listen_http(
+            listener_config,
+            root_resource,
+            self.version_string,
+            max_request_body_size(self.config),
+            self.tls_server_context_factory,
             reactor=self.get_reactor(),
         )
 
-        logger.info("Synapse worker now listening on port %d", port)
-
     def start_listening(self) -> None:
         for listener in self.config.worker.worker_listeners:
             if listener.type == "http":

synapse/app/homeserver.py

@@ -37,8 +37,7 @@
 from synapse.app import _base
 from synapse.app._base import (
     handle_startup_exception,
-    listen_ssl,
-    listen_tcp,
+    listen_http,
     max_request_body_size,
     redirect_stdio_to_logs,
     register_start,
@@ -53,7 +52,6 @@
     RootOptionsRedirectResource,
     StaticResource,
 )
-from synapse.http.site import SynapseSite
 from synapse.logging.context import LoggingContext
 from synapse.metrics import METRICS_PREFIX, MetricsResource, RegistryProxy
 from synapse.replication.http import REPLICATION_PREFIX, ReplicationRestResource
@@ -83,8 +81,6 @@ def _listener_http(
         self, config: HomeServerConfig, listener_config: ListenerConfig
     ) -> Iterable[Port]:
         port = listener_config.port
-        bind_addresses = listener_config.bind_addresses
-        tls = listener_config.tls
         # Must exist since this is an HTTP listener.
         assert listener_config.http_options is not None
         site_tag = listener_config.http_options.tag
@@ -140,37 +136,15 @@ def _listener_http(
         else:
             root_resource = OptionsResource()
 
-        site = SynapseSite(
-            "synapse.access.%s.%s" % ("https" if tls else "http", site_tag),
-            site_tag,
+        ports = listen_http(
             listener_config,
             create_resource_tree(resources, root_resource),
             self.version_string,
-            max_request_body_size=max_request_body_size(self.config),
+            max_request_body_size(self.config),
+            self.tls_server_context_factory,
             reactor=self.get_reactor(),
         )
 
-        if tls:
-            # refresh_certificate should have been called before this.
-            assert self.tls_server_context_factory is not None
-            ports = listen_ssl(
-                bind_addresses,
-                port,
-                site,
-                self.tls_server_context_factory,
-                reactor=self.get_reactor(),
-            )
-            logger.info("Synapse now listening on TCP port %d (TLS)", port)
-
-        else:
-            ports = listen_tcp(
-                bind_addresses,
-                port,
-                site,
-                reactor=self.get_reactor(),
-            )
-            logger.info("Synapse now listening on TCP port %d", port)
-
         return ports
 
     def _configure_named_resource(

synapse/config/workers.py

@@ -67,6 +67,7 @@ class InstanceLocationConfig:
 
     host: str
     port: int
+    tls: bool = False
 
 
 @attr.s
@@ -149,6 +150,12 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
         # The port on the main synapse for HTTP replication endpoint
         self.worker_replication_http_port = config.get("worker_replication_http_port")
 
+        # The tls mode on the main synapse for HTTP replication endpoint.
+        # For backward compatibility this defaults to False.
+        self.worker_replication_http_tls = config.get(
+            "worker_replication_http_tls", False
+        )
+
         # The shared secret used for authentication when connecting to the main synapse.
         self.worker_replication_secret = config.get("worker_replication_secret", None)
 

synapse/replication/http/_base.py

@@ -184,8 +184,10 @@ def make_client(cls, hs: "HomeServer") -> Callable:
         client = hs.get_simple_http_client()
         local_instance_name = hs.get_instance_name()
 
+        # The value of these option should match the replication listener settings
         master_host = hs.config.worker.worker_replication_host
         master_port = hs.config.worker.worker_replication_http_port
+        master_tls = hs.config.worker.worker_replication_http_tls
 
         instance_map = hs.config.worker.instance_map
 
@@ -205,9 +207,11 @@ async def send_request(*, instance_name: str = "master", **kwargs: Any) -> Any:
                 if instance_name == "master":
                     host = master_host
                     port = master_port
+                    tls = master_tls
                 elif instance_name in instance_map:
                     host = instance_map[instance_name].host
                     port = instance_map[instance_name].port
+                    tls = instance_map[instance_name].tls
                 else:
                     raise Exception(
                         "Instance %r not in 'instance_map' config" % (instance_name,)
@@ -238,7 +242,11 @@ async def send_request(*, instance_name: str = "master", **kwargs: Any) -> Any:
                         "Unknown METHOD on %s replication endpoint" % (cls.NAME,)
                     )
 
-                uri = "http://%s:%s/_synapse/replication/%s/%s" % (
+                # Here the protocol is hard coded to be http by default or https in case the replication
+                # port is set to have tls true.
+                scheme = "https" if tls else "http"
+                uri = "%s://%s:%s/_synapse/replication/%s/%s" % (
+                    scheme,
                     host,
                     port,
                     cls.NAME,