continuwuity/src/service/federation/execute.rs

use std::{borrow::Cow, fmt::Debug, mem, time::Instant};

use conduwuit::{
	Err, Error, Result, debug, debug_error, debug_info, debug_warn, err, metrics::KeyValue,
	trace, utils::response::LimitReadExt,
};
use ipaddress::IPAddress;
use reqwest::{Client, Method, Request, Response, Url};
use resolvematrix::resolution::Resolution;
use ruma::{
	ServerName,
	api::{
		EndpointError, IncomingResponseExt, OutgoingRequest, OutgoingRequestExt,
		SupportedVersions,
		auth_scheme::{AuthScheme, NoAuthentication},
		error::Error as RumaError,
		federation::authentication::{ServerSignatures, XMatrixSigningInput},
		path_builder::PathBuilder,
	},
};

use crate::SUPPORTED_VERSIONS;

impl super::Service {
	/// Sends a signed request to a remote server over federation.
	#[tracing::instrument(skip_all, name = "request", level = "debug")]
	pub async fn execute<'i, T>(
		&self,
		dest: &ServerName,
		request: T,
	) -> Result<T::IncomingResponse>
	where
		T: OutgoingRequest<
				Authentication = ServerSignatures,
				PathBuilder: PathBuilder<Input<'i>: FederationPathBuilderInput>,
			> + Debug
			+ Send,
	{
		let client = &self.services.client.federation;
		self.execute_signed(client, dest, request).await
	}

	/// Sends a signed request to a remote server over federation, with a
	/// significantly higher timeout to account for slow endpoints.
	#[tracing::instrument(skip_all, name = "request_slow", level = "debug")]
	pub async fn execute_slow<'i, T>(
		&self,
		dest: &ServerName,
		request: T,
	) -> Result<T::IncomingResponse>
	where
		T: OutgoingRequest<
				Authentication = ServerSignatures,
				PathBuilder: PathBuilder<Input<'i>: FederationPathBuilderInput>,
			> + Debug
			+ Send,
	{
		let client = &self.services.client.federation_slow;
		self.execute_signed(client, dest, request).await
	}

	/// Sends an unsigned (unauthenticated) request to a remote server over
	/// federation.
	pub async fn execute_unauthenticated<'i, T>(
		&self,
		dest: &ServerName,
		request: T,
	) -> Result<T::IncomingResponse>
	where
		T: OutgoingRequest<
				Authentication = NoAuthentication,
				PathBuilder: PathBuilder<Input<'i>: FederationPathBuilderInput>,
			> + Debug
			+ Send,
	{
		let client = &self.services.client.federation;

		self.execute_on(client, dest, request, ()).await
	}

	/// Sends a signed request to a remote server over federation, given the
	/// specific client.
	pub async fn execute_signed<'i, T>(
		&self,
		client: &Client,
		dest: &ServerName,
		request: T,
	) -> Result<T::IncomingResponse>
	where
		T: OutgoingRequest<
				Authentication = ServerSignatures,
				PathBuilder: PathBuilder<Input<'i>: FederationPathBuilderInput>,
			> + Send,
	{
		let authentication = XMatrixSigningInput::new(
			self.services.server.name.clone(),
			dest.to_owned(),
			self.services.server_keys.keypair(),
		);

		self.execute_on(client, dest, request, authentication).await
	}

	/// Prepares and sends a federation request with the given client.
	/// If federation is disabled, an error is returned here. Likewise, if the
	/// destination is forbidden.
	/// Preparation includes resolving the server before sending the request.
	#[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: <T::Authentication as AuthScheme>::Input<'_>,
	) -> Result<T::IncomingResponse>
	where
		T: OutgoingRequest<PathBuilder: PathBuilder<Input<'i> = 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."
			))));
		}

		self.ensure_remote_is_healthy(dest)?;

		let actual = self
			.services
			.client
			.matrix_resolver
			.resolve_server(dest.as_str())
			.await
			.inspect_err(|_| {
				self.hit_unhealthy(dest.to_owned());
				self.mark_destination_stale(dest);
			})?;

		let request = Request::try_from(request.try_into_http_request::<Vec<u8>>(
			actual.base_url().as_str(),
			authentication,
			PathBuilderInput::create(),
		)?)?;
		self.validate_url(request.url())?;
		self.services.server.check_running()?;

		self.perform::<T>(dest, &actual, request, client).await
	}

	/// Actually sends the federation request, handling the response before
	/// returning.
	///
	/// Timing info and request logs are available when debug logs are enabled.
	async fn perform<T>(
		&self,
		dest: &ServerName,
		actual: &Resolution,
		request: Request,
		client: &Client,
	) -> Result<T::IncomingResponse>
	where
		T: OutgoingRequest + Send,
	{
		let url = request.url().clone();
		let method = request.method().clone();

		debug_info!(
			"Sending request: {} matrix-federation://{dest}{}",
			method.as_str(),
			url.path(),
		);
		let start = Instant::now();
		let response = client.execute(request).await;
		let elapsed = start.elapsed();
		debug_info!(
			"Request finished in {elapsed:?}: {} matrix-federation://{dest}{}",
			method.as_str(),
			url.path(),
		);
		let outcome = match &response {
			| Ok(response) if response.status().is_success() => "success",
			| Ok(response) if response.status().is_client_error() => "client_error",
			| Ok(response) if response.status().is_server_error() => "server_error",
			| Ok(_) => "other",
			| Err(_) => "network_error",
		};
		self.services
			.server
			.metrics
			.instruments
			.federation_outgoing_request_duration
			.record(elapsed.as_secs_f64(), &[KeyValue::new("outcome", outcome)]);
		match response {
			| Ok(response) =>
				self.handle_response::<T>(dest, actual, &method, &url, response)
					.await,
			| Err(error) => {
				// This awful wrapping hack is required because `reqwest::Error`
				// does not implement `Clone`, but we need to convert it
				// into a local error type to pass to should_mark_stale, but
				// handle_error then itself expects the original
				// reqwest error.
				// handle_error *could* just take the wrapped error, but it'd
				// have to unwrap it anyway.
				let wrapped = Error::Reqwest(error.into());
				if self.should_mark_stale(&wrapped) {
					debug_info!("{dest} is unhealthy & stale due to a connect error");
					self.mark_destination_stale(dest);
					self.hit_unhealthy(dest.to_owned());
				}
				let Error::Reqwest(unwrapped) = wrapped else {
					unreachable!("wrapped reqwest error must unwrap to a reqwest error");
				};
				Err(handle_error(actual, &method, &url, unwrapped.into())
					.expect_err("always returns error"))
			},
		}
	}

	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:?}");
				if !self.services.client.valid_cidr_range(&ip) {
					return Err!(BadServerResponse("Not allowed to send requests to this IP"));
				}
			}
		}

		Ok(())
	}

	/// Handles a successful response from a remote server, reading and
	/// deserializing the response body (if any).
	async fn handle_response<T>(
		&self,
		dest: &ServerName,
		actual: &Resolution,
		method: &Method,
		url: &Url,
		response: Response,
	) -> Result<T::IncomingResponse>
	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?;

		let (parts, body) = response.into_parts();
		T::IncomingResponse::try_from_http_response(http::Response::from_parts(
			parts,
			body.as_ref(),
		))
		.inspect(|_| {
			self.mark_healthy(dest);
		})
		.map_err(|e| {
			// Bad 200 response is usually a sign of bad routing, so we'll mark
			// as stale. We won't mark as offline though since we technically
			// got a response.
			self.mark_destination_stale(dest);
			err!(BadServerResponse("Server returned bad 200 response: {e:?}"))
		})
	}
}

