Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-08-01 16:46:07 -04:00
Code Issues Projects Releases Packages Wiki Activity

support federation queries through http connect proxy (#10475)

Browse source 
Signed-off-by: Marcus Hoffmann <bubu@bubu1.eu>
Signed-off-by: Dirk Klimpel dirk@klimpel.org
This commit is contained in:
Dirk Klimpel 2021-08-11 16:34:59 +02:00 • committed by GitHub
parent 8c654b7309
commit 339c3918e1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 557 additions and 193 deletions
Download patch file Download diff file Expand all files Collapse all files

68
synapse/http/connectproxyclient.py
View file

@ -12,8 +12,11 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import base64
import logging
from typing import Optional
import attr
from zope.interface import implementer
from twisted.internet import defer, protocol
@ -21,7 +24,6 @@ from twisted.internet.error import ConnectError
from twisted.internet.interfaces import IReactorCore, IStreamClientEndpoint
from twisted.internet.protocol import ClientFactory, Protocol, connectionDone
from twisted.web import http
from twisted.web.http_headers import Headers
logger = logging.getLogger(__name__)
@ -30,6 +32,22 @@ class ProxyConnectError(ConnectError):
pass
@attr.s
class ProxyCredentials:
username_password = attr.ib(type=bytes)
def as_proxy_authorization_value(self) -> bytes:
"""
Return the value for a Proxy-Authorization header (i.e. 'Basic abdef==').
Returns:
A transformation of the authentication string the encoded value for
a Proxy-Authorization header.
"""
# Encode as base64 and prepend the authorization type
return b"Basic " + base64.encodebytes(self.username_password)
@implementer(IStreamClientEndpoint)
class HTTPConnectProxyEndpoint:
"""An Endpoint implementation which will send a CONNECT request to an http proxy
@ -46,7 +64,7 @@ class HTTPConnectProxyEndpoint:
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
headers: Extra HTTP headers to include in the CONNECT request
proxy_creds: credentials to authenticate at proxy
"""
def __init__(
@ -55,20 +73,20 @@ class HTTPConnectProxyEndpoint:
proxy_endpoint: IStreamClientEndpoint,
host: bytes,
port: int,
headers: Headers,
proxy_creds: Optional[ProxyCredentials],
):
self._reactor = reactor
self._proxy_endpoint = proxy_endpoint
self._host = host
self._port = port
self._headers = headers
self._proxy_creds = proxy_creds
def __repr__(self):
return "<HTTPConnectProxyEndpoint %s>" % (self._proxy_endpoint,)
def connect(self, protocolFactory: ClientFactory):
f = HTTPProxiedClientFactory(
self._host, self._port, protocolFactory, self._headers
self._host, self._port, protocolFactory, self._proxy_creds
)
d = self._proxy_endpoint.connect(f)
# once the tcp socket connects successfully, we need to wait for the
@ -87,7 +105,7 @@ class HTTPProxiedClientFactory(protocol.ClientFactory):
dst_host: hostname that we want to CONNECT to
dst_port: port that we want to connect to
wrapped_factory: The original Factory
headers: Extra HTTP headers to include in the CONNECT request
proxy_creds: credentials to authenticate at proxy
"""
def __init__(
@ -95,12 +113,12 @@ class HTTPProxiedClientFactory(protocol.ClientFactory):
dst_host: bytes,
dst_port: int,
wrapped_factory: ClientFactory,
headers: Headers,
proxy_creds: Optional[ProxyCredentials],
):
self.dst_host = dst_host
self.dst_port = dst_port
self.wrapped_factory = wrapped_factory
self.headers = headers
self.proxy_creds = proxy_creds
self.on_connection = defer.Deferred()
def startedConnecting(self, connector):
@ -114,7 +132,7 @@ class HTTPProxiedClientFactory(protocol.ClientFactory):
self.dst_port,
wrapped_protocol,
self.on_connection,
self.headers,
self.proxy_creds,
)
def clientConnectionFailed(self, connector, reason):
@ -145,7 +163,7 @@ class HTTPConnectProtocol(protocol.Protocol):
connected_deferred: a Deferred which will be callbacked with
wrapped_protocol when the CONNECT completes
headers: Extra HTTP headers to include in the CONNECT request
proxy_creds: credentials to authenticate at proxy
"""
def __init__(
@ -154,16 +172,16 @@ class HTTPConnectProtocol(protocol.Protocol):
port: int,
wrapped_protocol: Protocol,
connected_deferred: defer.Deferred,
headers: Headers,
proxy_creds: Optional[ProxyCredentials],
):
self.host = host
self.port = port
self.wrapped_protocol = wrapped_protocol
self.connected_deferred = connected_deferred
self.headers = headers
self.proxy_creds = proxy_creds
self.http_setup_client = HTTPConnectSetupClient(
self.host, self.port, self.headers
self.host, self.port, self.proxy_creds
)
self.http_setup_client.on_connected.addCallback(self.proxyConnected)
@ -205,30 +223,38 @@ class HTTPConnectSetupClient(http.HTTPClient):
Args:
host: The hostname to send in the CONNECT message
port: The port to send in the CONNECT message
headers: Extra headers to send with the CONNECT message
proxy_creds: credentials to authenticate at proxy
"""
def __init__(self, host: bytes, port: int, headers: Headers):
def __init__(
self,
host: bytes,
port: int,
proxy_creds: Optional[ProxyCredentials],
):
self.host = host
self.port = port
self.headers = headers
self.proxy_creds = proxy_creds
self.on_connected = defer.Deferred()
def connectionMade(self):
logger.debug("Connected to proxy, sending CONNECT")
self.sendCommand(b"CONNECT", b"%s:%d" % (self.host, self.port))
# Send any additional specified headers
for name, values in self.headers.getAllRawHeaders():
for value in values:
self.sendHeader(name, value)
# 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):
logger.debug("Got Status: %s %s %s", status, message, version)
if status != b"200":
raise ProxyConnectError("Unexpected status on CONNECT: %s" % status)
raise ProxyConnectError(f"Unexpected status on CONNECT: {status!s}")
def handleEndHeaders(self):
logger.debug("End Headers")

100
synapse/http/federation/matrix_federation_agent.py
View file

@ -14,6 +14,10 @@
import logging
import urllib.parse
from typing import Any, Generator, List, Optional
from urllib.request import ( # type: ignore[attr-defined]
getproxies_environment,
proxy_bypass_environment,
)
from netaddr import AddrFormatError, IPAddress, IPSet
from zope.interface import implementer
@ -30,9 +34,12 @@ from twisted.web.http_headers import Headers
from twisted.web.iweb import IAgent, IAgentEndpointFactory, IBodyProducer, IResponse
from synapse.crypto.context_factory import FederationPolicyForHTTPS
from synapse.http.client import BlacklistingAgentWrapper
from synapse.http import proxyagent
from synapse.http.client import BlacklistingAgentWrapper, BlacklistingReactorWrapper
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 import Clock
@ -57,6 +64,14 @@ class MatrixFederationAgent:
user_agent:
The user agent header to use for federation requests.
ip_whitelist: Allowed IP addresses.
ip_blacklist: Disallowed IP addresses.
proxy_reactor: twisted reactor to use for connections to the proxy server
reactor might have some blacklisting 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.
@ -71,11 +86,18 @@ class MatrixFederationAgent:
reactor: ISynapseReactor,
tls_client_options_factory: Optional[FederationPolicyForHTTPS],
user_agent: bytes,
ip_whitelist: IPSet,
ip_blacklist: IPSet,
_srv_resolver: Optional[SrvResolver] = None,
_well_known_resolver: Optional[WellKnownResolver] = None,
):
self._reactor = reactor
# proxy_reactor is not blacklisted
proxy_reactor = reactor
# We need to use a DNS resolver which filters out blacklisted IP
# addresses, to prevent DNS rebinding.
reactor = BlacklistingReactorWrapper(reactor, ip_whitelist, ip_blacklist)
self._clock = Clock(reactor)
self._pool = HTTPConnectionPool(reactor)
self._pool.retryAutomatically = False
@ -83,24 +105,27 @@ class MatrixFederationAgent:
self._pool.cachedConnectionTimeout = 2 * 60
self._agent = Agent.usingEndpointFactory(
self._reactor,
reactor,
MatrixHostnameEndpointFactory(
reactor, tls_client_options_factory, _srv_resolver
reactor,
proxy_reactor,
tls_client_options_factory,
_srv_resolver,
),
pool=self._pool,
)
self.user_agent = user_agent
if _well_known_resolver is None:
# Note that the name resolver has already been wrapped in a
# IPBlacklistingResolver by MatrixFederationHttpClient.
_well_known_resolver = WellKnownResolver(
self._reactor,
reactor,
agent=BlacklistingAgentWrapper(
Agent(
self._reactor,
ProxyAgent(
reactor,
proxy_reactor,
pool=self._pool,
contextFactory=tls_client_options_factory,
use_proxy=True,
),
ip_blacklist=ip_blacklist,
),
@ -200,10 +225,12 @@ class MatrixHostnameEndpointFactory:
def __init__(
self,
reactor: IReactorCore,
proxy_reactor: IReactorCore,
tls_client_options_factory: Optional[FederationPolicyForHTTPS],
srv_resolver: Optional[SrvResolver],
):
self._reactor = reactor
self._proxy_reactor = proxy_reactor
self._tls_client_options_factory = tls_client_options_factory
if srv_resolver is None:
@ -211,9 +238,10 @@ class MatrixHostnameEndpointFactory:
self._srv_resolver = srv_resolver
def endpointForURI(self, parsed_uri):
def endpointForURI(self, parsed_uri: URI):
return MatrixHostnameEndpoint(
self._reactor,
self._proxy_reactor,
self._tls_client_options_factory,
self._srv_resolver,
parsed_uri,
@ -227,23 +255,45 @@ class MatrixHostnameEndpoint:
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 blacklisting 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
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: Optional[FederationPolicyForHTTPS],
srv_resolver: SrvResolver,
parsed_uri: URI,
):
self._reactor = reactor
self._parsed_uri = parsed_uri
# http_proxy is not needed because federation is always over TLS
proxies = getproxies_environment()
https_proxy = proxies["https"].encode() if "https" in proxies else None
self.no_proxy = proxies["no"] if "no" in proxies else None
# 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(
https_proxy,
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
@ -273,9 +323,33 @@ class MatrixHostnameEndpoint:
host = server.host
port = server.port
should_skip_proxy = False
if self.no_proxy is not None:
should_skip_proxy = proxy_bypass_environment(
host.decode(),
proxies={"no": self.no_proxy},
)
endpoint: IStreamClientEndpoint
try:
logger.debug("Connecting to %s:%i", host.decode("ascii"), port)
endpoint = HostnameEndpoint(self._reactor, host, port)
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(

12
synapse/http/matrixfederationclient.py
View file

@ -59,7 +59,6 @@ from synapse.api.errors import (
from synapse.http import QuieterFileBodyProducer
from synapse.http.client import (
BlacklistingAgentWrapper,
BlacklistingReactorWrapper,
BodyExceededMaxSize,
ByteWriteable,
encode_query_args,
@ -69,7 +68,7 @@ from synapse.http.federation.matrix_federation_agent import MatrixFederationAgen
from synapse.logging import opentracing
from synapse.logging.context import make_deferred_yieldable
from synapse.logging.opentracing import set_tag, start_active_span, tags
from synapse.types import ISynapseReactor, JsonDict
from synapse.types import JsonDict
from synapse.util import json_decoder
from synapse.util.async_helpers import timeout_deferred
from synapse.util.metrics import Measure
@ -325,13 +324,7 @@ class MatrixFederationHttpClient:
self.signing_key = hs.signing_key
self.server_name = hs.hostname
# We need to use a DNS resolver which filters out blacklisted IP
# addresses, to prevent DNS rebinding.
self.reactor: ISynapseReactor = BlacklistingReactorWrapper(
hs.get_reactor(),
hs.config.federation_ip_range_whitelist,
hs.config.federation_ip_range_blacklist,
)
self.reactor = hs.get_reactor()
user_agent = hs.version_string
if hs.config.user_agent_suffix:
@ -342,6 +335,7 @@ class MatrixFederationHttpClient:
self.reactor,
tls_client_options_factory,
user_agent,
hs.config.federation_ip_range_whitelist,
hs.config.federation_ip_range_blacklist,
)

51
synapse/http/proxyagent.py
View file

@ -11,7 +11,6 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import base64
import logging
import re
from typing import Any, Dict, Optional, Tuple
@ -21,7 +20,6 @@ from urllib.request import ( # type: ignore[attr-defined]
proxy_bypass_environment,
)
import attr
from zope.interface import implementer
from twisted.internet import defer
@ -38,7 +36,7 @@ from twisted.web.error import SchemeNotSupported
from twisted.web.http_headers import Headers
from twisted.web.iweb import IAgent, IBodyProducer, IPolicyForHTTPS
from synapse.http.connectproxyclient import HTTPConnectProxyEndpoint
from synapse.http.connectproxyclient import HTTPConnectProxyEndpoint, ProxyCredentials
from synapse.types import ISynapseReactor
logger = logging.getLogger(__name__)
@ -46,22 +44,6 @@ logger = logging.getLogger(__name__)
_VALID_URI = re.compile(br"\A[\x21-\x7e]+\Z")
@attr.s
class ProxyCredentials:
username_password = attr.ib(type=bytes)
def as_proxy_authorization_value(self) -> bytes:
"""
Return the value for a Proxy-Authorization header (i.e. 'Basic abdef==').
Returns:
A transformation of the authentication string the encoded value for
a Proxy-Authorization header.
"""
# Encode as base64 and prepend the authorization type
return b"Basic " + base64.encodebytes(self.username_password)
@implementer(IAgent)
class ProxyAgent(_AgentBase):
"""An Agent implementation which will use an HTTP proxy if one was requested
@ -95,6 +77,7 @@ class ProxyAgent(_AgentBase):
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__(
@ -131,11 +114,11 @@ class ProxyAgent(_AgentBase):
https_proxy = proxies["https"].encode() if "https" in proxies else None
no_proxy = proxies["no"] if "no" in proxies else None
self.http_proxy_endpoint, self.http_proxy_creds = _http_proxy_endpoint(
self.http_proxy_endpoint, self.http_proxy_creds = http_proxy_endpoint(
http_proxy, self.proxy_reactor, contextFactory, **self._endpoint_kwargs
)
self.https_proxy_endpoint, self.https_proxy_creds = _http_proxy_endpoint(
self.https_proxy_endpoint, self.https_proxy_creds = http_proxy_endpoint(
https_proxy, self.proxy_reactor, contextFactory, **self._endpoint_kwargs
)
@ -224,22 +207,12 @@ class ProxyAgent(_AgentBase):
and self.https_proxy_endpoint
and not should_skip_proxy
):
connect_headers = Headers()
# Determine whether we need to set Proxy-Authorization headers
if self.https_proxy_creds:
# Set a Proxy-Authorization header
connect_headers.addRawHeader(
b"Proxy-Authorization",
self.https_proxy_creds.as_proxy_authorization_value(),
)
endpoint = HTTPConnectProxyEndpoint(
self.proxy_reactor,
self.https_proxy_endpoint,
parsed_uri.host,
parsed_uri.port,
headers=connect_headers,
self.https_proxy_creds,
)
else:
# not using a proxy
@ -268,10 +241,10 @@ class ProxyAgent(_AgentBase):
)
def _http_proxy_endpoint(
def http_proxy_endpoint(
proxy: Optional[bytes],
reactor: IReactorCore,
tls_options_factory: IPolicyForHTTPS,
tls_options_factory: Optional[IPolicyForHTTPS],
**kwargs,
) -> Tuple[Optional[IStreamClientEndpoint], Optional[ProxyCredentials]]:
"""Parses an http proxy setting and returns an endpoint for the proxy
@ -294,6 +267,7 @@ def _http_proxy_endpoint(
Raise:
ValueError if proxy has no hostname or unsupported scheme.
RuntimeError if no tls_options_factory is given for a https connection
"""
if proxy is None:
return None, None
@ -305,8 +279,13 @@ def _http_proxy_endpoint(
proxy_endpoint = HostnameEndpoint(reactor, host, port, **kwargs)
if scheme == b"https":
tls_options = tls_options_factory.creatorForNetloc(host, port)
proxy_endpoint = wrapClientTLS(tls_options, proxy_endpoint)
if tls_options_factory:
tls_options = tls_options_factory.creatorForNetloc(host, port)
proxy_endpoint = wrapClientTLS(tls_options, proxy_endpoint)
else:
raise RuntimeError(
f"No TLS options for a https connection via proxy {proxy!s}"
)
return proxy_endpoint, credentials