foxy/server/
mod.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! HTTP server implementation for Foxy.
6//!
7//! The server is a *thin* wrapper around **hyper-util**.  It owns the
8//! listening socket(s) and translates between Hyper's body types and the
9//! internal [`ProxyRequest`] / [`ProxyResponse`] generics that the core uses.
10//!
11//! **Protocol support**  
12//! Uses `hyper_util::server::conn::auto::Builder`, so the same
13//! connection transparently handles both HTTP/1.1 *and* HTTP/2.
14//!
15//! ## Body streaming
16//! Inbound bodies are **streamed** straight into the upstream connection; no
17//! intermediate buffering beyond the configured `server.body_limit` takes
18//! place.  This prevents unbounded memory usage when clients upload large
19//! files but still gives you a safety-valve.
20
21#[cfg(test)]
22mod tests;
23
24use std::sync::Arc;
25use std::net::SocketAddr;
26use std::convert::Infallible;
27use tokio::sync::RwLock;
28use hyper::body::Incoming;
29use hyper::{Request, Response};
30use hyper_util::server::conn::auto::Builder as AutoBuilder;
31use hyper_util::rt::TokioExecutor;
32use hyper::service::service_fn;
33use hyper_util::rt::TokioIo;
34use bytes::Bytes;
35use futures_util::TryStreamExt;
36use http_body_util::{BodyExt, Full};
37use reqwest::Body;
38use serde::{Serialize, Deserialize};
39use log::{debug};
40
41use crate::core::{ProxyCore, ProxyRequest, ProxyResponse, ProxyError, HttpMethod, RequestContext};
42
43/// Configuration for the HTTP server.
44#[derive(Debug, Clone, Serialize, Deserialize)]
45pub struct ServerConfig {
46    /// Host to bind to
47    #[serde(default = "default_host")]
48    pub host: String,
49
50    /// Port to listen on
51    #[serde(default = "default_port")]
52    pub port: u16,
53}
54
55fn default_host() -> String {
56    "127.0.0.1".to_string()
57}
58
59fn default_port() -> u16 {
60    8080
61}
62
63impl Default for ServerConfig {
64    fn default() -> Self {
65        Self {
66            host: default_host(),
67            port: default_port(),
68        }
69    }
70}
71
72/// HTTP server for the proxy.
73#[derive(Debug, Clone)]
74pub struct ProxyServer {
75    /// Server configuration
76    config: ServerConfig,
77    /// Proxy core
78    core: Arc<ProxyCore>,
79}
80
81impl ProxyServer {
82    /// Create a new proxy server with the given configuration and proxy core.
83    pub fn new(config: ServerConfig, core: Arc<ProxyCore>) -> Self {
84        Self { config, core }
85    }
86
87    /// Start the proxy server.
88    pub async fn start(&self) -> Result<(), ProxyError> {
89        let addr: SocketAddr = format!("{}:{}", self.config.host, self.config.port)
90            .parse()
91            .map_err(|e| ProxyError::Other(format!("Invalid server address: {}", e)))?;
92
93        let core = self.core.clone();
94
95        // Create a TCP listener
96        let listener = tokio::net::TcpListener::bind(addr).await
97            .map_err(|e| ProxyError::Other(format!("Failed to bind to address: {}", e)))?;
98        
99        log::info!("Foxy proxy server listening on http://{}", addr);
100        
101        // Accept connections
102        loop {
103            let (stream, remote_addr) = match listener.accept().await {
104                Ok(conn) => conn,
105                Err(e) => {
106                    log::error!("Failed to accept connection: {}", e);
107                    continue;
108                }
109            };
110            
111            let core = core.clone();
112            let client_ip = remote_addr.ip().to_string();
113            
114            // Spawn a task to handle the connection
115            tokio::spawn(async move {
116                let service = service_fn(move |req| {
117                    debug!("Incoming request over {:?}", req.version());
118                    let core = core.clone();
119                    let client_ip = client_ip.clone();
120                    handle_request(req, core, client_ip)
121                });
122
123                // Wrap the TcpStream with TokioIo for compatibility with hyper
124                let io = TokioIo::new(stream);
125
126                // Serve either HTTP/1.1 or HTTP/2, negotiated automatically.
127                if let Err(e) = AutoBuilder::new(TokioExecutor::new())
128                    .serve_connection(io, service)
129                    .await
130                {
131                    log::error!("Error serving connection: {}", e);
132                }
133            });
134        }
135    }
136}
137
138/// Convert a hyper request to a proxy request.
139async fn convert_hyper_request(
140    req: Request<Incoming>,
141    client_ip: String,
142) -> Result<ProxyRequest, ProxyError> {
143    let method = HttpMethod::from(req.method());
144    let uri = req.uri().clone();
145    let path = uri.path().to_owned();
146    let query = uri.query().map(|q| q.to_owned());
147    let headers = req.headers().clone();
148
149    // Incoming → Stream → reqwest::Body
150    let hyper_stream = req.into_body().into_data_stream();
151    let byte_stream = hyper_stream.map_ok(Bytes::from);
152    let body = reqwest::Body::wrap_stream(byte_stream);
153
154    Ok(ProxyRequest {
155        method,
156        path,
157        query,
158        headers,
159        body,
160        context: Arc::new(RwLock::new(RequestContext {
161            client_ip: Some(client_ip),
162            start_time: Some(std::time::Instant::now()),
163            attributes: std::collections::HashMap::new(),
164        })),
165    })
166}
167
168/// Convert a proxy response to a hyper response.
169fn convert_proxy_response(resp: ProxyResponse) -> Result<Response<Body>, ProxyError> {
170    let stream = resp
171        .body
172        .into_data_stream()
173        .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e));
174
175    let body = Body::wrap_stream(stream);
176
177    let mut builder = Response::builder().status(resp.status);
178    *builder
179        .headers_mut()
180        .ok_or_else(|| ProxyError::Other("unable to set headers".into()))? = resp.headers;
181
182    Ok(builder
183        .body(body)
184        .map_err(|e| ProxyError::Other(e.to_string()))?)
185}
186
187/// Handle an incoming HTTP request.
188async fn handle_request(
189    req: Request<Incoming>,
190    core: Arc<ProxyCore>,
191    client_ip: String,
192) -> Result<Response<Body>, Infallible> {
193    /* ---- convert Hyper → ProxyRequest ---- */
194    let proxy_req = match convert_hyper_request(req, client_ip).await {
195        Ok(r) => r,
196        Err(e) => {
197            log::error!("convert request: {e}");
198            return Ok(Response::builder()
199                .status(500)
200                .body(Body::from("Internal Server Error"))
201                .unwrap());
202        }
203    };
204
205    /* ---- core processing ---- */
206    match core.process_request(proxy_req).await {
207        Ok(proxy_resp) => match convert_proxy_response(proxy_resp) {
208            Ok(resp) => Ok(resp),
209            Err(e) => {
210                log::error!("convert response: {e}");
211                Ok(Response::builder()
212                    .status(500)
213                    .body(Body::from("Internal Server Error"))
214                    .unwrap())
215            }
216        },
217        Err(e) => {
218            log::error!("proxy error: {e}");
219            let (status, msg) = match e {
220                ProxyError::Timeout(d)     => (504, format!("Gateway Timeout after {d:?}")),
221                ProxyError::RoutingError(_) => (404, "Route not found".into()),
222                _                           => (500, "Internal Server Error".into()),
223            };
224            Ok(Response::builder()
225                .status(status)
226                .body(Body::from(msg))
227                .unwrap())
228        }
229    }
230}
foxy/server/mod.rs

foxy/server/
mod.rs
