diff --git a/src/service/federation/execute.rs b/src/service/federation/execute.rs index 0c147c1ef..ee588df7c 100644 --- a/src/service/federation/execute.rs +++ b/src/service/federation/execute.rs @@ -2,8 +2,7 @@ use std::{borrow::Cow, fmt::Debug, mem}; use bytes::Bytes; use conduwuit::{ - Err, Error, Result, debug, debug_error, debug_warn, err, implement, trace, - utils::response::LimitReadExt, + Err, Error, Result, debug, debug_error, debug_warn, err, trace, utils::response::LimitReadExt, }; use ipaddress::IPAddress; use reqwest::{Client, Method, Request, Response, Url}; @@ -20,181 +19,186 @@ use ruma::{ use crate::{SUPPORTED_VERSIONS, resolver::actual::ActualDest}; -/// Sends a request to a federation server -#[implement(super::Service)] -#[tracing::instrument(skip_all, name = "request", level = "debug")] -pub async fn execute<'i, T>(&self, dest: &ServerName, request: T) -> Result -where - T: OutgoingRequest< - Authentication = ServerSignatures, - PathBuilder: PathBuilder: FederationPathBuilderInput>, - > + Debug - + Send, -{ - let client = &self.services.client.federation; - self.execute_signed(client, dest, request).await -} - -/// Like execute() but with a very large timeout -#[implement(super::Service)] -#[tracing::instrument(skip_all, name = "synapse", level = "debug")] -pub async fn execute_synapse<'i, T>( - &self, - dest: &ServerName, - request: T, -) -> Result -where - T: OutgoingRequest< - Authentication = ServerSignatures, - PathBuilder: PathBuilder: FederationPathBuilderInput>, - > + Debug - + Send, -{ - let client = &self.services.client.synapse; - self.execute_signed(client, dest, request).await -} - -#[implement(super::Service)] -pub async fn execute_unauthenticated<'i, T>( - &self, - dest: &ServerName, - request: T, -) -> Result -where - T: OutgoingRequest< - Authentication = NoAuthentication, - PathBuilder: PathBuilder: FederationPathBuilderInput>, - > + Debug - + Send, -{ - let client = &self.services.client.federation; - - self.execute_on(client, dest, request, ()).await -} - -#[implement(super::Service)] -pub async fn execute_signed<'i, T>( - &self, - client: &Client, - dest: &ServerName, - request: T, -) -> Result -where - T: OutgoingRequest< - Authentication = ServerSignatures, - PathBuilder: PathBuilder: FederationPathBuilderInput>, - > + Send, -{ - let authentication = ServerSignaturesInput::new( - self.services.server.name.clone(), - dest.to_owned(), - self.services.server_keys.keypair(), - ); - - self.execute_on(client, dest, request, authentication).await -} - -#[implement(super::Service)] -#[tracing::instrument(name = "fed", level = "info", skip(self, client, request, authentication))] -pub async fn execute_on<'i, T, PathBuilderInput>( - &self, - client: &Client, - dest: &ServerName, - request: T, - authentication: ::Input<'_>, -) -> Result -where - T: OutgoingRequest = PathBuilderInput>> + Send, - PathBuilderInput: FederationPathBuilderInput, -{ - if !self.services.server.config.allow_federation { - return Err!(Config("allow_federation", "Federation is disabled.")); +impl super::Service { + /// Sends a request to a federation server + #[tracing::instrument(skip_all, name = "request", level = "debug")] + pub async fn execute<'i, T>( + &self, + dest: &ServerName, + request: T, + ) -> Result + where + T: OutgoingRequest< + Authentication = ServerSignatures, + PathBuilder: PathBuilder: FederationPathBuilderInput>, + > + Debug + + Send, + { + let client = &self.services.client.federation; + self.execute_signed(client, dest, request).await } - if self.services.moderation.is_remote_server_forbidden(dest) { - return Err!(Request(Forbidden(debug_warn!("Federation with {dest} is not allowed.")))); + /// Like execute() but with a very large timeout + #[tracing::instrument(skip_all, name = "synapse", level = "debug")] + pub async fn execute_synapse<'i, T>( + &self, + dest: &ServerName, + request: T, + ) -> Result + where + T: OutgoingRequest< + Authentication = ServerSignatures, + PathBuilder: PathBuilder: FederationPathBuilderInput>, + > + Debug + + Send, + { + let client = &self.services.client.synapse; + self.execute_signed(client, dest, request).await } - let actual = self.services.resolver.get_actual_dest(dest).await?; + pub async fn execute_unauthenticated<'i, T>( + &self, + dest: &ServerName, + request: T, + ) -> Result + where + T: OutgoingRequest< + Authentication = NoAuthentication, + PathBuilder: PathBuilder: FederationPathBuilderInput>, + > + Debug + + Send, + { + let client = &self.services.client.federation; - let request = Request::try_from(request.try_into_http_request::>( - actual.string().as_str(), - authentication, - PathBuilderInput::create(), - )?)?; - self.validate_url(request.url())?; - self.services.server.check_running()?; - - self.perform::(dest, &actual, request, client).await -} - -#[implement(super::Service)] -async fn perform( - &self, - dest: &ServerName, - actual: &ActualDest, - request: Request, - client: &Client, -) -> Result -where - T: OutgoingRequest + Send, -{ - let url = request.url().clone(); - let method = request.method().clone(); - - debug!(%method, %url, "Sending request"); - match client.execute(request).await { - | Ok(response) => - self.handle_response::(dest, actual, &method, &url, response) - .await, - | Err(error) => - Err(handle_error(actual, &method, &url, error).expect_err("always returns error")), + self.execute_on(client, dest, request, ()).await } -} -#[implement(super::Service)] -fn validate_url(&self, url: &Url) -> Result<()> { - if let Some(url_host) = url.host_str() { - if let Ok(ip) = IPAddress::parse(url_host) { - trace!("Checking request URL IP {ip:?}"); - self.services.resolver.validate_ip(&ip)?; + pub async fn execute_signed<'i, T>( + &self, + client: &Client, + dest: &ServerName, + request: T, + ) -> Result + where + T: OutgoingRequest< + Authentication = ServerSignatures, + PathBuilder: PathBuilder: FederationPathBuilderInput>, + > + Send, + { + let authentication = ServerSignaturesInput::new( + self.services.server.name.clone(), + dest.to_owned(), + self.services.server_keys.keypair(), + ); + + self.execute_on(client, dest, request, authentication).await + } + + #[tracing::instrument( + name = "fed", + level = "info", + skip(self, client, request, authentication) + )] + pub async fn execute_on<'i, T, PathBuilderInput>( + &self, + client: &Client, + dest: &ServerName, + request: T, + authentication: ::Input<'_>, + ) -> Result + where + T: OutgoingRequest = PathBuilderInput>> + Send, + PathBuilderInput: FederationPathBuilderInput, + { + if !self.services.server.config.allow_federation { + return Err!(Config("allow_federation", "Federation is disabled.")); + } + + if self.services.moderation.is_remote_server_forbidden(dest) { + return Err!(Request(Forbidden(debug_warn!( + "Federation with {dest} is not allowed." + )))); + } + + let actual = self.services.resolver.get_actual_dest(dest).await?; + + let request = Request::try_from(request.try_into_http_request::>( + actual.string().as_str(), + authentication, + PathBuilderInput::create(), + )?)?; + self.validate_url(request.url())?; + self.services.server.check_running()?; + + self.perform::(dest, &actual, request, client).await + } + + async fn perform( + &self, + dest: &ServerName, + actual: &ActualDest, + request: Request, + client: &Client, + ) -> Result + where + T: OutgoingRequest + Send, + { + let url = request.url().clone(); + let method = request.method().clone(); + + debug!(%method, %url, "Sending request"); + match client.execute(request).await { + | Ok(response) => + self.handle_response::(dest, actual, &method, &url, response) + .await, + | Err(error) => + Err(handle_error(actual, &method, &url, error).expect_err("always returns error")), } } - Ok(()) -} + fn validate_url(&self, url: &Url) -> Result<()> { + if let Some(url_host) = url.host_str() { + if let Ok(ip) = IPAddress::parse(url_host) { + trace!("Checking request URL IP {ip:?}"); + self.services.resolver.validate_ip(&ip)?; + } + } -#[implement(super::Service)] -async fn handle_response( - &self, - dest: &ServerName, - actual: &ActualDest, - method: &Method, - url: &Url, - response: Response, -) -> Result -where - T: OutgoingRequest + Send, -{ - const HUGE_ENDPOINTS: [&str; 2] = - ["/_matrix/federation/v2/send_join/", "/_matrix/federation/v2/state/"]; - let size_limit: u64 = if HUGE_ENDPOINTS.iter().any(|e| url.path().starts_with(e)) { - // Some federation endpoints can return huge response bodies, so we'll bump the - // limit for those endpoints specifically. - self.services - .server - .config - .max_request_size - .saturating_mul(10) - } else { - self.services.server.config.max_request_size + Ok(()) } - .try_into() - .expect("size_limit (usize) should fit within a u64"); - let response = into_http_response(dest, actual, method, url, response, size_limit).await?; - T::IncomingResponse::try_from_http_response(response) - .map_err(|e| err!(BadServerResponse("Server returned bad 200 response: {e:?}"))) + async fn handle_response( + &self, + dest: &ServerName, + actual: &ActualDest, + method: &Method, + url: &Url, + response: Response, + ) -> Result + where + T: OutgoingRequest + Send, + { + const HUGE_ENDPOINTS: [&str; 2] = + ["/_matrix/federation/v2/send_join/", "/_matrix/federation/v2/state/"]; + let size_limit: u64 = if HUGE_ENDPOINTS.iter().any(|e| url.path().starts_with(e)) { + // Some federation endpoints can return huge response bodies, so we'll bump the + // limit for those endpoints specifically. + self.services + .server + .config + .max_request_size + .saturating_mul(10) + } else { + self.services.server.config.max_request_size + } + .try_into() + .expect("size_limit (usize) should fit within a u64"); + let response = + into_http_response(dest, actual, method, url, response, size_limit).await?; + + T::IncomingResponse::try_from_http_response(response) + .map_err(|e| err!(BadServerResponse("Server returned bad 200 response: {e:?}"))) + } } async fn into_http_response(