async fn into_http_response(
	dest: &ServerName,
	actual: &Resolution,
	method: &Method,
	url: &Url,
	mut response: Response,
	max_size: u64,
) -> Result<http::Response<Vec<u8>>> {
	let status = response.status();
	trace!(
		%status, %method,
		request_url = %url,
		response_url = %response.url(),
		"Received response from {}",
		actual.base_url(),
	);

	if url.path() != "/.well-known/matrix/server" {
		// For some reason, the spec starts off saying federation responses MUST
		// include a `Content-Type` header of `application/json` on all
		// responses. Then, the definition for the well-known server endpoint
		// says that it's a SHOULD and that servers SHOULD just assume that the
		// response is JSON regardless... And then there's media wayyyy down in
		// the spec doc that just throws a `multipart/mixed` in there.
		//
		// Media has a special case regarding status codes. Only 200 OK
		// responses should have the multipart assertion, as errors are JSON.
		let ct = if url.path().starts_with("/_matrix/federation/v1/media/")
			&& response.status().is_success()
		{
			"multipart/mixed"
		} else {
			"application/json"
		};
		enforce_content_type(&response, ct)?;
	}

	let mut http_response_builder = http::Response::builder()
		.status(status)
		.version(response.version());

	mem::swap(
		response.headers_mut(),
		http_response_builder
			.headers_mut()
			.expect("http::response::Builder is usable"),
	);

	trace!("Waiting for response body...");
	let http_response = http_response_builder
		.body(response.limit_read(max_size).await.unwrap_or_default())
		.expect("reqwest body is valid http body");

	debug!("Got {status:?} for {method} {url}");
	if !status.is_success() {
		let (parts, body) = http_response.into_parts();
		return Err(Error::Federation(
			dest.to_owned(),
			RumaError::from_http_response(http::Response::from_parts(parts, body.as_ref())),
		));
	}

	Ok(http_response)
}

