diff --git a/src/auth/jwt.rs b/src/auth/jwt.rs index 2c2fd0e..1e3edf8 100644 --- a/src/auth/jwt.rs +++ b/src/auth/jwt.rs @@ -1,9 +1,51 @@ use crate::auth::types::AuthValidator; -use crate::utils::jwt::{check_jwt, JWT_TOKEN}; +use ahash::AHasher; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use base64::Engine; +use jsonwebtoken::{Algorithm, DecodingKey, Validation}; +use moka::sync::Cache; +use moka::Expiry; use pingora_proxy::Session; +use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use std::env; +use std::hash::{Hash, Hasher}; +use std::sync::{Arc, LazyLock}; +use std::time::{Duration, Instant, SystemTime}; use urlencoding::decode; +#[derive(Debug, Serialize, Deserialize)] +pub struct Claims { + pub master_key: String, + pub owner: String, + pub exp: u64, + pub random: Option, +} + +#[derive(Debug, Deserialize)] +struct Expired { + exp: Option, +} + +static JWT_VALIDATION: LazyLock = LazyLock::new(|| Validation::new(Algorithm::HS256)); + +pub static JWT_TOKEN: LazyLock>> = LazyLock::new(|| match env::var("JWT_KEY") { + Ok(key) if !key.is_empty() => Some(Arc::from(key.as_str())), + _ => None, +}); + +static JWT_CACHE: LazyLock> = LazyLock::new(|| Cache::builder().max_capacity(100_000).expire_after(JwtExpiry).build()); +struct JwtExpiry; +impl Expiry for JwtExpiry { + fn expire_after_create(&self, _key: &u64, value: &u64, _current_time: Instant) -> Option { + let now = SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + if *value > now { + Some(Duration::from_secs(value - now)) + } else { + Some(Duration::ZERO) + } + } +} pub struct JwtAuth(); #[async_trait::async_trait] impl AuthValidator for JwtAuth { @@ -40,3 +82,53 @@ fn get_query_param(session: &mut Session, key: &str) -> Option { .collect(); params.get(key).and_then(|v| decode(v).ok()).map(|s| s.to_string()) } + +pub fn check_jwt(token: &str, secret: &str) -> bool { + let key = hash_token(token, secret); + let now = SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + if let Some(exp) = JWT_CACHE.get(&key) { + if exp < now { + return false; + } + return true; + } + match is_expired(token, now) { + Ok(true) => return false, + Ok(false) => {} + Err(_) => return false, + } + + match jsonwebtoken::decode::(token, &DecodingKey::from_secret(secret.as_ref()), &JWT_VALIDATION) { + Ok(data) => { + let now = SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + if data.claims.exp > now { + JWT_CACHE.insert(key, data.claims.exp); + true + } else { + false + } + } + Err(_) => false, + } +} + +fn is_expired(token: &str, now: u64) -> Result> { + let parts: Vec<&str> = token.split('.').collect(); + if parts.len() != 3 { + return Err("Invalid JWT format".into()); + } + let decoded = URL_SAFE_NO_PAD.decode(parts[1])?; + let claims: Expired = serde_json::from_slice(&decoded)?; + if let Some(exp) = claims.exp { + Ok(exp < now) + } else { + Ok(true) + } +} + +fn hash_token(token: &str, secret: &str) -> u64 { + let mut hasher = AHasher::default(); + token.hash(&mut hasher); + secret.hash(&mut hasher); + hasher.finish() +} diff --git a/src/auth/types.rs b/src/auth/types.rs index f739125..2fcf3da 100644 --- a/src/auth/types.rs +++ b/src/auth/types.rs @@ -2,7 +2,7 @@ use crate::auth::apikey::ApiKeyAuth; use crate::auth::basic::BasicAuth; use crate::auth::forward::ForwardAuth; use crate::auth::jwt::JwtAuth; -use crate::utils::structs::InnerAuth; +use crate::utils::types::InnerAuth; use pingora_proxy::Session; #[async_trait::async_trait] diff --git a/src/web/bgservice.rs b/src/core/bgservice.rs similarity index 96% rename from src/web/bgservice.rs rename to src/core/bgservice.rs index 195ca9b..5a66e72 100644 --- a/src/web/bgservice.rs +++ b/src/core/bgservice.rs @@ -1,12 +1,12 @@ +use crate::core::logging::init_logging; +use crate::core::proxyhttp::LB; +use crate::ingress::{APIUpstreamProvider, ConsulProvider, Discovery, FromFileProvider, KubernetesProvider}; use crate::tls::acme::order::refresh_order; -use crate::utils::discovery::{APIUpstreamProvider, ConsulProvider, Discovery, FromFileProvider, KubernetesProvider}; use crate::utils::metrics::calc_cache_metrics; use crate::utils::parceyaml::load_configuration; -use crate::utils::structs::Configuration; use crate::utils::tools::*; +use crate::utils::types::Configuration; use crate::utils::*; -use crate::web::logging::init_logging; -use crate::web::proxyhttp::LB; use async_trait::async_trait; use dashmap::DashMap; use log::{error, info}; diff --git a/src/web/gethosts.rs b/src/core/gethosts.rs similarity index 98% rename from src/web/gethosts.rs rename to src/core/gethosts.rs index e95d90b..243b62c 100644 --- a/src/web/gethosts.rs +++ b/src/core/gethosts.rs @@ -1,5 +1,5 @@ -use crate::utils::structs::InnerMap; -use crate::web::proxyhttp::LB; +use crate::core::proxyhttp::LB; +use crate::utils::types::InnerMap; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/src/web/logging.rs b/src/core/logging.rs similarity index 97% rename from src/web/logging.rs rename to src/core/logging.rs index 54d3211..993acfe 100644 --- a/src/web/logging.rs +++ b/src/core/logging.rs @@ -1,5 +1,5 @@ use crate::utils::metrics::LOGGING_ERRORS; -use crate::utils::structs::AppConfig; +use crate::utils::types::AppConfig; use log::{error, info, warn, LevelFilter}; use log4rs::append::rolling_file::policy::compound::roll::fixed_window::FixedWindowRoller; use log4rs::append::rolling_file::policy::compound::trigger::size::SizeTrigger; @@ -16,7 +16,7 @@ use pingora_http::Version; use pingora_proxy::Session; use std::net::{IpAddr, Ipv4Addr}; use std::sync::OnceLock; -use tokio::sync::mpsc; // Add Logger import +use tokio::sync::mpsc; #[derive(Debug)] pub struct LogMessage { @@ -138,7 +138,6 @@ pub fn access_log(response_code: u16, summary: &str, session: &Session) { LogLevel::Access => true, LogLevel::None => false, LogLevel::Error => matches!(status, MatchStatus::Er5xx), - // LogLevel::Error => matches!(status, MatchStatus::Er4xx | MatchStatus::Er5xx), }; if !should_log { diff --git a/src/web.rs b/src/core/mod.rs similarity index 88% rename from src/web.rs rename to src/core/mod.rs index b7ba18d..859d4dd 100644 --- a/src/web.rs +++ b/src/core/mod.rs @@ -1,4 +1,3 @@ -pub mod acme; pub mod bgservice; pub mod gethosts; pub mod logging; diff --git a/src/web/proxyhttp.rs b/src/core/proxyhttp.rs similarity index 98% rename from src/web/proxyhttp.rs rename to src/core/proxyhttp.rs index a7ec6d2..16f046c 100644 --- a/src/web/proxyhttp.rs +++ b/src/core/proxyhttp.rs @@ -1,9 +1,9 @@ use crate::auth::authenticate; +use crate::core::gethosts::{GetHost, GetHostsReturHeaders}; +use crate::core::logging::access_log; use crate::utils::lazylock::{CACHE_LOCK, CACHE_TTL, EVICTION, LOCALHOST, MEM_CACHE, RATE_LIMITER, REQUESTS_4XX, REVERSE_STORE}; use crate::utils::metrics::*; -use crate::utils::structs::{AppConfig, Extraparams, Headers, InnerMap, UpstreamsDashMap, UpstreamsIdMap}; -use crate::web::gethosts::{GetHost, GetHostsReturHeaders}; -use crate::web::logging::access_log; +use crate::utils::types::{AppConfig, Extraparams, Headers, InnerMap, UpstreamsDashMap, UpstreamsIdMap}; use arc_swap::ArcSwap; use async_trait::async_trait; use axum::body::Bytes; diff --git a/src/web/start.rs b/src/core/start.rs similarity index 97% rename from src/web/start.rs rename to src/core/start.rs index 65a39ee..697d2a6 100644 --- a/src/web/start.rs +++ b/src/core/start.rs @@ -1,11 +1,12 @@ +use crate::core::logging::init_access_log; +use crate::core::proxyhttp::LB; use crate::tls::grades; use crate::tls::load; use crate::tls::load::CertificateConfig; +use crate::utils::folder_watch; use crate::utils::lazylock::{CACHE_TTL, EVICTION}; -use crate::utils::structs::Extraparams; use crate::utils::tools::*; -use crate::web::logging::init_access_log; -use crate::web::proxyhttp::LB; +use crate::utils::types::Extraparams; use arc_swap::ArcSwap; use dashmap::DashMap; use log::info; @@ -27,6 +28,7 @@ use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::Arc; use std::time::Duration; use std::{fs, thread}; + pub fn run() { default_provider().install_default().expect("Failed to install RusTLS crypto provider"); let parameters = Opt::parse_args(); @@ -113,7 +115,7 @@ pub fn run() { fs::create_dir_all(certs_path.clone()).unwrap(); } thread::spawn(move || { - watch_folder(certs_path, tx).unwrap(); + folder_watch(certs_path, tx).unwrap(); }); let certificate_configs = rx.recv().unwrap(); let first_set = load::Certificates::new(&certificate_configs, grade.as_str()).unwrap_or_else(|| panic!("Unable to load initial certificate info")); diff --git a/src/web/webserver.rs b/src/core/webserver.rs similarity index 97% rename from src/web/webserver.rs rename to src/core/webserver.rs index 76619ac..c803b55 100644 --- a/src/web/webserver.rs +++ b/src/core/webserver.rs @@ -1,9 +1,9 @@ -use crate::utils::discovery::APIUpstreamProvider; -use crate::utils::jwt::Claims; +use crate::auth::jwt::Claims; +use crate::ingress::APIUpstreamProvider; +use crate::tls::acme::http01::{acme_create, acme_order, http01_challenge}; use crate::utils::metrics::{get_memory_usage, get_open_files, MEMORY_USAGE, OPEN_FILES}; -use crate::utils::structs::{Config, Configuration, UpstreamsDashMap}; use crate::utils::tools::{upstreams_liveness_json, upstreams_to_json}; -use crate::web::acme::{acme_create, acme_order, http01_challenge}; +use crate::utils::types::{Config, Configuration, UpstreamsDashMap}; use axum::body::Body; use axum::extract::{Query, State}; use axum::http::{Response, StatusCode}; diff --git a/src/utils/consul.rs b/src/ingress/consul.rs similarity index 93% rename from src/utils/consul.rs rename to src/ingress/consul.rs index c374137..025221a 100644 --- a/src/utils/consul.rs +++ b/src/ingress/consul.rs @@ -1,7 +1,8 @@ +use crate::ingress::kuberconsul; +use crate::ingress::kuberconsul::{list_to_upstreams, ServiceDiscovery}; use crate::utils::httpclient; -use crate::utils::kuberconsul::{list_to_upstreams, ServiceDiscovery}; use crate::utils::parceyaml::build_headers; -use crate::utils::structs::{Configuration, GlobalServiceMapping, UpstreamsDashMap}; +use crate::utils::types::{Configuration, GlobalServiceMapping, UpstreamsDashMap}; use async_trait::async_trait; use dashmap::DashMap; use pingora::prelude::sleep; @@ -81,7 +82,7 @@ impl ServiceDiscovery for ConsulDiscovery { } } - if let Some(lt) = crate::utils::kuberconsul::clone_compare(&upstreams, &config).await { + if let Some(lt) = kuberconsul::clone_compare(&upstreams, &config).await { let _ = toreturn.send(lt).await; } diff --git a/src/utils/discovery.rs b/src/ingress/discovery.rs similarity index 82% rename from src/utils/discovery.rs rename to src/ingress/discovery.rs index fca615a..7629a35 100644 --- a/src/utils/discovery.rs +++ b/src/ingress/discovery.rs @@ -1,9 +1,9 @@ -use crate::utils::consul::ConsulDiscovery; -use crate::utils::filewatch; -use crate::utils::kuberconsul::ServiceDiscovery; -use crate::utils::kubernetes::KubernetesDiscovery; -use crate::utils::structs::{Configuration, UpstreamsDashMap}; -use crate::web::webserver; +use crate::core::webserver; +use crate::ingress::consul::ConsulDiscovery; +use crate::ingress::kuberconsul::ServiceDiscovery; +use crate::ingress::kubernetes::KubernetesDiscovery; +use crate::utils::types::{Configuration, UpstreamsDashMap}; +use crate::utils::watch; use async_trait::async_trait; use std::sync::Arc; use tokio::sync::mpsc::Sender; @@ -48,7 +48,7 @@ impl Discovery for APIUpstreamProvider { #[async_trait] impl Discovery for FromFileProvider { async fn start(&self, tx: Sender) { - tokio::spawn(filewatch::start(self.path.clone(), tx)); + tokio::spawn(watch::file_watch(self.path.clone(), tx)); } } diff --git a/src/utils/kuberconsul.rs b/src/ingress/kuberconsul.rs similarity index 96% rename from src/utils/kuberconsul.rs rename to src/ingress/kuberconsul.rs index 1fec46a..69a0e8b 100644 --- a/src/utils/kuberconsul.rs +++ b/src/ingress/kuberconsul.rs @@ -1,5 +1,5 @@ -use crate::utils::structs::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap}; use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams}; +use crate::utils::types::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap}; use async_trait::async_trait; use dashmap::DashMap; use std::sync::atomic::AtomicUsize; diff --git a/src/utils/kubernetes.rs b/src/ingress/kubernetes.rs similarity index 94% rename from src/utils/kubernetes.rs rename to src/ingress/kubernetes.rs index 2be6a3c..3359fe5 100644 --- a/src/utils/kubernetes.rs +++ b/src/ingress/kubernetes.rs @@ -1,7 +1,6 @@ use crate::utils::httpclient; -use crate::utils::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate}; use crate::utils::parceyaml::build_headers; -use crate::utils::structs::{Configuration, GlobalServiceMapping, UpstreamsDashMap}; +use crate::utils::types::{Configuration, GlobalServiceMapping, UpstreamsDashMap}; use async_trait::async_trait; use dashmap::DashMap; use log::error; @@ -9,7 +8,9 @@ use rand::RngExt; use std::env; use std::sync::Arc; -use crate::utils::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery}; +use crate::ingress::kuberconsul; +use crate::ingress::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery}; +use crate::ingress::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate}; use rustls::crypto::ring::default_provider; use rustls::crypto::CryptoProvider; use tokio::sync::mpsc::Sender; @@ -67,7 +68,7 @@ impl ServiceDiscovery for KubernetesDiscovery { let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 }; let server = servers.get(num).unwrap().to_string(); let path = kuber.tokenpath.unwrap_or_else(|| "/var/run/secrets/kubernetes.io/serviceaccount/token".to_string()); - let token = crate::utils::kuberconsul::read_token(path.as_str()).await; + let token = kuberconsul::read_token(path.as_str()).await; if CryptoProvider::get_default().is_none() { default_provider().install_default().expect("Failed to install rustls crypto provider"); diff --git a/src/utils/kubewatcher.rs b/src/ingress/kubewatcher.rs similarity index 100% rename from src/utils/kubewatcher.rs rename to src/ingress/kubewatcher.rs diff --git a/src/ingress/mod.rs b/src/ingress/mod.rs new file mode 100644 index 0000000..d2308dc --- /dev/null +++ b/src/ingress/mod.rs @@ -0,0 +1,7 @@ +pub mod consul; +pub mod discovery; +pub mod kuberconsul; +pub mod kubernetes; +pub mod kubewatcher; + +pub use discovery::*; diff --git a/src/lib.rs b/src/lib.rs index 1007601..14e02d2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +1,5 @@ pub mod auth; +pub mod core; +pub mod ingress; pub mod tls; pub mod utils; -pub mod web; diff --git a/src/main.rs b/src/main.rs index 80c719a..b90b3fb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,4 @@ -use aralez::web; +use aralez::core; #[global_allocator] static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc; @@ -10,5 +10,5 @@ fn main() { std::process::exit(0); } - web::start::run(); + core::start::run(); } diff --git a/src/tls/acme.rs b/src/tls/acme.rs deleted file mode 100644 index e2a28e0..0000000 --- a/src/tls/acme.rs +++ /dev/null @@ -1,2 +0,0 @@ -pub mod account; -pub mod order; diff --git a/src/web/acme.rs b/src/tls/acme/http01.rs similarity index 88% rename from src/web/acme.rs rename to src/tls/acme/http01.rs index 39f90c3..6d8db8b 100644 --- a/src/web/acme.rs +++ b/src/tls/acme/http01.rs @@ -1,3 +1,4 @@ +use crate::core::webserver; use crate::tls::acme::{account, order}; use crate::utils::lazylock::CHALLENGES; use axum::body::Body; @@ -6,7 +7,7 @@ use axum::http::{Response, StatusCode}; use axum::response::IntoResponse; #[allow(clippy::needless_return)] -pub async fn acme_create(State(state): State) -> impl IntoResponse { +pub async fn acme_create(State(state): State) -> impl IntoResponse { match account::load_or_create(state.cert_creds.as_str()).await { Ok(txt) => { return Response::builder() @@ -24,7 +25,7 @@ pub async fn acme_create(State(state): State) - }; } #[allow(clippy::needless_return)] -pub async fn acme_order(State(state): State, axum::extract::Path(domain): axum::extract::Path) -> impl IntoResponse { +pub async fn acme_order(State(state): State, axum::extract::Path(domain): axum::extract::Path) -> impl IntoResponse { let domain_clean = domain.trim_matches('/'); match order::order(domain_clean, state.cert_creds.as_str(), state.certs_dir).await { Ok(txt) => { diff --git a/src/tls/acme/mod.rs b/src/tls/acme/mod.rs new file mode 100644 index 0000000..523e3b7 --- /dev/null +++ b/src/tls/acme/mod.rs @@ -0,0 +1,6 @@ +pub mod account; +pub mod http01; +pub mod order; + +pub use account::*; +pub use order::*; diff --git a/src/tls.rs b/src/tls/mod.rs similarity index 72% rename from src/tls.rs rename to src/tls/mod.rs index d94b74b..77a7f1d 100644 --- a/src/tls.rs +++ b/src/tls/mod.rs @@ -1,3 +1,4 @@ pub mod acme; pub mod grades; pub mod load; +pub use load::*; diff --git a/src/utils/auth.rs b/src/utils/auth.rs deleted file mode 100644 index 7db5065..0000000 --- a/src/utils/auth.rs +++ /dev/null @@ -1,228 +0,0 @@ -/* -use crate::utils::jwt::{check_jwt, JWT_TOKEN}; -use crate::utils::structs::InnerAuth; -pub(crate) use crate::utils::tools::split_host_port; -use axum::http::StatusCode; -use base64::engine::general_purpose::STANDARD; -use base64::Engine; -use pingora::http::RequestHeader; -use pingora_core::connectors::http::Connector; -use pingora_core::upstreams::peer::HttpPeer; -use pingora_http::ResponseHeader; -use pingora_proxy::Session; -use std::collections::HashMap; -use std::sync::LazyLock; -use subtle::ConstantTimeEq; -use urlencoding::decode; - -#[async_trait::async_trait] -trait AuthValidator { - async fn validate(&self, session: &mut Session) -> bool; -} -struct BasicAuth<'a>(&'a str); -struct ApiKeyAuth<'a>(&'a str); -struct JwtAuth(); -struct ForwardAuth<'a>(&'a str); - -pub static AUTH_CONNECTOR: LazyLock = LazyLock::new(|| Connector::new(None)); - -#[async_trait::async_trait] -impl AuthValidator for ForwardAuth<'_> { - async fn validate(&self, session: &mut Session) -> bool { - let method = match session.req_header().method.as_str() { - "HEAD" => "HEAD", - _ => "GET", - }; - - let auth_url = self.0; - - let (plain, tls) = if let Some(p) = auth_url.strip_prefix("http://") { - (p, false) - } else if let Some(p) = auth_url.strip_prefix("https://") { - (p, true) - } else { - return false; - }; - - let (addr, uri) = if let Some(pos) = plain.find('/') { - (&plain[..pos], &plain[pos..]) - } else { - (plain, "/") - }; - - let hp = match split_host_port(addr, tls) { - Some(hp) => hp, - None => return false, - }; - - let peer = HttpPeer::new((hp.0, hp.1), tls, hp.0.to_string()); - - let (mut http_session, _) = match AUTH_CONNECTOR.get_http_session(&peer).await { - Ok(s) => s, - Err(e) => { - log::warn!("ForwardAuth: connect failed: {}", e); - return false; - } - }; - - let mut auth_req = match RequestHeader::build(method, uri.as_bytes(), None) { - Ok(r) => r, - Err(e) => { - log::warn!("ForwardAuth: failed to build request: {}", e); - return false; - } - }; - - // auth_req.headers = session.req_header().headers.clone(); - auth_req.insert_header("Host", addr).ok(); - auth_req.insert_header("X-Forwarded-Uri", uri).ok(); - auth_req.insert_header("X-Forwarded-Method", session.req_header().method.as_str()).ok(); - if let Some(auth) = session.req_header().headers.get("authorization") { - auth_req.insert_header("Authorization", auth.clone()).ok(); - } - - if let Some(cookie) = session.req_header().headers.get("cookie") { - auth_req.insert_header("Cookie", cookie.clone()).ok(); - } - - if tls { - auth_req.insert_header("X-Forwarded-Proto", "https").ok(); - } else { - auth_req.insert_header("X-Forwarded-Proto", "http").ok(); - } - - if let Err(e) = http_session.write_request_header(Box::new(auth_req)).await { - log::warn!("ForwardAuth: write failed: {}", e); - return false; - } - - let status = match http_session.read_response_header().await { - Ok(_) => http_session.response_header().map(|r| r.status.as_u16()).unwrap_or(500), - Err(e) => { - log::warn!("ForwardAuth: read failed: {}", e); - return false; - } - }; - - let auth_headers_to_forward: Vec<(String, String)> = if let Some(resp_header) = http_session.response_header() { - resp_header - .headers - .iter() - .filter_map(|(name, value)| { - let name_str = name.as_str(); - if name_str.starts_with("x-") || name_str.starts_with("remote-") || name_str.starts_with("locat") { - value.to_str().ok().map(|v| (name_str.to_string(), v.to_string())) - } else { - None - } - }) - .collect() - } else { - Vec::new() - }; - - AUTH_CONNECTOR.release_http_session(http_session, &peer, None).await; - - if (200..300).contains(&status) { - for (name, value) in auth_headers_to_forward { - session.req_header_mut().insert_header(name, value).ok(); - } - true - } else if status == 302 || status == 301 { - let resp = ResponseHeader::build(StatusCode::MOVED_PERMANENTLY, None); - match resp { - Ok(mut r) => { - for (name, value) in auth_headers_to_forward { - r.insert_header(name, value).ok(); - } - let _ = r.insert_header("Content-Length", "0"); - let _ = session.write_response_header(Box::new(r), true).await; - true - } - Err(_) => return false, - } - } else { - false - } - } -} - -#[async_trait::async_trait] -impl AuthValidator for BasicAuth<'_> { - async fn validate(&self, session: &mut Session) -> bool { - if let Some(header) = session.get_header("authorization") { - if let Ok(h) = header.to_str() { - if let Some((_, val)) = h.split_once(' ') { - if let Ok(decoded) = STANDARD.decode(val) { - if decoded.as_slice().ct_eq(self.0.as_bytes()).into() { - return true; - } - } - } - } - } - false - } -} - -#[async_trait::async_trait] -impl AuthValidator for ApiKeyAuth<'_> { - async fn validate(&self, session: &mut Session) -> bool { - if let Some(header) = session.get_header("x-api-key") { - if let Ok(h) = header.to_str() { - return h.as_bytes().ct_eq(self.0.as_bytes()).into(); - } - } - false - } -} - -#[async_trait::async_trait] -impl AuthValidator for JwtAuth { - async fn validate(&self, session: &mut Session) -> bool { - if let Some(jwtsecret) = JWT_TOKEN.clone() { - if let Some(tok) = get_query_param(session, "araleztoken") { - return check_jwt(tok.as_str(), jwtsecret.as_ref()); - } - if let Some(auth_header) = session.get_header("authorization") { - if let Ok(header_str) = auth_header.to_str() { - if let Some((scheme, token)) = header_str.split_once(' ') { - if scheme.eq_ignore_ascii_case("bearer") { - return check_jwt(token, jwtsecret.as_ref()); - } - } - } - } - } - false - } -} - -pub async fn authenticate(auth: &InnerAuth, session: &mut Session) -> bool { - match &*auth.auth_type { - "basic" => BasicAuth(&*auth.auth_cred).validate(session).await, - "apikey" => ApiKeyAuth(&*auth.auth_cred).validate(session).await, - "jwt" => JwtAuth().validate(session).await, - "forward" => ForwardAuth(&*auth.auth_cred).validate(session).await, - _ => { - log::warn!("Unsupported authentication mechanism : {}", &*auth.auth_type); - false - } - } -} - -pub fn get_query_param(session: &mut Session, key: &str) -> Option { - let query = session.req_header().uri.query()?; - - let params: HashMap<_, _> = query - .split('&') - .filter_map(|pair| { - let mut parts = pair.splitn(2, '='); - let k = parts.next()?; - let v = parts.next().unwrap_or(""); // Some params might have no value - Some((k, v)) - }) - .collect(); - params.get(key).and_then(|v| decode(v).ok()).map(|s| s.to_string()) -} -*/ diff --git a/src/utils/fordebug.rs b/src/utils/fordebug.rs index 65f0706..1774f12 100644 --- a/src/utils/fordebug.rs +++ b/src/utils/fordebug.rs @@ -23,9 +23,9 @@ unsafe impl GlobalAlloc for CountingAllocator { // Uncomment following lines and comment allocator in main.rs // #[global_allocator] // pub static A: CountingAllocator = CountingAllocator; -#[allow(dead_code)] -fn for_example() { - let before = crate::utils::fordebug::ALLOC_COUNT.load(Ordering::Relaxed); - let after = crate::utils::fordebug::ALLOC_COUNT.load(Ordering::Relaxed); - println!("Allocations : {}", after - before); -} +// #[allow(dead_code)] +// fn for_example() { +// let before = crate::utils::fordebug::ALLOC_COUNT.load(Ordering::Relaxed); +// let after = crate::utils::fordebug::ALLOC_COUNT.load(Ordering::Relaxed); +// println!("Allocations : {}", after - before); +// } diff --git a/src/utils/healthcheck.rs b/src/utils/healthcheck.rs index 58f38e4..72e5efb 100644 --- a/src/utils/healthcheck.rs +++ b/src/utils/healthcheck.rs @@ -1,7 +1,7 @@ use crate::utils::hcclient::httpclient; use crate::utils::lazylock::REVERSE_STORE; -use crate::utils::structs::{InnerMap, UpstreamsDashMap, UpstreamsIdMap}; use crate::utils::tools::*; +use crate::utils::types::{InnerMap, UpstreamsDashMap, UpstreamsIdMap}; use dashmap::DashMap; use std::sync::atomic::AtomicUsize; use std::sync::Arc; diff --git a/src/utils/httpclient.rs b/src/utils/httpclient.rs index cb8b1c1..8684d1c 100644 --- a/src/utils/httpclient.rs +++ b/src/utils/httpclient.rs @@ -1,7 +1,7 @@ -use crate::utils::consul::ConsulService; -use crate::utils::kuberconsul::match_path; -use crate::utils::kubernetes::KubeEndpointSliceList; -use crate::utils::structs::{GlobalServiceMapping, InnerMap}; +use crate::ingress::consul::ConsulService; +use crate::ingress::kuberconsul::match_path; +use crate::ingress::kubernetes::KubeEndpointSliceList; +use crate::utils::types::{GlobalServiceMapping, InnerMap}; use ahash::HashMap; use dashmap::DashMap; use pingora_core::connectors::http::Connector; diff --git a/src/utils/jwt.rs b/src/utils/jwt.rs deleted file mode 100644 index e1b003b..0000000 --- a/src/utils/jwt.rs +++ /dev/null @@ -1,93 +0,0 @@ -use ahash::AHasher; -use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; -use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation}; -use moka::sync::Cache; -use moka::Expiry; -use serde::{Deserialize, Serialize}; -use std::env; -use std::hash::{Hash, Hasher}; -use std::sync::{Arc, LazyLock}; -use std::time::{Duration, Instant, SystemTime}; - -#[derive(Debug, Serialize, Deserialize)] -pub struct Claims { - pub master_key: String, - pub owner: String, - pub exp: u64, - pub random: Option, -} - -#[derive(Debug, Deserialize)] -struct Expired { - exp: Option, -} - -static JWT_VALIDATION: LazyLock = LazyLock::new(|| Validation::new(Algorithm::HS256)); - -pub static JWT_TOKEN: LazyLock>> = LazyLock::new(|| match env::var("JWT_KEY") { - Ok(key) if !key.is_empty() => Some(Arc::from(key.as_str())), - _ => None, -}); - -static JWT_CACHE: LazyLock> = LazyLock::new(|| Cache::builder().max_capacity(100_000).expire_after(JwtExpiry).build()); -struct JwtExpiry; -impl Expiry for JwtExpiry { - fn expire_after_create(&self, _key: &u64, value: &u64, _current_time: Instant) -> Option { - let now = SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); - if *value > now { - Some(Duration::from_secs(value - now)) - } else { - Some(Duration::ZERO) - } - } -} - -pub fn check_jwt(token: &str, secret: &str) -> bool { - let key = hash_token(token, secret); - let now = SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); - if let Some(exp) = JWT_CACHE.get(&key) { - if exp < now { - return false; - } - return true; - } - match is_expired(token, now) { - Ok(true) => return false, - Ok(false) => {} - Err(_) => return false, - } - - match decode::(token, &DecodingKey::from_secret(secret.as_ref()), &JWT_VALIDATION) { - Ok(data) => { - let now = SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); - if data.claims.exp > now { - JWT_CACHE.insert(key, data.claims.exp); - true - } else { - false - } - } - Err(_) => false, - } -} - -fn is_expired(token: &str, now: u64) -> Result> { - let parts: Vec<&str> = token.split('.').collect(); - if parts.len() != 3 { - return Err("Invalid JWT format".into()); - } - let decoded = URL_SAFE_NO_PAD.decode(parts[1])?; - let claims: Expired = serde_json::from_slice(&decoded)?; - if let Some(exp) = claims.exp { - Ok(exp < now) - } else { - Ok(true) - } -} - -fn hash_token(token: &str, secret: &str) -> u64 { - let mut hasher = AHasher::default(); - token.hash(&mut hasher); - secret.hash(&mut hasher); - hasher.finish() -} diff --git a/src/utils/metrics.rs b/src/utils/metrics.rs index c6bf9ac..db68e97 100644 --- a/src/utils/metrics.rs +++ b/src/utils/metrics.rs @@ -117,13 +117,6 @@ pub fn calc_metrics(metric_types: &MetricTypes) { REQUESTS_BY_UPSTREAM.with_label_values(&[metric_types.upstream.as_ref()]).inc(); RESPONSE_LATENCY.observe(metric_types.latency.as_secs_f64()); - - // if let Some(eviction) = EVICTION.get() { - // CACHE_SIZE_BYTES.set(eviction.total_size() as i64); - // CACHE_ITEMS.set(eviction.total_items() as i64); - // CACHE_EVICTED_BYTES.set(eviction.evicted_size() as i64); - // CACHE_EVICTED_ITEMS.set(eviction.evicted_items() as i64); - // } } pub(crate) fn get_memory_usage() -> usize { diff --git a/src/utils.rs b/src/utils/mod.rs similarity index 52% rename from src/utils.rs rename to src/utils/mod.rs index 5ae7492..65f072f 100644 --- a/src/utils.rs +++ b/src/utils/mod.rs @@ -1,19 +1,14 @@ -pub mod auth; -pub mod consul; -pub mod discovery; -mod filewatch; pub mod fordebug; pub mod grpc; pub mod hcclient; pub mod healthcheck; pub mod httpclient; -pub mod jwt; -pub mod kuberconsul; -pub mod kubernetes; -pub mod kubewatcher; +// pub mod ingress; pub mod lazylock; pub mod metrics; pub mod parceyaml; pub mod state; -pub mod structs; pub mod tools; +pub mod types; +pub(crate) mod watch; +pub use watch::*; diff --git a/src/utils/parceyaml.rs b/src/utils/parceyaml.rs index f00829d..37dba8b 100644 --- a/src/utils/parceyaml.rs +++ b/src/utils/parceyaml.rs @@ -1,9 +1,9 @@ +use crate::core::logging::log_builder; use crate::utils::healthcheck; use crate::utils::lazylock::REVERSE_STORE; use crate::utils::state::{is_first_run, mark_not_first_run}; -use crate::utils::structs::*; use crate::utils::tools::{clone_dashmap, clone_dashmap_into, print_upstreams}; -use crate::web::logging::log_builder; +use crate::utils::types::*; use dashmap::DashMap; use log::{error, info, warn}; use std::collections::HashMap; diff --git a/src/utils/tools.rs b/src/utils/tools.rs index 4cd54f6..cb82d93 100644 --- a/src/utils/tools.rs +++ b/src/utils/tools.rs @@ -1,13 +1,11 @@ use crate::tls::load; use crate::tls::load::CertificateConfig; -use crate::utils::structs::{Extraparams, InnerMapForJson, UpstreamSnapshotForJson, UpstreamsDashMap, UpstreamsIdMap}; +use crate::utils::types::{Extraparams, InnerMapForJson, UpstreamSnapshotForJson, UpstreamsDashMap, UpstreamsIdMap}; use dashmap::DashMap; use log::{error, info}; -use notify::{event::ModifyKind, Config, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; use privdrop::PrivDrop; use serde_json::{json, Value}; use sha2::{Digest, Sha256}; -use std::any::type_name; use std::collections::{HashMap, HashSet}; use std::fmt::Write; use std::fs::OpenOptions; @@ -18,9 +16,7 @@ use std::os::unix::fs::MetadataExt; use std::os::unix::fs::OpenOptionsExt; use std::str::FromStr; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::mpsc::{channel, Sender}; use std::sync::Arc; -use std::time::{Duration, Instant}; use std::{fs, process, thread, time}; pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams) { @@ -47,24 +43,6 @@ pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams) } info!("\n{}", out.trim_end()); } -#[allow(dead_code)] -pub fn typeoff(_: T) { - let to = type_name::(); - println!("{:?}", to); -} - -#[allow(dead_code)] -pub fn string_to_bool(val: Option<&str>) -> Option { - match val { - Some(v) => match v { - "yes" => Some(true), - "true" => Some(true), - _ => Some(false), - }, - None => Some(false), - } -} - pub fn clone_dashmap(original: &UpstreamsDashMap) -> UpstreamsDashMap { let new_map: UpstreamsDashMap = DashMap::new(); @@ -209,33 +187,6 @@ pub fn listdir(dir: String) -> Vec { certificate_configs } -pub fn watch_folder(path: String, sender: Sender>) -> notify::Result<()> { - let (tx, rx) = channel(); - let mut watcher = RecommendedWatcher::new(tx, Config::default())?; - watcher.watch(path.as_ref(), RecursiveMode::Recursive)?; - info!("Watching for certificates in : {}", path); - let certificate_configs = listdir(path.clone()); - sender.send(certificate_configs)?; - let mut start = Instant::now(); - loop { - match rx.recv_timeout(Duration::from_secs(1)) { - Ok(Ok(event)) => match &event.kind { - EventKind::Modify(ModifyKind::Data(_)) | EventKind::Create(_) | EventKind::Remove(_) => { - if start.elapsed() > Duration::from_secs(1) { - start = Instant::now(); - let certificate_configs = listdir(path.clone()); - sender.send(certificate_configs)?; - info!("Certificate changed: {:?}, {:?}", event.kind, event.paths); - } - } - _ => {} - }, - Ok(Err(e)) => error!("Watch error: {:?}", e), - Err(_) => {} - } - } -} - pub fn drop_priv(user: String, group: String, http_addr: String, tls_addr: Option) { thread::sleep(time::Duration::from_millis(10)); loop { diff --git a/src/utils/structs.rs b/src/utils/types.rs similarity index 100% rename from src/utils/structs.rs rename to src/utils/types.rs diff --git a/src/utils/filewatch.rs b/src/utils/watch.rs similarity index 57% rename from src/utils/filewatch.rs rename to src/utils/watch.rs index 49f41e9..2cbaf7a 100644 --- a/src/utils/filewatch.rs +++ b/src/utils/watch.rs @@ -1,15 +1,18 @@ +use crate::tls::CertificateConfig; use crate::utils::parceyaml::load_configuration; -use crate::utils::structs::Configuration; -use log::error; +use crate::utils::tools::listdir; +use crate::utils::types::Configuration; +use log::{error, info}; use notify::event::ModifyKind; use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; use pingora::prelude::sleep; use std::path::Path; +use std::sync::mpsc::channel; use std::time::{Duration, Instant}; use tokio::sync::mpsc::Sender; use tokio::task; -pub async fn start(fp: String, toreturn: Sender) { +pub async fn file_watch(fp: String, toreturn: Sender) { sleep(Duration::from_millis(50)).await; // For having nice logs :-) let file_path = fp.as_str(); let parent_dir = Path::new(file_path).parent().unwrap(); @@ -50,3 +53,29 @@ pub async fn start(fp: String, toreturn: Sender) { } } } +pub fn folder_watch(path: String, sender: std::sync::mpsc::Sender>) -> notify::Result<()> { + let (tx, rx) = channel(); + let mut watcher = RecommendedWatcher::new(tx, Config::default())?; + watcher.watch(path.as_ref(), RecursiveMode::Recursive)?; + info!("Watching for certificates in : {}", path); + let certificate_configs = listdir(path.clone()); + sender.send(certificate_configs)?; + let mut start = Instant::now(); + loop { + match rx.recv_timeout(Duration::from_secs(1)) { + Ok(Ok(event)) => match &event.kind { + EventKind::Modify(ModifyKind::Data(_)) | EventKind::Create(_) | EventKind::Remove(_) => { + if start.elapsed() > Duration::from_secs(1) { + start = Instant::now(); + let certificate_configs = listdir(path.clone()); + sender.send(certificate_configs)?; + info!("Certificate changed: {:?}, {:?}", event.kind, event.paths); + } + } + _ => {} + }, + Ok(Err(e)) => error!("Watch error: {:?}", e), + Err(_) => {} + } + } +}