rama_http/layer/upgrade/http_proxy_connect/
service_matcher.rs

1use std::convert::Infallible;
2
3use crate::{Request, Response};
4use rama_core::{
5    extensions::ExtensionsRef,
6    matcher::service::{ServiceMatch, ServiceMatcher},
7    rt::Executor,
8    telemetry::tracing,
9};
10use rama_http_types::proxy::is_req_http_proxy_connect;
11use rama_net::{
12    client::ConnectorTarget,
13    proxy::IoForwardService,
14    user::{ProxyCredential, credentials::DpiProxyCredential},
15};
16
17#[derive(Debug, Clone)]
18/// Default matcher that can be used for Http proxy connects.
19///
20/// Request matches for an http proxy connect request return
21/// a [`HttpProxyConnectRelayServiceResponseMatcher`] instance which
22/// will match on any success responses...
23pub struct HttpProxyConnectRelayServiceRequestMatcher<S = IoForwardService> {
24    relay_svc: S,
25}
26
27impl Default for HttpProxyConnectRelayServiceRequestMatcher {
28    fn default() -> Self {
29        Self {
30            relay_svc: IoForwardService::default(),
31        }
32    }
33}
34
35impl HttpProxyConnectRelayServiceRequestMatcher {
36    /// Create a [`HttpProxyConnectRelayServiceRequestMatcher`] whose default
37    /// fallback relay observes graceful shutdown via the given [`Executor`].
38    ///
39    /// Prefer this over [`Self::default`] when you have an executor available
40    /// — it lets the proxy connect bridges unwind cleanly on shutdown.
41    #[must_use]
42    pub fn default_with_exec(exec: Executor) -> Self {
43        Self {
44            relay_svc: IoForwardService::new(exec),
45        }
46    }
47}
48
49impl<S> HttpProxyConnectRelayServiceRequestMatcher<S> {
50    #[inline(always)]
51    #[must_use]
52    /// Create a new [`HttpProxyConnectRelayServiceRequestMatcher`].
53    pub fn new(relay_svc: S) -> Self {
54        Self { relay_svc }
55    }
56}
57
58impl<S, Body> ServiceMatcher<Request<Body>> for HttpProxyConnectRelayServiceRequestMatcher<S>
59where
60    S: Clone + Send + Sync + 'static,
61    Body: Send + 'static,
62{
63    type Service = HttpProxyConnectRelayServiceResponseMatcher<S>;
64    type Error = Infallible;
65    type ModifiedInput = Request<Body>;
66
67    async fn match_service(
68        &self,
69        req: Request<Body>,
70    ) -> Result<ServiceMatch<Self::ModifiedInput, Self::Service>, Self::Error> {
71        let Self { relay_svc } = self;
72
73        let mut svc_match = ServiceMatch {
74            service: None,
75            input: req,
76        };
77
78        if !is_req_http_proxy_connect(&svc_match.input) {
79            tracing::debug!(
80                "no req http proxy connect match: target = {:?}; http version = {:?}; method = {:?}; uri = {:?}",
81                svc_match.input.extensions().get_ref::<ConnectorTarget>(),
82                svc_match.input.version(),
83                svc_match.input.method(),
84                svc_match.input.uri(),
85            );
86            return Ok(svc_match);
87        }
88
89        tracing::debug!(
90            "http proxy connect match: target = {:?}; http version = {:?}; method = {:?}; uri = {:?}",
91            svc_match.input.extensions().get_ref::<ConnectorTarget>(),
92            svc_match.input.version(),
93            svc_match.input.method(),
94            svc_match.input.uri(),
95        );
96
97        svc_match.service = Some(HttpProxyConnectRelayServiceResponseMatcher {
98            relay_svc: relay_svc.clone(),
99        });
100
101        Ok(svc_match)
102    }
103
104    async fn into_match_service(
105        self,
106        req: Request<Body>,
107    ) -> Result<ServiceMatch<Self::ModifiedInput, Self::Service>, Self::Error> {
108        let Self { relay_svc } = self;
109
110        let mut svc_match = ServiceMatch {
111            service: None,
112            input: req,
113        };
114
115        if !is_req_http_proxy_connect(&svc_match.input) {
116            tracing::debug!(
117                "no req http proxy connect match: target = {:?}; http version = {:?}; method = {:?}; uri = {:?}",
118                svc_match.input.extensions().get_ref::<ConnectorTarget>(),
119                svc_match.input.version(),
120                svc_match.input.method(),
121                svc_match.input.request_uri(),
122            );
123            return Ok(svc_match);
124        }
125
126        tracing::debug!(
127            "http proxy connect match: target = {:?}; http version = {:?}; method = {:?}; uri = {:?}",
128            svc_match.input.extensions().get_ref::<ConnectorTarget>(),
129            svc_match.input.version(),
130            svc_match.input.method(),
131            svc_match.input.request_uri(),
132        );
133
134        svc_match.service = Some(HttpProxyConnectRelayServiceResponseMatcher { relay_svc });
135
136        Ok(svc_match)
137    }
138}
139
140#[derive(Debug, Clone)]
141/// Created by [`HttpProxyConnectRelayServiceRequestMatcher`] for a valid http proxy connect
142/// request match, this response matcher half ensures the returned status code is successfull.
143pub struct HttpProxyConnectRelayServiceResponseMatcher<S> {
144    relay_svc: S,
145}
146
147impl<S, Body> ServiceMatcher<Response<Body>> for HttpProxyConnectRelayServiceResponseMatcher<S>
148where
149    S: Clone + Send + Sync + 'static,
150    Body: Send + 'static,
151{
152    type Service = S;
153    type Error = Infallible;
154    type ModifiedInput = Response<Body>;
155
156    async fn match_service(
157        &self,
158        res: Response<Body>,
159    ) -> Result<ServiceMatch<Self::ModifiedInput, Self::Service>, Self::Error> {
160        let Self { relay_svc } = self;
161
162        let mut svc_match = ServiceMatch {
163            service: None,
164            input: res,
165        };
166
167        if !svc_match.input.status().is_success() {
168            tracing::debug!(
169                "no res http proxy connect match: target = {:?}; http version = {:?}",
170                svc_match.input.extensions().get_ref::<ConnectorTarget>(),
171                svc_match.input.version(),
172            );
173            return Ok(svc_match);
174        }
175
176        let proxy_target = svc_match.input.extensions().get_ref::<ConnectorTarget>();
177        let proxy_credential_info =
178            svc_match
179                .input
180                .extensions()
181                .get_ref()
182                .map(|DpiProxyCredential(c)| match c {
183                    ProxyCredential::Basic(user_pass) => ("basic", user_pass.username()),
184                    ProxyCredential::Bearer(_) => ("bearer", "***"),
185                });
186        tracing::debug!(
187            "response matched by (HTTP) proxy request: proxy target = {proxy_target:?}; credials = {proxy_credential_info:?}"
188        );
189
190        svc_match.service = Some(relay_svc.clone());
191
192        Ok(svc_match)
193    }
194
195    async fn into_match_service(
196        self,
197        res: Response<Body>,
198    ) -> Result<ServiceMatch<Self::ModifiedInput, Self::Service>, Self::Error> {
199        let Self { relay_svc } = self;
200
201        let mut svc_match = ServiceMatch {
202            service: None,
203            input: res,
204        };
205
206        if !svc_match.input.status().is_success() {
207            tracing::debug!(
208                "no res http proxy connect match: target = {:?}; http version = {:?}",
209                svc_match.input.extensions().get_ref::<ConnectorTarget>(),
210                svc_match.input.version(),
211            );
212            return Ok(svc_match);
213        }
214
215        let proxy_target = svc_match.input.extensions().get_ref::<ConnectorTarget>();
216        let proxy_credential_info =
217            svc_match
218                .input
219                .extensions()
220                .get_ref()
221                .map(|DpiProxyCredential(c)| match c {
222                    ProxyCredential::Basic(user_pass) => ("basic", user_pass.username()),
223                    ProxyCredential::Bearer(_) => ("bearer", "***"),
224                });
225        tracing::debug!(
226            "response matched by (HTTP) proxy request: proxy target = {proxy_target:?}; credials = {proxy_credential_info:?}"
227        );
228
229        svc_match.service = Some(relay_svc);
230
231        Ok(svc_match)
232    }
233}
rama_http/layer/upgrade/http_proxy_connect/service_matcher.rs

rama_http/layer/upgrade/http_proxy_connect/
service_matcher.rs
