From 7f34eefa269cbda6085be851cdaab7bc08a1c61e Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 15 Oct 2026 11:34:53 +0000 Subject: [PATCH] Port federation server name resolution and policies to Rust The parts of the outbound federation agent that decide where to connect and what to trust, each a port of the Python it replaces, with Rust unit tests. Nothing uses them yet. - `resolve`: `_matrix-fed._tcp`, then `_matrix._tcp`, then port 8448, as `MatrixHostnameEndpoint._resolve_server`. SRV records are ordered as `_sort_server_list` (RFC 2782), cached while every record is live, and the stale answer is used when a refetch fails. A lone `.` target means the service is unavailable. SRV lookups go through hickory, and A/AAAA through the system resolver, as Twisted uses getaddrinfo. Parts derive from matrix-hyper-federation-client (Apache-2.0). - `well_known`: `WellKnownResolver`, with Python's cache periods, jitter, grace period, had-valid memory, size cap and retries, behind a `WellKnownFetcher` trait so the state machine is tested without HTTP. Two additions: each fetch attempt has a timeout (`client_timeout`), and concurrent lookups for one server name share a fetch. An `m.server` that is not a valid URI authority now counts as an invalid response. - `policy`: the IP range lists, refusing a name if any of its addresses is blocked (the DNS rebinding defence), and the `https_proxy` settings with Python's `no_proxy` matching. - `tls`: `FederationPolicyForHTTPS` on rustls: platform roots or the custom CA list, verification off (still sending SNI), the per-host verification whitelist. rustls has no TLS 1.0 or 1.1, so a minimum below 1.2 means 1.2. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb --- Cargo.lock | 277 +++++++- rust/Cargo.toml | 18 + rust/src/federation/agent/mod.rs | 34 + rust/src/federation/agent/policy.rs | 312 +++++++++ rust/src/federation/agent/resolve.rs | 590 +++++++++++++++++ rust/src/federation/agent/tls.rs | 258 ++++++++ rust/src/federation/agent/well_known.rs | 805 ++++++++++++++++++++++++ rust/src/federation/mod.rs | 3 + 8 files changed, 2292 insertions(+), 5 deletions(-) create mode 100644 rust/src/federation/agent/mod.rs create mode 100644 rust/src/federation/agent/policy.rs create mode 100644 rust/src/federation/agent/resolve.rs create mode 100644 rust/src/federation/agent/tls.rs create mode 100644 rust/src/federation/agent/well_known.rs diff --git a/Cargo.lock b/Cargo.lock index caab8ace95..a63ff20355 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -208,6 +208,36 @@ dependencies = [ "libc", ] +[[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + +[[package]] +name = "crossbeam-channel" +version = "0.5.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "98b0cc327b5bc766e7fda9c9260cc0fa81b43a8e240440422dff70788e3f9ef1" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" + [[package]] name = "crypto-common" version = "0.1.6" @@ -263,6 +293,12 @@ dependencies = [ "syn 2.0.104", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "deadpool" version = "0.12.3" @@ -370,6 +406,18 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" +[[package]] +name = "enum-as-inner" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1e6a265c649f3f5979b601d26f1d05ada116434c87741c9493cb56218f76cbc" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn 2.0.104", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -526,6 +574,18 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi 5.3.0", + "wasip2", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -535,7 +595,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "r-efi", + "r-efi 6.0.0", "rand_core 0.10.1", "wasm-bindgen", ] @@ -618,6 +678,52 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hickory-proto" +version = "0.25.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8a6fe56c0038198998a6f217ca4e7ef3a5e51f46163bd6dd60b5c71ca6c6502" +dependencies = [ + "async-trait", + "cfg-if", + "data-encoding", + "enum-as-inner", + "futures-channel", + "futures-io", + "futures-util", + "idna", + "ipnet", + "once_cell", + "rand 0.9.5", + "ring", + "thiserror", + "tinyvec", + "tokio", + "tracing", + "url", +] + +[[package]] +name = "hickory-resolver" +version = "0.25.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc62a9a99b0bfb44d2ab95a7208ac952d31060efc16241c87eaf36406fecf87a" +dependencies = [ + "cfg-if", + "futures-util", + "hickory-proto", + "ipconfig", + "moka", + "once_cell", + "parking_lot", + "rand 0.9.5", + "resolv-conf", + "smallvec", + "thiserror", + "tokio", + "tracing", +] + [[package]] name = "hmac" version = "0.13.0" @@ -905,6 +1011,19 @@ dependencies = [ "hashbrown 0.15.4", ] +[[package]] +name = "ipconfig" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d40460c0ce33d6ce4b0630ad68ff63d6661961c48b6dba35e5a4d81cfb48222" +dependencies = [ + "socket2 0.6.5", + "widestring", + "windows-registry", + "windows-result", + "windows-sys 0.61.2", +] + [[package]] name = "ipnet" version = "2.11.0" @@ -1053,6 +1172,23 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "moka" +version = "0.12.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4293f18e7567a1caf3c584855554377025c65e0aa445344d04171f5ad63d19b9" +dependencies = [ + "crossbeam-channel", + "crossbeam-epoch", + "crossbeam-utils", + "equivalent", + "parking_lot", + "portable-atomic", + "smallvec", + "tagptr", + "uuid", +] + [[package]] name = "num_cpus" version = "1.17.0" @@ -1086,6 +1222,10 @@ name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +dependencies = [ + "critical-section", + "portable-atomic", +] [[package]] name = "openssl-probe" @@ -1176,7 +1316,7 @@ dependencies = [ "hmac", "md-5", "memchr", - "rand", + "rand 0.10.2", "sha2 0.11.0", "stringprep", ] @@ -1202,6 +1342,15 @@ dependencies = [ "zerovec", ] +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "proc-macro2" version = "1.0.95" @@ -1321,7 +1470,7 @@ dependencies = [ "bytes", "getrandom 0.4.3", "lru-slab", - "rand", + "rand 0.10.2", "rand_pcg", "ring", "rustc-hash", @@ -1357,12 +1506,28 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "r-efi" version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + [[package]] name = "rand" version = "0.10.2" @@ -1374,6 +1539,16 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -1383,6 +1558,15 @@ dependencies = [ "getrandom 0.2.16", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "rand_core" version = "0.10.1" @@ -1478,6 +1662,12 @@ dependencies = [ "web-sys", ] +[[package]] +name = "resolv-conf" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7" + [[package]] name = "ring" version = "0.17.14" @@ -1831,8 +2021,10 @@ dependencies = [ "futures", "headers", "hex", + "hickory-resolver", "http", "http-body-util", + "httpdate", "icu_segmenter", "ipnet", "itertools", @@ -1847,6 +2039,7 @@ dependencies = [ "pyo3", "pyo3-log", "pythonize", + "rand 0.10.2", "regex", "reqwest", "rustc-hash", @@ -1860,7 +2053,9 @@ dependencies = [ "tokio", "tokio-postgres", "tokio-postgres-rustls", + "tokio-rustls", "ulid", + "url", ] [[package]] @@ -1883,6 +2078,12 @@ dependencies = [ "syn 2.0.104", ] +[[package]] +name = "tagptr" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" + [[package]] name = "target-lexicon" version = "0.13.5" @@ -2000,7 +2201,7 @@ dependencies = [ "pin-project-lite", "postgres-protocol", "postgres-types", - "rand", + "rand 0.10.2", "socket2 0.6.5", "tokio", "tokio-util", @@ -2126,7 +2327,7 @@ version = "3.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "947dde63b6d514cc5e044edad4e0ca7261afd1099d16c83d942cb2b2f348689c" dependencies = [ - "rand", + "rand 0.10.2", "web-time", ] @@ -2180,6 +2381,17 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" +[[package]] +name = "uuid" +version = "1.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cc1186384beb7dd8eedea376413fd654937285ea6c9cfbb928dc3043ea4b606" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "version_check" version = "0.9.5" @@ -2330,12 +2542,47 @@ dependencies = [ "web-sys", ] +[[package]] +name = "widestring" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471" + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-registry" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720" +dependencies = [ + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.52.0" @@ -2475,6 +2722,26 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zerocopy" +version = "0.8.63" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5fe1f8f1b06191a00962174c61aa5005e0bb391a6d80d07e24d115c01a92ed8" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.63" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "863ad3ac83293fb4d740aedbfdc9240dd8d1a50c1099acd76ce80ce7c7230c7f" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.104", +] + [[package]] name = "zerofrom" version = "0.1.6" diff --git a/rust/Cargo.toml b/rust/Cargo.toml index ad55b4ddde..dd3282a4ea 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -102,6 +102,24 @@ tokio-postgres-rustls = { version = "0.14.0", default-features = false, features "ring", ] } lru = "0.16.4" +# Outbound federation (`federation::agent`): the TLS policy. +tokio-rustls = { version = "0.26.2", default-features = false, features = [ + "ring", + "logging", + "tls12", +] } +# 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 = [ + "tokio", + "system-config", +] } +# SRV record ordering and cache period jitter. +rand = "0.10.2" +# The `Expires` header of `.well-known` responses. +httpdate = "1.0.3" +# Resolving `.well-known` redirects. +url = "2.5.4" # Signing outbound federation requests. 2.x rather than 3.x, which needs a # newer rustc than our MSRV. ed25519-dalek = "2.2.0" diff --git a/rust/src/federation/agent/mod.rs b/rust/src/federation/agent/mod.rs new file mode 100644 index 0000000000..1706a9f39d --- /dev/null +++ b/rust/src/federation/agent/mod.rs @@ -0,0 +1,34 @@ +/* + * 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: + * . + * + */ + +//! The outbound federation agent: Python's `MatrixFederationAgent` and its +//! `HTTPConnectionPool`. This holds its parts so far. +//! +//! - [`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. + +use std::sync::Arc; + +pub mod policy; +pub mod resolve; +pub mod tls; +pub mod well_known; + +/// 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>; diff --git a/rust/src/federation/agent/policy.rs b/rust/src/federation/agent/policy.rs new file mode 100644 index 0000000000..ff76d3d40c --- /dev/null +++ b/rust/src/federation/agent/policy.rs @@ -0,0 +1,312 @@ +/* + * 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: + * . + * + */ + +//! Which addresses outbound federation may connect to, and through which +//! proxy. + +use std::{net::IpAddr, sync::Arc}; + +use anyhow::{anyhow, bail, Context}; +use base64::{prelude::BASE64_STANDARD, Engine}; +use log::info; + +use super::resolve::{parse_ip_literal, DnsError, DnsResolver}; +use crate::{ + config::types::{IpRangeSet, ProxyConfig}, + federation::transport::TransportError, +}; + +/// The IP range block and allow lists, applied to every address we connect +/// to. This is Python's `_IPBlockingResolver` (via +/// `BlocklistingReactorWrapper`). +#[derive(Clone)] +pub struct IpPolicy { + blocklist: IpRangeSet, + allowlist: Option, +} + +impl IpPolicy { + pub fn new(blocklist: IpRangeSet, allowlist: Option) -> Self { + IpPolicy { + blocklist, + allowlist, + } + } + + /// Blocked if in the blocklist and not in the allowlist. + pub fn is_blocked(&self, ip: &IpAddr) -> bool { + self.blocklist.contains(ip) && !self.allowlist.as_ref().is_some_and(|a| a.contains(ip)) + } + + /// Whether an IP literal is blocked by the blocklist alone, ignoring the + /// allowlist. This is how Python's `BlocklistingAgentWrapper` treats the + /// `.well-known` fetch, since it is not given the allowlist. + pub fn is_blocked_ignoring_allowlist(&self, ip: &IpAddr) -> bool { + self.blocklist.contains(ip) + } + + /// Resolve `host` to the addresses we may connect to. + /// + /// If any address is blocked, none are returned: an attacker controlling + /// DNS could otherwise mix a blocked address in with good ones (DNS + /// rebinding). Python reports this as a failed DNS lookup, and so do we. + pub async fn resolve( + &self, + dns: &dyn DnsResolver, + host: &str, + ) -> Result, TransportError> { + let ips = match dns.lookup_ip(host).await { + Ok(ips) => ips, + Err(DnsError::NotFound) => { + return Err(TransportError::Dns(anyhow!("no addresses for {host}"))) + } + Err(DnsError::Failed(err)) => return Err(TransportError::Dns(err)), + }; + + if let Some(blocked) = ips.iter().find(|ip| self.is_blocked(ip)) { + info!("Blocked {blocked} from DNS resolution to {host}"); + return Err(TransportError::Dns(anyhow!( + "{host} resolved to blocked address {blocked}" + ))); + } + + Ok(ips) + } +} + +/// An `https_proxy` that federation connections tunnel through with +/// `CONNECT`, unless `no_proxy` exempts the host. +#[derive(Clone, Debug)] +pub struct Proxy { + /// Whether to speak TLS to the proxy itself. + pub tls: bool, + pub host: String, + pub port: u16, + /// The `Proxy-Authorization` header value, from credentials in the + /// proxy URL. + pub authorization: Option, + no_proxy: Arc, +} + +impl Proxy { + /// The proxy to use, if `https_proxy` is set. Parses the URL the way + /// Python's `http_proxy_endpoint` does: scheme `http` (the default) or + /// `https`, port defaulting to 1080, optional `user:password@`. + pub fn from_config(config: &ProxyConfig) -> anyhow::Result> { + let Some(https_proxy) = config.https_proxy.as_deref().filter(|p| !p.is_empty()) else { + return Ok(None); + }; + + let with_scheme = if https_proxy.contains("://") { + https_proxy.to_owned() + } else { + format!("http://{https_proxy}") + }; + let url = url::Url::parse(&with_scheme) + .with_context(|| format!("parsing https_proxy {https_proxy:?}"))?; + + let tls = match url.scheme() { + "http" => false, + "https" => true, + other => bail!("unsupported https_proxy scheme {other:?}"), + }; + let host = url + .host_str() + .with_context(|| format!("https_proxy {https_proxy:?} has no host"))? + .to_owned(); + let port = url.port().unwrap_or(1080); + + // Credentials are only used when both parts are present, and are + // sent as written in the URL (Python does not percent-decode them). + let authorization = match (url.username(), url.password()) { + (user, Some(password)) if !user.is_empty() && !password.is_empty() => Some(format!( + "Basic {}", + BASE64_STANDARD.encode(format!("{user}:{password}")) + )), + _ => None, + }; + + Ok(Some(Proxy { + tls, + host, + port, + authorization, + no_proxy: Arc::new(NoProxy::new( + config.no_proxy_hosts.as_deref().unwrap_or(&[]), + )), + })) + } + + /// Whether a connection to `host` should go through this proxy. + pub fn applies_to(&self, host: &str) -> bool { + !self.no_proxy.bypasses(host) + } +} + +/// The `no_proxy` host list, matched as Python's +/// `urllib.request.proxy_bypass_environment` does: `*` exempts everything, +/// otherwise an entry matches the host itself and its subdomains. CIDR +/// entries are only compared as strings, as in Python. +#[derive(Debug)] +pub struct NoProxy { + all: bool, + names: Vec, +} + +impl NoProxy { + pub fn new(hosts: &[String]) -> Self { + // Python joins the list with commas and compares the whole string + // to `*`. + let all = hosts.len() == 1 && hosts[0] == "*"; + let names = hosts + .iter() + .flat_map(|h| h.split(',')) + .map(|name| name.trim().trim_start_matches('.').to_lowercase()) + .filter(|name| !name.is_empty()) + .collect(); + NoProxy { all, names } + } + + pub fn bypasses(&self, host: &str) -> bool { + if self.all { + return true; + } + + let host = host.to_lowercase(); + // Python splits a trailing `:port` off; our hosts carry no port, but + // an IPv6 literal must not be mistaken for one. + let host_only = match host.rsplit_once(':') { + Some((h, port)) + if parse_ip_literal(&host).is_none() + && port.bytes().all(|b| b.is_ascii_digit()) => + { + h.to_owned() + } + _ => host.clone(), + }; + + self.names.iter().any(|name| { + let dotted = format!(".{name}"); + host_only == *name + || host == *name + || host_only.ends_with(&dotted) + || host.ends_with(&dotted) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::federation::agent::resolve::InMemoryResolver; + + fn policy(block: &[&str], allow: Option<&[&str]>) -> IpPolicy { + let set = + |nets: &[&str]| IpRangeSet::new(nets.iter().map(|n| n.parse().unwrap()).collect()); + IpPolicy::new(set(block), allow.map(set)) + } + + #[test] + fn allowlist_overrides_blocklist() { + let p = policy(&["10.0.0.0/8"], Some(&["10.1.0.0/16"])); + assert!(p.is_blocked(&"10.2.0.1".parse().unwrap())); + assert!(!p.is_blocked(&"10.1.0.1".parse().unwrap())); + assert!(!p.is_blocked(&"11.0.0.1".parse().unwrap())); + assert!(p.is_blocked_ignoring_allowlist(&"10.1.0.1".parse().unwrap())); + } + + #[tokio::test] + async fn one_blocked_address_blocks_the_name() { + let dns = InMemoryResolver::new(); + dns.add_host( + "mixed.example", + vec!["1.2.3.4".parse().unwrap(), "10.0.0.1".parse().unwrap()], + ); + dns.add_host("good.example", vec!["1.2.3.4".parse().unwrap()]); + let p = policy(&["10.0.0.0/8"], None); + + let err = p.resolve(&dns, "mixed.example").await.unwrap_err(); + assert!(err.is_dns(), "{err}"); + assert_eq!( + p.resolve(&dns, "good.example").await.unwrap(), + ["1.2.3.4".parse::().unwrap()] + ); + + // IP literals are checked too. + assert!(p.resolve(&dns, "10.0.0.1").await.unwrap_err().is_dns()); + assert!(p + .resolve(&dns, "unknown.example") + .await + .unwrap_err() + .is_dns()); + } + + #[test] + fn no_proxy_matching() { + let no_proxy = NoProxy::new(&[ + "example.com".to_owned(), + ".other.org".to_owned(), + " spaced.net ".to_owned(), + ]); + assert!(no_proxy.bypasses("example.com")); + assert!(no_proxy.bypasses("sub.example.com")); + assert!(no_proxy.bypasses("EXAMPLE.COM")); + assert!(!no_proxy.bypasses("notexample.com")); + assert!(no_proxy.bypasses("other.org")); + assert!(no_proxy.bypasses("a.other.org")); + assert!(no_proxy.bypasses("spaced.net")); + assert!(!no_proxy.bypasses("matrix.org")); + + assert!(NoProxy::new(&["*".to_owned()]).bypasses("anything")); + assert!(!NoProxy::new(&[]).bypasses("anything")); + } + + #[test] + fn proxy_url_parsing() { + let config = |p: &str| ProxyConfig { + http_proxy: None, + https_proxy: Some(p.to_owned()), + no_proxy_hosts: None, + }; + + let proxy = Proxy::from_config(&config("proxy.example")) + .unwrap() + .unwrap(); + assert!(!proxy.tls); + assert_eq!((proxy.host.as_str(), proxy.port), ("proxy.example", 1080)); + assert_eq!(proxy.authorization, None); + + let proxy = Proxy::from_config(&config("https://bob:pa%20ss@proxy.example:3128")) + .unwrap() + .unwrap(); + assert!(proxy.tls); + assert_eq!(proxy.port, 3128); + assert_eq!( + proxy.authorization.as_deref(), + Some(format!("Basic {}", BASE64_STANDARD.encode("bob:pa%20ss")).as_str()) + ); + + // A user without a password sends no credentials. + let proxy = Proxy::from_config(&config("http://bob@proxy.example")) + .unwrap() + .unwrap(); + assert_eq!(proxy.authorization, None); + + assert!(Proxy::from_config(&config("socks5://proxy.example")).is_err()); + assert!(Proxy::from_config(&ProxyConfig::default()) + .unwrap() + .is_none()); + } +} diff --git a/rust/src/federation/agent/resolve.rs b/rust/src/federation/agent/resolve.rs new file mode 100644 index 0000000000..ab112c64d9 --- /dev/null +++ b/rust/src/federation/agent/resolve.rs @@ -0,0 +1,590 @@ +/* + * 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 `src/server_resolver.rs` of + * . + * + * [This file includes modifications made by Element Creations Ltd] + */ + +//! Resolving a Matrix server name to the hosts and ports to connect to. +//! +//! This is `MatrixHostnameEndpoint._resolve_server` and `SrvResolver` from +//! Python. The `.well-known` lookup happens before this, in +//! [`super::well_known`], because it rewrites the request's `Host` header. + +use std::{ + collections::{BTreeMap, HashMap}, + net::IpAddr, + sync::{Arc, Mutex}, +}; + +use anyhow::anyhow; +use async_trait::async_trait; +use log::{debug, warn}; +use rand::{seq::SliceRandom, Rng, RngExt}; + +use crate::federation::transport::TransportError; + +/// The port federation uses when a server name gives none and has no SRV +/// record. +pub const DEFAULT_FEDERATION_PORT: u16 = 8448; + +/// One SRV record. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SrvRecord { + pub priority: u16, + pub weight: u16, + pub port: u16, + /// The target host, without a trailing dot. `.` for the root, which a + /// lone record uses to say the service is unavailable. + pub target: String, + /// The record's TTL, in seconds. + pub ttl: u32, +} + +/// Why a DNS lookup found nothing. +#[derive(Debug)] +pub enum DnsError { + /// The name does not exist, or has no records of this type. + NotFound, + /// The lookup itself failed (no answer, server failure, and so on). + Failed(anyhow::Error), +} + +/// The DNS lookups the agent makes. +#[async_trait] +pub trait DnsResolver: Send + Sync + 'static { + async fn lookup_srv(&self, name: &str) -> Result, DnsError>; + + /// The addresses of `host`. An IP literal resolves to itself. + async fn lookup_ip(&self, host: &str) -> Result, DnsError>; +} + +/// The production resolver. SRV records come from hickory, which reads the +/// system's `resolv.conf` as Twisted's DNS client does. Addresses come from +/// the system resolver (`getaddrinfo`), as Twisted's `HostnameEndpoint` uses, +/// so `/etc/hosts` and `nsswitch.conf` apply. +pub struct SystemResolver { + hickory: hickory_resolver::TokioResolver, +} + +impl SystemResolver { + pub fn new() -> anyhow::Result { + let hickory = hickory_resolver::TokioResolver::builder_tokio()?.build(); + Ok(SystemResolver { hickory }) + } +} + +#[async_trait] +impl DnsResolver for SystemResolver { + async fn lookup_srv(&self, name: &str) -> Result, DnsError> { + let lookup = match self.hickory.srv_lookup(name).await { + Ok(lookup) => lookup, + Err(err) if err.is_no_records_found() => return Err(DnsError::NotFound), + Err(err) => return Err(DnsError::Failed(err.into())), + }; + + let records = lookup + .as_lookup() + .record_iter() + .filter_map(|record| { + let srv = record.data().as_srv()?; + let target = srv.target(); + let target = if target.is_root() { + ".".to_owned() + } else { + target.to_ascii().trim_end_matches('.').to_owned() + }; + Some(SrvRecord { + priority: srv.priority(), + weight: srv.weight(), + port: srv.port(), + target, + ttl: record.ttl(), + }) + }) + .collect(); + + Ok(records) + } + + async fn lookup_ip(&self, host: &str) -> Result, DnsError> { + if let Some(ip) = parse_ip_literal(host) { + return Ok(vec![ip]); + } + + // `lookup_host` wants a port, which it ignores for our purposes. + let addrs = tokio::net::lookup_host((host, 0)).await.map_err(|err| { + // getaddrinfo's errors do not tell "no such name" apart from a + // failed lookup in a way std exposes, so treat all as failures. + DnsError::Failed(anyhow!(err).context(format!("resolving {host}"))) + })?; + + let ips: Vec = addrs.map(|addr| addr.ip()).collect(); + if ips.is_empty() { + return Err(DnsError::NotFound); + } + Ok(ips) + } +} + +/// A resolver that answers from fixed tables, for tests. +#[derive(Default)] +pub struct InMemoryResolver { + srv: Mutex, String>>>, + ips: Mutex>>, +} + +impl InMemoryResolver { + pub fn new() -> Self { + Self::default() + } + + pub fn add_srv(&self, name: &str, records: Vec) { + self.srv + .lock() + .expect("poisoned") + .insert(name.to_owned(), Ok(records)); + } + + /// Make SRV lookups of `name` fail (rather than find nothing). + pub fn fail_srv(&self, name: &str, message: &str) { + self.srv + .lock() + .expect("poisoned") + .insert(name.to_owned(), Err(message.to_owned())); + } + + pub fn add_host(&self, host: &str, ips: Vec) { + self.ips + .lock() + .expect("poisoned") + .insert(host.to_owned(), ips); + } +} + +#[async_trait] +impl DnsResolver for InMemoryResolver { + async fn lookup_srv(&self, name: &str) -> Result, DnsError> { + match self.srv.lock().expect("poisoned").get(name) { + Some(Ok(records)) if !records.is_empty() => Ok(records.clone()), + Some(Ok(_)) | None => Err(DnsError::NotFound), + Some(Err(message)) => Err(DnsError::Failed(anyhow!("{message}"))), + } + } + + async fn lookup_ip(&self, host: &str) -> Result, DnsError> { + if let Some(ip) = parse_ip_literal(host) { + return Ok(vec![ip]); + } + match self.ips.lock().expect("poisoned").get(host) { + Some(ips) if !ips.is_empty() => Ok(ips.clone()), + _ => Err(DnsError::NotFound), + } + } +} + +/// Parse an IP literal, as a URI host gives it (IPv6 in brackets) or bare. +pub fn parse_ip_literal(host: &str) -> Option { + host.strip_prefix('[') + .and_then(|h| h.strip_suffix(']')) + .unwrap_or(host) + .parse() + .ok() +} + +/// A host and port to try connecting to. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Server { + pub host: String, + pub port: u16, +} + +/// A cached SRV record, with the time it expires in seconds since the epoch. +#[derive(Clone, Debug)] +struct CachedRecord { + record: SrvRecord, + expires: u64, +} + +/// Resolves server names via SRV, caching the answers. +/// +/// The cache follows Python's `SrvResolver`: a cached answer is used only +/// while all its records are unexpired, and when a refetch fails the stale +/// answer is used instead. The cache belongs to the agent, where Python's is +/// module-global. +pub struct ServerResolver { + dns: Arc, + cache: Mutex>>, +} + +impl ServerResolver { + pub fn new(dns: Arc) -> Self { + ServerResolver { + dns, + cache: Mutex::new(HashMap::new()), + } + } + + pub fn dns(&self) -> &Arc { + &self.dns + } + + /// The servers to try for `host` and optional `port`, in order. + /// + /// An IP literal or an explicit port bypasses SRV. Otherwise we look up + /// `_matrix-fed._tcp`, then the deprecated `_matrix._tcp`, and fall back + /// to `host:8448`. + pub async fn resolve_server( + &self, + host: &str, + port: Option, + now_secs: u64, + ) -> Result, TransportError> { + if port.is_some() || parse_ip_literal(host).is_some() { + return Ok(vec![Server { + host: host.to_owned(), + port: port.unwrap_or(DEFAULT_FEDERATION_PORT), + }]); + } + + for service in ["_matrix-fed._tcp", "_matrix._tcp"] { + let servers = self + .resolve_service(&format!("{service}.{host}"), now_secs) + .await?; + if !servers.is_empty() { + debug!("Got {servers:?} from {service} SRV lookup for {host}"); + return Ok(servers); + } + } + + debug!("No SRV records for {host}"); + Ok(vec![Server { + host: host.to_owned(), + port: DEFAULT_FEDERATION_PORT, + }]) + } + + /// Look up one SRV name, returning the servers in RFC 2782 order, or + /// nothing if there are no records. + async fn resolve_service( + &self, + service_name: &str, + now_secs: u64, + ) -> Result, TransportError> { + let cached = self + .cache + .lock() + .expect("poisoned") + .get(service_name) + .cloned(); + + // An empty cached answer is refetched, as in Python. + if let Some(cached) = &cached { + if !cached.is_empty() && cached.iter().all(|c| c.expires > now_secs) { + let records = cached.iter().map(|c| c.record.clone()).collect(); + return Ok(sort_srv_records(records, &mut rand::rng())); + } + } + + let records = match self.dns.lookup_srv(service_name).await { + Ok(records) => records, + Err(DnsError::NotFound) => return Ok(Vec::new()), + Err(DnsError::Failed(err)) => { + return match cached { + Some(cached) if !cached.is_empty() => { + warn!("Failed to resolve {service_name}, falling back to cache: {err:#}"); + let records = cached.into_iter().map(|c| c.record).collect(); + Ok(sort_srv_records(records, &mut rand::rng())) + } + _ => Err(TransportError::Dns( + err.context(format!("SRV lookup of {service_name}")), + )), + }; + } + }; + + if let [only] = records.as_slice() { + if only.target == "." { + return Err(TransportError::ServiceUnavailable(service_name.to_owned())); + } + } + + let cached = records + .iter() + .map(|record| CachedRecord { + record: record.clone(), + expires: now_secs + u64::from(record.ttl), + }) + .collect(); + self.cache + .lock() + .expect("poisoned") + .insert(service_name.to_owned(), cached); + + Ok(sort_srv_records(records, &mut rand::rng())) + } +} + +/// Order SRV records as Python's `_sort_server_list` does: by priority, and +/// within a priority by a weighted random pick (RFC 2782, without its +/// off-by-one). Zero-weight records left over are shuffled onto the end. +pub fn sort_srv_records(records: Vec, rng: &mut R) -> Vec { + let mut by_priority: BTreeMap> = BTreeMap::new(); + for record in records { + by_priority.entry(record.priority).or_default().push(record); + } + + let mut results = Vec::new(); + for (_, mut records) in by_priority { + let mut total_weight: u32 = records.iter().map(|r| u32::from(r.weight)).sum(); + + while !records.is_empty() && total_weight > 0 { + let mut target = rng.random_range(1..=total_weight); + let mut chosen = records.len() - 1; + for (i, record) in records.iter().enumerate() { + target = target.saturating_sub(u32::from(record.weight)); + if target == 0 { + chosen = i; + break; + } + } + let record = records.remove(chosen); + total_weight -= u32::from(record.weight); + results.push(record); + } + + records.shuffle(rng); + results.extend(records); + } + + results + .into_iter() + .map(|record| Server { + host: record.target, + port: record.port, + }) + .collect() +} + +#[cfg(test)] +mod tests { + use rand::{rngs::StdRng, SeedableRng}; + + use super::*; + + fn srv(priority: u16, weight: u16, target: &str, port: u16, ttl: u32) -> SrvRecord { + SrvRecord { + priority, + weight, + port, + target: target.to_owned(), + ttl, + } + } + + fn server(host: &str, port: u16) -> Server { + Server { + host: host.to_owned(), + port, + } + } + + #[test] + fn sort_orders_by_priority() { + let mut rng = StdRng::seed_from_u64(1); + let sorted = sort_srv_records( + vec![ + srv(10, 0, "c", 1, 0), + srv(0, 0, "a", 1, 0), + srv(5, 0, "b", 1, 0), + ], + &mut rng, + ); + let hosts: Vec<_> = sorted.iter().map(|s| s.host.as_str()).collect(); + assert_eq!(hosts, ["a", "b", "c"]); + } + + /// Within a priority, a record is picked first in proportion to its + /// weight, and zero-weight records come last. + #[test] + fn sort_weights_within_a_priority() { + let mut rng = StdRng::seed_from_u64(2); + let mut heavy_first = 0; + for _ in 0..1000 { + let sorted = sort_srv_records( + vec![ + srv(0, 0, "zero", 1, 0), + srv(0, 1, "light", 1, 0), + srv(0, 9, "heavy", 1, 0), + ], + &mut rng, + ); + assert_eq!(sorted.len(), 3); + assert_eq!(sorted[2].host, "zero"); + if sorted[0].host == "heavy" { + heavy_first += 1; + } + } + // Expect ~900. + assert!((850..950).contains(&heavy_first), "{heavy_first}"); + } + + #[test] + fn sort_all_zero_weights_keeps_everything() { + let mut rng = StdRng::seed_from_u64(3); + let mut sorted = + sort_srv_records(vec![srv(0, 0, "a", 1, 0), srv(0, 0, "b", 1, 0)], &mut rng); + sorted.sort_by(|a, b| a.host.cmp(&b.host)); + assert_eq!(sorted, [server("a", 1), server("b", 1)]); + } + + fn resolver() -> (Arc, ServerResolver) { + let dns = Arc::new(InMemoryResolver::new()); + let resolver = ServerResolver::new(dns.clone()); + (dns, resolver) + } + + #[tokio::test] + async fn literal_or_port_bypasses_srv() { + let (dns, resolver) = resolver(); + dns.add_srv( + "_matrix-fed._tcp.example.com", + vec![srv(0, 0, "other", 1, 60)], + ); + + assert_eq!( + resolver + .resolve_server("example.com", Some(1234), 0) + .await + .unwrap(), + [server("example.com", 1234)] + ); + assert_eq!( + resolver.resolve_server("1.2.3.4", None, 0).await.unwrap(), + [server("1.2.3.4", 8448)] + ); + assert_eq!( + resolver.resolve_server("[::1]", None, 0).await.unwrap(), + [server("[::1]", 8448)] + ); + } + + #[tokio::test] + async fn srv_lookup_order() { + let (dns, resolver) = resolver(); + + // Nothing: host and the default port. + assert_eq!( + resolver + .resolve_server("example.com", None, 0) + .await + .unwrap(), + [server("example.com", 8448)] + ); + + // The deprecated record is used when it is the only one. + dns.add_srv("_matrix._tcp.example.com", vec![srv(0, 0, "old", 1111, 60)]); + assert_eq!( + resolver + .resolve_server("example.com", None, 0) + .await + .unwrap(), + [server("old", 1111)] + ); + + // `_matrix-fed._tcp` wins over it. + dns.add_srv( + "_matrix-fed._tcp.example.com", + vec![srv(0, 0, "new", 2222, 60)], + ); + assert_eq!( + resolver + .resolve_server("example.com", None, 0) + .await + .unwrap(), + [server("new", 2222)] + ); + } + + #[tokio::test] + async fn lone_root_target_is_unavailable() { + let (dns, resolver) = resolver(); + dns.add_srv("_matrix-fed._tcp.example.com", vec![srv(0, 0, ".", 0, 60)]); + + let err = resolver + .resolve_server("example.com", None, 0) + .await + .unwrap_err(); + assert!( + matches!(err, TransportError::ServiceUnavailable(_)), + "{err}" + ); + } + + #[tokio::test] + async fn cache_and_stale_fallback() { + let (dns, resolver) = resolver(); + let name = "_matrix-fed._tcp.example.com"; + dns.add_srv(name, vec![srv(0, 0, "first", 1, 60)]); + + assert_eq!( + resolver + .resolve_server("example.com", None, 1000) + .await + .unwrap(), + [server("first", 1)] + ); + + // Still cached: a changed answer is not seen. + dns.add_srv(name, vec![srv(0, 0, "second", 2, 60)]); + assert_eq!( + resolver + .resolve_server("example.com", None, 1059) + .await + .unwrap(), + [server("first", 1)] + ); + + // Expired: refetched. + assert_eq!( + resolver + .resolve_server("example.com", None, 1060) + .await + .unwrap(), + [server("second", 2)] + ); + + // Expired and the lookup fails: the stale answer is used. + dns.fail_srv(name, "SERVFAIL"); + assert_eq!( + resolver + .resolve_server("example.com", None, 5000) + .await + .unwrap(), + [server("second", 2)] + ); + + // A failure with nothing cached is a DNS error. + dns.fail_srv("_matrix-fed._tcp.other.com", "SERVFAIL"); + let err = resolver + .resolve_server("other.com", None, 0) + .await + .unwrap_err(); + assert!(err.is_dns(), "{err}"); + } +} diff --git a/rust/src/federation/agent/tls.rs b/rust/src/federation/agent/tls.rs new file mode 100644 index 0000000000..81b54efd0b --- /dev/null +++ b/rust/src/federation/agent/tls.rs @@ -0,0 +1,258 @@ +/* + * 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: + * . + * + */ + +//! The TLS policy for outbound federation: Python's +//! `FederationPolicyForHTTPS`. +//! +//! Certificates are checked against the platform roots, or the +//! `federation_custom_ca_list` instead when set. Verification can be turned +//! off entirely, or for hosts matching +//! `federation_certificate_verification_whitelist`. With verification off we +//! still send SNI. +//! +//! rustls has no TLS 1.0 or 1.1, so a `federation_client_minimum_tls_version` +//! below 1.2 means 1.2. + +use std::sync::Arc; + +use anyhow::{bail, Context}; +use log::warn; +use tokio_rustls::rustls::{ + self, + client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}, + crypto::{verify_tls12_signature, verify_tls13_signature, CryptoProvider}, + pki_types::{CertificateDer, ServerName, UnixTime}, + ClientConfig, DigitallySignedStruct, RootCertStore, SignatureScheme, +}; + +use crate::config::{types::PythonRegex, TlsConfig}; + +pub struct TlsPolicy { + verifying: Arc, + not_verifying: Arc, + verify: bool, + verification_whitelist: Vec, +} + +impl TlsPolicy { + pub fn new(config: &TlsConfig) -> anyhow::Result { + let provider = Arc::new(rustls::crypto::ring::default_provider()); + + let versions: &[&rustls::SupportedProtocolVersion] = + match config.federation_client_minimum_tls_version.as_str() { + "1" | "1.1" | "1.2" => &[&rustls::version::TLS12, &rustls::version::TLS13], + "1.3" => &[&rustls::version::TLS13], + other => bail!("unknown federation_client_minimum_tls_version {other:?}"), + }; + + let roots = match &config.federation_custom_ca_pems { + Some(pems) => custom_roots(pems)?, + None => platform_roots(), + }; + + let verifying = ClientConfig::builder_with_provider(provider.clone()) + .with_protocol_versions(versions)? + .with_root_certificates(roots) + .with_no_client_auth(); + + let not_verifying = ClientConfig::builder_with_provider(provider.clone()) + .with_protocol_versions(versions)? + .dangerous() + .with_custom_certificate_verifier(Arc::new(NoVerification(provider))) + .with_no_client_auth(); + + Ok(TlsPolicy { + verifying: Arc::new(verifying), + not_verifying: Arc::new(not_verifying), + verify: config.federation_verify_certificates, + verification_whitelist: config.federation_certificate_verification_whitelist.clone(), + }) + } + + /// The client config to connect to `host` with. `host` is the name we + /// verify the certificate against: the server name, or the delegated + /// name after a `.well-known` lookup, never the SRV target. + pub fn config_for(&self, host: &str) -> Arc { + let verify = self.verify + && !self + .verification_whitelist + .iter() + .any(|regex| regex.is_match_at_start(host)); + + if verify { + self.verifying.clone() + } else { + self.not_verifying.clone() + } + } +} + +/// The name to verify (and send as SNI) for `host`. IP literals verify +/// against the certificate's IP addresses and send no SNI. +pub fn server_name(host: &str) -> anyhow::Result> { + let bare = host + .strip_prefix('[') + .and_then(|h| h.strip_suffix(']')) + .unwrap_or(host); + ServerName::try_from(bare.to_owned()) + .with_context(|| format!("invalid TLS server name {host:?}")) +} + +fn custom_roots(pems: &[String]) -> anyhow::Result { + let mut roots = RootCertStore::empty(); + for pem in pems { + for cert in rustls_pemfile::certs(&mut pem.as_bytes()) { + let cert = cert.context("parsing federation_custom_ca_list")?; + roots + .add(cert) + .context("adding a federation_custom_ca_list certificate")?; + } + } + if roots.is_empty() { + bail!("federation_custom_ca_list contains no certificates"); + } + Ok(roots) +} + +fn platform_roots() -> RootCertStore { + let mut roots = RootCertStore::empty(); + let loaded = rustls_native_certs::load_native_certs(); + for err in &loaded.errors { + warn!("Error loading platform TLS roots: {err}"); + } + let (added, ignored) = roots.add_parsable_certificates(loaded.certs); + if ignored > 0 { + warn!("Ignored {ignored} unparsable platform TLS roots ({added} loaded)"); + } + roots +} + +/// Accepts any certificate. The handshake signatures are still checked, +/// since they are part of TLS itself rather than of trusting the peer. +#[derive(Debug)] +struct NoVerification(Arc); + +impl ServerCertVerifier for NoVerification { + fn verify_server_cert( + &self, + _end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _server_name: &ServerName<'_>, + _ocsp_response: &[u8], + _now: UnixTime, + ) -> Result { + Ok(ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + verify_tls12_signature( + message, + cert, + dss, + &self.0.signature_verification_algorithms, + ) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + verify_tls13_signature( + message, + cert, + dss, + &self.0.signature_verification_algorithms, + ) + } + + fn supported_verify_schemes(&self) -> Vec { + self.0.signature_verification_algorithms.supported_schemes() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn config(verify: bool, whitelist: &[&str]) -> TlsConfig { + TlsConfig { + federation_verify_certificates: verify, + federation_client_minimum_tls_version: "1".to_owned(), + federation_certificate_verification_whitelist: whitelist + .iter() + .map(|p| PythonRegex::from_python_pattern(p, true).unwrap()) + .collect(), + federation_custom_ca_pems: None, + } + } + + #[test] + fn whitelist_turns_off_verification_per_host() { + let policy = TlsPolicy::new(&config(true, &[r"\A(.{0,}\.insecure\.example)\Z"])).unwrap(); + assert!(Arc::ptr_eq( + &policy.config_for("matrix.org"), + &policy.verifying + )); + assert!(Arc::ptr_eq( + &policy.config_for("a.insecure.example"), + &policy.not_verifying + )); + + let policy = TlsPolicy::new(&config(false, &[])).unwrap(); + assert!(Arc::ptr_eq( + &policy.config_for("matrix.org"), + &policy.not_verifying + )); + } + + #[test] + fn minimum_version() { + let mut c = config(true, &[]); + c.federation_client_minimum_tls_version = "1.3".to_owned(); + assert!(TlsPolicy::new(&c).is_ok()); + c.federation_client_minimum_tls_version = "2".to_owned(); + assert!(TlsPolicy::new(&c).is_err()); + } + + #[test] + fn server_names() { + assert!(matches!( + server_name("example.com").unwrap(), + ServerName::DnsName(_) + )); + assert!(matches!( + server_name("1.2.3.4").unwrap(), + ServerName::IpAddress(_) + )); + assert!(matches!( + server_name("[::1]").unwrap(), + ServerName::IpAddress(_) + )); + } + + #[test] + fn custom_ca_list_must_hold_certificates() { + let mut c = config(true, &[]); + c.federation_custom_ca_pems = Some(vec!["not a pem".to_owned()]); + assert!(TlsPolicy::new(&c).is_err()); + } +} diff --git a/rust/src/federation/agent/well_known.rs b/rust/src/federation/agent/well_known.rs new file mode 100644 index 0000000000..a1fcfcf113 --- /dev/null +++ b/rust/src/federation/agent/well_known.rs @@ -0,0 +1,805 @@ +/* + * 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: + * . + * + */ + +//! `.well-known/matrix/server` delegation, as Python's `WellKnownResolver`. +//! +//! The caching rules are Python's, constant for constant. Two things are +//! new: each fetch attempt has a timeout (Python's has none), and +//! concurrent lookups of one server name share a single fetch. + +use std::{ + collections::HashMap, + sync::{Arc, Mutex}, + time::{Duration, UNIX_EPOCH}, +}; + +use anyhow::anyhow; +use async_trait::async_trait; +use bytes::Bytes; +use futures::{ + future::{BoxFuture, Shared}, + FutureExt, +}; +use http::{header, HeaderMap, StatusCode, Uri}; +use log::{info, warn}; +use rand::RngExt; + +use super::Now; +use crate::clock::Sleeper; + +/// How long to cache a `.well-known` result by default. +const DEFAULT_CACHE_PERIOD: f64 = 24.0 * 3600.0; +/// Jitter on the default and failure cache periods. +const CACHE_PERIOD_JITTER: f64 = 0.1; +/// How long to cache a failure to fetch. +const INVALID_CACHE_PERIOD: f64 = 3600.0; +/// How long to cache a failure to fetch if there was recently a valid result. +const DOWN_CACHE_PERIOD: f64 = 2.0 * 60.0; +/// How long to remember there was a valid result, after it expires. +const REMEMBER_DOMAIN_HAD_VALID: f64 = 2.0 * 3600.0; +/// Bounds on a cache period taken from the response headers. +const MAX_CACHE_PERIOD: f64 = 48.0 * 3600.0; +const MIN_CACHE_PERIOD: f64 = 5.0 * 60.0; +/// The largest `.well-known` body we accept. +pub const MAX_SIZE: usize = 50 * 1024; +/// Refetch this fraction of the TTL before a cached result expires. +const GRACE_PERIOD_FACTOR: f64 = 0.2; +/// Attempts to fetch for a server that recently had a valid result. +const RETRY_ATTEMPTS: u32 = 3; +/// The pause between those attempts. +const RETRY_DELAY: Duration = Duration::from_millis(500); +/// The most redirects we follow, as Twisted's `RedirectAgent`. +const MAX_REDIRECTS: usize = 20; + +/// One HTTP response, with the body read in full. +#[derive(Clone, Debug)] +pub struct FetchResponse { + pub status: StatusCode, + pub headers: HeaderMap, + pub body: Bytes, +} + +#[derive(Debug)] +pub enum FetchError { + /// The body was longer than [`MAX_SIZE`]. + TooLarge, + Other(anyhow::Error), +} + +/// Makes one `GET` request, without following redirects. The agent's +/// implementation goes through the agent's own connection pool. +#[async_trait] +pub trait WellKnownFetcher: Send + Sync + 'static { + async fn fetch(&self, uri: &Uri) -> Result; +} + +#[derive(Clone, Debug)] +struct CachedResult { + /// The delegated server name, if any. + delegated: Option, + expiry: f64, + ttl: f64, +} + +#[derive(Default)] +struct CacheState { + results: HashMap, + /// Whether each server recently had a valid `.well-known`, and until + /// when we remember that. + had_valid: HashMap, +} + +impl CacheState { + fn result(&self, server_name: &str, now: f64) -> Option<&CachedResult> { + self.results.get(server_name).filter(|r| r.expiry > now) + } + + fn had_valid(&self, server_name: &str, now: f64) -> bool { + self.had_valid + .get(server_name) + .is_some_and(|&(valid, expiry)| valid && expiry > now) + } + + fn insert_result(&mut self, server_name: &str, result: CachedResult, now: f64) { + // Drop expired entries, as Python's `TTLCache` does on every access, + // so the maps only hold live entries. + self.results.retain(|_, r| r.expiry > now); + self.results.insert(server_name.to_owned(), result); + } + + fn insert_had_valid(&mut self, server_name: &str, valid: bool, expiry: f64, now: f64) { + self.had_valid.retain(|_, &mut (_, e)| e > now); + self.had_valid + .insert(server_name.to_owned(), (valid, expiry)); + } +} + +/// Why a fetch failed. A temporary failure (no response, or a 5xx) keeps +/// serving a cached result. A permanent one (a bad response) does not. +#[derive(Debug)] +struct FetchFailure { + temporary: bool, +} + +pub struct WellKnownResolver { + fetcher: Arc, + sleeper: Arc, + now: Now, + attempt_timeout: Duration, + state: Mutex, + in_flight: Mutex>>>>, +} + +impl WellKnownResolver { + pub fn new( + fetcher: Arc, + sleeper: Arc, + now: Now, + attempt_timeout: Duration, + ) -> Self { + WellKnownResolver { + fetcher, + sleeper, + now, + attempt_timeout, + state: Mutex::new(CacheState::default()), + in_flight: Mutex::new(HashMap::new()), + } + } + + fn now_secs(&self) -> f64 { + (self.now)() as f64 / 1000.0 + } + + /// The server `server_name` delegates federation to, if any. + pub async fn get_well_known(self: &Arc, server_name: &str) -> Option { + let now = self.now_secs(); + let prev = { + let state = self.state.lock().expect("poisoned"); + match state.result(server_name, now) { + Some(cached) if now < cached.expiry - GRACE_PERIOD_FACTOR * cached.ttl => { + return cached.delegated.clone(); + } + Some(cached) => cached.delegated.clone(), + None => None, + } + }; + + let lookup = { + let mut in_flight = self.in_flight.lock().expect("poisoned"); + in_flight + .entry(server_name.to_owned()) + .or_insert_with(|| { + let this = self.clone(); + let server_name = server_name.to_owned(); + async move { + let result = this.lookup(&server_name, prev).await; + this.in_flight + .lock() + .expect("poisoned") + .remove(&server_name); + result + } + .boxed() + .shared() + }) + .clone() + }; + + lookup.await + } + + /// Fetch, and cache what we learn. `prev` is the cached delegation we + /// are refreshing, if any. + async fn lookup(&self, server_name: &str, prev: Option) -> Option { + let (result, cache_period) = match self.fetch_well_known(server_name).await { + Ok(fetched) => fetched, + Err(failure) => { + if failure.temporary && prev.is_some() { + // Keep serving the cached result; hopefully the server + // is back next time we ask. + return prev; + } + + let had_valid = self + .state + .lock() + .expect("poisoned") + .had_valid(server_name, self.now_secs()); + let period = if had_valid { + DOWN_CACHE_PERIOD + } else { + INVALID_CACHE_PERIOD + }; + (None, period * jitter()) + } + }; + + if cache_period > 0.0 { + let now = self.now_secs(); + self.state.lock().expect("poisoned").insert_result( + server_name, + CachedResult { + delegated: result.clone(), + expiry: now + cache_period, + ttl: cache_period, + }, + now, + ); + } + + result + } + + /// Fetch and parse, without consulting the result cache. Returns the + /// delegated server (if any) and how long to cache that for. + async fn fetch_well_known( + &self, + server_name: &str, + ) -> Result<(Option, f64), FetchFailure> { + let had_valid = self + .state + .lock() + .expect("poisoned") + .had_valid(server_name, self.now_secs()); + + let response = self.make_request(server_name, had_valid).await?; + + let delegated = parse_response(&response).map_err(|err| { + info!("Error parsing well-known for {server_name}: {err:#}"); + FetchFailure { temporary: false } + })?; + info!("Response from .well-known for {server_name}: {delegated:?}"); + + let now = self.now_secs(); + let cache_period = match cache_period_from_headers(&response.headers, now) { + None => DEFAULT_CACHE_PERIOD * jitter(), + Some(period) => period.clamp(MIN_CACHE_PERIOD, MAX_CACHE_PERIOD), + }; + + self.state.lock().expect("poisoned").insert_had_valid( + server_name, + delegated.is_some(), + now + cache_period + REMEMBER_DOMAIN_HAD_VALID, + now, + ); + + Ok((delegated, cache_period)) + } + + /// Fetch the file, retrying on failure (no response or a 5xx) only if + /// `retry`. A non-5xx response is returned whatever its status. + async fn make_request( + &self, + server_name: &str, + retry: bool, + ) -> Result { + let uri: Uri = format!("https://{server_name}/.well-known/matrix/server") + .parse() + .map_err(|err| { + info!("Invalid well-known URI for {server_name}: {err}"); + FetchFailure { temporary: true } + })?; + + let mut attempt = 0; + loop { + attempt += 1; + info!("Fetching {uri}"); + + let error = match tokio::time::timeout( + self.attempt_timeout, + self.fetch_following_redirects(&uri), + ) + .await + { + Ok(Ok(response)) if !response.status.is_server_error() => return Ok(response), + Ok(Ok(response)) => anyhow!("Non-200 response {}", response.status.as_u16()), + Ok(Err(FetchError::TooLarge)) => { + // Not worth fetching again, but may be temporary. + warn!("Requested .well-known file for {server_name} is too large > {MAX_SIZE} bytes"); + return Err(FetchFailure { temporary: true }); + } + Ok(Err(FetchError::Other(err))) => err, + Err(_) => anyhow!("timed out after {:?}", self.attempt_timeout), + }; + + if !retry || attempt >= RETRY_ATTEMPTS { + info!("Error fetching {uri}: {error:#}"); + return Err(FetchFailure { temporary: true }); + } + info!("Error fetching {uri}: {error:#}. Retrying"); + + // Pause briefly in the hope that it comes back up. + self.sleeper.sleep(RETRY_DELAY).await; + } + } + + /// One `GET`, following redirects as Twisted's `RedirectAgent` does. + async fn fetch_following_redirects(&self, uri: &Uri) -> Result { + let mut uri = uri.clone(); + for _ in 0..=MAX_REDIRECTS { + let response = self.fetcher.fetch(&uri).await?; + + if !matches!(response.status.as_u16(), 301 | 302 | 303 | 307 | 308) { + return Ok(response); + } + + let location = response + .headers + .get(header::LOCATION) + .and_then(|l| l.to_str().ok()) + .ok_or_else(|| FetchError::Other(anyhow!("redirect with no Location")))?; + let next = url::Url::parse(&uri.to_string()) + .and_then(|base| base.join(location)) + .map_err(|err| FetchError::Other(anyhow!("bad redirect {location:?}: {err}")))?; + uri = next + .as_str() + .parse() + .map_err(|err| FetchError::Other(anyhow!("bad redirect {location:?}: {err}")))?; + } + + Err(FetchError::Other(anyhow!("too many redirects"))) + } +} + +fn jitter() -> f64 { + rand::rng().random_range((1.0 - CACHE_PERIOD_JITTER)..=(1.0 + CACHE_PERIOD_JITTER)) +} + +/// The delegated server name in a `.well-known` response. +/// +/// Python also rejects a non-200 status and a missing, non-string or +/// non-ASCII `m.server`. We also reject a value that is not a valid URI +/// authority, which Python would only discover when the request built from +/// it failed. An empty `m.server` means no delegation. +fn parse_response(response: &FetchResponse) -> anyhow::Result> { + if response.status != StatusCode::OK { + anyhow::bail!("Non-200 response {}", response.status.as_u16()); + } + + let body: serde_json::Value = serde_json::from_slice(&response.body)?; + let server = body + .get("m.server") + .and_then(|s| s.as_str()) + .ok_or_else(|| anyhow!("no string m.server"))?; + + if server.is_empty() { + return Ok(None); + } + if !server.is_ascii() { + anyhow::bail!("non-ASCII m.server {server:?}"); + } + server + .parse::() + .map_err(|err| anyhow!("invalid m.server {server:?}: {err}"))?; + + Ok(Some(server.to_owned())) +} + +/// The cache period the response headers ask for, in seconds: `no-store` +/// is zero, then `max-age`, then `Expires`. As Python's +/// `_cache_period_from_headers`. +fn cache_period_from_headers(headers: &HeaderMap, now_secs: f64) -> Option { + let cache_control = parse_cache_control(headers); + + if cache_control.contains_key("no-store") { + return Some(0.0); + } + + if let Some(Some(max_age)) = cache_control.get("max-age") { + if let Ok(max_age) = max_age.parse::() { + return Some(max_age as f64); + } + } + + if let Some(expires) = headers.get_all(header::EXPIRES).iter().next_back() { + // An invalid date (notably "0") means already expired (RFC 7234). + return Some( + expires + .to_str() + .ok() + .and_then(|e| httpdate::parse_http_date(e).ok()) + .map(|expires| { + let expires = expires + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs_f64(); + expires - now_secs + }) + .unwrap_or(0.0), + ); + } + + None +} + +fn parse_cache_control(headers: &HeaderMap) -> HashMap> { + let mut directives = HashMap::new(); + for value in headers.get_all(header::CACHE_CONTROL) { + let Ok(value) = value.to_str() else { continue }; + for directive in value.split(',') { + let mut parts = directive.splitn(2, '=').map(str::trim); + let key = parts.next().unwrap_or("").to_lowercase(); + let value = parts.next().map(str::to_owned); + directives.insert(key, value); + } + } + directives +} + +/// For tests, which need a fixed `SystemTime` for `Expires` headers. +#[cfg(test)] +fn http_date(secs: u64) -> String { + httpdate::fmt_http_date(UNIX_EPOCH + Duration::from_secs(secs)) +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; + + use http::HeaderValue; + + use super::*; + use crate::clock::TokioSleeper; + + type Handler = dyn Fn(&Uri) -> Result + Send + Sync; + + struct FakeFetcher { + handler: Mutex>, + calls: AtomicUsize, + } + + impl FakeFetcher { + fn new( + handler: impl Fn(&Uri) -> Result + Send + Sync + 'static, + ) -> Arc { + Arc::new(FakeFetcher { + handler: Mutex::new(Box::new(handler)), + calls: AtomicUsize::new(0), + }) + } + + fn set( + &self, + handler: impl Fn(&Uri) -> Result + Send + Sync + 'static, + ) { + *self.handler.lock().unwrap() = Box::new(handler); + } + + fn calls(&self) -> usize { + self.calls.swap(0, Ordering::SeqCst) + } + } + + #[async_trait] + impl WellKnownFetcher for FakeFetcher { + async fn fetch(&self, uri: &Uri) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + // Yield, so concurrent lookups overlap. + tokio::task::yield_now().await; + (self.handler.lock().unwrap())(uri) + } + } + + fn response( + status: u16, + body: &str, + headers: &[(&str, &str)], + ) -> Result { + let mut map = HeaderMap::new(); + for (name, value) in headers { + map.append( + http::HeaderName::from_bytes(name.as_bytes()).unwrap(), + HeaderValue::from_str(value).unwrap(), + ); + } + Ok(FetchResponse { + status: StatusCode::from_u16(status).unwrap(), + headers: map, + body: Bytes::from(body.to_owned()), + }) + } + + /// A response delegating to `to`, cached for exactly two hours. + fn delegate(to: &'static str) -> impl Fn(&Uri) -> Result { + move |_| { + response( + 200, + &format!(r#"{{"m.server": "{to}"}}"#), + &[("cache-control", "max-age=7200")], + ) + } + } + + /// The same without cache headers, so cached for the jittered default. + fn delegate_default(to: &'static str) -> impl Fn(&Uri) -> Result { + move |_| response(200, &format!(r#"{{"m.server": "{to}"}}"#), &[]) + } + + fn make_resolver(fetcher: Arc) -> (Arc, Arc) { + let clock = Arc::new(AtomicU64::new(1_000_000_000)); + let now_clock = clock.clone(); + let resolver = Arc::new(WellKnownResolver::new( + fetcher, + Arc::new(TokioSleeper), + Arc::new(move || now_clock.load(Ordering::SeqCst)), + Duration::from_secs(60), + )); + (resolver, clock) + } + + fn advance(clock: &AtomicU64, secs: f64) { + clock.fetch_add((secs * 1000.0) as u64, Ordering::SeqCst); + } + + #[tokio::test(start_paused = true)] + async fn valid_result_is_cached_with_jitter() { + let fetcher = FakeFetcher::new(delegate_default("target:1234")); + let (resolver, clock) = make_resolver(fetcher.clone()); + + let uri_seen = Arc::new(Mutex::new(None)); + let seen = uri_seen.clone(); + fetcher.set(move |uri| { + *seen.lock().unwrap() = Some(uri.to_string()); + delegate_default("target:1234")(uri) + }); + + assert_eq!( + resolver.get_well_known("example.com").await.as_deref(), + Some("target:1234") + ); + assert_eq!(fetcher.calls(), 1); + assert_eq!( + uri_seen.lock().unwrap().as_deref(), + Some("https://example.com/.well-known/matrix/server") + ); + + // The default period is 24h ±10%, refetched in its last 20%: so no + // refetch at 24h × 0.9 × 0.8, and one by 24h × 1.1. + advance(&clock, 24.0 * 3600.0 * 0.9 * 0.8 - 1.0); + resolver.get_well_known("example.com").await; + assert_eq!(fetcher.calls(), 0); + + advance(&clock, 24.0 * 3600.0 * 0.4); + resolver.get_well_known("example.com").await; + assert_eq!(fetcher.calls(), 1); + } + + #[tokio::test(start_paused = true)] + async fn missing_well_known_is_cached_for_an_hour() { + let fetcher = FakeFetcher::new(|_| response(404, "", &[])); + let (resolver, clock) = make_resolver(fetcher.clone()); + + assert_eq!(resolver.get_well_known("example.com").await, None); + assert_eq!(fetcher.calls(), 1); + + // 1h ±10%, refetched in the last 20%. + advance(&clock, 3600.0 * 0.9 * 0.8 - 1.0); + assert_eq!(resolver.get_well_known("example.com").await, None); + assert_eq!(fetcher.calls(), 0); + + advance(&clock, 3600.0 * 0.4); + assert_eq!(resolver.get_well_known("example.com").await, None); + assert_eq!(fetcher.calls(), 1); + } + + #[tokio::test(start_paused = true)] + async fn headers_set_the_cache_period() { + let fetcher = FakeFetcher::new(|_| { + response( + 200, + r#"{"m.server": "t"}"#, + &[("cache-control", "max-age=7200")], + ) + }); + let (resolver, clock) = make_resolver(fetcher.clone()); + + resolver.get_well_known("example.com").await; + assert_eq!(fetcher.calls(), 1); + // Refetch starts at 80% of 2h. + advance(&clock, 7200.0 * 0.8 - 1.0); + resolver.get_well_known("example.com").await; + assert_eq!(fetcher.calls(), 0); + advance(&clock, 2.0); + resolver.get_well_known("example.com").await; + assert_eq!(fetcher.calls(), 1); + } + + #[test] + fn cache_period_parsing() { + let headers = |pairs: &[(&str, &str)]| response(200, "", pairs).unwrap().headers; + + assert_eq!(cache_period_from_headers(&headers(&[]), 0.0), None); + assert_eq!( + cache_period_from_headers(&headers(&[("cache-control", "max-age=60, public")]), 0.0), + Some(60.0) + ); + assert_eq!( + cache_period_from_headers(&headers(&[("Cache-Control", "No-Store, max-age=60")]), 0.0), + Some(0.0) + ); + // A bad max-age falls through to Expires. + assert_eq!( + cache_period_from_headers( + &headers(&[ + ("cache-control", "max-age=abc"), + ("expires", &http_date(1000)) + ]), + 400.0 + ), + Some(600.0) + ); + assert_eq!( + cache_period_from_headers(&headers(&[("expires", "0")]), 400.0), + Some(0.0) + ); + } + + /// A short (or zero) period from the headers is raised to the minimum. + #[tokio::test(start_paused = true)] + async fn short_periods_are_clamped() { + let fetcher = FakeFetcher::new(|_| { + response( + 200, + r#"{"m.server": "t"}"#, + &[("cache-control", "no-store")], + ) + }); + let (resolver, clock) = make_resolver(fetcher.clone()); + + resolver.get_well_known("example.com").await; + advance(&clock, 5.0 * 60.0 * 0.8 - 1.0); + resolver.get_well_known("example.com").await; + assert_eq!(fetcher.calls(), 1); + } + + /// Once a server has had a valid `.well-known`, a failure is cached for + /// only two minutes, fetching is retried, and a temporary failure keeps + /// the cached result. + #[tokio::test(start_paused = true)] + async fn failures_after_a_valid_result() { + let fetcher = FakeFetcher::new(delegate("target")); + let (resolver, clock) = make_resolver(fetcher.clone()); + + assert_eq!( + resolver.get_well_known("example.com").await.as_deref(), + Some("target") + ); + assert_eq!(fetcher.calls(), 1); + + // In the grace period, the server is down: three attempts, and the + // cached result is kept. + fetcher.set(|_| response(503, "", &[])); + advance(&clock, 7200.0 * 0.9); + assert_eq!( + resolver.get_well_known("example.com").await.as_deref(), + Some("target") + ); + assert_eq!(fetcher.calls(), 3); + + // After the result expires it is gone. We remember for another two + // hours that it was valid, so we retry, and cache the failure for + // 2m (±10%) rather than 1h. + advance(&clock, 7200.0 * 0.1 + 1.0); + assert_eq!(resolver.get_well_known("example.com").await, None); + assert_eq!(fetcher.calls(), 3); + + fetcher.set(delegate("target")); + advance(&clock, 2.0 * 60.0 * 0.9 * 0.8 - 1.0); + assert_eq!(resolver.get_well_known("example.com").await, None); + assert_eq!(fetcher.calls(), 0); + advance(&clock, 2.0 * 60.0 * 0.4); + assert_eq!( + resolver.get_well_known("example.com").await.as_deref(), + Some("target") + ); + assert_eq!(fetcher.calls(), 1); + } + + /// Without a recent valid result, a failure is tried once. + #[tokio::test(start_paused = true)] + async fn failures_without_a_valid_result_are_not_retried() { + let fetcher = FakeFetcher::new(|_| response(503, "", &[])); + let (resolver, _) = make_resolver(fetcher.clone()); + assert_eq!(resolver.get_well_known("example.com").await, None); + assert_eq!(fetcher.calls(), 1); + } + + /// A bad response is a permanent failure: the cached result is dropped. + #[tokio::test(start_paused = true)] + async fn invalid_response_replaces_cached_result() { + let fetcher = FakeFetcher::new(delegate("target")); + let (resolver, clock) = make_resolver(fetcher.clone()); + resolver.get_well_known("example.com").await; + + fetcher.set(|_| response(200, "not json", &[])); + advance(&clock, 7200.0 * 0.9); + assert_eq!(resolver.get_well_known("example.com").await, None); + assert_eq!(fetcher.calls(), 2); + } + + #[tokio::test(start_paused = true)] + async fn bad_bodies() { + for body in [ + r#"{}"#, + r#"{"m.server": 1}"#, + r#"{"m.server": "bad host/name"}"#, + r#"{"m.server": "é.example"}"#, + r#"{"m.server": ""}"#, + ] { + let fetcher = FakeFetcher::new(move |_| response(200, body, &[])); + let (resolver, _) = make_resolver(fetcher); + assert_eq!(resolver.get_well_known("example.com").await, None, "{body}"); + } + } + + #[tokio::test(start_paused = true)] + async fn too_large_is_not_retried() { + let fetcher = FakeFetcher::new(delegate("target")); + let (resolver, clock) = make_resolver(fetcher.clone()); + resolver.get_well_known("example.com").await; + fetcher.calls(); + + // Even though the server had a valid result. + fetcher.set(|_| Err(FetchError::TooLarge)); + advance(&clock, 7200.0 * 0.9); + assert_eq!( + resolver.get_well_known("example.com").await.as_deref(), + Some("target") + ); + assert_eq!(fetcher.calls(), 1); + } + + #[tokio::test(start_paused = true)] + async fn redirects_are_followed() { + let fetcher = FakeFetcher::new(|uri| match uri.path() { + "/.well-known/matrix/server" => response(302, "", &[("location", "/elsewhere")]), + "/elsewhere" => response(301, "", &[("location", "https://other.example/final")]), + "/final" => { + assert_eq!(uri.host(), Some("other.example")); + response(200, r#"{"m.server": "target"}"#, &[]) + } + other => panic!("unexpected {other}"), + }); + let (resolver, _) = make_resolver(fetcher.clone()); + + assert_eq!( + resolver.get_well_known("example.com").await.as_deref(), + Some("target") + ); + assert_eq!(fetcher.calls(), 3); + + // A redirect loop gives up. + let fetcher = FakeFetcher::new(|_| response(302, "", &[("location", "/loop")])); + let (resolver, _) = make_resolver(fetcher.clone()); + assert_eq!(resolver.get_well_known("example.com").await, None); + assert_eq!(fetcher.calls(), MAX_REDIRECTS + 1); + } + + #[tokio::test(start_paused = true)] + async fn concurrent_lookups_share_a_fetch() { + let fetcher = FakeFetcher::new(delegate("target")); + let (resolver, _) = make_resolver(fetcher.clone()); + + let (a, b) = tokio::join!( + resolver.get_well_known("example.com"), + resolver.get_well_known("example.com") + ); + assert_eq!(a.as_deref(), Some("target")); + assert_eq!(b.as_deref(), Some("target")); + assert_eq!(fetcher.calls(), 1); + assert!(resolver.in_flight.lock().unwrap().is_empty()); + } +} diff --git a/rust/src/federation/mod.rs b/rust/src/federation/mod.rs index 2816829d19..56be65e1cc 100644 --- a/rust/src/federation/mod.rs +++ b/rust/src/federation/mod.rs @@ -17,12 +17,15 @@ //! //! 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. //! - [`signing`]: the `X-Matrix` request signature. //! - [`transport`]: the seam between the request layer and the network, with //! a Python-backed implementation for tests. use pyo3::prelude::*; +pub mod agent; pub mod signing; pub mod transport;