From c8679caeb492deef626d7a8fc30402a318af858b 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 | 57 +++++++++++++++++-- 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 | 2 + src/service/federation/mod.rs | 41 ++++++++++++- 13 files changed, 292 insertions(+), 46 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 c69624620..32d081ae7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -871,6 +871,7 @@ dependencies = [ "log", "resolvematrix", "ruma", + "ruminuwuity", "serde-saphyr", "serde_json", "tokio", diff --git a/src/admin/Cargo.toml b/src/admin/Cargo.toml index 1f82978a4..6ce907fb2 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] @@ -87,6 +87,7 @@ lettre.workspace = true log.workspace = true assign.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 225874520..4aed80440 100644 --- a/src/admin/debug/commands.rs +++ b/src/admin/debug/commands.rs @@ -6,13 +6,14 @@ }; use conduwuit::{ - Err, Result, at, debug_error, err, info, + Err, Result, at, debug_error, debug_info, err, info, matrix::{ Event, pdu::{PduEvent, PduId, RawPduId}, }, trace, utils, utils::{ + random_string, stream::{IterStream, ReadyExt}, string::EMPTY, }, @@ -26,8 +27,10 @@ }; use ruma::{ CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, - OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId, UInt, - api::federation::event::get_room_state, events::AnyStateEvent, serde::Raw, + OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId, ServerName, UInt, + api::{error::ErrorKind, federation::event::get_room_state}, + events::AnyStateEvent, + serde::Raw, }; use service::rooms::{ short::{ShortEventId, ShortRoomId}, @@ -581,14 +584,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 6b972043d..d02e1abb5 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 9c98031db..f3adcde42 100644 --- a/src/ruminuwuity/mod.rs +++ b/src/ruminuwuity/mod.rs @@ -5,3 +5,5 @@ pub mod invite_permission_config; pub mod meowlnir_antispam; pub mod policy; + +pub mod api; diff --git a/src/service/federation/mod.rs b/src/service/federation/mod.rs index 703f939b2..870184735 100644 --- a/src/service/federation/mod.rs +++ b/src/service/federation/mod.rs @@ -1,13 +1,16 @@ mod execute; -use std::sync::Arc; -use conduwuit::{Result, Server}; +use std::{collections::HashMap, sync::Arc}; + +use conduwuit::{Err, Result, Server, SyncRwLock, err}; pub(crate) use execute::FederationPathBuilderInput; use crate::{Dep, client, moderation, server_keys}; pub struct Service { services: Services, + /// A map of {answer: channel} + pingpongs: SyncRwLock>>, } struct Services { @@ -26,8 +29,42 @@ fn build(args: crate::Args<'_>) -> Result> { server_keys: args.depend::("server_keys"), moderation: args.depend::("moderation"), }, + pingpongs: SyncRwLock::new(HashMap::new()), })) } fn name(&self) -> &str { crate::service::make_name(std::module_path!()) } } + +impl Service { + /// 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" + ))) + }) + } +}