From a2e294fa8d469dbddae72e96f015f15b58890ab9 Mon Sep 17 00:00:00 2001 From: timedout Date: Tue, 4 Aug 2026 00:07:23 +0100 Subject: [PATCH] feat: Implement MSC4524 --- Cargo.lock | 1 + src/admin/Cargo.toml | 79 +++++++++++++------------- src/admin/debug/commands.rs | 55 ++++++++++++++++-- src/api/router.rs | 2 + src/api/server/mod.rs | 6 +- src/api/server/ping.rs | 55 ++++++++++++++++++ src/api/server/pong.rs | 17 ++++++ src/ruminuwuity/api/federation/mod.rs | 2 + src/ruminuwuity/api/federation/ping.rs | 43 ++++++++++++++ src/ruminuwuity/api/federation/pong.rs | 30 ++++++++++ src/ruminuwuity/api/mod.rs | 3 + src/ruminuwuity/mod.rs | 1 + src/service/federation/mod.rs | 36 +++++++++++- 13 files changed, 285 insertions(+), 45 deletions(-) create mode 100644 src/api/server/ping.rs create mode 100644 src/api/server/pong.rs create mode 100644 src/ruminuwuity/api/federation/mod.rs create mode 100644 src/ruminuwuity/api/federation/ping.rs create mode 100644 src/ruminuwuity/api/federation/pong.rs create mode 100644 src/ruminuwuity/api/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 449d1cc25..7f2089671 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -950,6 +950,7 @@ dependencies = [ "log", "resolvematrix", "ruma", + "ruminuwuity", "serde-saphyr", "serde_json", "tokio", diff --git a/src/admin/Cargo.toml b/src/admin/Cargo.toml index 6495ad400..5d0d7d624 100644 --- a/src/admin/Cargo.toml +++ b/src/admin/Cargo.toml @@ -11,65 +11,65 @@ version.workspace = true [lib] path = "mod.rs" crate-type = [ - "rlib", -# "dylib", + "rlib", + # "dylib", ] [features] brotli_compression = [ - "conduwuit-api/brotli_compression", - "conduwuit-core/brotli_compression", - "conduwuit-service/brotli_compression", + "conduwuit-api/brotli_compression", + "conduwuit-core/brotli_compression", + "conduwuit-service/brotli_compression", ] gzip_compression = [ - "conduwuit-api/gzip_compression", - "conduwuit-core/gzip_compression", - "conduwuit-service/gzip_compression", + "conduwuit-api/gzip_compression", + "conduwuit-core/gzip_compression", + "conduwuit-service/gzip_compression", ] io_uring = [ - "conduwuit-api/io_uring", - "conduwuit-database/io_uring", - "conduwuit-service/io_uring", + "conduwuit-api/io_uring", + "conduwuit-database/io_uring", + "conduwuit-service/io_uring", ] jemalloc = [ - "conduwuit-api/jemalloc", - "conduwuit-core/jemalloc", - "conduwuit-database/jemalloc", - "conduwuit-service/jemalloc", + "conduwuit-api/jemalloc", + "conduwuit-core/jemalloc", + "conduwuit-database/jemalloc", + "conduwuit-service/jemalloc", ] jemalloc_conf = [ - "conduwuit-api/jemalloc_conf", - "conduwuit-core/jemalloc_conf", - "conduwuit-database/jemalloc_conf", - "conduwuit-service/jemalloc_conf", + "conduwuit-api/jemalloc_conf", + "conduwuit-core/jemalloc_conf", + "conduwuit-database/jemalloc_conf", + "conduwuit-service/jemalloc_conf", ] jemalloc_prof = [ - "conduwuit-api/jemalloc_prof", - "conduwuit-core/jemalloc_prof", - "conduwuit-database/jemalloc_prof", - "conduwuit-service/jemalloc_prof", + "conduwuit-api/jemalloc_prof", + "conduwuit-core/jemalloc_prof", + "conduwuit-database/jemalloc_prof", + "conduwuit-service/jemalloc_prof", ] jemalloc_stats = [ - "conduwuit-api/jemalloc_stats", - "conduwuit-core/jemalloc_stats", - "conduwuit-database/jemalloc_stats", - "conduwuit-service/jemalloc_stats", + "conduwuit-api/jemalloc_stats", + "conduwuit-core/jemalloc_stats", + "conduwuit-database/jemalloc_stats", + "conduwuit-service/jemalloc_stats", ] release_max_log_level = [ - "conduwuit-api/release_max_log_level", - "conduwuit-core/release_max_log_level", - "conduwuit-database/release_max_log_level", - "conduwuit-service/release_max_log_level", - "tracing/max_level_trace", - "tracing/release_max_level_info", - "log/max_level_trace", - "log/release_max_level_info", + "conduwuit-api/release_max_log_level", + "conduwuit-core/release_max_log_level", + "conduwuit-database/release_max_log_level", + "conduwuit-service/release_max_log_level", + "tracing/max_level_trace", + "tracing/release_max_level_info", + "log/max_level_trace", + "log/release_max_level_info", ] zstd_compression = [ - "conduwuit-api/zstd_compression", - "conduwuit-core/zstd_compression", - "conduwuit-database/zstd_compression", - "conduwuit-service/zstd_compression", + "conduwuit-api/zstd_compression", + "conduwuit-core/zstd_compression", + "conduwuit-database/zstd_compression", + "conduwuit-service/zstd_compression", ] [dependencies] @@ -86,6 +86,7 @@ futures.workspace = true lettre.workspace = true log.workspace = true ruma.workspace = true +ruminuwuity.workspace = true serde_json.workspace = true serde-saphyr.workspace = true tokio.workspace = true diff --git a/src/admin/debug/commands.rs b/src/admin/debug/commands.rs index 9fc69aae9..9194db6d9 100644 --- a/src/admin/debug/commands.rs +++ b/src/admin/debug/commands.rs @@ -6,7 +6,7 @@ }; use conduwuit::{ - Err, Result, at, debug_error, err, info, + Err, Result, at, debug_error, debug_info, err, info, matrix::{ Event, StateKey, pdu::{PduEvent, PduId, RawPduId}, @@ -14,6 +14,7 @@ state_res::EventTypeExt, trace, utils, utils::{ + random_string, stream::{IterStream, ReadyExt}, string::EMPTY, to_canonical_object, @@ -28,8 +29,8 @@ }; use ruma::{ CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, - OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId, UInt, - api::federation::event::get_room_state, + OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId, ServerName, UInt, + api::{error::ErrorKind, federation::event::get_room_state}, events::{AnyStateEvent, StateEventType}, serde::Raw, }; @@ -604,14 +605,60 @@ pub(super) async fn ping(&self, server: OwnedServerName) -> Result { if server == self.services.globals.server_name() { return Err!("Not allowed to send federation requests to ourselves."); } + let modern = self.modern_pingpong_ping(&server).await?; + if let Some(w) = modern { + self.write_str(&w).await + } else { + info!("{server} does not support msc4524, falling back to legacy ping"); + self.legacy_ping(&server).await + } + } + async fn modern_pingpong_ping(&self, server: &ServerName) -> Result> { + let timer = tokio::time::Instant::now(); + let question = &random_string(64); + + let response = self + .services + .sending + .send_federation_request( + server, + ruminuwuity::api::federation::ping::unstable::Request::new(question.to_owned()), + ) + .await; + + let answer = match response { + | Ok(resp) => resp.answer, + | Err(e) => + return if e.is_not_found() && e.kind() == ErrorKind::Unrecognized { + Ok(None) + } else { + Err!("Failed sending federation request to specified server:\n\n{e}") + }, + }; + let recv = self + .services + .federation + .register_ping_answer(answer.clone())?; + debug_info!("Waiting for a pong for up to 60 seconds"); + let result = tokio::time::timeout(std::time::Duration::from_mins(1), recv).await; + match result { + | Err(e) => Err!(BadServerResponse("Remote did not answer the ping in time: {e:?}")), + | Ok(Err(e)) => + Err!(BadServerResponse("Unexpectedly could not read from pong channel: {e:?}")), + | Ok(Ok(())) => + Ok(Some(format!("Ping got a pong in {:?}: {answer}", timer.elapsed()))), + } + } + + async fn legacy_ping(&self, server: &ServerName) -> Result { let timer = tokio::time::Instant::now(); match self .services .sending .send_unauthenticated_request( - &server, + server, ruma::api::federation::discovery::get_server_version::v1::Request::new(), ) .await diff --git a/src/api/router.rs b/src/api/router.rs index 1912f096a..f6ad94a27 100644 --- a/src/api/router.rs +++ b/src/api/router.rs @@ -229,6 +229,8 @@ pub fn build(router: Router, state: State) -> Router { .ruma_route(&server::well_known_server) .ruma_route(&server::get_content_route) .ruma_route(&server::get_content_thumbnail_route) + .ruma_route(&server::ping) + .ruma_route(&server::pong) .route("/_continuwuity/local_user_count", get(client::continuwuity_local_user_count)); } else { router = router diff --git a/src/api/server/mod.rs b/src/api/server/mod.rs index 5c1ff3f79..0ae163805 100644 --- a/src/api/server/mod.rs +++ b/src/api/server/mod.rs @@ -10,6 +10,8 @@ pub(super) mod make_leave; pub(super) mod media; pub(super) mod openid; +pub(super) mod ping; +pub(super) mod pong; pub(super) mod publicrooms; pub(super) mod query; pub(super) mod send; @@ -34,6 +36,8 @@ pub(super) use make_leave::*; pub(super) use media::*; pub(super) use openid::*; +pub(super) use ping::ping; +pub(super) use pong::pong; pub(super) use publicrooms::*; pub(super) use query::*; pub(super) use send::*; @@ -45,6 +49,6 @@ pub(super) use user::*; pub(super) use version::*; pub(super) use well_known::*; - mod utils; + use utils::AccessCheck; diff --git a/src/api/server/ping.rs b/src/api/server/ping.rs new file mode 100644 index 000000000..40aa84caf --- /dev/null +++ b/src/api/server/ping.rs @@ -0,0 +1,55 @@ +use std::{sync::Arc, time::Duration}; + +use axum::extract::State; +use conduwuit::{ + Result, info, + utils::{random_string, time::jitter}, + warn, +}; +use ruma::{OwnedServerName, api::error::ErrorKind}; +use ruminuwuity::api::federation::{ + ping::unstable::{Request, Response}, + pong, +}; + +use crate::Ruma; + +/// # `GET /_matrix/federation/unstable/uk.timedout.msc0000.tabletennis/ping` +/// +/// Initiates a ping +pub(crate) async fn ping( + State(services): State, + body: Ruma, +) -> Result { + // TODO: rate-limiting + // TODO: make answer stable based on question + let answer = random_string(64); + services.federation.register_ping_answer(answer.clone())?; + services + .server + .runtime() + .spawn(send_ping(Arc::new(services), body.identity, answer.clone())); + Ok(Response::new(answer)) +} + +async fn send_ping(services: Arc, target: OwnedServerName, answer: String) { + tokio::time::sleep(jitter(Duration::from_secs(1), 1.0..=10.0)).await; + info!(%answer, "Sending a federation pong to {target}"); + match services + .sending + .send_federation_request(&target, pong::unstable::Request::new(answer)) + .await + { + | Ok(_) => info!(%target, "Federation pong succeeded"), + | Err(e) => + if e.is_not_found() { + if e.kind() == ErrorKind::Unrecognized { + warn!(%target, "Remote requested a ping but does not recognize the pong."); + } else { + info!(%target, "Ping expired by the time we ponged."); + } + } else { + warn!(%target, "Pong failed"); + }, + } +} diff --git a/src/api/server/pong.rs b/src/api/server/pong.rs new file mode 100644 index 000000000..f255e69ac --- /dev/null +++ b/src/api/server/pong.rs @@ -0,0 +1,17 @@ +use axum::extract::State; +use conduwuit::Result; +use ruminuwuity::api::federation::pong; + +use crate::Ruma; + +/// # `GET /_matrix/federation/unstable/uk.timedout.msc0000.tabletennis/pong` +/// +/// Responds to a ping +pub(crate) async fn pong( + State(services): State, + body: Ruma, +) -> Result { + // TODO: verify incoming question is already registered + services.federation.answer_ping(&body.question)?; + Ok(pong::unstable::Response::default()) +} diff --git a/src/ruminuwuity/api/federation/mod.rs b/src/ruminuwuity/api/federation/mod.rs new file mode 100644 index 000000000..74151af2e --- /dev/null +++ b/src/ruminuwuity/api/federation/mod.rs @@ -0,0 +1,2 @@ +pub mod ping; +pub mod pong; diff --git a/src/ruminuwuity/api/federation/ping.rs b/src/ruminuwuity/api/federation/ping.rs new file mode 100644 index 000000000..38c90d3dc --- /dev/null +++ b/src/ruminuwuity/api/federation/ping.rs @@ -0,0 +1,43 @@ +//! `GET /_matrix/federation/*/ping` +//! +//! Initiate a federation ping +pub mod unstable { + use ruma::{ + OwnedServerName, + api::{federation::authentication::ServerSignatures, request, response}, + metadata, + }; + + metadata! { + method: POST, + rate_limited: true, + authentication: ServerSignatures, // TODO: Needs to be optional + path: "/_matrix/federation/unstable/uk.timedout.msc4524.tabletennis/ping" + } + + #[request] + pub struct Request { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub origin: Option, + + pub question: String, + } + + #[response] + pub struct Response { + pub answer: String, + + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub details: Vec, + } + + impl Request { + #[must_use] + pub fn new(question: String) -> Self { Self { question, origin: None } } + } + + impl Response { + #[must_use] + pub fn new(answer: String) -> Self { Self { answer, details: vec![] } } + } +} diff --git a/src/ruminuwuity/api/federation/pong.rs b/src/ruminuwuity/api/federation/pong.rs new file mode 100644 index 000000000..c936d048f --- /dev/null +++ b/src/ruminuwuity/api/federation/pong.rs @@ -0,0 +1,30 @@ +//! `GET /_matrix/federation/*/pong` +//! +//! Respond to a federation ping +pub mod unstable { + use ruma::{ + api::{federation::authentication::ServerSignatures, request, response}, + metadata, + }; + + metadata! { + method: POST, + rate_limited: false, + authentication: ServerSignatures, + path: "/_matrix/federation/unstable/uk.timedout.msc4524.tabletennis/ping" + } + + #[request] + pub struct Request { + pub question: String, + } + + #[response] + #[derive(Default)] + pub struct Response; + + impl Request { + #[must_use] + pub fn new(question: String) -> Self { Self { question } } + } +} diff --git a/src/ruminuwuity/api/mod.rs b/src/ruminuwuity/api/mod.rs new file mode 100644 index 000000000..d2f7e1016 --- /dev/null +++ b/src/ruminuwuity/api/mod.rs @@ -0,0 +1,3 @@ +pub mod federation; + +pub use federation::*; diff --git a/src/ruminuwuity/mod.rs b/src/ruminuwuity/mod.rs index 39623e1ca..1b41554a4 100644 --- a/src/ruminuwuity/mod.rs +++ b/src/ruminuwuity/mod.rs @@ -6,3 +6,4 @@ pub mod meowlnir_antispam; pub mod policy; pub mod pushers; +pub mod api; diff --git a/src/service/federation/mod.rs b/src/service/federation/mod.rs index 446516772..5ab60e792 100644 --- a/src/service/federation/mod.rs +++ b/src/service/federation/mod.rs @@ -8,7 +8,7 @@ use assign::assign; use async_trait::async_trait; -use conduwuit::{Error, Result, Server, SyncRwLock, debug, utils::millis_since_unix_epoch}; +use conduwuit::{Err, Error, Result, Server, SyncRwLock, debug, err, error, utils::millis_since_unix_epoch}; pub(crate) use execute::FederationPathBuilderInput; use http::StatusCode; use ruma::{ @@ -22,6 +22,8 @@ pub struct Service { services: Services, pub remote_health: SyncRwLock>, pub stale_destinations: SyncRwLock>, + /// A map of {answer: channel} + pingpongs: SyncRwLock>>, } struct Services { @@ -43,6 +45,7 @@ fn build(args: crate::Args<'_>) -> Result> { }, remote_health: SyncRwLock::new(HashMap::new()), stale_destinations: SyncRwLock::new(HashSet::new()), + pingpongs: SyncRwLock::new(HashMap::new()), })) } @@ -207,4 +210,35 @@ pub fn remote_health(&self) -> HashMap { pub fn stale_destinations(&self) -> HashSet { self.stale_destinations.read().clone() } + + /// Registers an outbound ping waiter based on the answer returned by the + /// remote. Returns a channel that is written to when the remote pongs. + pub fn register_ping_answer( + &self, + expected_answer: String, + ) -> Result> { + let mut pingpongs = self.pingpongs.write(); + if pingpongs.contains_key(&expected_answer) { + return Err!(Request(InvalidParam("Duplicate answer"))); + } + let (tx, rx) = tokio::sync::oneshot::channel(); + pingpongs.insert(expected_answer, tx); + Ok(rx) + } + + /// "Answers" a registered outbound ping by sending an event to it. This is + /// called when the remote server that was pinged calls /pong. + /// + /// `M_NOT_FOUND` is returned if the answer is not recognised. + pub fn answer_ping(&self, answer: &str) -> Result<()> { + let mut pingpongs = self.pingpongs.write(); + let Some(tx) = pingpongs.remove(answer) else { + return Err!(Request(NotFound("Unknown answer"))); + }; + tx.send(()).map_err(|e| { + err!(BadServerResponse(error!( + error=?e, "Failed to handle pong" + ))) + }) + } }