Experimental Unix socket support (#15353)

* Add IReactorUNIX to ISynapseReactor type hint.

* Create listen_unix().

Two options, 'path' to the file and 'mode' of permissions(not umask, recommend 666 as default as
nginx/other reverse proxies write to it and it's setup as user www-data)

For the moment, leave the option to always create a PID lockfile turned on by default

* Create UnixListenerConfig and wire it up.

Rename ListenerConfig to TCPListenerConfig, then Union them together into ListenerConfig.
This spidered around a bit, but I think I got it all. Metrics and manhole have been placed
behind a conditional in case of accidental putting them onto a unix socket.

Use new helpers to get if a listener is configured for TLS, and to help create a site tag
for logging.

There are 2 TODO things in parse_listener_def() to finish up at a later point.

* Refactor SynapseRequest to handle logging correctly when using a unix socket.

This prevents an exception when an IP address can not be retrieved for a request.

* Make the 'Synapse now listening on Unix socket' log line a little prettier.

* No silent failures on generic workers when trying to use a unix socket with metrics or manhole.

* Inline variables in app/_base.py

* Update docstring for listen_unix() to remove reference to a hardcoded permission of 0o666 and add a few comments saying where the default IS declared.

* Disallow both a unix socket and a ip/port combo on the same listener resource

* Linting

* Changelog

* review: simplify how listen_unix returns(and get rid of a type: ignore)

* review: fix typo from ConfigError in app/homeserver.py

* review: roll conditional for http_options.tag into get_site_tag() helper(and add docstring)

* review: enhance the conditionals for checking if a port or path is valid, remove a TODO line

* review: Try updating comment in get_client_ip_if_available to clarify what is being retrieved and why

* Pretty up how 'Synapse now listening on Unix Socket' looks by decoding the byte string.

* review: In parse_listener_def(), raise ConfigError if neither socket_path nor port is declared(and fix a typo)

2 files changed, 98 insertions, 33 deletions

diff --git a/synapse/config/server.py b/synapse/config/server.py
index 0e46b849cf..386c3194b8 100644
--- a/synapse/config/server.py
+++ b/synapse/config/server.py
@@ -214,17 +214,52 @@ class HttpListenerConfig:
 
 
 @attr.s(slots=True, frozen=True, auto_attribs=True)
-class ListenerConfig:
-    """Object describing the configuration of a single listener."""
+class TCPListenerConfig:
+    """Object describing the configuration of a single TCP listener."""
 
     port: int = attr.ib(validator=attr.validators.instance_of(int))
-    bind_addresses: List[str]
+    bind_addresses: List[str] = attr.ib(validator=attr.validators.instance_of(List))
     type: str = attr.ib(validator=attr.validators.in_(KNOWN_LISTENER_TYPES))
     tls: bool = False
 
     # http_options is only populated if type=http
     http_options: Optional[HttpListenerConfig] = None
 
+    def get_site_tag(self) -> str:
+        """Retrieves http_options.tag if it exists, otherwise the port number."""
+        if self.http_options and self.http_options.tag is not None:
+            return self.http_options.tag
+        else:
+            return str(self.port)
+
+    def is_tls(self) -> bool:
+        return self.tls
+
+
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class UnixListenerConfig:
+    """Object describing the configuration of a single Unix socket listener."""
+
+    # Note: unix sockets can not be tls encrypted, so HAVE to be behind a tls-handling
+    # reverse proxy
+    path: str = attr.ib()
+    # A default(0o666) for this is set in parse_listener_def() below
+    mode: int
+    type: str = attr.ib(validator=attr.validators.in_(KNOWN_LISTENER_TYPES))
+
+    # http_options is only populated if type=http
+    http_options: Optional[HttpListenerConfig] = None
+
+    def get_site_tag(self) -> str:
+        return "unix"
+
+    def is_tls(self) -> bool:
+        """Unix sockets can't have TLS"""
+        return False
+
+
+ListenerConfig = Union[TCPListenerConfig, UnixListenerConfig]
+
 
 @attr.s(slots=True, frozen=True, auto_attribs=True)
 class ManholeConfig:
@@ -531,12 +566,12 @@ class ServerConfig(Config):
 
         self.listeners = [parse_listener_def(i, x) for i, x in enumerate(listeners)]
 
-        # no_tls is not really supported any more, but let's grandfather it in
-        # here.
+        # no_tls is not really supported anymore, but let's grandfather it in here.
         if config.get("no_tls", False):
             l2 = []
             for listener in self.listeners:
-                if listener.tls:
+                if isinstance(listener, TCPListenerConfig) and listener.tls:
+                    # Use isinstance() as the assertion this *has* a listener.port
                     logger.info(
                         "Ignoring TLS-enabled listener on port %i due to no_tls",
                         listener.port,
@@ -577,7 +612,7 @@ class ServerConfig(Config):
             )
 
             self.listeners.append(
-                ListenerConfig(
+                TCPListenerConfig(
                     port=bind_port,
                     bind_addresses=[bind_host],
                     tls=True,
@@ -589,7 +624,7 @@ class ServerConfig(Config):
             unsecure_port = config.get("unsecure_port", bind_port - 400)
             if unsecure_port:
                 self.listeners.append(
-                    ListenerConfig(
+                    TCPListenerConfig(
                         port=unsecure_port,
                         bind_addresses=[bind_host],
                         tls=False,
@@ -601,7 +636,7 @@ class ServerConfig(Config):
         manhole = config.get("manhole")
         if manhole:
             self.listeners.append(
-                ListenerConfig(
+                TCPListenerConfig(
                     port=manhole,
                     bind_addresses=["127.0.0.1"],
                     type="manhole",
@@ -648,7 +683,7 @@ class ServerConfig(Config):
             logger.warning(METRICS_PORT_WARNING)
 
             self.listeners.append(
-                ListenerConfig(
+                TCPListenerConfig(
                     port=metrics_port,
                     bind_addresses=[config.get("metrics_bind_host", "127.0.0.1")],
                     type="http",
@@ -724,7 +759,7 @@ class ServerConfig(Config):
             self.delete_stale_devices_after = None
 
     def has_tls_listener(self) -> bool:
-        return any(listener.tls for listener in self.listeners)
+        return any(listener.is_tls() for listener in self.listeners)
 
     def generate_config_section(
         self,
@@ -904,25 +939,25 @@ def parse_listener_def(num: int, listener: Any) -> ListenerConfig:
         raise ConfigError(DIRECT_TCP_ERROR, ("listeners", str(num), "type"))
 
     port = listener.get("port")
-    if type(port) is not int:
+    socket_path = listener.get("path")
+    # Either a port or a path should be declared at a minimum. Using both would be bad.
+    if port is not None and not isinstance(port, int):
         raise ConfigError("Listener configuration is lacking a valid 'port' option")
+    if socket_path is not None and not isinstance(socket_path, str):
+        raise ConfigError("Listener configuration is lacking a valid 'path' option")
+    if port and socket_path:
+        raise ConfigError(
+            "Can not have both a UNIX socket and an IP/port declared for the same "
+            "resource!"
+        )
+    if port is None and socket_path is None:
+        raise ConfigError(
+            "Must have either a UNIX socket or an IP/port declared for a given "
+            "resource!"
+        )
 
     tls = listener.get("tls", False)
 
-    bind_addresses = listener.get("bind_addresses", [])
-    bind_address = listener.get("bind_address")
-    # if bind_address was specified, add it to the list of addresses
-    if bind_address:
-        bind_addresses.append(bind_address)
-
-    # if we still have an empty list of addresses, use the default list
-    if not bind_addresses:
-        if listener_type == "metrics":
-            # the metrics listener doesn't support IPv6
-            bind_addresses.append("0.0.0.0")
-        else:
-            bind_addresses.extend(DEFAULT_BIND_ADDRESSES)
-
     http_config = None
     if listener_type == "http":
         try:
@@ -932,8 +967,12 @@ def parse_listener_def(num: int, listener: Any) -> ListenerConfig:
         except ValueError as e:
             raise ConfigError("Unknown listener resource") from e
 
+        # For a unix socket, default x_forwarded to True, as this is the only way of
+        # getting a client IP.
+        # Note: a reverse proxy is required anyway, as there is no way of exposing a
+        # unix socket to the internet.
         http_config = HttpListenerConfig(
-            x_forwarded=listener.get("x_forwarded", False),
+            x_forwarded=listener.get("x_forwarded", (True if socket_path else False)),
             resources=resources,
             additional_resources=listener.get("additional_resources", {}),
             tag=listener.get("tag"),
@@ -941,7 +980,30 @@ def parse_listener_def(num: int, listener: Any) -> ListenerConfig:
             experimental_cors_msc3886=listener.get("experimental_cors_msc3886", False),
         )
 
-    return ListenerConfig(port, bind_addresses, listener_type, tls, http_config)
+    if socket_path:
+        # TODO: Add in path validation, like if the directory exists and is writable?
+        # Set a default for the permission, in case it's left out
+        socket_mode = listener.get("mode", 0o666)
+
+        return UnixListenerConfig(socket_path, socket_mode, listener_type, http_config)
+
+    else:
+        assert port is not None
+        bind_addresses = listener.get("bind_addresses", [])
+        bind_address = listener.get("bind_address")
+        # if bind_address was specified, add it to the list of addresses
+        if bind_address:
+            bind_addresses.append(bind_address)
+
+        # if we still have an empty list of addresses, use the default list
+        if not bind_addresses:
+            if listener_type == "metrics":
+                # the metrics listener doesn't support IPv6
+                bind_addresses.append("0.0.0.0")
+            else:
+                bind_addresses.extend(DEFAULT_BIND_ADDRESSES)
+
+        return TCPListenerConfig(port, bind_addresses, listener_type, tls, http_config)
 
 
 _MANHOLE_SETTINGS_SCHEMA = {
diff --git a/synapse/config/workers.py b/synapse/config/workers.py
index 2580660b6c..1dfbe27e89 100644
--- a/synapse/config/workers.py
+++ b/synapse/config/workers.py
@@ -19,15 +19,18 @@ from typing import Any, Dict, List, Union
 
 import attr
 
-from synapse.types import JsonDict
-
-from ._base import (
+from synapse.config._base import (
     Config,
     ConfigError,
     RoutableShardedWorkerHandlingConfig,
     ShardedWorkerHandlingConfig,
 )
-from .server import DIRECT_TCP_ERROR, ListenerConfig, parse_listener_def
+from synapse.config.server import (
+    DIRECT_TCP_ERROR,
+    TCPListenerConfig,
+    parse_listener_def,
+)
+from synapse.types import JsonDict
 
 _DEPRECATED_WORKER_DUTY_OPTION_USED = """
 The '%s' configuration option is deprecated and will be removed in a future
@@ -161,7 +164,7 @@ class WorkerConfig(Config):
         manhole = config.get("worker_manhole")
         if manhole:
             self.worker_listeners.append(
-                ListenerConfig(
+                TCPListenerConfig(
                     port=manhole,
                     bind_addresses=["127.0.0.1"],
                     type="manhole",