volo_http/client/layer/
http_proxy.rs

1//! HTTP/1.1 Proxy defined by [RFC 7230][rfc7230]
2//!
3//! [rfc7230]: https://datatracker.ietf.org/doc/html/rfc7230
4
5use std::fmt;
6
7use bytes::Bytes;
8use http::{
9    uri::{Authority, PathAndQuery, Scheme, Uri},
10    version::Version,
11};
12use motore::{layer::Layer, service::Service};
13use volo::{client::Apply, context::Context};
14
15use crate::{
16    client::{Target, target::RemoteHost, utils::is_default_port},
17    context::ClientContext,
18    error::{
19        ClientError,
20        client::{Result, request_error},
21    },
22    request::Request,
23};
24
25/// A [`Layer`] implements HTTP/1.1 proxy defined by [RFC 7230][rfc7230].
26///
27/// [rfc7230]: https://datatracker.ietf.org/doc/html/rfc7230
28pub struct HttpProxy {
29    target: Option<Target>,
30}
31
32impl<S> Layer<S> for HttpProxy {
33    type Service = HttpProxyService<S>;
34
35    fn layer(self, inner: S) -> Self::Service {
36        HttpProxyService {
37            inner,
38            target: self.target,
39        }
40    }
41}
42
43/// [`ClientError`] during accessing proxy.
44#[derive(Debug)]
45pub struct GatewayError {
46    inner: ClientError,
47    gateway: Target,
48}
49
50impl fmt::Display for GatewayError {
51    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
52        write!(
53            f,
54            "error while using proxy {}: {}",
55            self.gateway, self.inner
56        )
57    }
58}
59
60impl std::error::Error for GatewayError {
61    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
62        Some(&self.inner)
63    }
64}
65
66fn parse_uri(uri: Uri) -> Option<Target> {
67    if let Some(scheme) = uri.scheme() {
68        if scheme != &Scheme::HTTP {
69            tracing::warn!(
70                "[Volo-HTTP] HttpProxy: `{uri}` is not a valid proxy url, only HTTP protocol is \
71                 supported"
72            );
73            return None;
74        }
75    }
76    let target = match Target::from_uri(&uri) {
77        Ok(target) => target,
78        Err(err) => {
79            tracing::warn!("[Volo-HTTP] HttpProxy: failed to parse uri `{uri}`: {err}");
80            return None;
81        }
82    };
83
84    tracing::info!("[Volo-HTTP] HttpProxy: `{uri}` is used as http proxy");
85    Some(target)
86}
87
88fn parse_env() -> Option<Uri> {
89    let env = if let Ok(uri) = std::env::var("http_proxy") {
90        uri
91    } else if let Ok(uri) = std::env::var("HTTP_PROXY") {
92        uri
93    } else {
94        return None;
95    };
96    Uri::from_maybe_shared(env).ok()
97}
98
99impl HttpProxy {
100    /// Create a [`HttpProxy`] via the proxy server from environment variable `http_proxy` or
101    /// `HTTP_PROXY`.
102    ///
103    /// If there is no valid value in environment variable, the layer will do nothing.
104    pub fn env() -> Self {
105        let target = parse_env().and_then(parse_uri);
106        Self { target }
107    }
108
109    /// Create a [`HttpProxy`] via the `uri` as proxy server.
110    ///
111    /// If the argument `uri` is not a valid uri, the layer will do nothing.
112    pub fn new<U>(uri: U) -> Self
113    where
114        U: TryInto<Uri>,
115        U::Error: std::error::Error,
116    {
117        let proxy_uri = match uri.try_into() {
118            Ok(uri) => Some(uri),
119            Err(e) => {
120                tracing::warn!("[Volo-HTTP] HttpProxy: failed to build http proxy: {e}");
121                None
122            }
123        };
124        let target = proxy_uri.and_then(parse_uri);
125        Self { target }
126    }
127}
128
129/// [`Service`] generated by [`HttpProxy`].
130///
131/// Refer to [`HttpProxy`] for more details.
132pub struct HttpProxyService<S> {
133    inner: S,
134    target: Option<Target>,
135}
136
137impl<S> HttpProxyService<S> {
138    fn update_req<B>(&self, cx: &mut ClientContext, req: &mut Request<B>) -> Option<Target> {
139        let Some(target) = &self.target else {
140            return None;
141        };
142
143        // A configured proxy is a routing requirement. Unsupported requests must not silently
144        // fall back to a direct connection.
145        if req.version() != Version::HTTP_11 {
146            tracing::info!("[Volo-HTTP] HttpProxy only works for HTTP/1.1");
147            return None;
148        }
149        if let Some(scheme) = cx.target().scheme() {
150            if scheme != &Scheme::HTTP {
151                tracing::info!("[Volo-HTTP] HttpProxy only supports HTTP protocol");
152                return None;
153            }
154        }
155
156        // Generate authority from the logical upstream target, then rewrite the HTTP/1.1
157        // request-target to absolute-form for the proxy.
158        let Some(authority) = gen_authority(cx.target()) else {
159            tracing::warn!(
160                "[Volo-HTTP] HttpProxy: failed to gen authority by {:?}",
161                cx.target()
162            );
163            return None;
164        };
165        let authority = match Authority::from_maybe_shared(Bytes::from(authority)) {
166            Ok(authority) => authority,
167            Err(e) => {
168                tracing::warn!("[Volo-HTTP] HttpProxy: failed to parse authority: {e}");
169                return None;
170            }
171        };
172        let mut parts = req.uri().to_owned().into_parts();
173        parts.scheme = Some(Scheme::HTTP);
174        parts.authority = Some(authority);
175        parts.path_and_query = Some(
176            parts
177                .path_and_query
178                .unwrap_or(PathAndQuery::from_static("/")),
179        );
180        let uri = match Uri::from_parts(parts) {
181            Ok(uri) => uri,
182            Err(e) => {
183                tracing::warn!("[Volo-HTTP] HttpProxy: failed to build uri: {e}");
184                return None;
185            }
186        };
187        *req.uri_mut() = uri;
188
189        // Only the transport target becomes the proxy. HttpProxy::call owns restoring the logical
190        // upstream after the inner call returns.
191        cx.rpc_info_mut().callee_mut().clear();
192        let old_target = target
193            .to_owned()
194            .apply_and_replace(cx)
195            .expect("infallible: failed to parse target in HttpProxy");
196
197        Some(old_target)
198    }
199}
200
201fn gen_authority(target: &Target) -> Option<String> {
202    let rt = match target {
203        Target::None => return None,
204        Target::Remote(rt) => rt,
205        #[cfg(target_family = "unix")]
206        Target::Local(_) => return None,
207    };
208    let default_port = is_default_port(&rt.scheme, rt.port);
209    let host = match &rt.host {
210        RemoteHost::Ip(ip) => {
211            if default_port {
212                if ip.is_ipv4() {
213                    format!("{ip}")
214                } else {
215                    format!("[{ip}]")
216                }
217            } else {
218                let port = rt.port;
219                if ip.is_ipv4() {
220                    format!("{ip}:{port}")
221                } else {
222                    format!("[{ip}]:{port}")
223                }
224            }
225        }
226        RemoteHost::Name(name) => {
227            let port = rt.port;
228            if default_port {
229                name.as_str().to_owned()
230            } else {
231                format!("{name}:{port}")
232            }
233        }
234    };
235    Some(host)
236}
237
238fn restore_target(cx: &mut ClientContext, target: Target) -> Result<()> {
239    cx.rpc_info_mut().callee_mut().clear();
240    target.apply(cx)
241}
242
243impl<B, S> Service<ClientContext, Request<B>> for HttpProxyService<S>
244where
245    B: Send,
246    S: Service<ClientContext, Request<B>, Error = ClientError> + Send + Sync,
247{
248    type Response = S::Response;
249    type Error = S::Error;
250
251    async fn call(
252        &self,
253        cx: &mut ClientContext,
254        mut req: Request<B>,
255    ) -> Result<Self::Response, Self::Error> {
256        let old_target = self.update_req(cx, &mut req);
257        let result = self.inner.call(cx, req).await;
258
259        // HttpProxy only borrows cx.target() as a transport target. Restore the logical upstream
260        // before returning to FollowRedirect or any other outer layer. Do this for both success and
261        // error responses.
262        if let Some(target) = old_target {
263            restore_target(cx, target)?;
264        };
265
266        match result {
267            Ok(resp) => Ok(resp),
268            Err(e) => {
269                if let Some(target) = &self.target {
270                    let err = GatewayError {
271                        inner: e,
272                        gateway: target.to_owned(),
273                    };
274                    Err(request_error(err))
275                } else {
276                    Err(e)
277                }
278            }
279        }
280    }
281}
volo_http/client/layer/http_proxy.rs

volo_http/client/layer/
http_proxy.rs
