diff --git a/Cargo.lock b/Cargo.lock index a63ff20355..c238030f4f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2025,6 +2025,8 @@ dependencies = [ "http", "http-body-util", "httpdate", + "hyper", + "hyper-util", "icu_segmenter", "ipnet", "itertools", @@ -2054,6 +2056,7 @@ dependencies = [ "tokio-postgres", "tokio-postgres-rustls", "tokio-rustls", + "tower-service", "ulid", "url", ] diff --git a/rust/Cargo.toml b/rust/Cargo.toml index dd3282a4ea..98a4c35f8b 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -102,12 +102,22 @@ tokio-postgres-rustls = { version = "0.14.0", default-features = false, features "ring", ] } lru = "0.16.4" -# Outbound federation (`federation::agent`): the TLS policy. +# Outbound federation (`federation::agent`). We drive hyper's connection pool +# with our own connector, which does Matrix server name resolution, the IP +# range policy and the TLS policy. +hyper = { version = "1.6.0", features = ["client", "http1"] } +hyper-util = { version = "0.1.16", features = [ + "client", + "client-legacy", + "http1", + "tokio", +] } tokio-rustls = { version = "0.26.2", default-features = false, features = [ "ring", "logging", "tls12", ] } +tower-service = "0.3.3" # SRV lookups. A/AAAA go through the system resolver, as in Python. 0.25 # rather than 0.26, which needs a newer rustc than our MSRV. hickory-resolver = { version = "0.25.2", default-features = false, features = [ diff --git a/rust/src/federation/agent/connect.rs b/rust/src/federation/agent/connect.rs new file mode 100644 index 0000000000..5319f4b46d --- /dev/null +++ b/rust/src/federation/agent/connect.rs @@ -0,0 +1,466 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + * + * Originally licensed under the Apache License, Version 2.0: + * . + * + * Derived from `MatrixConnector` in `src/server_resolver.rs` of + * . + * + * [This file includes modifications made by Element Creations Ltd] + */ + +//! The connector hyper's connection pool calls to open a connection. +//! +//! For a `matrix-federation://[:port]` URI it resolves the server name +//! (SRV, see [`super::resolve`]), tries each server in turn, and wraps the +//! first connection that opens in TLS, verified against ``. The pool +//! keys connections on the URI's scheme and authority, so it pools per +//! (delegated) server name as Python does. `https://` URIs (the +//! `.well-known` fetch) connect straight to the host. +//! +//! This replaces Python's `MatrixHostnameEndpoint`, `HostnameEndpoint` and +//! `HTTPConnectProxyEndpoint`. + +use std::{ + collections::HashMap, + future::Future, + io, + net::{IpAddr, SocketAddr}, + pin::Pin, + sync::Arc, + task::{Context, Poll}, + time::Duration, +}; + +use anyhow::{anyhow, bail, Context as _}; +use futures::{stream::FuturesUnordered, StreamExt}; +use http::Uri; +use hyper_util::{ + client::legacy::connect::{Connected, Connection}, + rt::TokioIo, +}; +use log::{debug, info}; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::TcpStream, +}; +use tokio_rustls::TlsConnector; + +use super::{ + policy::{IpPolicy, Proxy}, + resolve::{parse_ip_literal, Server, ServerResolver}, + tls::{server_name, TlsPolicy}, + Now, +}; +use crate::federation::transport::TransportError; + +/// The scheme of federation request URIs. +pub const MATRIX_FEDERATION_SCHEME: &str = "matrix-federation"; + +/// How long to wait for a TCP connection, as Twisted's `HostnameEndpoint`. +pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); + +/// How long to wait for one address before also trying the next, as +/// Twisted's `HostnameEndpoint` does. +const ATTEMPT_DELAY: Duration = Duration::from_millis(300); + +/// The longest `CONNECT` response head we read from a proxy. +const MAX_PROXY_RESPONSE: usize = 16 * 1024; + +pub(super) struct ConnectorInner { + pub servers: ServerResolver, + pub ip_policy: IpPolicy, + pub tls: TlsPolicy, + pub proxy: Option, + pub now: Now, + /// Test-only: connect to these ports in place of the ones resolution + /// picks, so tests can serve "port 443" and "port 8448" from an + /// unprivileged listener. Empty in production. + pub port_overrides: HashMap, +} + +/// A `tower_service::Service` for hyper's legacy client. +#[derive(Clone)] +pub struct FederationConnector { + pub(super) inner: Arc, +} + +impl tower_service::Service for FederationConnector { + type Response = FederationStream; + type Error = TransportError; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, uri: Uri) -> Self::Future { + let inner = self.inner.clone(); + Box::pin(async move { inner.connect(uri).await }) + } +} + +impl ConnectorInner { + async fn connect(&self, uri: Uri) -> Result { + let host = uri + .host() + .ok_or_else(|| TransportError::InvalidRequest(anyhow!("URI {uri} has no host")))?; + + let servers = match uri.scheme_str() { + Some(MATRIX_FEDERATION_SCHEME) => { + let now_secs = (self.now)() / 1000; + self.servers + .resolve_server(host, uri.port_u16(), now_secs) + .await? + } + Some("https") => vec![Server { + host: host.to_owned(), + port: uri.port_u16().unwrap_or(443), + }], + other => { + return Err(TransportError::InvalidRequest(anyhow!( + "unsupported scheme {other:?} in {uri}" + ))) + } + }; + + // We verify the certificate against the host in the URI: for + // federation that is the server name or its `.well-known` delegate, + // not the SRV target. + let tls_name = server_name(host).map_err(TransportError::Tls)?; + let tls_config = self.tls.config_for(host); + + let mut first_error = None; + for server in servers { + match self.connect_to(&server).await { + Ok(tcp) => { + let connector = TlsConnector::from(tls_config.clone()); + return match connector.connect(tls_name.clone(), tcp).await { + Ok(tls) => Ok(FederationStream::new(Box::new(tls))), + Err(err) => Err(TransportError::Tls( + anyhow!(err).context(format!("TLS handshake with {host}")), + )), + }; + } + Err(err) => { + info!( + "Failed to connect to {}:{}: {err}", + server.host, server.port + ); + first_error.get_or_insert(err); + } + } + } + + // The first failure is probably the most interesting. + Err(first_error.unwrap_or_else(|| { + TransportError::Connect(anyhow!("no servers to connect to for {host}")) + })) + } + + /// Open a byte stream to one server: TCP, possibly tunnelled through + /// the proxy. + async fn connect_to(&self, server: &Server) -> Result, TransportError> { + let port = self + .port_overrides + .get(&server.port) + .copied() + .unwrap_or(server.port); + + if let Some(proxy) = self.proxy.as_ref().filter(|p| p.applies_to(&server.host)) { + debug!( + "Connecting to {}:{} via proxy {}:{}", + server.host, server.port, proxy.host, proxy.port + ); + return self.connect_via_proxy(proxy, &server.host, port).await; + } + + debug!("Connecting to {}:{}", server.host, server.port); + let ips = self + .ip_policy + .resolve(self.servers.dns().as_ref(), &server.host) + .await?; + let tcp = connect_tcp(ips, port).await?; + Ok(Box::new(tcp)) + } + + async fn connect_via_proxy( + &self, + proxy: &Proxy, + host: &str, + port: u16, + ) -> Result, TransportError> { + // The proxy itself is not subject to the IP range policy, as in + // Python, which connects to it with the unfiltered reactor. + let ips = match self.servers.dns().lookup_ip(&proxy.host).await { + Ok(ips) => ips, + Err(err) => { + return Err(TransportError::Dns(anyhow!( + "resolving proxy {}: {err:?}", + proxy.host + ))) + } + }; + let tcp = connect_tcp(ips, proxy.port).await?; + + let mut stream: Box = if proxy.tls { + let name = server_name(&proxy.host).map_err(TransportError::Tls)?; + let connector = TlsConnector::from(self.tls.config_for(&proxy.host)); + let tls = connector.connect(name, tcp).await.map_err(|err| { + TransportError::Tls(anyhow!(err).context(format!("TLS to proxy {}", proxy.host))) + })?; + Box::new(tls) + } else { + Box::new(tcp) + }; + + proxy_connect(&mut stream, host, port, proxy.authorization.as_deref()) + .await + .map_err(TransportError::Connect)?; + + Ok(stream) + } +} + +/// Connect to the first of `ips` that answers. Like Twisted's +/// `HostnameEndpoint`, a new attempt starts every [`ATTEMPT_DELAY`] (or as +/// soon as one fails) without abandoning the earlier ones. +async fn connect_tcp(ips: Vec, port: u16) -> Result { + let attempt = async move { + let mut pending = ips.into_iter().map(|ip| SocketAddr::new(ip, port)); + let mut attempts = FuturesUnordered::new(); + let mut first_error: Option = None; + + let start = |attempts: &mut FuturesUnordered<_>, addr: SocketAddr| { + attempts.push(async move { (addr, TcpStream::connect(addr).await) }); + }; + + match pending.next() { + Some(addr) => start(&mut attempts, addr), + None => bail!("no addresses to connect to"), + } + + loop { + let delay = tokio::time::sleep(ATTEMPT_DELAY); + tokio::select! { + result = attempts.next() => match result { + Some((_, Ok(stream))) => return Ok(stream), + Some((addr, Err(err))) => { + debug!("Connection attempt to {addr} failed: {err}"); + first_error.get_or_insert(err); + if let Some(addr) = pending.next() { + start(&mut attempts, addr); + } else if attempts.is_empty() { + let err = first_error.expect("set above"); + return Err(anyhow!(err)); + } + } + None => unreachable!("attempts is never empty here"), + }, + _ = delay, if pending.len() > 0 => { + if let Some(addr) = pending.next() { + start(&mut attempts, addr); + } + } + } + } + }; + + match tokio::time::timeout(CONNECT_TIMEOUT, attempt).await { + Ok(Ok(stream)) => { + let _ = stream.set_nodelay(true); + Ok(stream) + } + Ok(Err(err)) => Err(TransportError::Connect(err)), + Err(_) => Err(TransportError::Connect(anyhow!( + "timed out after {CONNECT_TIMEOUT:?}" + ))), + } +} + +/// Ask an HTTP proxy for a tunnel to `host:port`. +async fn proxy_connect( + stream: &mut S, + host: &str, + port: u16, + authorization: Option<&str>, +) -> anyhow::Result<()> { + // An IPv6 literal needs brackets in an authority. + let authority = match parse_ip_literal(host) { + Some(IpAddr::V6(_)) if !host.starts_with('[') => format!("[{host}]:{port}"), + _ => format!("{host}:{port}"), + }; + + let mut request = format!("CONNECT {authority} HTTP/1.1\r\nHost: {authority}\r\n"); + if let Some(authorization) = authorization { + request.push_str(&format!("Proxy-Authorization: {authorization}\r\n")); + } + request.push_str("\r\n"); + stream.write_all(request.as_bytes()).await?; + stream.flush().await?; + + // Read the response head byte by byte, so we never consume bytes that + // belong to the tunnelled connection. + let mut head = Vec::with_capacity(256); + while !head.ends_with(b"\r\n\r\n") { + if head.len() >= MAX_PROXY_RESPONSE { + bail!("proxy response head too long"); + } + let byte = stream + .read_u8() + .await + .context("proxy closed the connection")?; + head.push(byte); + } + + let status_line = head + .split(|&b| b == b'\n') + .next() + .map(|line| String::from_utf8_lossy(line).trim().to_owned()) + .unwrap_or_default(); + let status = status_line.split_whitespace().nth(1); + if !status_line.starts_with("HTTP/1.") || status != Some("200") { + bail!("proxy refused CONNECT to {authority}: {status_line}"); + } + + Ok(()) +} + +/// A byte stream the agent can speak TLS (or HTTP) over. +pub(super) trait Io: AsyncRead + AsyncWrite + Send + Unpin {} +impl Io for T {} + +/// An open federation connection, as handed to hyper. +pub struct FederationStream { + io: TokioIo>, +} + +impl FederationStream { + fn new(io: Box) -> Self { + FederationStream { + io: TokioIo::new(io), + } + } +} + +impl Connection for FederationStream { + fn connected(&self) -> Connected { + Connected::new() + } +} + +impl hyper::rt::Read for FederationStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: hyper::rt::ReadBufCursor<'_>, + ) -> Poll> { + Pin::new(&mut self.io).poll_read(cx, buf) + } +} + +impl hyper::rt::Write for FederationStream { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.io).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.io).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.io).poll_shutdown(cx) + } +} + +#[cfg(test)] +mod tests { + use tokio::io::duplex; + + use super::*; + + #[tokio::test] + async fn proxy_connect_handshake() { + let (mut client, mut proxy) = duplex(4096); + + let server = tokio::spawn(async move { + let mut buf = vec![0u8; 1024]; + let n = proxy.read(&mut buf).await.unwrap(); + let request = String::from_utf8(buf[..n].to_vec()).unwrap(); + proxy + .write_all(b"HTTP/1.1 200 Connection established\r\n\r\ntunnelled") + .await + .unwrap(); + request + }); + + proxy_connect(&mut client, "remote.example", 8448, Some("Basic eA==")) + .await + .unwrap(); + + // The bytes after the response head are the tunnel's, untouched. + let mut rest = [0u8; 9]; + client.read_exact(&mut rest).await.unwrap(); + assert_eq!(&rest, b"tunnelled"); + + let request = server.await.unwrap(); + assert_eq!( + request, + "CONNECT remote.example:8448 HTTP/1.1\r\nHost: remote.example:8448\r\n\ + Proxy-Authorization: Basic eA==\r\n\r\n" + ); + } + + #[tokio::test] + async fn proxy_connect_refused() { + let (mut client, mut proxy) = duplex(4096); + tokio::spawn(async move { + let mut buf = vec![0u8; 1024]; + let _ = proxy.read(&mut buf).await.unwrap(); + proxy + .write_all(b"HTTP/1.1 403 Forbidden\r\n\r\n") + .await + .unwrap(); + }); + + let err = proxy_connect(&mut client, "::1", 8448, None) + .await + .unwrap_err(); + assert!(err.to_string().contains("403"), "{err}"); + } + + #[tokio::test] + async fn tcp_tries_each_address() { + // Nothing listens on port 1 of the first address, so the second wins. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let accept = tokio::spawn(async move { listener.accept().await.unwrap() }); + + let stream = connect_tcp( + vec!["127.0.0.2".parse().unwrap(), "127.0.0.1".parse().unwrap()], + port, + ) + .await + .unwrap(); + assert_eq!(stream.peer_addr().unwrap().port(), port); + accept.await.unwrap(); + + let err = connect_tcp(vec![], port).await.unwrap_err(); + assert!(matches!(err, TransportError::Connect(_)), "{err}"); + } +} diff --git a/rust/src/federation/agent/mod.rs b/rust/src/federation/agent/mod.rs index 1706a9f39d..e2d56d6d50 100644 --- a/rust/src/federation/agent/mod.rs +++ b/rust/src/federation/agent/mod.rs @@ -14,21 +14,276 @@ */ //! The outbound federation agent: Python's `MatrixFederationAgent` and its -//! `HTTPConnectionPool`. This holds its parts so far. +//! `HTTPConnectionPool`. +//! +//! [`Agent::request`] takes a `matrix-federation:///` +//! request, applies `.well-known` delegation (rewriting the URI and the +//! `Host` header), and sends it through a hyper connection pool whose +//! [`connect::FederationConnector`] does SRV resolution, the IP range policy, +//! the proxy and TLS. It does no retries, signing or response parsing: those +//! belong to the request layer above. //! //! - [`resolve`]: SRV lookups and the order to try servers in. //! - [`well_known`]: `.well-known` delegation and its cache. //! - [`policy`]: the IP range lists and the proxy. //! - [`tls`]: certificate verification. +//! - [`connect`]: opening connections for the pool. -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc, time::Duration}; +use anyhow::{anyhow, Context as _}; +use async_trait::async_trait; +use bytes::Bytes; +use http::{header, HeaderValue, Request, Response, Uri}; +use http_body_util::{BodyExt, Full, Limited}; +use hyper::body::Incoming; +use hyper_util::{ + client::legacy::Client, + rt::{TokioExecutor, TokioTimer}, +}; + +use crate::{ + clock::Sleeper, + config::SynapseHomeServerConfig, + federation::transport::{FederationTransport, ResponseBody, TransportError}, +}; + +pub mod connect; pub mod policy; +pub mod python; pub mod resolve; pub mod tls; pub mod well_known; +use connect::{ConnectorInner, FederationConnector, MATRIX_FEDERATION_SCHEME}; +use policy::{IpPolicy, Proxy}; +use resolve::{parse_ip_literal, DnsResolver, ServerResolver}; +use tls::TlsPolicy; +use well_known::{FetchError, FetchResponse, WellKnownFetcher, WellKnownResolver}; + /// The current time in milliseconds since the epoch. Production reads the /// homeserver's [`Clock`](crate::clock::Clock). Rust tests use their own, /// since the virtual clock is process-global. pub type Now = Arc u64 + Send + Sync>; + +/// Idle connections kept per host, as Python's pool. +const MAX_IDLE_PER_HOST: usize = 5; +/// How long an idle connection is kept, as Python's pool. +const IDLE_TIMEOUT: Duration = Duration::from_secs(120); + +/// Everything the agent needs to know, taken from the homeserver config. +#[derive(Clone)] +pub struct AgentConfig { + pub user_agent: String, + pub ip_policy: IpPolicy, + pub tls: crate::config::TlsConfig, + pub proxy: crate::config::types::ProxyConfig, + /// The timeout for each `.well-known` fetch attempt. + pub well_known_timeout: Duration, +} + +impl AgentConfig { + pub fn from_homeserver_config(config: &SynapseHomeServerConfig, user_agent: String) -> Self { + AgentConfig { + user_agent, + ip_policy: IpPolicy::new( + config.server.federation_ip_range_blocklist.clone(), + config.server.federation_ip_range_allowlist.clone(), + ), + tls: config.tls.clone(), + proxy: config.server.proxy_config.clone(), + well_known_timeout: Duration::from_millis(config.federation.client_timeout_ms), + } + } +} + +type PoolClient = Client>; + +pub struct Agent { + client: PoolClient, + well_known: Arc, + user_agent: HeaderValue, +} + +impl Agent { + /// Build an agent. `port_overrides` is for tests (see + /// [`ConnectorInner::port_overrides`]) and is empty in production. + pub fn new( + config: AgentConfig, + dns: Arc, + sleeper: Arc, + now: Now, + port_overrides: HashMap, + ) -> anyhow::Result { + let user_agent = HeaderValue::from_str(&config.user_agent).context("invalid user agent")?; + let ip_policy = config.ip_policy.clone(); + + let connector = FederationConnector { + inner: Arc::new(ConnectorInner { + servers: ServerResolver::new(dns), + ip_policy: config.ip_policy, + tls: TlsPolicy::new(&config.tls)?, + proxy: Proxy::from_config(&config.proxy)?, + now: now.clone(), + port_overrides, + }), + }; + + let client = Client::builder(TokioExecutor::new()) + .pool_timer(TokioTimer::new()) + .pool_idle_timeout(IDLE_TIMEOUT) + .pool_max_idle_per_host(MAX_IDLE_PER_HOST) + // Python's pool does not retry either: the request layer decides. + .retry_canceled_requests(false) + .build(connector); + + // The `.well-known` fetch shares the pool, as in Python. + let fetcher = Arc::new(PoolFetcher { + client: client.clone(), + user_agent: user_agent.clone(), + ip_policy, + }); + let well_known = Arc::new(WellKnownResolver::new( + fetcher, + sleeper, + now, + config.well_known_timeout, + )); + + Ok(Agent { + client, + well_known, + user_agent, + }) + } + + /// Send one request and return the response head, with the body still + /// to stream. The response is returned whatever its status. + pub async fn request( + &self, + mut request: Request>, + ) -> Result, TransportError> { + let uri = request.uri().clone(); + let host = uri + .host() + .ok_or_else(|| TransportError::InvalidRequest(anyhow!("URI {uri} has no host")))?; + + // `.well-known` applies to a federation server name with no port + // that is not an IP literal. It has to happen here rather than in + // the connector, since it changes the `Host` header. + if uri.scheme_str() == Some(MATRIX_FEDERATION_SCHEME) + && uri.port().is_none() + && parse_ip_literal(host).is_none() + { + if let Some(delegated) = self.well_known.get_well_known(host).await { + let mut parts = uri.into_parts(); + parts.authority = Some(delegated.parse().map_err(|err| { + TransportError::InvalidRequest(anyhow!( + "bad delegated server {delegated:?}: {err}" + )) + })?); + *request.uri_mut() = Uri::from_parts(parts) + .map_err(|err| TransportError::InvalidRequest(anyhow!(err)))?; + } + } + + // The `Host` header is the (delegated) server name, with its port + // if it has one. hyper would leave a non-default port off. + if !request.headers().contains_key(header::HOST) { + let authority = request + .uri() + .authority() + .expect("checked above") + .as_str() + .to_owned(); + let value = HeaderValue::from_str(&authority) + .map_err(|err| TransportError::InvalidRequest(anyhow!(err)))?; + request.headers_mut().insert(header::HOST, value); + } + if !request.headers().contains_key(header::USER_AGENT) { + request + .headers_mut() + .insert(header::USER_AGENT, self.user_agent.clone()); + } + + self.client + .request(request) + .await + .map_err(from_client_error) + } +} + +#[async_trait] +impl FederationTransport for Agent { + async fn send( + &self, + request: Request, + ) -> Result, TransportError> { + let response = self.request(request.map(Full::new)).await?; + Ok(response.map(|body| body.map_err(Into::into).boxed_unsync())) + } +} + +/// Recover our own error from a failed connect, else wrap hyper's. +fn from_client_error(err: hyper_util::client::legacy::Error) -> TransportError { + let mut source = std::error::Error::source(&err); + while let Some(inner) = source { + if let Some(ours) = inner.downcast_ref::() { + return ours.duplicate(); + } + source = inner.source(); + } + + if err.is_connect() { + TransportError::Connect(anyhow!(err)) + } else { + TransportError::Request(anyhow!(err)) + } +} + +/// Fetches `.well-known` files through the agent's pool. +struct PoolFetcher { + client: PoolClient, + user_agent: HeaderValue, + ip_policy: IpPolicy, +} + +#[async_trait] +impl WellKnownFetcher for PoolFetcher { + async fn fetch(&self, uri: &Uri) -> Result { + // Python wraps this fetch in a `BlocklistingAgentWrapper` that is + // given only the blocklist, so an IP literal (reachable through a + // redirect) is blocked even if the allowlist covers it. + if let Some(ip) = uri.host().and_then(parse_ip_literal) { + if self.ip_policy.is_blocked_ignoring_allowlist(&ip) { + return Err(FetchError::Other(anyhow!("IP address {ip} blocked"))); + } + } + + let request = Request::get(uri.clone()) + .header(header::USER_AGENT, self.user_agent.clone()) + .body(Full::new(Bytes::new())) + .map_err(|err| FetchError::Other(anyhow!(err)))?; + + let response = self + .client + .request(request) + .await + .map_err(|err| FetchError::Other(anyhow!(from_client_error(err))))?; + + let (parts, body) = response.into_parts(); + let body = match Limited::new(body, well_known::MAX_SIZE).collect().await { + Ok(collected) => collected.to_bytes(), + Err(err) if err.is::() => { + return Err(FetchError::TooLarge) + } + Err(err) => return Err(FetchError::Other(anyhow!(err))), + }; + + Ok(FetchResponse { + status: parts.status, + headers: parts.headers, + body, + }) + } +} diff --git a/rust/src/federation/agent/python.rs b/rust/src/federation/agent/python.rs new file mode 100644 index 0000000000..189a94cdb1 --- /dev/null +++ b/rust/src/federation/agent/python.rs @@ -0,0 +1,166 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + * + */ + +//! A Python handle on the agent, for tests only. Nothing in Synapse sends +//! through the agent yet: the request layer that will is the next step. + +use std::{collections::HashMap, net::IpAddr, sync::Arc}; + +use http::{HeaderName, HeaderValue, Method, Request}; +use http_body_util::{BodyExt, Full}; +use pyo3::{ + exceptions::{PyRuntimeError, PyValueError}, + prelude::*, +}; +use tokio::runtime::Handle; + +use super::{ + resolve::{InMemoryResolver, SrvRecord}, + Agent, AgentConfig, +}; +use crate::{clock::Clock, homeserver::HomeServer}; + +/// What `request_blocking` returns: status, headers and body. +type BlockingResponse = (u16, Vec<(String, Vec)>, Vec); + +/// An [`Agent`] built from the homeserver's config, resolving names from +/// fixed tables, with a blocking `request` method. +/// +/// `request_blocking` blocks the calling thread (the reactor thread, under +/// trial) until the response arrives, so the server it talks to must run on +/// another thread. The agent sleeps on the virtual clock, so a request that +/// needs to sleep (a `.well-known` retry) would never finish. Tests avoid +/// those paths. +#[pyclass(frozen)] +pub struct RustFederationAgentForTest { + agent: Arc, + tokio: Handle, +} + +#[pymethods] +impl RustFederationAgentForTest { + /// `srv_records` maps an SRV name to `(priority, weight, port, target)` + /// tuples, `host_records` maps a host to its addresses, and + /// `port_overrides` maps the ports resolution picks to the ports to + /// really connect to. + #[new] + #[pyo3(signature = (hs, srv_records, host_records, port_overrides, user_agent = "Synapse-test".to_owned()))] + fn py_new( + py: Python<'_>, + hs: HomeServer, + srv_records: HashMap>, + host_records: HashMap>, + port_overrides: HashMap, + user_agent: String, + ) -> PyResult { + let config = hs.config(py)?; + let tokio = hs.get_rust_runtime(py)?.tokio_handle()?; + + let dns = InMemoryResolver::new(); + for (name, records) in srv_records { + dns.add_srv( + &name, + records + .into_iter() + .map(|(priority, weight, port, target)| SrvRecord { + priority, + weight, + port, + target, + ttl: 3600, + }) + .collect(), + ); + } + for (host, ips) in host_records { + let ips = ips + .iter() + .map(|ip| ip.parse::()) + .collect::, _>>() + .map_err(|err| PyValueError::new_err(err.to_string()))?; + dns.add_host(&host, ips); + } + + let clock = Clock::new(); + let agent = Agent::new( + AgentConfig::from_homeserver_config(&config, user_agent), + Arc::new(dns), + Arc::new(Clock::new()), + Arc::new(move || clock.now_millis()), + port_overrides, + )?; + + Ok(RustFederationAgentForTest { + agent: Arc::new(agent), + tokio, + }) + } + + /// Send a request and wait for the whole response. A failure raises + /// `RuntimeError(": ")`, where `kind` is one of the + /// `TransportError` kinds. + #[pyo3(signature = (method, uri, headers = Vec::new(), body = Vec::new()))] + fn request_blocking( + &self, + py: Python<'_>, + method: &str, + uri: &str, + headers: Vec<(String, String)>, + body: Vec, + ) -> PyResult { + let mut request = Request::builder() + .method( + Method::from_bytes(method.as_bytes()) + .map_err(|err| PyValueError::new_err(err.to_string()))?, + ) + .uri(uri) + .body(Full::new(bytes::Bytes::from(body))) + .map_err(|err| PyValueError::new_err(err.to_string()))?; + for (name, value) in headers { + request.headers_mut().append( + HeaderName::from_bytes(name.as_bytes()) + .map_err(|err| PyValueError::new_err(err.to_string()))?, + HeaderValue::from_str(&value) + .map_err(|err| PyValueError::new_err(err.to_string()))?, + ); + } + + let agent = self.agent.clone(); + let result = py.detach(|| { + self.tokio.block_on(async move { + let response = agent + .request(request) + .await + .map_err(|err| format!("{}: {err}", err.kind()))?; + let (parts, body) = response.into_parts(); + let body = body + .collect() + .await + .map_err(|err| format!("request: reading body: {err}"))? + .to_bytes(); + Ok::<_, String>((parts, body)) + }) + }); + + let (parts, body) = result.map_err(PyRuntimeError::new_err)?; + let headers = parts + .headers + .iter() + .map(|(name, value)| (name.as_str().to_owned(), value.as_bytes().to_vec())) + .collect(); + + Ok((parts.status.as_u16(), headers, body.to_vec())) + } +} diff --git a/rust/src/federation/mod.rs b/rust/src/federation/mod.rs index 56be65e1cc..46db2725d9 100644 --- a/rust/src/federation/mod.rs +++ b/rust/src/federation/mod.rs @@ -17,8 +17,8 @@ //! //! This is being built up from the network end, one layer at a time. //! -//! - [`agent`]: the parts of the agent that will send requests: server -//! name resolution, and the IP range and TLS policies. +//! - [`agent`]: sends requests: server name resolution, the IP range and +//! TLS policies, and the connection pool. //! - [`signing`]: the `X-Matrix` request signature. //! - [`transport`]: the seam between the request layer and the network, with //! a Python-backed implementation for tests. @@ -80,6 +80,7 @@ fn build_auth_header_for_test( pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { let child_module = PyModule::new(py, "federation")?; child_module.add_function(wrap_pyfunction!(build_auth_header_for_test, &child_module)?)?; + child_module.add_class::()?; m.add_submodule(&child_module)?; diff --git a/rust/src/federation/transport.rs b/rust/src/federation/transport.rs index 36928b4c79..8c7f4703a6 100644 --- a/rust/src/federation/transport.rs +++ b/rust/src/federation/transport.rs @@ -83,6 +83,40 @@ impl TransportError { pub fn is_dns(&self) -> bool { matches!(self, TransportError::Dns(_)) } + + /// A short name for the kind of failure. + pub fn kind(&self) -> &'static str { + match self { + TransportError::Dns(_) => "dns", + TransportError::ServiceUnavailable(_) => "service_unavailable", + TransportError::Connect(_) => "connect", + TransportError::Tls(_) => "tls", + TransportError::InvalidRequest(_) => "invalid_request", + TransportError::Request(_) => "request", + TransportError::Timeout => "timeout", + TransportError::Python(_) => "python", + } + } + + /// A copy of this error of the same kind, with the source chain + /// flattened into the message. hyper only lends us the error our + /// connector returned, so this is how we get it back out. + pub fn duplicate(&self) -> TransportError { + let flat = |e: &anyhow::Error| anyhow::anyhow!("{e:#}"); + match self { + TransportError::Dns(e) => TransportError::Dns(flat(e)), + TransportError::ServiceUnavailable(name) => { + TransportError::ServiceUnavailable(name.clone()) + } + TransportError::Connect(e) => TransportError::Connect(flat(e)), + TransportError::Tls(e) => TransportError::Tls(flat(e)), + TransportError::InvalidRequest(e) => TransportError::InvalidRequest(flat(e)), + TransportError::Request(e) => TransportError::Request(flat(e)), + TransportError::Timeout => TransportError::Timeout, + // Cloning a `PyErr` needs the GIL, which we may not hold. + TransportError::Python(e) => TransportError::Request(anyhow::anyhow!("{e}")), + } + } } impl fmt::Display for TransportError { diff --git a/synapse/synapse_rust/federation.pyi b/synapse/synapse_rust/federation.pyi index 4c865daef8..6060c8eca8 100644 --- a/synapse/synapse_rust/federation.pyi +++ b/synapse/synapse_rust/federation.pyi @@ -10,8 +10,11 @@ # See the GNU Affero General Public License for more details: # . +from typing import Mapping, Sequence + from signedjson.types import SigningKey +from synapse.server import HomeServer from synapse.types import JsonDict def build_auth_header_for_test( @@ -26,3 +29,43 @@ def build_auth_header_for_test( """Build the `X-Matrix` `Authorization` header the Rust request layer sends. Tests compare it with `MatrixFederationHttpClient.build_auth_headers`. """ + +class RustFederationAgentForTest: + """The Rust federation agent, built from the homeserver's config, resolving + names from fixed tables. For tests only. + + `request_blocking` blocks the calling thread until the response arrives, so + the server it talks to must run on another thread. + """ + + def __init__( + self, + hs: HomeServer, + srv_records: Mapping[str, Sequence[tuple[int, int, int, str]]], + host_records: Mapping[str, Sequence[str]], + port_overrides: Mapping[int, int], + user_agent: str = "Synapse-test", + ) -> None: + """ + Args: + hs: The homeserver whose config and Rust runtime to use. + srv_records: SRV name to `(priority, weight, port, target)` tuples. + host_records: Host name to IP addresses. + port_overrides: Ports that resolution picks, mapped to the ports to + really connect to. + user_agent: The `User-Agent` header to send. + """ + + def request_blocking( + self, + method: str, + uri: str, + headers: Sequence[tuple[str, str]] = (), + body: bytes = b"", + ) -> tuple[int, list[tuple[str, bytes]], bytes]: + """Send a request and return `(status, headers, body)`. + + Raises: + RuntimeError: `": "` if no response arrived, where + `kind` is e.g. `dns`, `connect`, `tls` or `service_unavailable`. + """ diff --git a/tests/http/federation/test_rust_agent.py b/tests/http/federation/test_rust_agent.py new file mode 100644 index 0000000000..35bf3f8adb --- /dev/null +++ b/tests/http/federation/test_rust_agent.py @@ -0,0 +1,299 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# +"""Tests for the Rust federation agent, against a real TLS server. + +The agent resolves names from fixed tables, and connects to the server below +whatever port resolution picks, so one listener can play every host. +""" + +import json +import ssl +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +import attr + +from twisted.internet.testing import MemoryReactor + +from synapse.server import HomeServer +from synapse.synapse_rust.federation import RustFederationAgentForTest +from synapse.util.clock import Clock + +from tests import unittest +from tests.http import create_test_cert_file, get_test_ca_cert_file, get_test_key_file + +# The names the test certificate is valid for. `nocert-server` resolves to the +# test server too, but is not in the certificate. +CERT_NAMES = [b"DNS:testserv", b"DNS:target-server", b"IP:127.0.0.1"] +HOSTS = ["testserv", "target-server", "srv-target", "nocert-server"] + + +@attr.s(auto_attribs=True, frozen=True) +class SeenRequest: + connection: int + sni: str | None + method: str + path: str + host: str | None + user_agent: str | None + + +class _TestServer: + """An HTTPS server on a thread, recording what it is sent.""" + + def __init__(self) -> None: + self.requests: list[SeenRequest] = [] + # Path to (status, body). + self.routes: dict[str, tuple[int, bytes]] = {} + self._connections = 0 + self._lock = threading.Lock() + + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(create_test_cert_file(CERT_NAMES), get_test_key_file()) + sni_by_socket: dict[int, str | None] = {} + + def sni_callback(sock: ssl.SSLObject, name: str | None, _: Any) -> None: + sni_by_socket[id(sock)] = name + + context.sni_callback = sni_callback + + server = self + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def setup(self) -> None: + super().setup() + with server._lock: + server._connections += 1 + self.connection_id = server._connections + self.sni = sni_by_socket.get(id(self.request)) + + def do_GET(self) -> None: + self._respond() + + def do_PUT(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + self.rfile.read(length) + self._respond() + + def _respond(self) -> None: + with server._lock: + server.requests.append( + SeenRequest( + connection=self.connection_id, + sni=self.sni, + method=self.command, + path=self.path, + host=self.headers.get("Host"), + user_agent=self.headers.get("User-Agent"), + ) + ) + status, body = server.routes.get(self.path, (404, b"")) + self.send_response(status) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: Any) -> None: + pass + + self._httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self._httpd.daemon_threads = True + self._httpd.socket = context.wrap_socket(self._httpd.socket, server_side=True) + self.port = self._httpd.server_address[1] + self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True) + self._thread.start() + + def stop(self) -> None: + self._httpd.shutdown() + self._httpd.server_close() + + +class RustFederationAgentTestCase(unittest.HomeserverTestCase): + def default_config(self) -> dict[str, Any]: + config = super().default_config() + # `setdefault`, since the base class has already applied any + # `override_config`. + config.setdefault("federation_custom_ca_list", [get_test_ca_cert_file()]) + config.setdefault("ip_range_whitelist", ["127.0.0.1"]) + return config + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.server = _TestServer() + self.addCleanup(self.server.stop) + self.srv_records: dict[str, list[tuple[int, int, int, str]]] = {} + + def _agent(self) -> RustFederationAgentForTest: + port = self.server.port + return RustFederationAgentForTest( + self.hs, + self.srv_records, + {host: ["127.0.0.1"] for host in HOSTS}, + {443: port, 8448: port, 8443: port}, + ) + + def _seen(self) -> list[SeenRequest]: + seen = list(self.server.requests) + self.server.requests.clear() + return seen + + def test_explicit_port(self) -> None: + """An explicit port skips `.well-known` and SRV.""" + self.server.routes["/foo"] = (200, b"hello") + + status, _, body = self._agent().request_blocking( + "GET", "matrix-federation://testserv:8448/foo" + ) + + self.assertEqual((status, body), (200, b"hello")) + (request,) = self._seen() + self.assertEqual(request.path, "/foo") + self.assertEqual(request.host, "testserv:8448") + self.assertEqual(request.sni, "testserv") + self.assertEqual(request.user_agent, "Synapse-test") + + def test_no_delegation(self) -> None: + """With no `.well-known` and no SRV record, we use the name on 8448.""" + self.server.routes["/foo"] = (200, b"") + + status, _, _ = self._agent().request_blocking( + "GET", "matrix-federation://testserv/foo" + ) + + self.assertEqual(status, 200) + well_known, request = self._seen() + self.assertEqual(well_known.path, "/.well-known/matrix/server") + self.assertEqual(well_known.host, "testserv") + self.assertEqual(well_known.sni, "testserv") + self.assertEqual(request.path, "/foo") + self.assertEqual(request.host, "testserv") + self.assertEqual(request.sni, "testserv") + + def test_well_known_delegation(self) -> None: + """A `.well-known` rewrites the host, the Host header and the TLS name, + and is cached.""" + self.server.routes["/.well-known/matrix/server"] = ( + 200, + json.dumps({"m.server": "target-server"}).encode(), + ) + self.server.routes["/foo"] = (200, b"") + agent = self._agent() + + agent.request_blocking("GET", "matrix-federation://testserv/foo") + well_known, request = self._seen() + self.assertEqual(well_known.host, "testserv") + self.assertEqual(request.host, "target-server") + self.assertEqual(request.sni, "target-server") + + # The second request needs no `.well-known` fetch, and reuses the + # pooled connection. + agent.request_blocking("PUT", "matrix-federation://testserv/foo", body=b"{}") + (second,) = self._seen() + self.assertEqual(second.method, "PUT") + self.assertEqual(second.host, "target-server") + self.assertEqual(second.connection, request.connection) + + def test_well_known_delegation_with_port(self) -> None: + self.server.routes["/.well-known/matrix/server"] = ( + 200, + json.dumps({"m.server": "target-server:8443"}).encode(), + ) + self.server.routes["/foo"] = (200, b"") + + self._agent().request_blocking("GET", "matrix-federation://testserv/foo") + + _, request = self._seen() + self.assertEqual(request.host, "target-server:8443") + self.assertEqual(request.sni, "target-server") + + def test_srv(self) -> None: + """An SRV record changes where we connect, but not the Host header or the + name the certificate must match.""" + self.srv_records["_matrix-fed._tcp.testserv"] = [(0, 0, 8443, "srv-target")] + self.server.routes["/foo"] = (200, b"") + + self._agent().request_blocking("GET", "matrix-federation://testserv/foo") + + _, request = self._seen() + self.assertEqual(request.host, "testserv") + self.assertEqual(request.sni, "testserv") + + def test_srv_service_unavailable(self) -> None: + self.srv_records["_matrix-fed._tcp.testserv"] = [(0, 0, 0, ".")] + + with self.assertRaisesRegex(RuntimeError, "^service_unavailable: "): + self._agent().request_blocking("GET", "matrix-federation://testserv/foo") + + def test_ip_literal(self) -> None: + """An IP literal is verified against the certificate's IP addresses, and + sends no SNI.""" + self.server.routes["/foo"] = (200, b"") + + self._agent().request_blocking("GET", "matrix-federation://127.0.0.1:8448/foo") + + (request,) = self._seen() + self.assertEqual(request.host, "127.0.0.1:8448") + self.assertIsNone(request.sni) + + @unittest.override_config({"ip_range_whitelist": []}) + def test_blocked_address(self) -> None: + """An address in the blocklist is refused, as a failed DNS lookup.""" + with self.assertRaisesRegex(RuntimeError, "^dns: "): + self._agent().request_blocking( + "GET", "matrix-federation://testserv:8448/foo" + ) + with self.assertRaisesRegex(RuntimeError, "^dns: "): + self._agent().request_blocking( + "GET", "matrix-federation://127.0.0.1:8448/foo" + ) + self.assertEqual(self._seen(), []) + + def test_bad_certificate(self) -> None: + with self.assertRaisesRegex(RuntimeError, "^tls: "): + self._agent().request_blocking( + "GET", "matrix-federation://nocert-server:8448/foo" + ) + + @unittest.override_config({"federation_verify_certificates": False}) + def test_verification_off(self) -> None: + self.server.routes["/foo"] = (200, b"") + status, _, _ = self._agent().request_blocking( + "GET", "matrix-federation://nocert-server:8448/foo" + ) + self.assertEqual(status, 200) + + # SNI is still sent. + (request,) = self._seen() + self.assertEqual(request.sni, "nocert-server") + + @unittest.override_config( + {"federation_certificate_verification_whitelist": ["nocert-*"]} + ) + def test_verification_whitelist(self) -> None: + self.server.routes["/foo"] = (200, b"") + status, _, _ = self._agent().request_blocking( + "GET", "matrix-federation://nocert-server:8448/foo" + ) + self.assertEqual(status, 200) + + def test_status_returned_as_is(self) -> None: + """The agent returns error responses rather than raising.""" + self.server.routes["/missing"] = (404, b'{"errcode": "M_NOT_FOUND"}') + status, _, body = self._agent().request_blocking( + "GET", "matrix-federation://testserv:8448/missing" + ) + self.assertEqual(status, 404) + self.assertEqual(json.loads(body), {"errcode": "M_NOT_FOUND"})