dandri/synapse
1
0 Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-11-04 06:57:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Port federation server name resolution and policies to Rust

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb
This commit is contained in:
Erik JohnstonandClaude Opus 5.5 committed 2026-10-15 14:54:15 +00:00
1 parent db4febd65a
commit 7f34eefa26
8 files changed
+2292 -5
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+272 -5
View File
@@ -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"
rust/Cargo.toml
+18
View File
@@ -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"
rust/src/federation/agent/mod.rs
+34
View File
@@ -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:
* <https://www.gnu.org/licenses/agpl-3.0.html>.
*
*/
//! 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<dyn Fn() -> u64 + Send + Sync>;
rust/src/federation/agent/policy.rs
+312
View File
@@ -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:
* <https://www.gnu.org/licenses/agpl-3.0.html>.
*
*/
//! 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<IpRangeSet>,
}
impl IpPolicy {
pub fn new(blocklist: IpRangeSet, allowlist: Option<IpRangeSet>) -> 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<Vec<IpAddr>, 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<String>,
no_proxy: Arc<NoProxy>,
}
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<Option<Self>> {
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<String>,
}
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::<IpAddr>().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());
}
}
rust/src/federation/agent/resolve.rs
+590
View File
@@ -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:
* <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 `src/server_resolver.rs` of
* <https://github.com/erikjohnston/matrix-hyper-federation-client>.
*
* [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<Vec<SrvRecord>, DnsError>;
/// The addresses of `host`. An IP literal resolves to itself.
async fn lookup_ip(&self, host: &str) -> Result<Vec<IpAddr>, 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<Self> {
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<Vec<SrvRecord>, 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<Vec<IpAddr>, 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<IpAddr> = 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<HashMap<String, Result<Vec<SrvRecord>, String>>>,
ips: Mutex<HashMap<String, Vec<IpAddr>>>,
}
impl InMemoryResolver {
pub fn new() -> Self {
Self::default()
}
pub fn add_srv(&self, name: &str, records: Vec<SrvRecord>) {
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<IpAddr>) {
self.ips
.lock()
.expect("poisoned")
.insert(host.to_owned(), ips);
}
}
#[async_trait]
impl DnsResolver for InMemoryResolver {
async fn lookup_srv(&self, name: &str) -> Result<Vec<SrvRecord>, 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<Vec<IpAddr>, 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<IpAddr> {
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<dyn DnsResolver>,
cache: Mutex<HashMap<String, Vec<CachedRecord>>>,
}
impl ServerResolver {
pub fn new(dns: Arc<dyn DnsResolver>) -> Self {
ServerResolver {
dns,
cache: Mutex::new(HashMap::new()),
}
}
pub fn dns(&self) -> &Arc<dyn DnsResolver> {
&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<u16>,
now_secs: u64,
) -> Result<Vec<Server>, 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<Vec<Server>, 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<R: Rng + ?Sized>(records: Vec<SrvRecord>, rng: &mut R) -> Vec<Server> {
let mut by_priority: BTreeMap<u16, Vec<SrvRecord>> = 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<InMemoryResolver>, 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}");
}
}
rust/src/federation/agent/tls.rs
+258
View File
@@ -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:
* <https://www.gnu.org/licenses/agpl-3.0.html>.
*
*/
//! 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<ClientConfig>,
not_verifying: Arc<ClientConfig>,
verify: bool,
verification_whitelist: Vec<PythonRegex>,
}
impl TlsPolicy {
pub fn new(config: &TlsConfig) -> anyhow::Result<Self> {
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<ClientConfig> {
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<ServerName<'static>> {
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<RootCertStore> {
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<CryptoProvider>);
impl ServerCertVerifier for NoVerification {
fn verify_server_cert(
&self,
_end_entity: &CertificateDer<'_>,
_intermediates: &[CertificateDer<'_>],
_server_name: &ServerName<'_>,
_ocsp_response: &[u8],
_now: UnixTime,
) -> Result<ServerCertVerified, rustls::Error> {
Ok(ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
message: &[u8],
cert: &CertificateDer<'_>,
dss: &DigitallySignedStruct,
) -> Result<HandshakeSignatureValid, rustls::Error> {
verify_tls12_signature(
message,
cert,
dss,
&self.0.signature_verification_algorithms,
)
}
fn verify_tls13_signature(
&self,
message: &[u8],
cert: &CertificateDer<'_>,
dss: &DigitallySignedStruct,
) -> Result<HandshakeSignatureValid, rustls::Error> {
verify_tls13_signature(
message,
cert,
dss,
&self.0.signature_verification_algorithms,
)
}
fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
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());
}
}
rust/src/federation/agent/well_known.rs
+805
View File
@@ -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:
* <https://www.gnu.org/licenses/agpl-3.0.html>.
*
*/
//! `.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<FetchResponse, FetchError>;
}
#[derive(Clone, Debug)]
struct CachedResult {
/// The delegated server name, if any.
delegated: Option<String>,
expiry: f64,
ttl: f64,
}
#[derive(Default)]
struct CacheState {
results: HashMap<String, CachedResult>,
/// Whether each server recently had a valid `.well-known`, and until
/// when we remember that.
had_valid: HashMap<String, (bool, f64)>,
}
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<dyn WellKnownFetcher>,
sleeper: Arc<dyn Sleeper>,
now: Now,
attempt_timeout: Duration,
state: Mutex<CacheState>,
in_flight: Mutex<HashMap<String, Shared<BoxFuture<'static, Option<String>>>>>,
}
impl WellKnownResolver {
pub fn new(
fetcher: Arc<dyn WellKnownFetcher>,
sleeper: Arc<dyn Sleeper>,
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<Self>, server_name: &str) -> Option<String> {
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<String>) -> Option<String> {
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<String>, 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<FetchResponse, FetchFailure> {
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<FetchResponse, FetchError> {
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<Option<String>> {
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::<http::uri::Authority>()
.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<f64> {
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::<i64>() {
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<String, Option<String>> {
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<FetchResponse, FetchError> + Send + Sync;
struct FakeFetcher {
handler: Mutex<Box<Handler>>,
calls: AtomicUsize,
}
impl FakeFetcher {
fn new(
handler: impl Fn(&Uri) -> Result<FetchResponse, FetchError> + Send + Sync + 'static,
) -> Arc<Self> {
Arc::new(FakeFetcher {
handler: Mutex::new(Box::new(handler)),
calls: AtomicUsize::new(0),
})
}
fn set(
&self,
handler: impl Fn(&Uri) -> Result<FetchResponse, FetchError> + 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<FetchResponse, FetchError> {
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<FetchResponse, FetchError> {
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<FetchResponse, FetchError> {
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<FetchResponse, FetchError> {
move |_| response(200, &format!(r#"{{"m.server": "{to}"}}"#), &[])
}
fn make_resolver(fetcher: Arc<FakeFetcher>) -> (Arc<WellKnownResolver>, Arc<AtomicU64>) {
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());
}
}
rust/src/federation/mod.rs
+3
View File
@@ -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;