relay_core_lib/proxy/
http.rs

1use std::convert::Infallible;
2use std::net::SocketAddr;
3use std::sync::Arc;
4use tokio::sync::{mpsc::Sender, watch};
5
6use hyper::{Request, Response, Method, StatusCode};
7use hyper::body::{Bytes, Incoming, Body};
8use http_body_util::{BodyExt, Full};
9use relay_core_api::flow::{FlowUpdate, Layer, Direction};
10use relay_core_api::policy::ProxyPolicy;
11use crate::interceptor::{Interceptor, InterceptionResult, RequestAction, ResponseAction, HttpBody, BoxError};
12use crate::tls::CertificateAuthority;
13use crate::proxy::http_utils::{
14    create_initial_flow, 
15    mock_to_response, parse_request_meta, create_error_response, HttpsClient,
16    build_forward_request, update_flow_with_response_headers,
17};
18use crate::proxy::tunnel;
19use crate::proxy::websocket::handle_websocket_handshake;
20use crate::capture::loop_detection::LoopDetector;
21use crate::proxy::tap::TapBody;
22
23/// Main entry point for HTTP Proxy handling
24#[allow(clippy::too_many_arguments)]
25pub async fn handle_request(
26    req: Request<Incoming>,
27    client_addr: SocketAddr,
28    on_flow: Sender<FlowUpdate>,
29    ca: Arc<CertificateAuthority>,
30    client: Arc<HttpsClient>,
31    interceptor: Arc<dyn Interceptor>,
32    target_addr: Option<SocketAddr>,
33    policy_rx: watch::Receiver<ProxyPolicy>,
34    loop_detector: Arc<LoopDetector>,
35) -> Result<Response<HttpBody>, Infallible>
36{
37    if req.method() == Method::CONNECT {
38        // Handle CONNECT (HTTPS Tunnel)
39        // Extract host from authority
40        let host = if let Some(authority) = req.uri().authority() {
41            authority.to_string()
42        } else {
43            // Fallback: try to get from Host header
44             req.headers().get("Host")
45                .and_then(|v| v.to_str().ok())
46                .map(|s| s.to_string())
47                .unwrap_or_else(|| "unknown".to_string())
48        };
49
50        if host == "unknown" {
51             return Ok(create_error_response(StatusCode::BAD_REQUEST, "CONNECT must have authority"));
52        }
53
54        let loop_detector = loop_detector.clone();
55        let policy_rx = policy_rx.clone();
56
57        tokio::task::spawn(async move {
58            match hyper::upgrade::on(req).await {
59                Ok(upgraded) => {
60                    if let Err(e) = tunnel::handle_tunnel(
61                        upgraded,
62                        host,
63                        client_addr,
64                        ca,
65                        on_flow,
66                        client,
67                        interceptor,
68                        policy_rx,
69                        target_addr,
70                        loop_detector,
71                    ).await {
72                        tracing::error!("Tunnel error: {}", e);
73                    }
74                },
75                Err(e) => tracing::error!("Upgrade error: {}", e),
76            }
77        });
78        return Ok(Response::new(Full::new(Bytes::new()).map_err(|e| e.into()).boxed()));
79    }
80
81    // Handle Standard HTTP / WebSocket
82    handle_http_request(req, client_addr, on_flow, client, interceptor, false, policy_rx, target_addr, loop_detector).await
83}
84
85#[allow(clippy::too_many_arguments)]
86pub(crate) async fn handle_http_request<B>(
87    req: Request<B>,
88    client_addr: SocketAddr,
89    on_flow: Sender<FlowUpdate>,
90    client: Arc<HttpsClient>,
91    interceptor: Arc<dyn Interceptor>,
92    is_mitm: bool,
93    policy_rx: watch::Receiver<ProxyPolicy>,
94    target_addr: Option<SocketAddr>,
95    loop_detector: Arc<LoopDetector>,
96) -> Result<Response<HttpBody>, Infallible>
97where
98    B: Body + Send + Sync + Unpin + 'static,
99    B::Data: Send + Into<Bytes>,
100    B::Error: Into<BoxError>,
101{
102    let policy = policy_rx.borrow().clone();
103    
104    // Check Content-Length against policy
105    if let Some(cl) = req.headers().get(hyper::header::CONTENT_LENGTH)
106        && let Ok(len) = cl.to_str().unwrap_or_default().parse::<usize>()
107            && len > policy.max_body_size {
108                return Ok(create_error_response(StatusCode::PAYLOAD_TOO_LARGE, "Request body too large"));
109            }
110
111    // Create Flow
112    let meta = parse_request_meta(&req, is_mitm);
113    
114    // Note: We don't read body here for streaming support
115    let mut flow = create_initial_flow(meta, None, client_addr, is_mitm, false);
116    
117    // Check for WebSocket
118    if hyper_tungstenite::is_upgrade_request(&req) {
119        return handle_websocket_handshake(req, client_addr, on_flow, client, interceptor, is_mitm, policy_rx, target_addr, loop_detector).await;
120    }
121
122    if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
123        tracing::error!("Failed to send flow update: {}", e);
124    }
125
126    // Phase 1: Request Headers Interception
127    match interceptor.on_request_headers(&mut flow).await {
128        InterceptionResult::Continue => {},
129        InterceptionResult::Drop => {
130             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
131                 tracing::error!("Failed to send flow update on drop: {}", e);
132             }
133             return Ok(create_error_response(StatusCode::FORBIDDEN, "Request dropped by policy"));
134        },
135        InterceptionResult::MockResponse(resp) => {
136             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
137                 tracing::error!("Failed to send flow update on mock: {}", e);
138             }
139             return Ok(mock_to_response(resp));
140        },
141        InterceptionResult::ModifiedRequest(_) => {},
142        InterceptionResult::ModifiedResponse(res) => {
143             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
144                 tracing::error!("Failed to send flow update on modified response: {}", e);
145             }
146             return Ok(mock_to_response(res));
147        },
148        _ => {}
149    }
150
151    // Phase 2: Request Body Streaming & Interception
152    let (parts, body) = req.into_parts();
153    let body: HttpBody = body.map_frame(|f| f.map_data(|d| d.into())).map_err(|e| e.into()).boxed();
154    
155    // Wrap in TapBody for streaming visualization BEFORE interception
156    let req_headers = if let Layer::Http(http) = &flow.layer {
157        http.request.headers.clone()
158    } else {
159        vec![]
160    };
161
162    let tap_body = TapBody::new(
163        body,
164        flow.id.to_string(),
165        on_flow.clone(),
166        Direction::ClientToServer,
167        policy.max_body_size,
168        req_headers,
169    );
170    let mut current_body = tap_body.boxed();
171    
172    match interceptor.on_request(&mut flow, current_body).await {
173        Ok(RequestAction::Continue(new_body)) => {
174            current_body = new_body;
175        },
176        Ok(RequestAction::Drop) => {
177             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
178                 tracing::error!("Failed to send flow update on request drop: {}", e);
179             }
180             return Ok(create_error_response(StatusCode::FORBIDDEN, "Request dropped by interceptor"));
181        },
182        Ok(RequestAction::MockResponse(res)) => {
183             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
184                 tracing::error!("Failed to send flow update on request mock: {}", e);
185             }
186             let (parts, body) = res.into_parts();
187             return Ok(Response::from_parts(parts, body));
188        },
189        Err(e) => {
190             tracing::error!("Interceptor error on_request: {}", e);
191             return Ok(create_error_response(StatusCode::INTERNAL_SERVER_ERROR, format!("Interceptor Error: {}", e)));
192        }
193    }
194    
195    let forward_req = match build_forward_request(&mut flow, current_body, &parts, target_addr, &policy, &loop_detector) {
196        Ok(req) => req,
197        Err(res) => return Ok(res),
198    };
199    
200    // Send Request
201    let res = match tokio::time::timeout(std::time::Duration::from_millis(policy.request_timeout_ms), client.request(forward_req)).await {
202        Ok(Ok(res)) => res,
203        Ok(Err(e)) => {
204            tracing::error!("Upstream request failed: {}", e);
205             if let Layer::Http(http) = &mut flow.layer {
206                http.error = Some(format!("Upstream Error: {}", e));
207            }
208            if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
209                tracing::error!("Failed to send flow update on upstream error: {}", e);
210            }
211            return Ok(create_error_response(StatusCode::BAD_GATEWAY, format!("Upstream Error: {}", e)));
212        },
213        Err(_) => {
214            tracing::error!("Upstream request timed out");
215             if let Layer::Http(http) = &mut flow.layer {
216                http.error = Some("Upstream Request Timed Out".to_string());
217            }
218            if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
219                tracing::error!("Failed to send flow update on upstream timeout: {}", e);
220            }
221            return Ok(create_error_response(StatusCode::GATEWAY_TIMEOUT, "Upstream Request Timed Out"));
222        }
223    };
224    
225    // Phase 3: Response Headers Interception
226    let (mut res_parts, res_body) = res.into_parts();
227
228    // Apply QUIC Downgrade
229    apply_quic_downgrade(&mut res_parts, &mut flow, &policy);
230    
231    update_flow_with_response_headers(&mut flow, res_parts.status, res_parts.version, &res_parts.headers);
232    
233    match interceptor.on_response_headers(&mut flow).await {
234        InterceptionResult::Continue => {},
235        InterceptionResult::Drop => {
236             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
237                 tracing::error!("Failed to send flow update on response drop: {}", e);
238             }
239             return Ok(create_error_response(StatusCode::FORBIDDEN, "Response dropped by policy"));
240        },
241        InterceptionResult::MockResponse(resp) => {
242             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
243                 tracing::error!("Failed to send flow update on response mock: {}", e);
244             }
245             return Ok(mock_to_response(resp));
246        },
247        InterceptionResult::ModifiedResponse(resp) => {
248             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
249                 tracing::error!("Failed to send flow update on response modification: {}", e);
250             }
251             return Ok(mock_to_response(resp));
252        },
253        _ => {}
254    }
255    
256    // Phase 4: Response Body Streaming & Interception
257    let res_body: HttpBody = res_body.map_frame(|f| f.map_data(|d| d)).map_err(|e| e.into()).boxed();
258    
259    // Wrap in TapBody for streaming visualization BEFORE interception
260    let res_headers = if let Layer::Http(http) = &flow.layer {
261        http.response.as_ref().map(|r| r.headers.clone()).unwrap_or_default()
262    } else {
263        vec![]
264    };
265
266    let tap_res_body = TapBody::new(
267        res_body,
268        flow.id.to_string(),
269        on_flow.clone(),
270        Direction::ServerToClient,
271        policy.max_body_size,
272        res_headers,
273    );
274    let mut current_res_body = tap_res_body.boxed();
275
276    match interceptor.on_response(&mut flow, current_res_body).await {
277        Ok(ResponseAction::Continue(new_body)) => {
278            current_res_body = new_body;
279        },
280        Ok(ResponseAction::Drop) => {
281             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
282                 tracing::error!("Failed to send flow update on response body drop: {}", e);
283             }
284             return Ok(create_error_response(StatusCode::FORBIDDEN, "Response dropped by interceptor"));
285        },
286        Ok(ResponseAction::ModifiedResponse(res)) => {
287             if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
288                 tracing::error!("Failed to send flow update on response body modification: {}", e);
289             }
290             let (parts, body) = res.into_parts();
291             return Ok(Response::from_parts(parts, body));
292        },
293        Err(e) => {
294             tracing::error!("Interceptor error on_response: {}", e);
295             return Ok(create_error_response(StatusCode::INTERNAL_SERVER_ERROR, format!("Interceptor Error: {}", e)));
296        }
297    }
298
299    if let Err(e) = on_flow.send(FlowUpdate::Full(Box::new(flow.clone()))).await {
300        tracing::error!("Failed to send final flow update: {}", e);
301    }
302    
303    Ok(Response::from_parts(res_parts, current_res_body))
304}
305
306pub(crate) fn apply_quic_downgrade(parts: &mut hyper::http::response::Parts, flow: &mut relay_core_api::flow::Flow, policy: &ProxyPolicy) {
307    use relay_core_api::policy::QuicMode;
308    if policy.quic_mode == QuicMode::Downgrade {
309         if parts.headers.remove("Alt-Svc").is_some() {
310             flow.tags.push("quic-downgraded".to_string());
311         }
312         if policy.quic_downgrade_clear_cache {
313             parts.headers.insert("Clear-Site-Data", hyper::header::HeaderValue::from_static("\"cache\""));
314        }
315    }
316}
317
318#[cfg(test)]
319mod http_tests;
relay_core_lib/proxy/http.rs

relay_core_lib/proxy/
http.rs
