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

foxy/core/
mod.rs
