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 std::borrow::Cow;
19use thiserror::Error;
20use crate::security::{ProviderConfig, SecurityChain, SecurityProvider, SecurityStage};
21use tokio::sync::RwLock;
22use tokio::time::timeout;
23use serde::{Serialize, Deserialize};
24
25use crate::config::Config;
26
27#[cfg(feature = "opentelemetry")]
28use opentelemetry::{
29    global,
30    trace::Tracer,
31    KeyValue,
32    Context,
33    context::FutureExt,
34    trace::{Span, SpanBuilder, SpanKind, TraceContextExt, Status}
35};
36#[cfg(feature = "opentelemetry")]
37use opentelemetry_http::HeaderInjector;
38#[cfg(feature = "opentelemetry")]
39use opentelemetry_semantic_conventions::attribute::HTTP_RESPONSE_STATUS_CODE;
40use crate::{debug_fmt, error_fmt, info_fmt, trace_fmt, warn_fmt};
41
42/// Errors that can occur during proxy operations.
43#[derive(Error, Debug)]
44pub enum ProxyError {
45    /// HTTP client error
46    #[error("HTTP client error: {0}")]
47    ClientError(#[from] reqwest::Error),
48
49    /// IO error
50    #[error("IO error: {0}")]
51    IoError(#[from] std::io::Error),
52
53    /// Timeout error
54    #[error("request timed out after {0:?}")]
55    Timeout(Duration),
56
57    /// Router error
58    #[error("routing error: {0}")]
59    RoutingError(String),
60
61    /// Filter error
62    #[error("filter error: {0}")]
63    FilterError(String),
64
65    /// Configuration error
66    #[error("configuration error: {0}")]
67    ConfigError(String),
68
69    /// Security provider error
70    #[error("security error: {0}")]
71    SecurityError(String),
72
73    /// Generic error
74    #[error("{0}")]
75    Other(String),
76}
77
78impl From<crate::config::error::ConfigError> for ProxyError {
79    fn from(err: crate::config::error::ConfigError) -> Self {
80        ProxyError::ConfigError(err.to_string())
81    }
82}
83
84impl From<globset::Error> for ProxyError {
85    fn from(e: globset::Error) -> Self {
86        ProxyError::SecurityError(e.to_string())
87    }
88}
89
90impl From<jsonwebtoken::errors::Error> for ProxyError {
91    fn from(e: jsonwebtoken::errors::Error) -> Self {
92        ProxyError::SecurityError(e.to_string())
93    }
94}
95
96/// HTTP methods supported by the proxy.
97#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
98#[serde(rename_all = "UPPERCASE")]
99pub enum HttpMethod {
100    Get,
101    Post,
102    Put,
103    Delete,
104    Head,
105    Options,
106    Patch,
107    Trace,
108    Connect,
109}
110
111impl fmt::Display for HttpMethod {
112    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
113        match self {
114            HttpMethod::Get => write!(f, "GET"),
115            HttpMethod::Post => write!(f, "POST"),
116            HttpMethod::Put => write!(f, "PUT"),
117            HttpMethod::Delete => write!(f, "DELETE"),
118            HttpMethod::Head => write!(f, "HEAD"),
119            HttpMethod::Options => write!(f, "OPTIONS"),
120            HttpMethod::Patch => write!(f, "PATCH"),
121            HttpMethod::Trace => write!(f, "TRACE"),
122            HttpMethod::Connect => write!(f, "CONNECT"),
123        }
124    }
125}
126
127impl From<&reqwest::Method> for HttpMethod {
128    fn from(method: &reqwest::Method) -> Self {
129        match *method {
130            reqwest::Method::GET => HttpMethod::Get,
131            reqwest::Method::POST => HttpMethod::Post,
132            reqwest::Method::PUT => HttpMethod::Put,
133            reqwest::Method::DELETE => HttpMethod::Delete,
134            reqwest::Method::HEAD => HttpMethod::Head,
135            reqwest::Method::OPTIONS => HttpMethod::Options,
136            reqwest::Method::PATCH => HttpMethod::Patch,
137            reqwest::Method::TRACE => HttpMethod::Trace,
138            reqwest::Method::CONNECT => HttpMethod::Connect,
139            _ => HttpMethod::Get, // Default to GET for unsupported methods
140        }
141    }
142}
143
144impl From<HttpMethod> for reqwest::Method {
145    fn from(method: HttpMethod) -> Self {
146        match method {
147            HttpMethod::Get => reqwest::Method::GET,
148            HttpMethod::Post => reqwest::Method::POST,
149            HttpMethod::Put => reqwest::Method::PUT,
150            HttpMethod::Delete => reqwest::Method::DELETE,
151            HttpMethod::Head => reqwest::Method::HEAD,
152            HttpMethod::Options => reqwest::Method::OPTIONS,
153            HttpMethod::Patch => reqwest::Method::PATCH,
154            HttpMethod::Trace => reqwest::Method::TRACE,
155            HttpMethod::Connect => reqwest::Method::CONNECT,
156        }
157    }
158}
159
160/// Represents an HTTP request that can be processed by the proxy.
161#[derive(Debug)]
162pub struct ProxyRequest {
163    pub method: HttpMethod,
164    pub path: String,
165    pub query: Option<String>,
166    pub headers: reqwest::header::HeaderMap,
167    pub body: reqwest::Body,
168    pub context: Arc<RwLock<RequestContext>>,
169    pub custom_target: Option<String>,
170}
171
172impl Clone for ProxyRequest {
173    fn clone(&self) -> Self {
174        // A streaming body can't be duplicated.  Give filters an empty one.
175        Self {
176            method:   self.method,
177            path:     self.path.clone(),
178            query:    self.query.clone(),
179            headers:  self.headers.clone(),
180            body:     reqwest::Body::from(""),
181            context:  self.context.clone(),
182            custom_target: self.custom_target.clone(),
183        }
184    }
185}
186
187/// Represents an HTTP response returned by the proxy.
188#[derive(Debug)]
189pub struct ProxyResponse {
190    pub status: u16,
191    pub headers: reqwest::header::HeaderMap,
192    pub body: reqwest::Body,
193    pub context: Arc<RwLock<ResponseContext>>,
194}
195
196/// Context data that can be attached to a request and accessed by filters.
197#[derive(Debug, Default, Clone)]
198pub struct RequestContext {
199    /// The original client's IP address
200    pub client_ip: Option<String>,
201    /// The start time of the request
202    pub start_time: Option<std::time::Instant>,
203    /// Custom attributes that can be set by filters
204    pub attributes: std::collections::HashMap<String, serde_json::Value>,
205}
206
207/// Context data that can be attached to a response and accessed by filters.
208#[derive(Debug, Default, Clone)]
209pub struct ResponseContext {
210    /// The time when the response was received from the target
211    pub receive_time: Option<std::time::Instant>,
212    /// Custom attributes that can be set by filters
213    pub attributes: std::collections::HashMap<String, serde_json::Value>,
214}
215
216/// Core proxy server implementation.
217#[derive(Debug)]
218pub struct ProxyCore {
219    /// Configuration for the proxy
220    pub config: Arc<Config>,
221    /// HTTP client for making outbound requests
222    pub client: reqwest::Client,
223    /// Router for matching requests to routes
224    pub router: Arc<dyn Router>,
225    /// Global filters that apply to all routes
226    pub global_filters: Arc<RwLock<Vec<Arc<dyn Filter>>>>,
227    /// Security chain that applies to all routes
228    pub security_chain: Arc<RwLock<SecurityChain>>,
229}
230
231impl ProxyCore {
232    /// Create a new proxy core with the given configuration and router.
233    pub async fn new(config: Arc<Config>, router: Arc<dyn Router>) -> Result<Self, ProxyError> {
234        // Configure the HTTP client based on the configuration
235        let timeout_secs: u64 = config.get_or_default("proxy.timeout", 30_u64)?;
236
237        let client_builder = reqwest::Client::builder();
238        let client = client_builder
239            .timeout(Duration::from_secs(timeout_secs))
240            .build()
241            .map_err(ProxyError::ClientError)?;
242
243        let actual_security_config: Vec<ProviderConfig> = match config.get("proxy.security_chain") {
244            Ok(Some(sc)) => sc,
245            Ok(None) => Vec::new(), // No security chain configured
246            Err(e) => {
247                warn_fmt!("Core", "Could not parse 'proxy.security_chain', defaulting to empty: {}", e);
248                Vec::new() // Default to empty on error
249            }
250        };
251
252        let security_chain = SecurityChain::from_configs(actual_security_config).await?;
253
254        Ok(Self {
255            config,
256            client,
257            router,
258            global_filters: Arc::new(RwLock::new(Vec::new())),
259            security_chain: Arc::new(RwLock::new(security_chain)),
260        })
261    }
262
263    /// Add a global filter.
264    pub async fn add_global_filter(&self, filter: Arc<dyn Filter>) {
265        let mut filters = self.global_filters.write().await;
266        filters.push(filter);
267    }
268
269    /// Add a security filter to the chain.
270    pub async fn add_security_provider(&self, p: Arc<dyn SecurityProvider>) {
271        self.security_chain.write().await.add(p);
272    }
273
274    /// Process a request through the proxy.
275    pub async fn process_request(
276        &self,
277        request: ProxyRequest,
278        #[cfg(feature = "opentelemetry")]
279        parent_context: Option<Context>,
280    ) -> Result<ProxyResponse, ProxyError> {
281        let overall_start = Instant::now();
282        let method = request.method.to_string();
283        let path = request.path.clone();
284
285        trace_fmt!("Core", "Processing request: {} {}", method, path);
286
287        #[cfg(feature = "opentelemetry")]
288        let span_context = {
289            let parent  = parent_context
290                .as_ref()
291                .cloned()
292                .unwrap_or_else(Context::current);
293
294            let mut span = global::tracer("foxy::proxy")
295                .build_with_context(SpanBuilder {
296                    name: Cow::from(format!("{method} {path}")),
297                    span_kind: Some(SpanKind::Client),
298                    ..Default::default()
299                }, &parent);
300
301            let span_context = &Context::current_with_span(span);
302            span_context.clone()
303        };
304
305        /* ---------- Security chain pre auth ---------- */
306        let mut request = match self.security_chain.read().await.apply_pre(request).await {
307            Ok(req) => {
308                trace_fmt!("Core", "Security pre-auth passed for {} {}", method, path);
309                req
310            },
311            Err(e) => {
312                warn_fmt!("Core", "Security pre-auth failed for {} {}: {}", method, path, e);
313
314                #[cfg(feature = "opentelemetry")]
315                {
316                    span_context.span().set_status(Status::Error {description: Cow::from(e.to_string()) });
317                    span_context.span().end();
318                }
319
320                return Err(e);
321            }
322        };
323
324        /* ---------- PRE-filters ---------- */
325        for f in self.global_filters.read().await.iter() {
326            if f.filter_type().is_pre() || f.filter_type().is_both() {
327                trace_fmt!("Core", "Applying global pre-filter: {}", f.name());
328                match f.pre_filter(request).await {
329                    Ok(req) => request = req,
330                    Err(e) => {
331                        error_fmt!("Core", "Global pre-filter '{}' failed: {}", f.name(), e);
332
333                        #[cfg(feature = "opentelemetry")]
334                        {
335                            span_context.span().set_status(Status::Error {description: Cow::from(e.to_string()) });
336                            span_context.span().end();
337                        }
338
339                        return Err(e);
340                    }
341                }
342            }
343        }
344
345        let mut route = match self.router.route(&request).await {
346            Ok(r) => {
347                debug_fmt!("Core", "Request {} {} matched route: {}", method, path, r.id);
348                r
349            },
350            Err(e) => {
351                warn_fmt!("Core", "No route found for {} {}: {}", method, path, e);
352
353                #[cfg(feature = "opentelemetry")]
354                {
355                    span_context.span().set_status(Status::Error {description: Cow::from(e.to_string()) });
356                    span_context.span().end();
357                }
358
359                return Err(e);
360            }
361        };
362
363        let route_filters = route.filters.clone().unwrap_or_default();
364        for f in &route_filters {
365            if f.filter_type().is_pre() || f.filter_type().is_both() {
366                trace_fmt!("Core", "Applying route pre-filter: {}", f.name());
367                match f.pre_filter(request).await {
368                    Ok(req) => request = req,
369                    Err(e) => {
370                        error_fmt!("Core", "Route pre-filter '{}' failed: {}", f.name(), e);
371                        return Err(e);
372                    }
373                }
374            }
375        }
376
377        info_fmt!("Core", "Initial target: {}", route.target_base_url);
378
379        /* ---------- build outbound req ---------- */
380        if(request.custom_target.is_some()){
381            debug_fmt!("Core", "Attempting to dynamically set target base Url: {}", route.target_base_url);
382            route.target_base_url = request.custom_target.clone().unwrap();
383            debug_fmt!("Core", "Dynamically set target base Url to: {}", route.target_base_url);
384        }
385        
386        let url = format!("{}{}", route.target_base_url, request.path);
387        debug_fmt!("Core", "Forwarding to target: {}", url);
388        let outbound_body = mem::replace(&mut request.body, reqwest::Body::from(""));
389
390        let mut outbound_headers = request.headers.clone();
391        #[cfg(feature = "opentelemetry")]
392        {
393            span_context.span().set_attribute(KeyValue::new("target", url.clone()));
394
395            global::get_text_map_propagator(|prop| {
396                prop.inject_context(&span_context, &mut HeaderInjector(&mut outbound_headers));
397            });
398        }
399
400        let mut builder = self
401            .client
402            .request(request.method.into(), &url)
403            .headers(outbound_headers)
404            .body(outbound_body);
405
406        if let Some(q) = &request.query {
407            builder = builder.query(&[(q, "")]);
408        }
409
410        /* ---------- send with timeout ---------- */
411        // The client already has a timeout configured.
412        // If a per-request timeout from a TimeoutFilter is present in context, it should override.
413        // For now, let's rely on the client's global timeout.
414        // A more advanced implementation could check request.context for a specific timeout.
415        let request_specific_timeout_ms: Option<u64> = request.context.read().await
416            .attributes
417            .get("timeout_ms")
418            .and_then(|v| v.as_u64());
419
420        let timeout_duration = if let Some(ms) = request_specific_timeout_ms {
421            Duration::from_millis(ms)
422        } else {
423            // Fallback to the client's configured timeout, or a default if not available.
424            // However, the client *is* configured with a timeout in ProxyCore::new.
425            // For consistency, we could fetch it from config again or store it in ProxyCore.
426            // For now, let's assume the client's timeout is sufficient.
427            // If we want to re-fetch:
428            self.config.get_or_default("proxy.timeout", 30_u64).map(Duration::from_secs)?
429        };
430
431        let upstream_start = Instant::now();
432        trace_fmt!("Core", "Sending request to upstream with timeout: {:?}", timeout_duration);
433
434        let resp = match timeout(timeout_duration, builder.send()).await {
435            Ok(result) => match result {
436                Ok(response) => response,
437                Err(e) => {
438                    error_fmt!("Core", "Upstream request failed: {}", e);
439
440                    #[cfg(feature = "opentelemetry")]
441                    {
442                        span_context.span().set_status(Status::Error {description: Cow::from(e.to_string()) });
443                        span_context.span().end();
444                    }
445
446                    return Err(ProxyError::ClientError(e));
447                }
448            },
449            Err(_) => {
450                warn_fmt!("Core", "Request to {} timed out after {:?}", url, timeout_duration);
451
452                #[cfg(feature = "opentelemetry")]
453                {
454                    span_context.span().set_status(Status::Error {description: Cow::from("Request timed out") });
455                    span_context.span().end();
456                }
457
458                return Err(ProxyError::Timeout(timeout_duration));
459            }
460        };
461
462        #[cfg(feature = "opentelemetry")]
463        {
464            let client_span = span_context.span();
465
466            client_span.set_attribute(KeyValue::new(
467                HTTP_RESPONSE_STATUS_CODE,
468                resp.status().as_u16() as i64,
469            ));
470            client_span.end();
471        }
472
473        let upstream_elapsed = upstream_start.elapsed();
474        trace_fmt!("Core", "Received response from upstream in {:?}", upstream_elapsed);
475
476        /* ---------- wrap streaming response ---------- */
477        let status = resp.status().as_u16();
478        let headers = resp.headers().clone();
479        let body = reqwest::Body::wrap_stream(resp.bytes_stream());
480
481        let mut proxy_resp = ProxyResponse {
482            status,
483            headers,
484            body,
485            context: Arc::new(RwLock::new(ResponseContext::default())),
486        };
487        proxy_resp.context.write().await.receive_time = Some(Instant::now());
488
489        debug_fmt!("Core", "Upstream responded with status: {}", status);
490
491        /* ---------- POST-filters ---------- */
492        for f in &route_filters {
493            if f.filter_type().is_post() || f.filter_type().is_both() {
494                trace_fmt!("Core", "Applying route post-filter: {}", f.name());
495                match f.post_filter(request.clone(), proxy_resp).await {
496                    Ok(resp) => proxy_resp = resp,
497                    Err(e) => {
498                        error_fmt!("Core", "Route post-filter '{}' failed: {}", f.name(), e);
499                        return Err(e);
500                    }
501                }
502            }
503        }
504
505        for f in self.global_filters.read().await.iter() {
506            if f.filter_type().is_post() || f.filter_type().is_both() {
507                trace_fmt!("Core", "Applying global post-filter: {}", f.name());
508                match f.post_filter(request.clone(), proxy_resp).await {
509                    Ok(resp) => proxy_resp = resp,
510                    Err(e) => {
511                        error_fmt!("Core", "Global post-filter '{}' failed: {}", f.name(), e);
512                        return Err(e);
513                    }
514                }
515            }
516        }
517
518        /* ---------- Security chain post auth ---------- */
519        proxy_resp = match self.security_chain.read().await.apply_post(request.clone(), proxy_resp).await {
520            Ok(resp) => {
521                trace_fmt!("Core", "Security post-auth passed for {} {}", method, path);
522                resp
523            },
524            Err(e) => {
525                warn_fmt!("Core", "Security post-auth failed for {} {}: {}", method, path, e);
526                return Err(e);
527            }
528        };
529
530        /* ---------- timing log ---------- */
531        let overall_elapsed = overall_start.elapsed();
532        let internal_elapsed = overall_elapsed.saturating_sub(upstream_elapsed);
533
534        debug_fmt!("Core", 
535            "[timing] {} {} -> {} | total={:?} upstream={:?} internal={:?}",
536            request.method,
537            request.path,
538            proxy_resp.status,
539            overall_elapsed,
540            upstream_elapsed,
541            internal_elapsed
542        );
543
544        Ok(proxy_resp)
545    }
546}
547
548/// Describes when a filter should be applied.
549#[derive(Debug, Clone, Copy, PartialEq, Eq)]
550pub enum FilterType {
551    /// Filter applied before the request is sent to the target
552    Pre,
553    /// Filter applied after the response is received from the target
554    Post,
555    /// Filter applied both before and after
556    Both,
557}
558
559impl FilterType {
560    /// Returns true if this is a pre-filter or both.
561    pub fn is_pre(&self) -> bool {
562        matches!(self, FilterType::Pre | FilterType::Both)
563    }
564
565    /// Returns true if this is a post-filter or both.
566    pub fn is_post(&self) -> bool {
567        matches!(self, FilterType::Post | FilterType::Both)
568    }
569
570    /// Returns true if this is both a pre and post filter.
571    pub fn is_both(&self) -> bool {
572        matches!(self, FilterType::Both)
573    }
574}
575
576/// A filter that processes requests and responses.
577#[async_trait::async_trait]
578pub trait Filter: fmt::Debug + Send + Sync {
579    /// Get the filter type.
580    fn filter_type(&self) -> FilterType;
581
582    /// Get the filter name.
583    fn name(&self) -> &str;
584
585    /// Process a request before it is sent to the target.
586    async fn pre_filter(&self, request: ProxyRequest) -> Result<ProxyRequest, ProxyError> {
587        // Default implementation: pass through the request unchanged
588        Ok(request)
589    }
590
591    /// Process a response after it is received from the target.
592    async fn post_filter(&self, _request: ProxyRequest, response: ProxyResponse) -> Result<ProxyResponse, ProxyError> {
593        // Default implementation: pass through the response unchanged
594        Ok(response)
595    }
596}
597
598/// A route that the proxy can forward requests to.
599#[derive(Debug, Clone)]
600pub struct Route {
601    /// The ID of the route (for logging and reference)
602    pub id: String,
603    /// The base URL of the target
604    pub target_base_url: String,
605    /// The path pattern that this route matches
606    pub path_pattern: String,
607    /// The filters that should be applied to this route
608    pub filters: Option<Vec<Arc<dyn Filter>>>,
609}
610
611/// A router that matches requests to routes.
612#[async_trait::async_trait]
613pub trait Router: fmt::Debug + Send + Sync {
614    /// Find a route for the given request.
615    async fn route(&self, request: &ProxyRequest) -> Result<Route, ProxyError>;
616
617    /// Get all routes managed by this router.
618    async fn get_routes(&self) -> Vec<Route>;
619
620    /// Add a new route to the router.
621    async fn add_route(&self, route: Route) -> Result<(), ProxyError>;
622
623    /// Remove a route from the router.
624    async fn remove_route(&self, route_id: &str) -> Result<(), ProxyError>;
625}
foxy/core/mod.rs

foxy/core/
mod.rs