fn handle_error(
	actual: &Resolution,
	method: &Method,
	url: &Url,
	mut e: reqwest::Error,
) -> Result {
	if e.is_timeout() || e.is_connect() {
		e = e.without_url();
		debug_warn!("{e:?}");
	} else if e.is_redirect() {
		debug_error!(
			%method,
			%url,
			final_url = e.url().map(tracing::field::display),
			"Redirect loop {}: {}",
			actual.host,
			e,
		);
	} else {
		debug_error!("{e:?}");
	}

	Err(e.into())
}

/// Enforces that the `Content-Type` response header is present, and that the
/// first part matches the expected header. This does not enforce that multipart
/// responses have a boundary as only the text left of a semicolon, if present,
/// is compared.
///
/// Returns `M_UNKNOWN` with HTTP 415 (Unsupported Media Type) if the response
/// header is missing or mismatched.
fn enforce_content_type(response: &Response, expected: &str) -> Result<()> {
	// N.B. Content type's value is case-insensitive per RFC 9110. The Matrix
	// spec explicitly prescribes `application/json` and `multipart/mixed` etc,
	// but it's assumed they're intended to be case-insensitive.
	debug_assert_eq!(
		expected,
		expected.to_lowercase(),
		"Content type checks should use the lowercase content type"
	); // ^ Saves us double-folding
	match response
		.headers()
		.get(&http::header::CONTENT_TYPE)
		.and_then(|h| h.to_str().ok())
		.map(|h| h.split_once(';').unwrap_or((h, "")).0.to_lowercase())
	{
		| Some(ct) if ct == expected => Ok(()),
		| Some(ct) => Err!(Request(
			Unknown(
				"Remote server returned the wrong Content-Type: expected {expected}, got {ct:?}"
			),
			UNSUPPORTED_MEDIA_TYPE
		)),
		| None => Err!(Request(
			Unknown("Remote server did not return a Content-Type header"),
			UNSUPPORTED_MEDIA_TYPE
		)),
	}
}

/// A trait for the input types of acceptable path builders for outgoing
/// federation requests.
///
/// Ruma uses Rust's type system to encode the versioning scheme of endpoints in
/// the Matrix spec. Every endpoint has a `PathBuilder` associated type, which
/// has an `Input` associated type. Endpoints with multiple versions have
/// `VersionHistory` as their `PathBuilder`, which has `SupportedVersions`
/// as its `Input` type. Endpoints with no version have `SinglePath` as their
/// `PathBuilder`, which has `()` as its `Input` type. Both `SupportedVersions`
/// and `()` can be created out of thin air using static data (or no data at
/// all). This property is what the `FederationPathBuilderInput` trait
/// represents.
///
/// This trait allows the federation sender service's functions to accept
/// requests for either versioned or unversioned endpoints, by requiring that
/// the `Input` of the `PathBuilder` of the endpoint implements
/// `FederationPathBuilderInput`.
pub trait FederationPathBuilderInput {
	fn create() -> Self;
}

impl FederationPathBuilderInput for () {
	fn create() -> Self {}
}

impl FederationPathBuilderInput for Cow<'_, SupportedVersions> {
	fn create() -> Self { Cow::Borrowed(&SUPPORTED_VERSIONS) }
}