tuwunel_service/pusher/
request.rs

1use std::{fmt::Debug, mem::swap};
2
3use bytes::BytesMut;
4use http::Response as HttpResponse;
5use reqwest::{Error as ReqwestError, Response};
6use ruma::api::{
7	IncomingResponse, OutgoingRequest, OutgoingRequestExt, auth_scheme::AuthScheme,
8	path_builder::PathBuilder,
9};
10use tuwunel_core::{
11	Err, Result, debug_warn, err, error::error_chain, implement, trace, utils::string_from_bytes,
12	warn,
13};
14
15use crate::client::read_response_capped;
16
17#[implement(super::Service)]
18#[tracing::instrument(level = "debug", skip_all)]
19pub(super) async fn send_request<T>(&self, dest: &str, request: T) -> Result<T::IncomingResponse>
20where
21	T: OutgoingRequest + Debug + Send,
22	for<'a> T::Authentication: AuthScheme<Input<'a> = ()>,
23	for<'a> T::PathBuilder: PathBuilder<Input<'a> = ()>,
24{
25	let dest = dest.replace(&self.services.config.notification_push_path, "");
26	trace!("Push gateway destination: {dest}");
27
28	let http_request = request
29		.try_into_http_request::<BytesMut>(&dest, (), ())
30		.map_err(|e| {
31			err!(BadServerResponse(warn!(
32				"Failed to find destination {dest} for push gateway: {e}"
33			)))
34		})?
35		.map(BytesMut::freeze);
36
37	let reqwest_request = reqwest::Request::try_from(http_request)?;
38
39	if self
40		.services
41		.client
42		.proxy
43		.resolver_alias(reqwest_request.url())
44	{
45		return Err!(BadServerResponse(
46			"Not allowed to request a locally resolved proxy endpoint"
47		));
48	}
49
50	trace!("Checking request URL for IP");
51	if !self
52		.services
53		.client
54		.valid_cidr_range_url(reqwest_request.url())
55	{
56		return Err!(BadServerResponse("Not allowed to send requests to this IP"));
57	}
58
59	match self
60		.services
61		.client
62		.pusher
63		.execute(reqwest_request)
64		.await
65	{
66		| Err(error) => handler_err(&dest, error),
67		| Ok(response) => self.handle_ok::<T>(&dest, response).await,
68	}
69}
70
71#[implement(super::Service)]
72async fn handle_ok<T>(&self, dest: &str, mut response: Response) -> Result<T::IncomingResponse>
73where
74	T: OutgoingRequest,
75{
76	trace!("Checking response destination's IP");
77	if let Some(remote_addr) = response.remote_addr()
78		&& !self
79			.services
80			.client
81			.valid_cidr_range_ip(remote_addr.ip())
82		&& !self.services.client.proxied(response.url())
83	{
84		return Err!(BadServerResponse("Not allowed to send requests to this IP"));
85	}
86
87	let status = response.status();
88	let mut http_response_builder = HttpResponse::builder()
89		.status(status)
90		.version(response.version());
91
92	swap(
93		response.headers_mut(),
94		http_response_builder
95			.headers_mut()
96			.expect("http::response::Builder is usable"),
97	);
98
99	let limit = self.services.config.max_response_size;
100	let body = read_response_capped(response, limit).await?;
101
102	if !status.is_success() {
103		debug_warn!(body = ?string_from_bytes(&body), "Push gateway response");
104		return Err!(BadServerResponse(warn!(
105			"Push gateway {dest} returned unsuccessful HTTP response: {status}"
106		)));
107	}
108
109	let response = T::IncomingResponse::try_from_http_response(
110		http_response_builder
111			.body(body)
112			.expect("reqwest body is valid http body"),
113	);
114
115	response.map_err(|e| {
116		err!(BadServerResponse(warn!("Push gateway {dest} returned invalid response: {e}")))
117	})
118}
119
120fn handler_err<R>(dest: &str, error: ReqwestError) -> Result<R> {
121	warn!(%dest, chain = %error_chain(&error), "Could not send request to pusher");
122	Err(error.into())
123}
tuwunel_service/pusher/request.rs

tuwunel_service/pusher/
request.rs
