Add a Rust federation agent

The Rust replacement for `MatrixFederationAgent` and its connection pool,
as a library: nothing in Synapse sends through it yet. The request layer
(signing, retries, backoff) comes next and will be its first caller.

`Agent::request` takes a `matrix-federation://` request, applies
`.well-known` delegation to the URI and the `Host` header, and sends it
through hyper-util's pooled client (5 idle connections per host, 120s
idle timeout, HTTP/1.1, no automatic retries), as Python's pool does.

The pool calls our connector, derived from matrix-hyper-federation-
client's `MatrixConnector` (Apache-2.0). It resolves the server name
(SRV), tries each server in order and returns the first error if all
fail. For each server it resolves addresses through the IP range policy,
connects TCP with Twisted's staggered attempts and 30s timeout, or
tunnels through `https_proxy` with `CONNECT` unless `no_proxy` exempts
the host, then does TLS verified against the delegated server name,
never the SRV target. The `.well-known` fetch shares the pool and, as in
Python, blocks IP literals by the blocklist alone.

Failures are a typed `TransportError` (DNS, connect, TLS, and so on), so
the request layer can tell a DNS failure from the rest. The agent
implements `FederationTransport`.

`RustFederationAgentForTest` exposes a blocking `request` to Python. The
trial tests drive it against a real TLS server on a thread, signed by
the test CA, covering the delegation, SRV, SNI, IP literal, blocklist,
certificate verification and connection reuse cases.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb

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",
 ]

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 = [

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:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ * Originally licensed under the Apache License, Version 2.0:
+ * <http://www.apache.org/licenses/LICENSE-2.0>.
+ *
+ * Derived from `MatrixConnector` in `src/server_resolver.rs` of
+ * <https://github.com/erikjohnston/matrix-hyper-federation-client>.
+ *
+ * [This file includes modifications made by Element Creations Ltd]
+ */
+
+//! The connector hyper's connection pool calls to open a connection.
+//!
+//! For a `matrix-federation://<host>[: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 `<host>`. 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<Proxy>,
+    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<u16, u16>,
+}
+
+/// A `tower_service::Service<Uri>` for hyper's legacy client.
+#[derive(Clone)]
+pub struct FederationConnector {
+    pub(super) inner: Arc<ConnectorInner>,
+}
+
+impl tower_service::Service<Uri> for FederationConnector {
+    type Response = FederationStream;
+    type Error = TransportError;
+    type Future = Pin<Box<dyn Future<Output = Result<FederationStream, TransportError>> + Send>>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        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<FederationStream, TransportError> {
+        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<Box<dyn Io>, 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<Box<dyn Io>, 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<dyn Io> = 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<IpAddr>, port: u16) -> Result<TcpStream, TransportError> {
+    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<io::Error> = 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<S: AsyncRead + AsyncWrite + Unpin + ?Sized>(
+    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<T: AsyncRead + AsyncWrite + Send + Unpin> Io for T {}
+
+/// An open federation connection, as handed to hyper.
+pub struct FederationStream {
+    io: TokioIo<Box<dyn Io>>,
+}
+
+impl FederationStream {
+    fn new(io: Box<dyn Io>) -> 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<io::Result<()>> {
+        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<io::Result<usize>> {
+        Pin::new(&mut self.io).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.io).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        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}");
+    }
+}

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://<server name>/<path>`
+//! 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<dyn Fn() -> 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<FederationConnector, Full<Bytes>>;
+
+pub struct Agent {
+    client: PoolClient,
+    well_known: Arc<WellKnownResolver>,
+    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<dyn DnsResolver>,
+        sleeper: Arc<dyn Sleeper>,
+        now: Now,
+        port_overrides: HashMap<u16, u16>,
+    ) -> anyhow::Result<Self> {
+        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<Full<Bytes>>,
+    ) -> Result<Response<Incoming>, 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<Bytes>,
+    ) -> Result<Response<ResponseBody>, 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::<TransportError>() {
+            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<FetchResponse, FetchError> {
+        // 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::<http_body_util::LengthLimitError>() => {
+                return Err(FetchError::TooLarge)
+            }
+            Err(err) => return Err(FetchError::Other(anyhow!(err))),
+        };
+
+        Ok(FetchResponse {
+            status: parts.status,
+            headers: parts.headers,
+            body,
+        })
+    }
+}

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:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ */
+
+//! 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<u8>)>, Vec<u8>);
+
+/// 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<Agent>,
+    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<String, Vec<(u16, u16, u16, String)>>,
+        host_records: HashMap<String, Vec<String>>,
+        port_overrides: HashMap<u16, u16>,
+        user_agent: String,
+    ) -> PyResult<Self> {
+        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::<IpAddr>())
+                .collect::<Result<Vec<_>, _>>()
+                .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("<kind>: <message>")`, 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<u8>,
+    ) -> PyResult<BlockingResponse> {
+        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()))
+    }
+}

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::<agent::python::RustFederationAgentForTest>()?;
 
     m.add_submodule(&child_module)?;
 

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 {

synapse/synapse_rust/federation.pyi

@@ -10,8 +10,11 @@
 # See the GNU Affero General Public License for more details:
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 
+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: `"<kind>: <message>"` if no response arrived, where
+                `kind` is e.g. `dns`, `connect`, `tls` or `service_unavailable`.
+        """

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+"""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"})