foxy/core/
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//! Core primitives – requests, responses, filters & routing.
6//!
7//! Everything that physically moves through the proxy pipeline is defined
8//! in this module.  No protocol-level logic lives here; that sits in
9//! `server.rs` (IO) and `filters.rs` (behaviour).
10
11#[cfg(test)]
12mod tests;
13
14use std::collections::HashMap;
15use std::sync::Arc;
16use std::time::{Duration, Instant};
17use std::{fmt, mem};
18use thiserror::Error;
19use crate::security::{ProviderConfig, SecurityChain, SecurityProvider, SecurityStage};
20use tokio::sync::RwLock;
21use tokio::time::timeout;
22use serde::{Serialize, Deserialize};
23
24use crate::config::Config;
25
26/// Errors that can occur during proxy operations.
27#[derive(Error, Debug)]
28pub enum ProxyError {
29    /// HTTP client error
30    #[error("HTTP client error: {0}")]
31    ClientError(#[from] reqwest::Error),
32
33    /// IO error
34    #[error("IO error: {0}")]
35    IoError(#[from] std::io::Error),
36
37    /// Timeout error
38    #[error("request timed out after {0:?}")]
39    Timeout(Duration),
40
41    /// Router error
42    #[error("routing error: {0}")]
43    RoutingError(String),
44
45    /// Filter error
46    #[error("filter error: {0}")]
47    FilterError(String),
48
49    /// Configuration error
50    #[error("configuration error: {0}")]
51    ConfigError(String),
52
53    /// Security provider error
54    #[error("security error: {0}")]
55    SecurityError(String),
56
57    /// Generic error
58    #[error("{0}")]
59    Other(String),
60}
61
62impl From<crate::config::error::ConfigError> for ProxyError {
63    fn from(err: crate::config::error::ConfigError) -> Self {
64        ProxyError::ConfigError(err.to_string())
65    }
66}
67
68impl From<globset::Error> for ProxyError {
69    fn from(e: globset::Error) -> Self {
70        ProxyError::SecurityError(e.to_string())
71    }
72}
73
74impl From<jsonwebtoken::errors::Error> for ProxyError {
75    fn from(e: jsonwebtoken::errors::Error) -> Self {
76        ProxyError::SecurityError(e.to_string())
77    }
78}
79
80/// HTTP methods supported by the proxy.
81#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
82#[serde(rename_all = "UPPERCASE")]
83pub enum HttpMethod {
84    Get,
85    Post,
86    Put,
87    Delete,
88    Head,
89    Options,
90    Patch,
91    Trace,
92    Connect,
93}
94
95impl fmt::Display for HttpMethod {
96    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
97        match self {
98            HttpMethod::Get => write!(f, "GET"),
99            HttpMethod::Post => write!(f, "POST"),
100            HttpMethod::Put => write!(f, "PUT"),
101            HttpMethod::Delete => write!(f, "DELETE"),
102            HttpMethod::Head => write!(f, "HEAD"),
103            HttpMethod::Options => write!(f, "OPTIONS"),
104            HttpMethod::Patch => write!(f, "PATCH"),
105            HttpMethod::Trace => write!(f, "TRACE"),
106            HttpMethod::Connect => write!(f, "CONNECT"),
107        }
108    }
109}
110
111impl From<&reqwest::Method> for HttpMethod {
112    fn from(method: &reqwest::Method) -> Self {
113        match *method {
114            reqwest::Method::GET => HttpMethod::Get,
115            reqwest::Method::POST => HttpMethod::Post,
116            reqwest::Method::PUT => HttpMethod::Put,
117            reqwest::Method::DELETE => HttpMethod::Delete,
118            reqwest::Method::HEAD => HttpMethod::Head,
119            reqwest::Method::OPTIONS => HttpMethod::Options,
120            reqwest::Method::PATCH => HttpMethod::Patch,
121            reqwest::Method::TRACE => HttpMethod::Trace,
122            reqwest::Method::CONNECT => HttpMethod::Connect,
123            _ => HttpMethod::Get, // Default to GET for unsupported methods
124        }
125    }
126}
127
128impl From<HttpMethod> for reqwest::Method {
129    fn from(method: HttpMethod) -> Self {
130        match method {
131            HttpMethod::Get => reqwest::Method::GET,
132            HttpMethod::Post => reqwest::Method::POST,
133            HttpMethod::Put => reqwest::Method::PUT,
134            HttpMethod::Delete => reqwest::Method::DELETE,
135            HttpMethod::Head => reqwest::Method::HEAD,
136            HttpMethod::Options => reqwest::Method::OPTIONS,
137            HttpMethod::Patch => reqwest::Method::PATCH,
138            HttpMethod::Trace => reqwest::Method::TRACE,
139            HttpMethod::Connect => reqwest::Method::CONNECT,
140        }
141    }
142}
143
144/// Represents an HTTP request that can be processed by the proxy.
145#[derive(Debug)]
146pub struct ProxyRequest {
147    pub method: HttpMethod,
148    pub path: String,
149    pub query: Option<String>,
150    pub headers: reqwest::header::HeaderMap,
151    pub body: reqwest::Body,
152    pub context: Arc<RwLock<RequestContext>>,
153}
154
155impl Clone for ProxyRequest {
156    fn clone(&self) -> Self {
157        // A streaming body can't be duplicated.  Give filters an empty one.
158        Self {
159            method:   self.method,
160            path:     self.path.clone(),
161            query:    self.query.clone(),
162            headers:  self.headers.clone(),
163            body:     reqwest::Body::from(""),
164            context:  self.context.clone(),
165        }
166    }
167}
168
169/// Represents an HTTP response returned by the proxy.
170#[derive(Debug)]
171pub struct ProxyResponse {
172    pub status: u16,
173    pub headers: reqwest::header::HeaderMap,
174    pub body: reqwest::Body,
175    pub context: Arc<RwLock<ResponseContext>>,
176}
177
178/// Context data that can be attached to a request and accessed by filters.
179#[derive(Debug, Default, Clone)]
180pub struct RequestContext {
181    /// The original client's IP address
182    pub client_ip: Option<String>,
183    /// The start time of the request
184    pub start_time: Option<std::time::Instant>,
185    /// Custom attributes that can be set by filters
186    pub attributes: std::collections::HashMap<String, serde_json::Value>,
187}
188
189/// Context data that can be attached to a response and accessed by filters.
190#[derive(Debug, Default, Clone)]
191pub struct ResponseContext {
192    /// The time when the response was received from the target
193    pub receive_time: Option<std::time::Instant>,
194    /// Custom attributes that can be set by filters
195    pub attributes: std::collections::HashMap<String, serde_json::Value>,
196}
197
198/// Core proxy server implementation.
199#[derive(Debug)]
200pub struct ProxyCore {
201    /// Configuration for the proxy
202    pub config: Arc<Config>,
203    /// HTTP client for making outbound requests
204    pub client: reqwest::Client,
205    /// Router for matching requests to routes
206    pub router: Arc<dyn Router>,
207    /// Global filters that apply to all routes
208    pub global_filters: Arc<RwLock<Vec<Arc<dyn Filter>>>>,
209    /// Security chain that applies to all routes
210    pub security_chain: Arc<RwLock<SecurityChain>>,
211}
212
213impl ProxyCore {
214    /// Create a new proxy core with the given configuration and router.
215    pub async fn new(config: Arc<Config>, router: Arc<dyn Router>) -> Result<Self, ProxyError> {
216        // Configure the HTTP client based on the configuration
217        let timeout_secs: u64 = config.get_or_default("proxy.timeout", 30)?;
218
219        let client = reqwest::Client::builder()
220            .timeout(Duration::from_secs(timeout_secs))
221            .build()
222            .map_err(ProxyError::ClientError)?;
223
224        let security_config = config
225            .get::<Vec<ProviderConfig>>("proxy.security_chain")
226            .unwrap_or_default();
227
228        let security_chain = SecurityChain::from_configs(
229            security_config.unwrap_or_default()
230        ).await?;
231
232        Ok(Self {
233            config,
234            client,
235            router,
236            global_filters: Arc::new(RwLock::new(Vec::new())),
237            security_chain: Arc::new(RwLock::new(security_chain)),
238        })
239    }
240
241    /// Add a global filter.
242    pub async fn add_global_filter(&self, filter: Arc<dyn Filter>) {
243        let mut filters = self.global_filters.write().await;
244        filters.push(filter);
245    }
246
247    /// Add a global OpenTelemetry filter if configured
248    #[cfg(feature = "opentelemetry")]
249    pub async fn add_opentelemetry_filter(&self, config: &crate::opentelemetry::OpenTelemetryConfig) -> Result<(), ProxyError> {
250        let filter = Arc::new(crate::opentelemetry::OpenTelemetryFilter::new(config.clone()));
251        self.add_global_filter(filter).await;
252        Ok(())
253    }
254
255    /// Add a security filter to the chain.
256    pub async fn add_security_provider(&self, p: Arc<dyn SecurityProvider>) {
257        self.security_chain.write().await.add(p);
258    }
259
260    /// Process a request through the proxy.
261    pub async fn process_request(
262        &self,
263        request: ProxyRequest,
264    ) -> Result<ProxyResponse, ProxyError> {
265        let overall_start = Instant::now();
266        let method = request.method.to_string();
267        let path = request.path.clone();
268
269        log::trace!("Processing request: {} {}", method, path);
270
271        /* ---------- Security chain pre auth ---------- */
272        let mut request = match self.security_chain.read().await.apply_pre(request).await {
273            Ok(req) => {
274                log::trace!("Security pre-auth passed for {} {}", method, path);
275                req
276            },
277            Err(e) => {
278                log::warn!("Security pre-auth failed for {} {}: {}", method, path, e);
279                return Err(e);
280            }
281        };
282
283        /* ---------- PRE-filters ---------- */
284        for f in self.global_filters.read().await.iter() {
285            if f.filter_type().is_pre() || f.filter_type().is_both() {
286                log::trace!("Applying global pre-filter: {}", f.name());
287                match f.pre_filter(request).await {
288                    Ok(req) => request = req,
289                    Err(e) => {
290                        log::error!("Global pre-filter '{}' failed: {}", f.name(), e);
291                        return Err(e);
292                    }
293                }
294            }
295        }
296        
297        let route = match self.router.route(&request).await {
298            Ok(r) => {
299                log::debug!("Request {} {} matched route: {}", method, path, r.id);
300                r
301            },
302            Err(e) => {
303                log::warn!("No route found for {} {}: {}", method, path, e);
304                return Err(e);
305            }
306        };
307        
308        let route_filters = route.filters.clone().unwrap_or_default();
309        for f in &route_filters {
310            if f.filter_type().is_pre() || f.filter_type().is_both() {
311                log::trace!("Applying route pre-filter: {}", f.name());
312                match f.pre_filter(request).await {
313                    Ok(req) => request = req,
314                    Err(e) => {
315                        log::error!("Route pre-filter '{}' failed: {}", f.name(), e);
316                        return Err(e);
317                    }
318                }
319            }
320        }
321
322        /* ---------- build outbound req ---------- */
323        let url = format!("{}{}", route.target_base_url, request.path);
324        log::debug!("Forwarding to target: {}", url);
325        let outbound_body = mem::replace(&mut request.body, reqwest::Body::from(""));
326
327        let mut builder = self
328            .client
329            .request(request.method.into(), &url)
330            .headers(request.headers.clone())
331            .body(outbound_body);
332
333        if let Some(q) = &request.query {
334            builder = builder.query(&[(q, "")]);
335        }
336
337        /* ---------- send with timeout ---------- */
338        let timeout_dur =
339            Duration::from_secs(self.config.get_or_default("proxy.timeout", 30).unwrap_or_else(|e| {
340                log::error!("Failed to get timeout config: {}", e);
341                30 // Default to 30 seconds on error
342            }));
343
344        let upstream_start = Instant::now();
345        log::trace!("Sending request to upstream with timeout: {:?}", timeout_dur);
346        
347        let resp = match timeout(timeout_dur, builder.send()).await {
348            Ok(result) => match result {
349                Ok(response) => response,
350                Err(e) => {
351                    log::error!("Upstream request failed: {}", e);
352                    return Err(ProxyError::ClientError(e));
353                }
354            },
355            Err(_) => {
356                log::warn!("Request to {} timed out after {:?}", url, timeout_dur);
357                return Err(ProxyError::Timeout(timeout_dur));
358            }
359        };
360        
361        let upstream_elapsed = upstream_start.elapsed();
362        log::trace!("Received response from upstream in {:?}", upstream_elapsed);
363
364        /* ---------- wrap streaming response ---------- */
365        let status = resp.status().as_u16();
366        let headers = resp.headers().clone();
367        let body = reqwest::Body::wrap_stream(resp.bytes_stream());
368
369        let mut proxy_resp = ProxyResponse {
370            status,
371            headers,
372            body,
373            context: Arc::new(RwLock::new(ResponseContext::default())),
374        };
375        proxy_resp.context.write().await.receive_time = Some(Instant::now());
376
377        log::debug!("Upstream responded with status: {}", status);
378
379        /* ---------- POST-filters ---------- */
380        for f in &route_filters {
381            if f.filter_type().is_post() || f.filter_type().is_both() {
382                log::trace!("Applying route post-filter: {}", f.name());
383                match f.post_filter(request.clone(), proxy_resp).await {
384                    Ok(resp) => proxy_resp = resp,
385                    Err(e) => {
386                        log::error!("Route post-filter '{}' failed: {}", f.name(), e);
387                        return Err(e);
388                    }
389                }
390            }
391        }
392        
393        for f in self.global_filters.read().await.iter() {
394            if f.filter_type().is_post() || f.filter_type().is_both() {
395                log::trace!("Applying global post-filter: {}", f.name());
396                match f.post_filter(request.clone(), proxy_resp).await {
397                    Ok(resp) => proxy_resp = resp,
398                    Err(e) => {
399                        log::error!("Global post-filter '{}' failed: {}", f.name(), e);
400                        return Err(e);
401                    }
402                }
403            }
404        }
405
406        /* ---------- Security chain post auth ---------- */
407        proxy_resp = match self.security_chain.read().await.apply_post(request.clone(), proxy_resp).await {
408            Ok(resp) => {
409                log::trace!("Security post-auth passed for {} {}", method, path);
410                resp
411            },
412            Err(e) => {
413                log::warn!("Security post-auth failed for {} {}: {}", method, path, e);
414                return Err(e);
415            }
416        };
417        
418        /* ---------- timing log ---------- */
419        let overall_elapsed = overall_start.elapsed();
420        let internal_elapsed = overall_elapsed.saturating_sub(upstream_elapsed);
421
422        log::debug!(
423            "[timing] {} {} -> {} | total={:?} upstream={:?} internal={:?}",
424            request.method,
425            request.path,
426            proxy_resp.status,
427            overall_elapsed,
428            upstream_elapsed,
429            internal_elapsed
430        );
431
432        Ok(proxy_resp)
433    }
434}
435
436/// Describes when a filter should be applied.
437#[derive(Debug, Clone, Copy, PartialEq, Eq)]
438pub enum FilterType {
439    /// Filter applied before the request is sent to the target
440    Pre,
441    /// Filter applied after the response is received from the target
442    Post,
443    /// Filter applied both before and after
444    Both,
445}
446
447impl FilterType {
448    /// Returns true if this is a pre-filter or both.
449    pub fn is_pre(&self) -> bool {
450        matches!(self, FilterType::Pre | FilterType::Both)
451    }
452
453    /// Returns true if this is a post-filter or both.
454    pub fn is_post(&self) -> bool {
455        matches!(self, FilterType::Post | FilterType::Both)
456    }
457
458    /// Returns true if this is both a pre and post filter.
459    pub fn is_both(&self) -> bool {
460        matches!(self, FilterType::Both)
461    }
462}
463
464/// A filter that processes requests and responses.
465#[async_trait::async_trait]
466pub trait Filter: fmt::Debug + Send + Sync {
467    /// Get the filter type.
468    fn filter_type(&self) -> FilterType;
469
470    /// Get the filter name.
471    fn name(&self) -> &str;
472
473    /// Process a request before it is sent to the target.
474    async fn pre_filter(&self, request: ProxyRequest) -> Result<ProxyRequest, ProxyError> {
475        // Default implementation: pass through the request unchanged
476        Ok(request)
477    }
478
479    /// Process a response after it is received from the target.
480    async fn post_filter(&self, _request: ProxyRequest, response: ProxyResponse) -> Result<ProxyResponse, ProxyError> {
481        // Default implementation: pass through the response unchanged
482        Ok(response)
483    }
484}
485
486/// A route that the proxy can forward requests to.
487#[derive(Debug, Clone)]
488pub struct Route {
489    /// The ID of the route (for logging and reference)
490    pub id: String,
491    /// The base URL of the target
492    pub target_base_url: String,
493    /// The path pattern that this route matches
494    pub path_pattern: String,
495    /// The filters that should be applied to this route
496    pub filters: Option<Vec<Arc<dyn Filter>>>,
497}
498
499/// A router that matches requests to routes.
500#[async_trait::async_trait]
501pub trait Router: fmt::Debug + Send + Sync {
502    /// Find a route for the given request.
503    async fn route(&self, request: &ProxyRequest) -> Result<Route, ProxyError>;
504
505    /// Get all routes managed by this router.
506    async fn get_routes(&self) -> Vec<Route>;
507
508    /// Add a new route to the router.
509    async fn add_route(&self, route: Route) -> Result<(), ProxyError>;
510
511    /// Remove a route from the router.
512    async fn remove_route(&self, route_id: &str) -> Result<(), ProxyError>;
513}
foxy/core/mod.rs

foxy/core/
mod.rs
