Github/aralez
1
0
Fork 0
mirror of https://github.com/sadoyan/aralez.git synced 2026-05-30 03:44:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Cleanup. Making clippy happy.

Browse Source
This commit is contained in:
Ara Sadoyan
2026-05-08 16:35:20 +02:00
parent 783ffb27e1
commit fec9d5f1d6
19 changed files with 213 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
Cargo.lock generated
View File

@@ -3892,9 +3892,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
[[package]]
name = "tokio"
version = "1.52.2"
version = "1.52.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "110a78583f19d5cdb2c5ccf321d1290344e71313c6c37d43520d386027d18386"
checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
dependencies = [
"bytes",
"libc",

10
Cargo.toml
View File

@@ -11,7 +11,7 @@ panic = "abort"
strip = true
[dependencies]
tokio = { version = "1.52.1", features = ["full"] }
tokio = { version = "1.52.3", features = ["full"] }
pingora = { version = "0.8.0", features = ["lb", "openssl"] } # openssl, rustls, boringssl
serde = { version = "1.0.228", features = ["derive"] }
dashmap = "7.0.0-rc2"
@@ -23,9 +23,9 @@ async-trait = "0.1.89"
env_logger = "0.11.10"
log = "0.4.29"
futures = "0.3.32"
notify = "9.0.0-rc.3"
notify = "9.0.0-rc.4"
axum = { version = "0.8.9" }
reqwest = { version = "0.13.2", features = ["json", "stream", "blocking"] }
reqwest = { version = "0.13.3", features = ["json", "stream", "blocking"] }
serde_yml = "0.0.12"
rand = "0.10.1"
base64 = "0.22.1"
@@ -39,12 +39,12 @@ mimalloc = { version = "0.1.50", default-features = false }
prometheus = "0.14.0"
x509-parser = "0.18.1"
rustls-pemfile = "2.2.0"
tower-http = { version = "0.6.8", features = ["fs"] }
tower-http = { version = "0.6.10", features = ["fs"] }
privdrop = "0.5.6"
ctrlc = "3.5.2"
serde_json = "1.0.149"
subtle = "2.6.1"
moka = { version = "0.12.1", features = ["sync"] }
moka = { version = "0.12.15", features = ["sync"] }
ahash = "0.8.12"
instant-acme = "0.8.5"
rcgen = "0.14.7"

3
Makefile
View File

@@ -7,9 +7,6 @@ features:
checkup:
cargo clippy --workspace --all-targets --all-features -- -D warnings
cargo check --workspace --all-targets --all-features
# cargo shear
# cargo machete
cargo audit
fix:
cargo fix

74
README.md
View File

@@ -27,16 +27,16 @@ Built on Rust, on top of **Cloudflare’s Pingora engine**, **Aralez** delivers
- **Upstreams TLS detection** — Aralez will automatically detect if upstreams uses secure connection.
- **Built in rate limiter** — Globar or route limit requests to upstreams.
- **Authentication** — Supports Basic Auth, API tokens, and JWT verification.
- **Basic Auth**
- **API Key** via `x-api-key` header
- **JWT Auth**, with tokens issued by Aralez itself via `/jwt` API
- **Forward Auth**, Sends requests to an authentication server.
- **Basic Auth**
- **API Key** via `x-api-key` header
- **JWT Auth**, with tokens issued by Aralez itself via `/jwt` API
- **Forward Auth**, Sends requests to an authentication server.
- **Load Balancing** Round-robin, health checks, optional sticky sessions.
- **Built in file server** — Build in minimalistic file server for serving static files, should be added as upstreams for public access.
- **Upstream Providers:**
- `file` Upstreams are declared in config file.
- `consul` Upstreams are dynamically updated from Hashicorp Consul.
- `kubernetes` Upstreams are dynamically updated from kubernetes api server.
- `file` Upstreams are declared in config file.
- `consul` Upstreams are dynamically updated from Hashicorp Consul.
- `kubernetes` Upstreams are dynamically updated from kubernetes api server.
- **Automatic WebSocket Support:** WS connection upgrades are handled automatically.
- **Automatic gRPC Support:** gRPC detected and handled automatically.
- **Header Injection:** Global and per-route server/client headers injection.
@@ -84,26 +84,16 @@ Make the binary executable `chmod 755 ./aralez-VERSION` and run.
File names:
| File Name | Description |
|---------------------------------|--------------------------------------------------------------------------|
| `aralez-x86_64-musl.gz` | Static Linux x86_64 binary, without any system dependency |
| `aralez-x86_64-glibc.gz` | Dynamic Linux x86_64 binary, with minimal system dependencies |
| `aralez-x86_64-compat-musl.gz` | Static Linux x86_64 binary, compatible with old pre Haswell CPUs |
| `aralez-x86_64-compat-glibc.gz` | Dynamic Linux x86_64 binary, compatible with old pre Haswell CPUs |
| `aralez-aarch64-musl.gz` | Static Linux ARM64 binary, without any system dependency |
| `aralez-aarch64-glibc.gz` | Dynamic Linux ARM64 binary, with minimal system dependencies |
| File Name | Description |
|---------------------------------|----------------------------------------------------------------------------|
| `aralez-x86_64-musl.gz` | Static Linux x86_64 binary, without any system dependency |
| `aralez-x86_64-glibc.gz` | Dynamic Linux x86_64 binary, with minimal system dependencies |
| `aralez-x86_64-compat-musl.gz` | Static Linux x86_64 binary, compatible with old pre Haswell CPUs |
| `aralez-x86_64-compat-glibc.gz` | Dynamic Linux x86_64 binary, compatible with old pre Haswell CPUs |
| `aralez-aarch64-musl.gz` | Static Linux ARM64 binary, without any system dependency |
| `aralez-aarch64-glibc.gz` | Dynamic Linux ARM64 binary, with minimal system dependencies |
| `sadoyan/aralez` | Docker image on Debian 13 slim (<https://hub.docker.com/r/sadoyan/aralez>) |
**Via docker**
```shell
docker run -d \
-v /local/path/to/config:/etc/aralez:ro \
-p 80:80 \
-p 443:443 \
sadoyan/aralez
```
## About binaries
**glibc** builds are in general faster, but have few, basic, Glibc dependencies:
@@ -116,6 +106,16 @@ The most intensive tests shows 107k-110k requests per second on **Glibc** binari
For running **Aralez** on very old hardware, CPUs prior Haswell, (launched before 2013) use `aralez-x86_64-compat-*.gz`
For getting the best performance on newer hardware use `aralez-x86_64-*.gz`.
**Via docker**
```shell
docker run -d \
-v /local/path/to/config:/etc/aralez:ro \
-p 80:80 \
-p 443:443 \
sadoyan/aralez
```
## Running the Proxy
```bash
@@ -210,9 +210,9 @@ myhost.mydomain.com:
- All upstreams will receive custom headers : `X-Forwarded-Proto:https` and `X-Forwarded-Port:443`
- Additionally, myhost.mydomain.com with path `/` will receive custom headers : `X-Another-Header:Hohohohoho` and `X-Something-Else:Foobar`
- Requests to each hosted domains will be limited to 10 requests per second per virtualhost.
- Requests limits are calculated per requester ip plus requested virtualhost.
- If the requester exceeds the limit it will receive `429 Too Many Requests` error.
- Optional. Rate limiter will be disabled if the parameter is entirely removed from config.
- Requests limits are calculated per requester ip plus requested virtualhost.
- If the requester exceeds the limit it will receive `429 Too Many Requests` error.
- Optional. Rate limiter will be disabled if the parameter is entirely removed from config.
- Requests to `myhost.mydomain.com/` will be limited to 20 requests per second.
- Requests to `myhost.mydomain.com/` will be proxied to `127.0.0.1` and `127.0.0.2`.
- Plain HTTP to `myhost.mydomain.com/foo` will get 301 redirect to configured TLS port of Aralez.
@@ -220,14 +220,14 @@ myhost.mydomain.com:
- Requests to `myhost.mydomain.com/foo` will be proxied to `127.0.0.4` and `127.0.0.5`.
- Requests to `myhost.mydomain.com/.well-known/acme-challenge` will be proxied to `127.0.0.1:8001`, but healthcheks are disabled.
- SSL/TLS for upstreams is detected automatically, no need to set any config parameter.
- Assuming the `127.0.0.5:8443` is SSL protected. The inner traffic will use TLS.
- Self-signed certificates are silently accepted.
- Assuming the `127.0.0.5:8443` is SSL protected. The inner traffic will use TLS.
- Self-signed certificates are silently accepted.
- Global headers (CORS for this case) will be injected to all upstreams.
- Additional headers will be injected into the request for `myhost.mydomain.com`.
- You can choose any path, deep nested paths are supported, the best match chosen.
- All requests to servers will require JWT token authentication (You can comment out the authorization to disable it),
- Firs parameter specifies the mechanism of authorisation `jwt`
- Second is the secret key for validating `jwt` tokens
- Firs parameter specifies the mechanism of authorisation `jwt`
- Second is the secret key for validating `jwt` tokens
---
@@ -244,7 +244,7 @@ To enable TLS for the proxy server.
- Set `proxy_address_tls` in `main.yaml`
- Provide at least on `tls_certificate/tls_key_file` pair.
- First pair is required tp create the TLS listener.
- First pair is required to create the TLS listener.
- This pair can be anything, even self-signed with dummy domain.
- After getting normal certificate it can be deleted
@@ -269,10 +269,10 @@ curl -XPOST --data-binary @./etc/upstreams.txt 127.0.0.1:3000/conf?key=${MASTERK
- `apikey` : Authentication via `x-api-key` header, which should match the value in config.
- `jwt`: JWT authentication implemented via `araleztoken=` url parameter. `/some/url?araleztoken=TOKEN`
- `jwt`: JWT authentication implemented via `Authorization: Bearer <token>` header.
- To obtain JWT a token, you should send **generate** request to built in api server's `/jwt` endpoint.
- `master_key`: should match configured `masterkey` in `main.yaml` and `upstreams.yaml`.
- `owner` : Just a placeholder, can be anything.
- `valid` : Time in minutes during which the generated token will be valid.
- To obtain JWT a token, you should send **generate** request to built in api server's `/jwt` endpoint.
- `master_key`: should match configured `masterkey` in `main.yaml` and `upstreams.yaml`.
- `owner` : Just a placeholder, can be anything.
- `valid` : Time in minutes during which the generated token will be valid.
**Example JWT token generation request**

17
src/tls/acme/order.rs
View File

@@ -25,7 +25,7 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String) -> Result<S
let crt = certs_dir.clone() + "/" + domain + ".crt";
let key = certs_dir.clone() + "/" + domain + ".key";
if let None = DOMAINS.get(domain) {
if DOMAINS.get(domain).is_none() {
DOMAINS.insert(domain.to_string(), true);
let mut newlist: Vec<String> = Vec::new();
for item in DOMAINS.iter() {
@@ -40,15 +40,12 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String) -> Result<S
}
}
let _ = match cert_expiry(crt.as_str()) {
Ok(expiry) => {
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs();
if expiry > now + 30 * 24 * 3600 {
// println!("Fresh certificate exists. Not renewing !");
return Ok("Fresh certificate exists. Not renewing ! \n".to_string());
}
if let Ok(expiry) = cert_expiry(crt.as_str()) {
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs();
if expiry > now + 30 * 24 * 3600 {
// println!("Fresh certificate exists. Not renewing !");
return Ok("Fresh certificate exists. Not renewing ! \n".to_string());
}
Err(_) => {}
};
let account = get_account(credsfile).await?;
@@ -73,7 +70,7 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String) -> Result<S
let private_key = KeyPair::generate()?;
let signing_request = params.serialize_request(&private_key)?;
let csr_der = signing_request.der();
order.finalize_csr(&csr_der).await?;
order.finalize_csr(csr_der).await?;
// poll for certificate
let cert_chain_pem = order.poll_certificate(&RetryPolicy::default()).await?;

26
src/tls/grades.rs
View File

@@ -16,17 +16,17 @@ const CIPHERS: CipherSuite = CipherSuite {
#[derive(Debug)]
pub enum TlsGrade {
HIGH,
MEDIUM,
LEGACY,
High,
Medium,
Legacy,
}
impl TlsGrade {
pub fn from_str(s: &str) -> Option<Self> {
match s.to_ascii_lowercase().as_str() {
"high" => Some(TlsGrade::HIGH),
"medium" => Some(TlsGrade::MEDIUM),
"unsafe" => Some(TlsGrade::LEGACY),
"high" => Some(TlsGrade::High),
"medium" => Some(TlsGrade::Medium),
"unsafe" => Some(TlsGrade::Legacy),
_ => None,
}
}
@@ -41,22 +41,22 @@ pub fn prefer_h2<'a>(_ssl: &mut SslRef, alpn_in: &'a [u8]) -> Result<&'a [u8], A
pub fn set_tsl_grade(tls_settings: &mut TlsSettings, grade: &str) {
let config_grade = TlsGrade::from_str(grade);
match config_grade {
Some(TlsGrade::HIGH) => {
Some(TlsGrade::High) => {
let _ = tls_settings.set_min_proto_version(Some(SslVersion::TLS1_2));
// let _ = tls_settings.set_max_proto_version(Some(SslVersion::TLS1_3));
let _ = tls_settings.set_cipher_list(CIPHERS.high);
// let _ = tls_settings.set_ciphersuites(CIPHERS.high);
let _ = tls_settings.set_cipher_list(CIPHERS.high);
info!("TLS grade: {:?}, => HIGH", tls_settings.options());
info!("TLS grade: {:?}, => High", tls_settings.options());
}
Some(TlsGrade::MEDIUM) => {
Some(TlsGrade::Medium) => {
let _ = tls_settings.set_min_proto_version(Some(SslVersion::TLS1));
let _ = tls_settings.set_cipher_list(CIPHERS.medium);
// let _ = tls_settings.set_ciphersuites(CIPHERS.medium);
let _ = tls_settings.set_cipher_list(CIPHERS.medium);
info!("TLS grade: {:?}, => MEDIUM", tls_settings.options());
info!("TLS grade: {:?}, => Medium", tls_settings.options());
}
Some(TlsGrade::LEGACY) => {
Some(TlsGrade::Legacy) => {
let _ = tls_settings.set_min_proto_version(Some(SslVersion::SSL3));
let _ = tls_settings.set_cipher_list(CIPHERS.legacy);
// let _ = tls_settings.set_ciphersuites(CIPHERS.legacy);
@@ -64,12 +64,12 @@ pub fn set_tsl_grade(tls_settings: &mut TlsSettings, grade: &str) {
warn!("TLS grade: {:?}, => UNSAFE", tls_settings.options());
}
None => {
// Defaults to MEDIUM
// Defaults to Medium
let _ = tls_settings.set_min_proto_version(Some(SslVersion::TLS1));
let _ = tls_settings.set_cipher_list(CIPHERS.medium);
// let _ = tls_settings.set_ciphersuites(CIPHERS.medium);
let _ = tls_settings.set_cipher_list(CIPHERS.medium);
warn!("TLS grade is not detected defaulting top MEDIUM");
warn!("TLS grade is not detected defaulting top Medium");
}
}
}

4
src/tls/load.rs
View File

@@ -60,7 +60,7 @@ impl Certificates {
}
}
Some(Self {
name_map: name_map,
name_map,
configs: cert_infos,
default_cert_path: default_cert.cert_path.clone(),
default_key_path: default_cert.key_path.clone(),
@@ -93,7 +93,7 @@ impl Certificates {
if let Some(name) = server_name {
match self.find_ssl_context(name) {
Some(ctx) => {
ssl_ref.set_ssl_context(&*ctx).map_err(|_| SniError::ALERT_FATAL)?;
ssl_ref.set_ssl_context(&ctx).map_err(|_| SniError::ALERT_FATAL)?;
}
None => {
log::debug!("No matching server name found");

7
src/utils/auth.rs
View File

@@ -153,9 +153,9 @@ impl AuthValidator for ForwardAuth<'_> {
impl AuthValidator for BasicAuth<'_> {
async fn validate(&self, session: &mut Session) -> bool {
if let Some(header) = session.get_header("authorization") {
if let Some(h) = header.to_str().ok() {
if let Ok(h) = header.to_str() {
if let Some((_, val)) = h.split_once(' ') {
if let Some(decoded) = STANDARD.decode(val).ok() {
if let Ok(decoded) = STANDARD.decode(val) {
if decoded.as_slice().ct_eq(self.0.as_bytes()).into() {
return true;
}
@@ -171,7 +171,7 @@ impl AuthValidator for BasicAuth<'_> {
impl AuthValidator for ApiKeyAuth<'_> {
async fn validate(&self, session: &mut Session) -> bool {
if let Some(header) = session.get_header("x-api-key") {
if let Some(h) = header.to_str().ok() {
if let Ok(h) = header.to_str() {
return h.as_bytes().ct_eq(self.0.as_bytes()).into();
}
}
@@ -227,6 +227,7 @@ pub fn get_query_param(session: &mut Session, key: &str) -> Option<String> {
params.get(key).and_then(|v| decode(v).ok()).map(|s| s.to_string())
}
#[allow(clippy::needless_return)]
fn split_host_port(addr: &str, tls: bool) -> Option<(&str, u16, bool, &str)> {
match addr.split_once(':') {
Some((h, p)) => match p.parse::<u16>() {

17
src/utils/filewatch.rs
View File

@@ -37,17 +37,12 @@ pub async fn start(fp: String, mut toreturn: Sender<Configuration>) {
match event {
Ok(e) => match e.kind {
EventKind::Modify(ModifyKind::Data(_)) | EventKind::Create(..) | EventKind::Remove(..) => {
if e.paths[0].to_str().unwrap().ends_with("yaml") {
if start.elapsed() > Duration::from_secs(2) {
start = Instant::now();
// info!("Config File changed :=> {:?}", e);
let snd = load_configuration(file_path, "filepath").await.0;
match snd {
Some(snd) => {
toreturn.send(snd).await.unwrap();
}
None => {}
}
if e.paths[0].to_str().unwrap().ends_with("yaml") && start.elapsed() > Duration::from_secs(2) {
start = Instant::now();
// info!("Config File changed :=> {:?}", e);
let snd = load_configuration(file_path, "filepath").await.0;
if let Some(snd) = snd {
toreturn.send(snd).await.unwrap();
}
}
}

18
src/utils/healthcheck.rs
View File

@@ -52,8 +52,8 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str, client: &Clie
let path = path_entry.key();
let mut innervec = Vec::new();
for (_, upstream) in path_entry.value().0.iter().enumerate() {
let tls = detect_tls(&upstream.address.to_string(), &upstream.port, &client).await;
for upstream in path_entry.value().0.iter() {
let tls = detect_tls(upstream.address.as_ref(), &upstream.port, client).await;
let is_h2 = matches!(tls.1, Some(Version::HTTP_2));
let link = if tls.0 {
@@ -75,7 +75,7 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str, client: &Clie
};
if scheme.healthcheck.unwrap_or(true) {
let resp = http_request(&link, method, "", &client).await;
let resp = http_request(&link, method, "", client).await;
if resp.0 {
if resp.1 {
scheme.is_http2 = is_h2; // could be adjusted further
@@ -109,12 +109,12 @@ async fn http_request(url: &str, method: &str, payload: &str, client: &Client) -
}
}
match send_request(&client, method, url, payload).await {
match send_request(client, method, url, payload).await {
Some(response) => {
let status = response.status().as_u16();
((99..499).contains(&status), false)
}
None => (ping_grpc(&url).await, true),
None => (ping_grpc(url).await, true),
}
}
@@ -128,12 +128,8 @@ pub async fn ping_grpc(addr: &str) -> bool {
async fn detect_tls(ip: &str, port: &u16, client: &Client) -> (bool, Option<Version>) {
let https_url = format!("https://{}:{}", ip, port);
match client.get(&https_url).send().await {
Ok(response) => {
// println!("{} => {:?} (HTTPS)", https_url, response.version());
return (true, Some(response.version()));
}
_ => {}
if let Ok(response) = client.get(&https_url).send().await {
return (true, Some(response.version()));
}
let http_url = format!("http://{}:{}", ip, port);
match client.get(&http_url).send().await {

11
src/utils/httpclient.rs
View File

@@ -23,11 +23,8 @@ pub async fn for_consul(url: String, token: Option<String>, conf: &GlobalService
let upstreams: DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)> = DashMap::new();
let endpoints: Vec<ConsulService> = resp.json().await.ok()?;
for subsets in endpoints {
// let addr = subsets.tagged_addresses.get("lan_ipv4").unwrap().address.clone();
// let prt = subsets.tagged_addresses.get("lan_ipv4").unwrap().port.clone();
let addr = subsets.tagged_addresses.get("lan_ipv4").unwrap().address.clone();
let prt = subsets.tagged_addresses.get("lan_ipv4").unwrap().port.clone();
// let redirect_link = conf.redirect_to.as_ref().map(|www| Arc::from(www.as_str()));
let prt = subsets.tagged_addresses.get("lan_ipv4").unwrap().port;
let to_add = Arc::from(InnerMap {
address: Arc::from(&*addr),
port: prt,
@@ -41,7 +38,7 @@ pub async fn for_consul(url: String, token: Option<String>, conf: &GlobalService
});
inner_vec.push(to_add);
}
match_path(&conf, &upstreams, inner_vec.clone());
match_path(conf, &upstreams, inner_vec);
Some(upstreams)
}
@@ -66,7 +63,7 @@ pub async fn for_kuber(url: &str, token: &str, conf: &GlobalServiceMapping) -> O
// let redirect_link = conf.redirect_to.as_ref().map(|www| Arc::from(www.as_str()));
let to_add = Arc::from(InnerMap {
address: Arc::from(addr.ip.clone()),
port: port.port.clone(),
port: port.port,
is_ssl: false,
is_http2: false,
to_https: conf.to_https.unwrap_or(false),
@@ -78,7 +75,7 @@ pub async fn for_kuber(url: &str, token: &str, conf: &GlobalServiceMapping) -> O
inner_vec.push(to_add);
}
}
match_path(&conf, &upstreams, inner_vec.clone());
match_path(conf, &upstreams, inner_vec.clone());
}
}
}

11
src/utils/kuberconsul.rs
View File

@@ -52,12 +52,13 @@ pub struct ConsulTaggedAddress {
#[serde(rename = "Port")]
pub port: u16,
}
#[allow(clippy::type_complexity)]
pub fn list_to_upstreams(lt: Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>>, upstreams: &UpstreamsDashMap, i: &GlobalServiceMapping) {
if let Some(list) = lt {
match upstreams.get(&*i.hostname.clone()) {
Some(upstr) => {
for (k, v) in list {
upstr.value().insert(Arc::from(k.to_owned()), v);
upstr.value().insert(k.to_owned(), v);
}
}
None => {
@@ -134,7 +135,7 @@ impl ServiceDiscovery for KubernetesDiscovery {
}
let url = format!("https://{}/api/v1/namespaces/{}/endpoints/{}", server, namespace, service.hostname);
// let url = format!("https://{}/api/v1/namespaces/{}/endpoints?labelSelector=app", server, namespace);
let list = httpclient::for_kuber(&*url, &*token, &service).await;
let list = httpclient::for_kuber(&url, &token, &service).await;
// println!("{:?}", list);
list_to_upstreams(list, &upstreams, &service);
}
@@ -209,7 +210,7 @@ impl ServiceDiscovery for ConsulDiscovery {
}
}
async fn clone_compare(upstreams: &UpstreamsDashMap, prev_upstreams: &UpstreamsDashMap, config: &Arc<Configuration>) -> Option<Configuration> {
if !compare_dashmaps(&upstreams, &prev_upstreams) {
if !compare_dashmaps(upstreams, prev_upstreams) {
let tosend: Configuration = Configuration {
upstreams: Default::default(),
client_headers: config.client_headers.clone(),
@@ -219,8 +220,8 @@ async fn clone_compare(upstreams: &UpstreamsDashMap, prev_upstreams: &UpstreamsD
typecfg: config.typecfg.clone(),
extraparams: config.extraparams.clone(),
};
clone_dashmap_into(&upstreams, &prev_upstreams);
clone_dashmap_into(&upstreams, &tosend.upstreams);
clone_dashmap_into(upstreams, prev_upstreams);
clone_dashmap_into(upstreams, &tosend.upstreams);
print_upstreams(&tosend.upstreams);
return Some(tosend);
};

10
src/utils/metrics.rs
View File

@@ -52,11 +52,11 @@ pub fn calc_metrics(metric_types: &MetricTypes) {
let timer = REQUEST_LATENCY.start_timer();
timer.observe_duration();
let version_str = match &metric_types.version {
&Version::HTTP_11 => "HTTP/1.1",
&Version::HTTP_2 => "HTTP/2.0",
&Version::HTTP_3 => "HTTP/3.0",
&Version::HTTP_10 => "HTTP/1.0",
let version_str = match metric_types.version {
Version::HTTP_11 => "HTTP/1.1",
Version::HTTP_2 => "HTTP/2.0",
Version::HTTP_3 => "HTTP/3.0",
Version::HTTP_10 => "HTTP/1.0",
_ => "Unknown",
};
REQUESTS_BY_VERSION.with_label_values(&[&version_str]).inc();

8
src/utils/parceyaml.rs
View File

@@ -10,7 +10,7 @@ use std::sync::atomic::AtomicUsize;
use std::sync::{Arc, LazyLock};
use std::{env, fs};
pub static DOMAINS: LazyLock<DashMap<String, bool>> = LazyLock::new(|| DashMap::new());
pub static DOMAINS: LazyLock<DashMap<String, bool>> = LazyLock::new(DashMap::new);
pub async fn load_configuration(d: &str, kind: &str) -> (Option<Configuration>, String) {
let mut conf_files = Vec::new();
@@ -21,7 +21,7 @@ pub async fn load_configuration(d: &str, kind: &str) -> (Option<Configuration>,
let mut autocfg = Path::new(d).parent().unwrap().to_path_buf();
autocfg.push("autoconfigs");
if !fs::metadata(autocfg.clone()).is_ok() {
if fs::metadata(autocfg.clone()).is_err() {
fs::create_dir_all(autocfg.clone()).ok();
}
autocfg.push("domains.json");
@@ -228,8 +228,8 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) {
pub fn parce_main_config(path: &str) -> AppConfig {
let data = fs::read_to_string(path).unwrap();
let reply = DashMap::new();
let cfg: HashMap<String, String> = serde_yml::from_str(&*data).expect("Failed to parse main config file");
let mut cfo: AppConfig = serde_yml::from_str(&*data).expect("Failed to parse main config file");
let cfg: HashMap<String, String> = serde_yml::from_str(&data).expect("Failed to parse main config file");
let mut cfo: AppConfig = serde_yml::from_str(&data).expect("Failed to parse main config file");
log_builder(&cfo);
cfo.hc_method = cfo.hc_method.to_uppercase();
for (k, v) in cfg {

30
src/utils/tools.rs
View File

@@ -150,7 +150,7 @@ pub fn merge_headers(target: &DashMap<Arc<str>, Vec<(String, Arc<str>)>>, source
for entry in source.iter() {
let global_key = entry.key().clone();
let global_values = entry.value().clone();
let mut target_entry = target.entry(global_key).or_insert_with(Vec::new);
let mut target_entry = target.entry(global_key).or_default();
target_entry.extend(global_values);
}
}
@@ -198,7 +198,7 @@ pub fn clone_idmap_into(original: &UpstreamsDashMap, cloned: &UpstreamsIdMap) {
authorization: None,
};
cloned.insert(id, Arc::from(to_add));
cloned.insert(hh, Arc::from(x.to_owned()));
cloned.insert(hh, x.to_owned());
// println!("CLONNED :===========> {:?}", cloned);
}
new_inner_map.insert(path.clone(), new_vec);
@@ -268,14 +268,14 @@ pub fn drop_priv(user: String, group: String, http_addr: String, tls_addr: Optio
thread::sleep(time::Duration::from_millis(10));
loop {
thread::sleep(time::Duration::from_millis(10));
if port_is_available(http_addr.clone()) {
if TcpListener::bind(&http_addr).is_err() {
break;
}
}
if let Some(tls_addr) = tls_addr {
loop {
thread::sleep(time::Duration::from_millis(10));
if port_is_available(tls_addr.clone()) {
if TcpListener::bind(&tls_addr).is_err() {
break;
}
}
@@ -287,24 +287,14 @@ pub fn drop_priv(user: String, group: String, http_addr: String, tls_addr: Optio
}
}
fn port_is_available(addr: String) -> bool {
match TcpListener::bind(addr) {
Ok(_) => false,
Err(_) => true,
}
}
pub fn check_priv(addr: &str) {
let port = SocketAddr::from_str(addr).map(|sa| sa.port()).unwrap();
match port < 1024 {
true => {
let meta = std::fs::metadata("/proc/self").map(|m| m.uid()).unwrap();
if meta != 0 {
error!("Running on privileged port requires to start as ROOT");
process::exit(1)
}
if port < 1024 {
let meta = std::fs::metadata("/proc/self").map(|m| m.uid()).unwrap();
if meta != 0 {
error!("Running on privileged port requires to start as ROOT");
process::exit(1)
}
false => {}
}
}
@@ -397,7 +387,7 @@ pub fn prepend(prefix: &str, val: &Option<Arc<str>>, uri: &str, port: &str) -> O
let mut buf = String::with_capacity(32);
buf.push_str(prefix);
buf.push_str(s);
buf.push_str(":");
buf.push(':');
buf.push_str(port);
buf.push_str(uri);
buf

107
src/web/bgservice.rs
View File

@@ -32,19 +32,20 @@ impl BackgroundService for LB {
let file_load = FromFileProvider {
path: self.config.upstreams_conf.clone(),
};
let _ = tokio::spawn(async move { file_load.start(tx).await });
// let _ = tokio::spawn(async move { file_load.start(tx).await });
drop(tokio::spawn(async move { file_load.start(tx).await }));
}
"kubernetes" => {
info!("Running Kubernetes discovery, requested type is: {}", config.typecfg);
let cf = Arc::from(config);
let kuber_load = KubernetesProvider { config: cf.clone() };
let _ = tokio::spawn(async move { kuber_load.start(tx).await });
drop(tokio::spawn(async move { kuber_load.start(tx).await }));
}
"consul" => {
info!("Running Consul discovery, requested type is: {}", config.typecfg);
let cf = Arc::from(config);
let consul_load = ConsulProvider { config: cf.clone() };
let _ = tokio::spawn(async move { consul_load.start(tx).await });
drop(tokio::spawn(async move { consul_load.start(tx).await }));
}
_ => {
error!("Unknown discovery type: {}", config.typecfg);
@@ -57,7 +58,7 @@ impl BackgroundService for LB {
let api_load = APIUpstreamProvider {
address: self.config.config_address.clone(),
masterkey: self.config.master_key.clone(),
config_api_enabled: self.config.config_api_enabled.clone(),
config_api_enabled: self.config.config_api_enabled,
// certs_dir: self.config.proxy_certificates.clone().unwrap_or_else(|| "/tmp".to_string()),
config_dir: confdir.clone(),
certs_dir: certdir.clone(),
@@ -71,14 +72,16 @@ impl BackgroundService for LB {
};
// let crtdir = api_load.certs_dir.clone();
// let tx_api = tx.clone();
let _ = tokio::spawn(async move { api_load.start(tx_api).await });
drop(tokio::spawn(async move { api_load.start(tx_api).await }));
let uu = self.ump_upst.clone();
let ff = self.ump_full.clone();
let im = self.ump_byid.clone();
let (hc_method, hc_interval) = (self.config.hc_method.clone(), self.config.hc_interval);
let _ = tokio::spawn(async move { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await });
let _ = tokio::spawn(async move { refresh_order(certdir, confdir).await });
drop(tokio::spawn(async move {
healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await
}));
drop(tokio::spawn(async move { refresh_order(certdir, confdir).await }));
loop {
tokio::select! {
@@ -86,57 +89,49 @@ impl BackgroundService for LB {
break;
}
val = rx.next() => {
match val {
Some(ss) => {
clone_dashmap_into(&ss.upstreams, &self.ump_full);
clone_dashmap_into(&ss.upstreams, &self.ump_upst);
let current = self.extraparams.load_full();
let mut new = (*current).clone();
new.to_https = ss.extraparams.to_https;
new.sticky_sessions = ss.extraparams.sticky_sessions;
new.authentication = ss.extraparams.authentication.clone();
new.rate_limit = ss.extraparams.rate_limit;
self.extraparams.store(Arc::new(new));
self.client_headers.clear();
self.server_headers.clear();
if let Some(ss) = val {
clone_dashmap_into(&ss.upstreams, &self.ump_full);
clone_dashmap_into(&ss.upstreams, &self.ump_upst);
let current = self.extraparams.load_full();
let mut new = (*current).clone();
new.to_https = ss.extraparams.to_https;
new.sticky_sessions = ss.extraparams.sticky_sessions;
new.authentication = ss.extraparams.authentication.clone();
new.rate_limit = ss.extraparams.rate_limit;
self.extraparams.store(Arc::new(new));
self.client_headers.clear();
self.server_headers.clear();
for entry in ss.upstreams.iter() {
let global_key = entry.key().clone();
let client_global_values = DashMap::new();
let server_global_values = DashMap::new();
for entry in ss.upstreams.iter() {
let global_key = entry.key().clone();
let client_global_values = DashMap::new();
let server_global_values = DashMap::new();
let mut client_target_entry = ss.client_headers.entry(global_key.clone()).or_insert_with(DashMap::new);
client_target_entry.extend(client_global_values);
let mut server_target_entry = ss.server_headers.entry(global_key).or_insert_with(DashMap::new);
server_target_entry.extend(server_global_values);
self.server_headers.insert(server_target_entry.key().to_owned(), server_target_entry.value().to_owned());
}
for path in ss.client_headers.iter() {
let path_key = path.key().clone();
let path_headers = path.value().clone();
self.client_headers.insert(path_key.clone(), path_headers);
if let Some(global_headers) = ss.client_headers.get("GLOBAL_CLIENT_HEADERS") {
if let Some(existing_headers) = self.client_headers.get_mut(&path_key) {
merge_headers(&existing_headers, &global_headers);
}
}
}
for path in ss.server_headers.iter() {
let path_key = path.key().clone();
let path_headers = path.value().clone();
self.server_headers.insert(path_key.clone(), path_headers);
if let Some(global_headers) = ss.server_headers.get("GLOBAL_SERVER_HEADERS") {
if let Some(existing_headers) = self.server_headers.get_mut(&path_key) {
merge_headers(&existing_headers, &global_headers);
}
}
}
// info!("Upstreams list is changed, updating to:");
// print_upstreams(&self.ump_full);
let mut client_target_entry = ss.client_headers.entry(global_key.clone()).or_insert_with(DashMap::new);
client_target_entry.extend(client_global_values);
let mut server_target_entry = ss.server_headers.entry(global_key).or_insert_with(DashMap::new);
server_target_entry.extend(server_global_values);
self.server_headers.insert(server_target_entry.key().to_owned(), server_target_entry.value().to_owned());
}
for path in ss.client_headers.iter() {
let path_key = path.key().clone();
let path_headers = path.value().clone();
self.client_headers.insert(path_key.clone(), path_headers);
if let Some(global_headers) = ss.client_headers.get("GLOBAL_CLIENT_HEADERS") {
if let Some(existing_headers) = self.client_headers.get_mut(&path_key) {
merge_headers(&existing_headers, &global_headers);
}
}
}
for path in ss.server_headers.iter() {
let path_key = path.key().clone();
let path_headers = path.value().clone();
self.server_headers.insert(path_key.clone(), path_headers);
if let Some(global_headers) = ss.server_headers.get("GLOBAL_SERVER_HEADERS") {
if let Some(existing_headers) = self.server_headers.get_mut(&path_key) {
merge_headers(&existing_headers, &global_headers);
}
}
}
None => {}
}
}
}

6
src/web/proxyhttp.rs
View File

@@ -25,7 +25,7 @@ use tokio::time::Instant;
// static RATE_LIMITER: Lazy<Rate> = Lazy::new(|| Rate::new(Duration::from_secs(1)));
// static REVERSE_STORE: Lazy<DashMap<String, String>> = Lazy::new(|| DashMap::new());
static REVERSE_STORE: LazyLock<DashMap<String, String>> = LazyLock::new(|| DashMap::new());
static REVERSE_STORE: LazyLock<DashMap<String, String>> = LazyLock::new(DashMap::new);
thread_local! {static IP_BUFFER: RefCell<String> = RefCell::new(String::with_capacity(50));}
pub static RATE_LIMITER: LazyLock<Rate> = LazyLock::new(|| Rate::new(Duration::from_secs(1)));
@@ -132,8 +132,8 @@ impl ProxyHttp for LB {
s.push_str("https://");
s.push_str(host);
if port != "443" {
s.push_str(":");
s.push_str(&port);
s.push(':');
s.push_str(port);
}
s.push_str(uri);
let mut resp = ResponseHeader::build(StatusCode::MOVED_PERMANENTLY, None)?;

80
src/web/start.rs
View File

@@ -18,7 +18,7 @@ use std::sync::Arc;
use std::{fs, thread};
pub fn run() {
// default_provider().install_default().expect("Failed to install rustls crypto provider");
let parameters = Some(Opt::parse_args()).unwrap();
let parameters = Opt::parse_args();
let file = parameters.conf.clone().unwrap();
let maincfg = crate::utils::parceyaml::parce_main_config(file.as_str());
@@ -60,50 +60,44 @@ pub fn run() {
check_priv(bind_address_http.as_str());
match bind_address_tls {
Some(bind_address_tls) => {
check_priv(bind_address_tls.as_str());
let (tx, rx): (Sender<Vec<CertificateConfig>>, Receiver<Vec<CertificateConfig>>) = channel();
let certs_path = cfg.proxy_configs.clone().unwrap() + "/certificates";
if let Some(bind_address_tls) = bind_address_tls {
check_priv(bind_address_tls.as_str());
let (tx, rx): (Sender<Vec<CertificateConfig>>, Receiver<Vec<CertificateConfig>>) = channel();
let certs_path = cfg.proxy_configs.clone().unwrap() + "/certificates";
if !fs::metadata(certs_path.clone()).is_ok() {
fs::create_dir_all(certs_path.clone()).unwrap();
}
thread::spawn(move || {
watch_folder(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"));
let certificates = Arc::new(ArcSwap::from_pointee(first_set));
let certs_for_callback = certificates.clone();
let certs_for_watcher = certificates.clone();
let new_certs = load::Certificates::new(&certificate_configs, grade.as_str());
certs_for_watcher.store(Arc::new(new_certs.unwrap()));
let mut tls_settings =
TlsSettings::intermediate(&certs_for_callback.load().default_cert_path, &certs_for_callback.load().default_key_path).expect("unable to load or parse cert/key");
grades::set_tsl_grade(&mut tls_settings, grade.as_str());
tls_settings.set_servername_callback(move |ssl_ref: &mut SslRef, ssl_alert: &mut SslAlert| certs_for_callback.load().server_name_callback(ssl_ref, ssl_alert));
tls_settings.set_alpn_select_callback(grades::prefer_h2);
proxy.add_tls_with_settings(&bind_address_tls, None, tls_settings);
let certs_for_watcher = certificates.clone();
thread::spawn(move || {
while let Ok(new_configs) = rx.recv() {
let new_certs = load::Certificates::new(&new_configs, grade.as_str());
match new_certs {
Some(new_certs) => {
certs_for_watcher.store(Arc::new(new_certs));
}
None => {}
};
}
});
if fs::metadata(certs_path.clone()).is_err() {
fs::create_dir_all(certs_path.clone()).unwrap();
}
None => {}
thread::spawn(move || {
watch_folder(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"));
let certificates = Arc::new(ArcSwap::from_pointee(first_set));
let certs_for_callback = certificates.clone();
let certs_for_watcher = certificates.clone();
let new_certs = load::Certificates::new(&certificate_configs, grade.as_str());
certs_for_watcher.store(Arc::new(new_certs.unwrap()));
let mut tls_settings =
TlsSettings::intermediate(&certs_for_callback.load().default_cert_path, &certs_for_callback.load().default_key_path).expect("unable to load or parse cert/key");
grades::set_tsl_grade(&mut tls_settings, grade.as_str());
tls_settings.set_servername_callback(move |ssl_ref: &mut SslRef, ssl_alert: &mut SslAlert| certs_for_callback.load().server_name_callback(ssl_ref, ssl_alert));
tls_settings.set_alpn_select_callback(grades::prefer_h2);
proxy.add_tls_with_settings(&bind_address_tls, None, tls_settings);
let certs_for_watcher = certificates.clone();
thread::spawn(move || {
while let Ok(new_configs) = rx.recv() {
let new_certs = load::Certificates::new(&new_configs, grade.as_str());
if let Some(new_certs) = new_certs {
certs_for_watcher.store(Arc::new(new_certs));
};
}
});
}
info!("Running HTTP listener on :{}", bind_address_http.as_str());
proxy.add_tcp(bind_address_http.as_str());

17
src/web/webserver.rs
View File

@@ -85,7 +85,7 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: Sender<Conf
let static_files = ServeDir::new(folder);
let static_serve: Router = Router::new().fallback_service(static_files);
let static_listen = TcpListener::bind(address).await.unwrap();
let _ = tokio::spawn(async move { axum::serve(static_listen, static_serve).await.unwrap() });
drop(tokio::spawn(async move { axum::serve(static_listen, static_serve).await.unwrap() }));
}
let listener = TcpListener::bind(config.address.clone()).await.unwrap();
@@ -103,7 +103,7 @@ async fn conf(State(st): State<AppState>, Query(params): Query<HashMap<String, S
let parsed = serde_yml::from_str::<Config>(strcontent);
match parsed {
Ok(_) => {
let _ = tokio::spawn(async move { apply_config(content.as_str(), st).await });
drop(tokio::spawn(async move { apply_config(content.as_str(), st).await }));
return Response::builder().status(StatusCode::OK).body(Body::from("Accepted! Applying in background\n")).unwrap();
}
Err(err) => {
@@ -172,8 +172,9 @@ async fn metrics() -> impl IntoResponse {
.unwrap()
}
#[allow(clippy::needless_return)]
async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String, String>>) -> impl IntoResponse {
if let Some(_) = params.get("live") {
if params.contains_key("live") {
let r = upstreams_liveness_json(&st.full_upstreams, &st.current_upstreams);
return Response::builder()
.status(StatusCode::OK)
@@ -181,7 +182,7 @@ async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String,
.body(Body::from(format!("{}", r)))
.unwrap();
}
if let Some(_) = params.get("all") {
if params.contains_key("all") {
let resp = upstreams_to_json(&st.current_upstreams);
match resp {
Ok(j) => {
@@ -201,16 +202,17 @@ async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String,
}
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from(format!("{}", "Parameter mismatch")))
.body(Body::from("Parameter mismatch"))
.unwrap()
}
#[allow(clippy::needless_return)]
async fn acme_create(State(state): State<AppState>, Query(params): Query<HashMap<String, String>>, headers: HeaderMap) -> impl IntoResponse {
if !key_authorization(&headers, &params, &state.master_key) {
return Response::builder().status(StatusCode::FORBIDDEN).body(Body::from("Access Denied !\n")).unwrap();
}
let _ = match account::load_or_create(state.cert_creds.as_str()).await {
match account::load_or_create(state.cert_creds.as_str()).await {
Ok(txt) => {
return Response::builder()
.status(StatusCode::OK)
@@ -226,6 +228,7 @@ async fn acme_create(State(state): State<AppState>, Query(params): Query<HashMap
}
};
}
#[allow(clippy::needless_return)]
async fn acme_order(
State(state): State<AppState>,
axum::extract::Path(domain): axum::extract::Path<String>,
@@ -237,7 +240,7 @@ async fn acme_order(
}
let domain_clean = domain.trim_matches('/');
let _ = match order::order(domain_clean, state.cert_creds.as_str(), state.certs_dir).await {
match order::order(domain_clean, state.cert_creds.as_str(), state.certs_dir).await {
Ok(txt) => {
return Response::builder()
.status(StatusCode::OK)