From fb47ad149cceeaca31371f76d52ac74d212b3e56 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sat, 22 Aug 2026 16:22:07 +0300 Subject: [PATCH] Rustfmt --- src/api/config_edit.rs | 11 +- src/api/config_edit/tests.rs | 12 +- src/api/config_store.rs | 6 +- src/api/config_store/persistence.rs | 13 +- src/api/config_store/tests.rs | 10 +- src/api/users/create.rs | 1 - src/api/users/lifecycle.rs | 2 - src/api/users/links.rs | 1 - src/api/users/tests.rs | 566 +++++++++--------- src/api/users/update.rs | 2 - src/api/users/view.rs | 1 - src/config/hot_reload/diff.rs | 3 - src/config/hot_reload/fields.rs | 2 - src/config/hot_reload/tests.rs | 484 +++++++-------- src/config/hot_reload/watcher.rs | 2 - src/config/load.rs | 6 +- src/config/load/decode.rs | 14 +- src/config/load/effective.rs | 6 +- src/config/load/validate_core.rs | 3 +- src/config/load/validate_me.rs | 9 +- src/config/load/validate_runtime.rs | 12 +- src/config/load/validate_server.rs | 3 +- src/config/mod.rs | 2 +- .../tests/load_basic_tests/api_tests.rs | 2 - .../tests/load_basic_tests/conntrack_tests.rs | 1 - .../load_basic_tests/defaults_access_tests.rs | 1 - .../load_basic_tests/legacy_policy_tests.rs | 1 - .../tests/load_basic_tests/me_route_tests.rs | 1 - .../load_basic_tests/me_startup_tests.rs | 1 - .../load_basic_tests/synlimit_mss_tests.rs | 2 - .../tests/load_basic_tests/tls_fetch_tests.rs | 1 - src/config/types.rs | 17 +- src/config/types/links.rs | 1 - src/config/types/policies.rs | 1 - src/maestro/bootstrap.rs | 4 +- src/maestro/listeners/accept.rs | 5 +- src/maestro/listeners/bind.rs | 13 +- src/maestro/listeners/control.rs | 4 +- src/maestro/listeners/plan.rs | 5 +- src/maestro/listeners/unix.rs | 3 +- src/maestro/orchestrator.rs | 4 +- src/maestro/reload.rs | 8 +- src/maestro/reload_supervisor.rs | 7 +- src/maestro/runtime_build.rs | 8 +- src/maestro/runtime_startup.rs | 10 +- src/proxy/handshake.rs | 9 +- src/proxy/handshake/auth_candidates.rs | 24 +- src/proxy/handshake/auth_probe.rs | 19 +- src/proxy/handshake/mtproto.rs | 1 - src/proxy/handshake/nonce.rs | 1 - src/proxy/handshake/session.rs | 1 - src/proxy/handshake/tls_handshake.rs | 2 - src/synlimit_control/mod.rs | 4 +- src/synlimit_control/pf.rs | 12 +- src/synlimit_control/pf_runtime_tests.rs | 5 +- .../socket/fragmented_send_wire_tests.rs | 32 +- src/transport/socket/tests.rs | 4 +- 57 files changed, 675 insertions(+), 710 deletions(-) diff --git a/src/api/config_edit.rs b/src/api/config_edit.rs index 984bb4d..b9a8fd2 100644 --- a/src/api/config_edit.rs +++ b/src/api/config_edit.rs @@ -58,10 +58,10 @@ pub(super) async fn patch_config( let reservation = if let Some(request) = reload_request.filter(|_| resolved.runtime_changed) { Some( shared - .reload_control - .reserve(prepared.response.revision.clone(), request) - .await - .map_err(reload_submit_failure)?, + .reload_control + .reserve(prepared.response.revision.clone(), request) + .await + .map_err(reload_submit_failure)?, ) } else { None @@ -176,8 +176,7 @@ async fn prepare_patch_to_path( for section in &touched { if *section == "server" { let rendered = render_server_listeners(&requested_cfg)?; - owner_contents = - upsert_toml_table(&owner_contents, "server.listeners", &rendered); + owner_contents = upsert_toml_table(&owner_contents, "server.listeners", &rendered); } else { let rendered = render_top_level_section(&requested_cfg, section)?; owner_contents = upsert_toml_table(&owner_contents, section, &rendered); diff --git a/src/api/config_edit/tests.rs b/src/api/config_edit/tests.rs index a5e9f52..f738449 100644 --- a/src/api/config_edit/tests.rs +++ b/src/api/config_edit/tests.rs @@ -51,8 +51,7 @@ async fn patch_revision_conflict() { #[tokio::test] async fn patch_sni_reports_restart_required() { - let (path, _d) = - temp_config("[censorship]\ntls_domain = \"a.com\"\n[server]\nport = 443\n"); + let (path, _d) = temp_config("[censorship]\ntls_domain = \"a.com\"\n[server]\nport = 443\n"); let patch: Json = serde_json::json!({"censorship": {"tls_domain": "b.com"}}); let resp = apply_patch_to_path(&path, &patch, None).await.unwrap(); assert!(resp.restart_required); @@ -252,9 +251,7 @@ async fn patch_rejects_multiple_source_owners_without_writing() { "censorship": {"tls_domain": "new.example"} }); - let error = apply_patch_to_path(&root, &patch, None) - .await - .unwrap_err(); + let error = apply_patch_to_path(&root, &patch, None).await.unwrap_err(); assert_eq!(error.code, "config_patch_not_atomic"); assert_eq!(tokio::fs::read_to_string(&root).await.unwrap(), root_body); @@ -276,10 +273,7 @@ async fn unavailable_reload_coordinator_is_detected_before_config_write() { drop(receiver); let error = match control - .reserve( - prepared.response.revision.clone(), - ReloadRequest::default(), - ) + .reserve(prepared.response.revision.clone(), ReloadRequest::default()) .await { Ok(_) => panic!("closed reload coordinator must reject the reservation"), diff --git a/src/api/config_store.rs b/src/api/config_store.rs index ff40032..a034882 100644 --- a/src/api/config_store.rs +++ b/src/api/config_store.rs @@ -11,14 +11,12 @@ use super::model::ApiFailure; // Source-preserving TOML rendering and atomic persistence helpers. mod persistence; +#[cfg(test)] +use persistence::{find_toml_table_bounds, render_access_section, save_sections_to_disk}; pub(in crate::api) use persistence::{ render_server_listeners, render_top_level_section, save_access_sections_to_disk, upsert_toml_table, write_atomic, }; -#[cfg(test)] -use persistence::{ - find_toml_table_bounds, render_access_section, save_sections_to_disk, -}; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum AccessSection { diff --git a/src/api/config_store/persistence.rs b/src/api/config_store/persistence.rs index f7d14a8..8893e9e 100644 --- a/src/api/config_store/persistence.rs +++ b/src/api/config_store/persistence.rs @@ -7,13 +7,13 @@ use serde::Serialize; use crate::config::{ProxyConfig, RateLimitBps}; +#[cfg(test)] +use super::compute_revision; use super::{ AccessSection, compute_snapshot_revision, load_candidate_snapshot, load_config_snapshot, resolve_single_source_owner, toml_path_exists, }; use crate::api::model::ApiFailure; -#[cfg(test)] -use super::compute_revision; /// Re-render the given top-level tables from `cfg` and upsert each into the /// on-disk file, preserving every untouched section (and its comments). @@ -80,9 +80,7 @@ pub(in crate::api) fn render_top_level_section( } /// Renders normalized listener entries as nested array-of-table blocks. -pub(in crate::api) fn render_server_listeners( - cfg: &ProxyConfig, -) -> Result { +pub(in crate::api) fn render_server_listeners(cfg: &ProxyConfig) -> Result { let mut out = String::new(); for listener in &cfg.server.listeners { out.push_str("[[server.listeners]]\n"); @@ -339,10 +337,7 @@ fn header_belongs_to(header: &str, table_name: &str) -> bool { /// [`find_all_table_blocks`] for the full set of (possibly scattered) blocks. /// Locates one complete TOML table block in source text. #[cfg(test)] -pub(super) fn find_toml_table_bounds( - source: &str, - table_name: &str, -) -> Option<(usize, usize)> { +pub(super) fn find_toml_table_bounds(source: &str, table_name: &str) -> Option<(usize, usize)> { find_all_table_blocks(source, table_name).into_iter().next() } diff --git a/src/api/config_store/tests.rs b/src/api/config_store/tests.rs index 85f53cb..ef35909 100644 --- a/src/api/config_store/tests.rs +++ b/src/api/config_store/tests.rs @@ -31,8 +31,7 @@ async fn save_sections_preserves_other_tables_and_comments() { #[test] fn find_bounds_matches_array_of_tables() { - let src = - "[server]\nport = 1\n\n[[upstreams]]\nkind = \"a\"\n\n[[upstreams]]\nkind = \"b\"\n"; + let src = "[server]\nport = 1\n\n[[upstreams]]\nkind = \"a\"\n\n[[upstreams]]\nkind = \"b\"\n"; let bounds = find_toml_table_bounds(src, "upstreams"); assert!(bounds.is_some(), "should locate [[upstreams]] block start"); let (start, end) = bounds.unwrap(); @@ -242,8 +241,7 @@ async fn access_mutation_writes_only_the_single_included_owner() { let root = dir.path().join("config.toml"); let included = dir.path().join("users.toml"); let root_body = "include = \"users.toml\"\n[censorship]\ntls_domain = \"example.com\"\n"; - let included_body = - "[access.users]\nalice = \"00000000000000000000000000000000\"\n"; + let included_body = "[access.users]\nalice = \"00000000000000000000000000000000\"\n"; tokio::fs::write(&root, root_body).await.unwrap(); tokio::fs::write(&included, included_body).await.unwrap(); let mut cfg = load_config_from_disk(&root).await.unwrap(); @@ -303,8 +301,8 @@ fn render_user_rate_limits_section() { }, ); - let rendered = render_access_section(&cfg, AccessSection::UserRateLimits) - .expect("section must render"); + let rendered = + render_access_section(&cfg, AccessSection::UserRateLimits).expect("section must render"); assert!(rendered.starts_with("[access.user_rate_limits]\n")); assert!(rendered.contains("alice = { up_bps = 1024, down_bps = 2048 }")); diff --git a/src/api/users/create.rs b/src/api/users/create.rs index e123b4b..a46ed59 100644 --- a/src/api/users/create.rs +++ b/src/api/users/create.rs @@ -175,4 +175,3 @@ pub(in crate::api) async fn create_user( Ok((CreateUserResponse { user, secret }, revision)) } - diff --git a/src/api/users/lifecycle.rs b/src/api/users/lifecycle.rs index 1866e86..fd0a281 100644 --- a/src/api/users/lifecycle.rs +++ b/src/api/users/lifecycle.rs @@ -56,7 +56,6 @@ pub(in crate::api) async fn rotate_secret( )) } - pub(in crate::api) async fn delete_user( user: &str, expected_revision: Option, @@ -115,4 +114,3 @@ pub(in crate::api) async fn delete_user( Ok((user.to_string(), revision)) } - diff --git a/src/api/users/links.rs b/src/api/users/links.rs index 2d2cb96..abf4ed8 100644 --- a/src/api/users/links.rs +++ b/src/api/users/links.rs @@ -197,4 +197,3 @@ fn resolve_extra_tls_domains(cfg: &ProxyConfig) -> Vec<&str> { } domains } - diff --git a/src/api/users/tests.rs b/src/api/users/tests.rs index 45e2ec5..48fade7 100644 --- a/src/api/users/tests.rs +++ b/src/api/users/tests.rs @@ -1,292 +1,292 @@ - use super::*; - use crate::ip_tracker::UserIpTracker; - use crate::stats::Stats; +use super::*; +use crate::ip_tracker::UserIpTracker; +use crate::stats::Stats; - #[tokio::test] - async fn users_from_config_reports_effective_tcp_limit_with_global_fallback() { - let mut cfg = ProxyConfig::default(); +#[tokio::test] +async fn users_from_config_reports_effective_tcp_limit_with_global_fallback() { + let mut cfg = ProxyConfig::default(); + cfg.access.users.insert( + "alice".to_string(), + "0123456789abcdef0123456789abcdef".to_string(), + ); + cfg.access.user_max_tcp_conns_global_each = 7; + + let stats = Stats::new(); + let tracker = UserIpTracker::new(); + + let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; + let alice = users + .iter() + .find(|entry| entry.username == "alice") + .expect("alice must be present"); + assert!(!alice.in_runtime); + assert_eq!(alice.max_tcp_conns, Some(7)); + + cfg.access.user_max_tcp_conns.insert("alice".to_string(), 5); + let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; + let alice = users + .iter() + .find(|entry| entry.username == "alice") + .expect("alice must be present"); + assert!(!alice.in_runtime); + assert_eq!(alice.max_tcp_conns, Some(5)); + + cfg.access.user_max_tcp_conns.insert("alice".to_string(), 0); + let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; + let alice = users + .iter() + .find(|entry| entry.username == "alice") + .expect("alice must be present"); + assert!(!alice.in_runtime); + assert_eq!(alice.max_tcp_conns, Some(7)); + + cfg.access.user_max_tcp_conns_global_each = 0; + let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; + let alice = users + .iter() + .find(|entry| entry.username == "alice") + .expect("alice must be present"); + assert!(!alice.in_runtime); + assert_eq!(alice.max_tcp_conns, None); +} + +#[tokio::test] +async fn users_from_config_reports_user_rate_limits() { + let mut cfg = ProxyConfig::default(); + cfg.access.users.insert( + "alice".to_string(), + "0123456789abcdef0123456789abcdef".to_string(), + ); + cfg.access.user_rate_limits.insert( + "alice".to_string(), + RateLimitBps { + up_bps: 1024, + down_bps: 0, + }, + ); + + let stats = Stats::new(); + let tracker = UserIpTracker::new(); + + let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; + let alice = users + .iter() + .find(|entry| entry.username == "alice") + .expect("alice must be present"); + + assert_eq!(alice.rate_limit_up_bps, Some(1024)); + assert_eq!(alice.rate_limit_down_bps, None); +} + +#[tokio::test] +async fn users_from_config_reports_user_enabled_default_and_override() { + let mut cfg = ProxyConfig::default(); + cfg.access.users.insert( + "alice".to_string(), + "0123456789abcdef0123456789abcdef".to_string(), + ); + cfg.access.users.insert( + "bob".to_string(), + "fedcba9876543210fedcba9876543210".to_string(), + ); + cfg.access.user_enabled.insert("bob".to_string(), false); + + let stats = Stats::new(); + let tracker = UserIpTracker::new(); + let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; + let alice = users + .iter() + .find(|entry| entry.username == "alice") + .expect("alice must be present"); + let bob = users + .iter() + .find(|entry| entry.username == "bob") + .expect("bob must be present"); + + assert!(alice.enabled); + assert!(!bob.enabled); + + cfg.access.user_enabled.insert("bob".to_string(), true); + let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; + let bob = users + .iter() + .find(|entry| entry.username == "bob") + .expect("bob must be present"); + assert!(bob.enabled); +} + +#[tokio::test] +async fn users_from_config_marks_runtime_membership_when_snapshot_is_provided() { + let mut disk_cfg = ProxyConfig::default(); + disk_cfg.access.users.insert( + "alice".to_string(), + "0123456789abcdef0123456789abcdef".to_string(), + ); + disk_cfg.access.users.insert( + "bob".to_string(), + "fedcba9876543210fedcba9876543210".to_string(), + ); + + let mut runtime_cfg = ProxyConfig::default(); + runtime_cfg.access.users.insert( + "alice".to_string(), + "0123456789abcdef0123456789abcdef".to_string(), + ); + + let stats = Stats::new(); + let tracker = UserIpTracker::new(); + let users = + users_from_config(&disk_cfg, &stats, &tracker, None, None, Some(&runtime_cfg)).await; + + let alice = users + .iter() + .find(|entry| entry.username == "alice") + .expect("alice must be present"); + let bob = users + .iter() + .find(|entry| entry.username == "bob") + .expect("bob must be present"); + + assert!(alice.in_runtime); + assert!(!bob.in_runtime); +} + +#[tokio::test] +async fn users_from_config_returns_tls_link_for_each_tls_domain() { + let mut cfg = ProxyConfig::default(); + cfg.access.users.insert( + "alice".to_string(), + "0123456789abcdef0123456789abcdef".to_string(), + ); + cfg.general.modes.classic = false; + cfg.general.modes.secure = false; + cfg.general.modes.tls = true; + cfg.general.links.public_host = Some("proxy.example.net".to_string()); + cfg.general.links.public_port = Some(443); + cfg.censorship.tls_domain = "front-a.example.com".to_string(); + cfg.censorship.tls_domains = vec![ + "front-b.example.com".to_string(), + "front-c.example.com".to_string(), + "front-b.example.com".to_string(), + "front-a.example.com".to_string(), + ]; + + let stats = Stats::new(); + let tracker = UserIpTracker::new(); + let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; + let alice = users + .iter() + .find(|entry| entry.username == "alice") + .expect("alice must be present"); + + assert_eq!(alice.links.tls.len(), 3); + assert!( + alice + .links + .tls + .iter() + .any(|link| link.ends_with(&hex::encode("front-a.example.com"))) + ); + assert!( + alice + .links + .tls + .iter() + .any(|link| link.ends_with(&hex::encode("front-b.example.com"))) + ); + assert!( + alice + .links + .tls + .iter() + .any(|link| link.ends_with(&hex::encode("front-c.example.com"))) + ); + assert_eq!(alice.links.tls_domains.len(), 2); + assert!( + alice + .links + .tls_domains + .iter() + .any(|entry| entry.domain == "front-b.example.com" + && entry.link.ends_with(&hex::encode("front-b.example.com"))) + ); + assert!( + alice + .links + .tls_domains + .iter() + .any(|entry| entry.domain == "front-c.example.com" + && entry.link.ends_with(&hex::encode("front-c.example.com"))) + ); + assert!( + !alice + .links + .tls_domains + .iter() + .any(|entry| entry.domain == "front-a.example.com") + ); +} + +#[test] +fn build_user_quota_list_skips_users_without_positive_quota_and_sorts_by_username() { + let mut cfg = ProxyConfig::default(); + cfg.access.users.insert( + "alice".to_string(), + "0123456789abcdef0123456789abcdef".to_string(), + ); + cfg.access.users.insert( + "bob".to_string(), + "fedcba9876543210fedcba9876543210".to_string(), + ); + cfg.access.users.insert( + "carol".to_string(), + "aaaabbbbccccddddeeeeffff00001111".to_string(), + ); + // alice has a positive quota and should be listed. + cfg.access + .user_data_quota + .insert("alice".to_string(), 1 << 20); + // bob has no quota entry at all (None) — should be skipped. + // carol has an explicit zero quota — should be skipped. + cfg.access.user_data_quota.insert("carol".to_string(), 0); + + let stats = Stats::new(); + // Charge some traffic against alice; carol gets traffic too but should + // still be filtered out by the quota check. + let alice_stats = stats.get_or_create_user_stats_handle("alice"); + stats.quota_charge_post_write(&alice_stats, 4096); + let carol_stats = stats.get_or_create_user_stats_handle("carol"); + stats.quota_charge_post_write(&carol_stats, 99); + + let data = build_user_quota_list(&cfg, &stats); + + assert_eq!(data.users.len(), 1); + let entry = &data.users[0]; + assert_eq!(entry.username, "alice"); + assert_eq!(entry.data_quota_bytes, 1 << 20); + assert_eq!(entry.used_bytes, 4096); + assert_eq!(entry.last_reset_epoch_secs, 0); +} + +#[test] +fn build_user_quota_list_orders_multiple_users_by_username_ascending() { + let mut cfg = ProxyConfig::default(); + for name in ["charlie", "alice", "bob"] { cfg.access.users.insert( - "alice".to_string(), + name.to_string(), "0123456789abcdef0123456789abcdef".to_string(), ); - cfg.access.user_max_tcp_conns_global_each = 7; - - let stats = Stats::new(); - let tracker = UserIpTracker::new(); - - let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; - let alice = users - .iter() - .find(|entry| entry.username == "alice") - .expect("alice must be present"); - assert!(!alice.in_runtime); - assert_eq!(alice.max_tcp_conns, Some(7)); - - cfg.access.user_max_tcp_conns.insert("alice".to_string(), 5); - let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; - let alice = users - .iter() - .find(|entry| entry.username == "alice") - .expect("alice must be present"); - assert!(!alice.in_runtime); - assert_eq!(alice.max_tcp_conns, Some(5)); - - cfg.access.user_max_tcp_conns.insert("alice".to_string(), 0); - let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; - let alice = users - .iter() - .find(|entry| entry.username == "alice") - .expect("alice must be present"); - assert!(!alice.in_runtime); - assert_eq!(alice.max_tcp_conns, Some(7)); - - cfg.access.user_max_tcp_conns_global_each = 0; - let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; - let alice = users - .iter() - .find(|entry| entry.username == "alice") - .expect("alice must be present"); - assert!(!alice.in_runtime); - assert_eq!(alice.max_tcp_conns, None); + cfg.access.user_data_quota.insert(name.to_string(), 1 << 30); } - #[tokio::test] - async fn users_from_config_reports_user_rate_limits() { - let mut cfg = ProxyConfig::default(); - cfg.access.users.insert( - "alice".to_string(), - "0123456789abcdef0123456789abcdef".to_string(), - ); - cfg.access.user_rate_limits.insert( - "alice".to_string(), - RateLimitBps { - up_bps: 1024, - down_bps: 0, - }, - ); + let stats = Stats::new(); + let data = build_user_quota_list(&cfg, &stats); - let stats = Stats::new(); - let tracker = UserIpTracker::new(); - - let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; - let alice = users - .iter() - .find(|entry| entry.username == "alice") - .expect("alice must be present"); - - assert_eq!(alice.rate_limit_up_bps, Some(1024)); - assert_eq!(alice.rate_limit_down_bps, None); - } - - #[tokio::test] - async fn users_from_config_reports_user_enabled_default_and_override() { - let mut cfg = ProxyConfig::default(); - cfg.access.users.insert( - "alice".to_string(), - "0123456789abcdef0123456789abcdef".to_string(), - ); - cfg.access.users.insert( - "bob".to_string(), - "fedcba9876543210fedcba9876543210".to_string(), - ); - cfg.access.user_enabled.insert("bob".to_string(), false); - - let stats = Stats::new(); - let tracker = UserIpTracker::new(); - let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; - let alice = users - .iter() - .find(|entry| entry.username == "alice") - .expect("alice must be present"); - let bob = users - .iter() - .find(|entry| entry.username == "bob") - .expect("bob must be present"); - - assert!(alice.enabled); - assert!(!bob.enabled); - - cfg.access.user_enabled.insert("bob".to_string(), true); - let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; - let bob = users - .iter() - .find(|entry| entry.username == "bob") - .expect("bob must be present"); - assert!(bob.enabled); - } - - #[tokio::test] - async fn users_from_config_marks_runtime_membership_when_snapshot_is_provided() { - let mut disk_cfg = ProxyConfig::default(); - disk_cfg.access.users.insert( - "alice".to_string(), - "0123456789abcdef0123456789abcdef".to_string(), - ); - disk_cfg.access.users.insert( - "bob".to_string(), - "fedcba9876543210fedcba9876543210".to_string(), - ); - - let mut runtime_cfg = ProxyConfig::default(); - runtime_cfg.access.users.insert( - "alice".to_string(), - "0123456789abcdef0123456789abcdef".to_string(), - ); - - let stats = Stats::new(); - let tracker = UserIpTracker::new(); - let users = - users_from_config(&disk_cfg, &stats, &tracker, None, None, Some(&runtime_cfg)).await; - - let alice = users - .iter() - .find(|entry| entry.username == "alice") - .expect("alice must be present"); - let bob = users - .iter() - .find(|entry| entry.username == "bob") - .expect("bob must be present"); - - assert!(alice.in_runtime); - assert!(!bob.in_runtime); - } - - #[tokio::test] - async fn users_from_config_returns_tls_link_for_each_tls_domain() { - let mut cfg = ProxyConfig::default(); - cfg.access.users.insert( - "alice".to_string(), - "0123456789abcdef0123456789abcdef".to_string(), - ); - cfg.general.modes.classic = false; - cfg.general.modes.secure = false; - cfg.general.modes.tls = true; - cfg.general.links.public_host = Some("proxy.example.net".to_string()); - cfg.general.links.public_port = Some(443); - cfg.censorship.tls_domain = "front-a.example.com".to_string(); - cfg.censorship.tls_domains = vec![ - "front-b.example.com".to_string(), - "front-c.example.com".to_string(), - "front-b.example.com".to_string(), - "front-a.example.com".to_string(), - ]; - - let stats = Stats::new(); - let tracker = UserIpTracker::new(); - let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await; - let alice = users - .iter() - .find(|entry| entry.username == "alice") - .expect("alice must be present"); - - assert_eq!(alice.links.tls.len(), 3); - assert!( - alice - .links - .tls - .iter() - .any(|link| link.ends_with(&hex::encode("front-a.example.com"))) - ); - assert!( - alice - .links - .tls - .iter() - .any(|link| link.ends_with(&hex::encode("front-b.example.com"))) - ); - assert!( - alice - .links - .tls - .iter() - .any(|link| link.ends_with(&hex::encode("front-c.example.com"))) - ); - assert_eq!(alice.links.tls_domains.len(), 2); - assert!( - alice - .links - .tls_domains - .iter() - .any(|entry| entry.domain == "front-b.example.com" - && entry.link.ends_with(&hex::encode("front-b.example.com"))) - ); - assert!( - alice - .links - .tls_domains - .iter() - .any(|entry| entry.domain == "front-c.example.com" - && entry.link.ends_with(&hex::encode("front-c.example.com"))) - ); - assert!( - !alice - .links - .tls_domains - .iter() - .any(|entry| entry.domain == "front-a.example.com") - ); - } - - #[test] - fn build_user_quota_list_skips_users_without_positive_quota_and_sorts_by_username() { - let mut cfg = ProxyConfig::default(); - cfg.access.users.insert( - "alice".to_string(), - "0123456789abcdef0123456789abcdef".to_string(), - ); - cfg.access.users.insert( - "bob".to_string(), - "fedcba9876543210fedcba9876543210".to_string(), - ); - cfg.access.users.insert( - "carol".to_string(), - "aaaabbbbccccddddeeeeffff00001111".to_string(), - ); - // alice has a positive quota and should be listed. - cfg.access - .user_data_quota - .insert("alice".to_string(), 1 << 20); - // bob has no quota entry at all (None) — should be skipped. - // carol has an explicit zero quota — should be skipped. - cfg.access.user_data_quota.insert("carol".to_string(), 0); - - let stats = Stats::new(); - // Charge some traffic against alice; carol gets traffic too but should - // still be filtered out by the quota check. - let alice_stats = stats.get_or_create_user_stats_handle("alice"); - stats.quota_charge_post_write(&alice_stats, 4096); - let carol_stats = stats.get_or_create_user_stats_handle("carol"); - stats.quota_charge_post_write(&carol_stats, 99); - - let data = build_user_quota_list(&cfg, &stats); - - assert_eq!(data.users.len(), 1); - let entry = &data.users[0]; - assert_eq!(entry.username, "alice"); - assert_eq!(entry.data_quota_bytes, 1 << 20); - assert_eq!(entry.used_bytes, 4096); + let names: Vec<&str> = data.users.iter().map(|e| e.username.as_str()).collect(); + assert_eq!(names, vec!["alice", "bob", "charlie"]); + for entry in &data.users { + assert_eq!(entry.used_bytes, 0); assert_eq!(entry.last_reset_epoch_secs, 0); + assert_eq!(entry.data_quota_bytes, 1 << 30); } - - #[test] - fn build_user_quota_list_orders_multiple_users_by_username_ascending() { - let mut cfg = ProxyConfig::default(); - for name in ["charlie", "alice", "bob"] { - cfg.access.users.insert( - name.to_string(), - "0123456789abcdef0123456789abcdef".to_string(), - ); - cfg.access.user_data_quota.insert(name.to_string(), 1 << 30); - } - - let stats = Stats::new(); - let data = build_user_quota_list(&cfg, &stats); - - let names: Vec<&str> = data.users.iter().map(|e| e.username.as_str()).collect(); - assert_eq!(names, vec!["alice", "bob", "charlie"]); - for entry in &data.users { - assert_eq!(entry.used_bytes, 0); - assert_eq!(entry.last_reset_epoch_secs, 0); - assert_eq!(entry.data_quota_bytes, 1 << 30); - } - } +} diff --git a/src/api/users/update.rs b/src/api/users/update.rs index 06164da..bfd1af2 100644 --- a/src/api/users/update.rs +++ b/src/api/users/update.rs @@ -194,7 +194,6 @@ pub(in crate::api) async fn patch_user( Ok((user_info, revision)) } - pub(in crate::api) async fn set_user_enabled( user: &str, enabled: bool, @@ -243,4 +242,3 @@ pub(in crate::api) async fn set_user_enabled( Ok((user_info, revision)) } - diff --git a/src/api/users/view.rs b/src/api/users/view.rs index d37a039..f38ce75 100644 --- a/src/api/users/view.rs +++ b/src/api/users/view.rs @@ -110,4 +110,3 @@ pub(in crate::api) fn build_user_quota_list(cfg: &ProxyConfig, stats: &Stats) -> } UserQuotaListData { users } } - diff --git a/src/config/hot_reload/diff.rs b/src/config/hot_reload/diff.rs index f060bdd..575c4fe 100644 --- a/src/config/hot_reload/diff.rs +++ b/src/config/hot_reload/diff.rs @@ -1,6 +1,5 @@ use super::*; - fn canonicalize_json(value: &mut serde_json::Value) { match value { serde_json::Value::Object(map) => { @@ -44,7 +43,6 @@ fn listeners_equal( serde_json::to_value(lhs).ok() == serde_json::to_value(rhs).ok() } - /// Warns when the requested snapshot contains fields that require restart. pub(super) fn warn_non_hot_changes(old: &ProxyConfig, new: &ProxyConfig, non_hot_changed: bool) { let mut warned = false; @@ -231,7 +229,6 @@ pub(super) fn warn_non_hot_changes(old: &ProxyConfig, new: &ProxyConfig, non_hot /// 3. `detected_ip_v6` — fallback /// 4. `"UNKNOWN"` — warn the user to set `public_host` - /// Which top-level config sections changed and whether any require a restart. #[derive(Debug, Default, Clone, serde::Serialize)] pub struct ChangeClassification { diff --git a/src/config/hot_reload/fields.rs b/src/config/hot_reload/fields.rs index f7af2f9..5084b49 100644 --- a/src/config/hot_reload/fields.rs +++ b/src/config/hot_reload/fields.rs @@ -1,6 +1,5 @@ use super::*; - /// Fields that are safe to swap without restarting listeners. #[derive(Debug, Clone, PartialEq)] pub struct HotFields { @@ -223,7 +222,6 @@ impl HotFields { } } - pub(super) fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig { let mut cfg = old.clone(); diff --git a/src/config/hot_reload/tests.rs b/src/config/hot_reload/tests.rs index b7c9ed5..2ae7fe9 100644 --- a/src/config/hot_reload/tests.rs +++ b/src/config/hot_reload/tests.rs @@ -1,284 +1,290 @@ - use super::*; +use super::*; - fn sample_config() -> ProxyConfig { - ProxyConfig::default() - } +fn sample_config() -> ProxyConfig { + ProxyConfig::default() +} - fn write_reload_config(path: &Path, ad_tag: Option<&str>, server_port: Option) { - let mut config = String::from( - r#" +fn write_reload_config(path: &Path, ad_tag: Option<&str>, server_port: Option) { + let mut config = String::from( + r#" [censorship] tls_domain = "example.com" [access.users] user = "00000000000000000000000000000000" "#, - ); + ); - if ad_tag.is_some() { - config.push_str("\n[general]\n"); - if let Some(tag) = ad_tag { - config.push_str(&format!("ad_tag = \"{tag}\"\n")); - } + if ad_tag.is_some() { + config.push_str("\n[general]\n"); + if let Some(tag) = ad_tag { + config.push_str(&format!("ad_tag = \"{tag}\"\n")); } - - if let Some(port) = server_port { - config.push_str("\n[server]\n"); - config.push_str(&format!("port = {port}\n")); - } - - std::fs::write(path, config).unwrap(); } - fn temp_config_path(prefix: &str) -> PathBuf { - let nonce = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_nanos(); - std::env::temp_dir().join(format!("{prefix}_{nonce}.toml")) + if let Some(port) = server_port { + config.push_str("\n[server]\n"); + config.push_str(&format!("port = {port}\n")); } - #[test] - fn overlay_applies_hot_and_preserves_non_hot() { - let old = sample_config(); - let mut new = old.clone(); - new.general.hardswap = !old.general.hardswap; - new.server.port = old.server.port.saturating_add(1); + std::fs::write(path, config).unwrap(); +} - let applied = overlay_hot_fields(&old, &new); - assert_eq!(applied.general.hardswap, new.general.hardswap); - assert_eq!(applied.server.port, old.server.port); - } +fn temp_config_path(prefix: &str) -> PathBuf { + let nonce = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::env::temp_dir().join(format!("{prefix}_{nonce}.toml")) +} - #[test] - fn non_hot_only_change_does_not_change_hot_snapshot() { - let old = sample_config(); - let mut new = old.clone(); - new.server.port = old.server.port.saturating_add(1); +#[test] +fn overlay_applies_hot_and_preserves_non_hot() { + let old = sample_config(); + let mut new = old.clone(); + new.general.hardswap = !old.general.hardswap; + new.server.port = old.server.port.saturating_add(1); - let applied = overlay_hot_fields(&old, &new); - assert_eq!( - HotFields::from_config(&old), - HotFields::from_config(&applied) - ); - assert_eq!(applied.server.port, old.server.port); - } + let applied = overlay_hot_fields(&old, &new); + assert_eq!(applied.general.hardswap, new.general.hardswap); + assert_eq!(applied.server.port, old.server.port); +} - #[test] - fn bind_stale_mode_is_hot() { - let old = sample_config(); - let mut new = old.clone(); - new.general.me_bind_stale_mode = match old.general.me_bind_stale_mode { - MeBindStaleMode::Never => MeBindStaleMode::Ttl, - MeBindStaleMode::Ttl => MeBindStaleMode::Always, - MeBindStaleMode::Always => MeBindStaleMode::Never, - }; +#[test] +fn non_hot_only_change_does_not_change_hot_snapshot() { + let old = sample_config(); + let mut new = old.clone(); + new.server.port = old.server.port.saturating_add(1); - let applied = overlay_hot_fields(&old, &new); - assert_eq!( - applied.general.me_bind_stale_mode, - new.general.me_bind_stale_mode - ); - assert_ne!( - HotFields::from_config(&old), - HotFields::from_config(&applied) - ); - } + let applied = overlay_hot_fields(&old, &new); + assert_eq!( + HotFields::from_config(&old), + HotFields::from_config(&applied) + ); + assert_eq!(applied.server.port, old.server.port); +} - #[test] - fn keepalive_is_not_hot() { - let old = sample_config(); - let mut new = old.clone(); - new.general.me_keepalive_interval_secs = old.general.me_keepalive_interval_secs + 5; +#[test] +fn bind_stale_mode_is_hot() { + let old = sample_config(); + let mut new = old.clone(); + new.general.me_bind_stale_mode = match old.general.me_bind_stale_mode { + MeBindStaleMode::Never => MeBindStaleMode::Ttl, + MeBindStaleMode::Ttl => MeBindStaleMode::Always, + MeBindStaleMode::Always => MeBindStaleMode::Never, + }; - let applied = overlay_hot_fields(&old, &new); - assert_eq!( - applied.general.me_keepalive_interval_secs, - old.general.me_keepalive_interval_secs - ); - assert_eq!( - HotFields::from_config(&old), - HotFields::from_config(&applied) - ); - } + let applied = overlay_hot_fields(&old, &new); + assert_eq!( + applied.general.me_bind_stale_mode, + new.general.me_bind_stale_mode + ); + assert_ne!( + HotFields::from_config(&old), + HotFields::from_config(&applied) + ); +} - #[test] - fn mixed_hot_and_non_hot_change_applies_only_hot_subset() { - let old = sample_config(); - let mut new = old.clone(); - new.general.hardswap = !old.general.hardswap; - new.general.use_middle_proxy = !old.general.use_middle_proxy; +#[test] +fn keepalive_is_not_hot() { + let old = sample_config(); + let mut new = old.clone(); + new.general.me_keepalive_interval_secs = old.general.me_keepalive_interval_secs + 5; - let applied = overlay_hot_fields(&old, &new); - assert_eq!(applied.general.hardswap, new.general.hardswap); - assert_eq!( - applied.general.use_middle_proxy, - old.general.use_middle_proxy - ); - assert!(!config_equal(&applied, &new)); - } + let applied = overlay_hot_fields(&old, &new); + assert_eq!( + applied.general.me_keepalive_interval_secs, + old.general.me_keepalive_interval_secs + ); + assert_eq!( + HotFields::from_config(&old), + HotFields::from_config(&applied) + ); +} - #[test] - fn listener_synlimit_fields_are_process_owned() { - let mut old = sample_config(); - old.server.listeners.push(ListenerConfig { - ip: "0.0.0.0".parse().unwrap(), - port: Some(443), - client_mss: None, - synlimit: SynLimitMode::Iptables, - synlimit_seconds: 60, - synlimit_hitcount: 48, - synlimit_burst: 1, - synlimit_ios_seconds: 1, - synlimit_ios_hitcount: 12, - synlimit_ios_burst: 24, - synlimit_hashlimit_expire_ms: 60_000, - synlimit_hashlimit_size: 32_768, - announce: None, - announce_ip: None, - proxy_protocol: None, - reuse_allow: false, - }); - let mut new = old.clone(); - new.server.port = 8443; - new.server.listeners[0].synlimit_seconds = 120; - new.server.listeners[0].synlimit_hitcount = 96; - new.server.listeners[0].synlimit_burst = 2; - new.server.listeners[0].synlimit_ios_seconds = 2; - new.server.listeners[0].synlimit_ios_hitcount = 18; - new.server.listeners[0].synlimit_ios_burst = 36; - new.server.listeners[0].synlimit_hashlimit_expire_ms = 90_000; - new.server.listeners[0].synlimit_hashlimit_size = 65_536; +#[test] +fn mixed_hot_and_non_hot_change_applies_only_hot_subset() { + let old = sample_config(); + let mut new = old.clone(); + new.general.hardswap = !old.general.hardswap; + new.general.use_middle_proxy = !old.general.use_middle_proxy; - let applied = overlay_hot_fields(&old, &new); - let listener = &applied.server.listeners[0]; - assert_eq!(applied.server.port, old.server.port); - assert_eq!(listener.synlimit_seconds, old.server.listeners[0].synlimit_seconds); - assert_eq!( - listener.synlimit_hitcount, - old.server.listeners[0].synlimit_hitcount - ); - assert_eq!(listener.synlimit_burst, old.server.listeners[0].synlimit_burst); - assert_eq!( - listener.synlimit_hashlimit_size, - old.server.listeners[0].synlimit_hashlimit_size - ); - assert!(classify_config_changes(&old, &new).restart_required); - } + let applied = overlay_hot_fields(&old, &new); + assert_eq!(applied.general.hardswap, new.general.hardswap); + assert_eq!( + applied.general.use_middle_proxy, + old.general.use_middle_proxy + ); + assert!(!config_equal(&applied, &new)); +} - #[test] - fn reload_applies_hot_change_on_first_observed_snapshot() { - let initial_tag = "11111111111111111111111111111111"; - let final_tag = "22222222222222222222222222222222"; - let path = temp_config_path("telemt_hot_reload_stable"); +#[test] +fn listener_synlimit_fields_are_process_owned() { + let mut old = sample_config(); + old.server.listeners.push(ListenerConfig { + ip: "0.0.0.0".parse().unwrap(), + port: Some(443), + client_mss: None, + synlimit: SynLimitMode::Iptables, + synlimit_seconds: 60, + synlimit_hitcount: 48, + synlimit_burst: 1, + synlimit_ios_seconds: 1, + synlimit_ios_hitcount: 12, + synlimit_ios_burst: 24, + synlimit_hashlimit_expire_ms: 60_000, + synlimit_hashlimit_size: 32_768, + announce: None, + announce_ip: None, + proxy_protocol: None, + reuse_allow: false, + }); + let mut new = old.clone(); + new.server.port = 8443; + new.server.listeners[0].synlimit_seconds = 120; + new.server.listeners[0].synlimit_hitcount = 96; + new.server.listeners[0].synlimit_burst = 2; + new.server.listeners[0].synlimit_ios_seconds = 2; + new.server.listeners[0].synlimit_ios_hitcount = 18; + new.server.listeners[0].synlimit_ios_burst = 36; + new.server.listeners[0].synlimit_hashlimit_expire_ms = 90_000; + new.server.listeners[0].synlimit_hashlimit_size = 65_536; - write_reload_config(&path, Some(initial_tag), None); - let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap()); - let initial_hash = ProxyConfig::load_with_metadata(&path) - .unwrap() - .rendered_hash; - let (config_tx, _config_rx) = watch::channel(initial_cfg.clone()); - let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone()); - let mut reload_state = ReloadState::new(Some(initial_hash)); + let applied = overlay_hot_fields(&old, &new); + let listener = &applied.server.listeners[0]; + assert_eq!(applied.server.port, old.server.port); + assert_eq!( + listener.synlimit_seconds, + old.server.listeners[0].synlimit_seconds + ); + assert_eq!( + listener.synlimit_hitcount, + old.server.listeners[0].synlimit_hitcount + ); + assert_eq!( + listener.synlimit_burst, + old.server.listeners[0].synlimit_burst + ); + assert_eq!( + listener.synlimit_hashlimit_size, + old.server.listeners[0].synlimit_hashlimit_size + ); + assert!(classify_config_changes(&old, &new).restart_required); +} - write_reload_config(&path, Some(final_tag), None); - reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap(); - assert_eq!( - config_tx.borrow().general.ad_tag.as_deref(), - Some(final_tag) - ); +#[test] +fn reload_applies_hot_change_on_first_observed_snapshot() { + let initial_tag = "11111111111111111111111111111111"; + let final_tag = "22222222222222222222222222222222"; + let path = temp_config_path("telemt_hot_reload_stable"); - let _ = std::fs::remove_file(path); - } + write_reload_config(&path, Some(initial_tag), None); + let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap()); + let initial_hash = ProxyConfig::load_with_metadata(&path) + .unwrap() + .rendered_hash; + let (config_tx, _config_rx) = watch::channel(initial_cfg.clone()); + let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone()); + let mut reload_state = ReloadState::new(Some(initial_hash)); - #[test] - fn reload_keeps_hot_apply_when_non_hot_fields_change() { - let initial_tag = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; - let final_tag = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; - let path = temp_config_path("telemt_hot_reload_mixed"); + write_reload_config(&path, Some(final_tag), None); + reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap(); + assert_eq!( + config_tx.borrow().general.ad_tag.as_deref(), + Some(final_tag) + ); - write_reload_config(&path, Some(initial_tag), None); - let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap()); - let initial_hash = ProxyConfig::load_with_metadata(&path) - .unwrap() - .rendered_hash; - let (config_tx, _config_rx) = watch::channel(initial_cfg.clone()); - let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone()); - let mut reload_state = ReloadState::new(Some(initial_hash)); + let _ = std::fs::remove_file(path); +} - write_reload_config(&path, Some(final_tag), Some(initial_cfg.server.port + 1)); - reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap(); +#[test] +fn reload_keeps_hot_apply_when_non_hot_fields_change() { + let initial_tag = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + let final_tag = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + let path = temp_config_path("telemt_hot_reload_mixed"); - let applied = config_tx.borrow().clone(); - assert_eq!(applied.general.ad_tag.as_deref(), Some(final_tag)); - assert_eq!(applied.server.port, initial_cfg.server.port); + write_reload_config(&path, Some(initial_tag), None); + let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap()); + let initial_hash = ProxyConfig::load_with_metadata(&path) + .unwrap() + .rendered_hash; + let (config_tx, _config_rx) = watch::channel(initial_cfg.clone()); + let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone()); + let mut reload_state = ReloadState::new(Some(initial_hash)); - let _ = std::fs::remove_file(path); - } + write_reload_config(&path, Some(final_tag), Some(initial_cfg.server.port + 1)); + reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap(); - #[test] - fn classify_sni_change_requires_restart() { - // censorship.* is not in overlay_hot_fields -> restart. - let old = ProxyConfig::default(); - let mut new = ProxyConfig::default(); - new.censorship.tls_domain = "front.example".to_string(); + let applied = config_tx.borrow().clone(); + assert_eq!(applied.general.ad_tag.as_deref(), Some(final_tag)); + assert_eq!(applied.server.port, initial_cfg.server.port); - let class = classify_config_changes(&old, &new); - assert!(class.restart_required); - assert!(class.changed.iter().any(|c| c == "censorship")); - } + let _ = std::fs::remove_file(path); +} - #[test] - fn classify_dns_overrides_change_is_hot() { - // network.dns_overrides IS in overlay_hot_fields -> no restart. - let old = ProxyConfig::default(); - let mut new = ProxyConfig::default(); - new.network.dns_overrides.push("1.1.1.1".to_string()); +#[test] +fn classify_sni_change_requires_restart() { + // censorship.* is not in overlay_hot_fields -> restart. + let old = ProxyConfig::default(); + let mut new = ProxyConfig::default(); + new.censorship.tls_domain = "front.example".to_string(); - let class = classify_config_changes(&old, &new); - assert!(!class.restart_required); - assert!(class.changed.iter().any(|c| c == "network")); - } + let class = classify_config_changes(&old, &new); + assert!(class.restart_required); + assert!(class.changed.iter().any(|c| c == "censorship")); +} - #[test] - fn classify_timeouts_change_requires_restart() { - // timeouts.* is NOT in overlay_hot_fields -> restart. - let old = ProxyConfig::default(); - let mut new = ProxyConfig::default(); - new.timeouts.client_handshake = old.timeouts.client_handshake + 1; +#[test] +fn classify_dns_overrides_change_is_hot() { + // network.dns_overrides IS in overlay_hot_fields -> no restart. + let old = ProxyConfig::default(); + let mut new = ProxyConfig::default(); + new.network.dns_overrides.push("1.1.1.1".to_string()); - let class = classify_config_changes(&old, &new); - assert!(class.restart_required); - } + let class = classify_config_changes(&old, &new); + assert!(!class.restart_required); + assert!(class.changed.iter().any(|c| c == "network")); +} - #[test] - fn reload_recovers_after_parse_error_on_next_attempt() { - let initial_tag = "cccccccccccccccccccccccccccccccc"; - let final_tag = "dddddddddddddddddddddddddddddddd"; - let path = temp_config_path("telemt_hot_reload_parse_recovery"); +#[test] +fn classify_timeouts_change_requires_restart() { + // timeouts.* is NOT in overlay_hot_fields -> restart. + let old = ProxyConfig::default(); + let mut new = ProxyConfig::default(); + new.timeouts.client_handshake = old.timeouts.client_handshake + 1; - write_reload_config(&path, Some(initial_tag), None); - let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap()); - let initial_hash = ProxyConfig::load_with_metadata(&path) - .unwrap() - .rendered_hash; - let (config_tx, _config_rx) = watch::channel(initial_cfg.clone()); - let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone()); - let mut reload_state = ReloadState::new(Some(initial_hash)); + let class = classify_config_changes(&old, &new); + assert!(class.restart_required); +} - std::fs::write(&path, "[access.users\nuser = \"broken\"\n").unwrap(); - assert!(reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).is_none()); - assert_eq!( - config_tx.borrow().general.ad_tag.as_deref(), - Some(initial_tag) - ); +#[test] +fn reload_recovers_after_parse_error_on_next_attempt() { + let initial_tag = "cccccccccccccccccccccccccccccccc"; + let final_tag = "dddddddddddddddddddddddddddddddd"; + let path = temp_config_path("telemt_hot_reload_parse_recovery"); - write_reload_config(&path, Some(final_tag), None); - reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap(); - assert_eq!( - config_tx.borrow().general.ad_tag.as_deref(), - Some(final_tag) - ); + write_reload_config(&path, Some(initial_tag), None); + let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap()); + let initial_hash = ProxyConfig::load_with_metadata(&path) + .unwrap() + .rendered_hash; + let (config_tx, _config_rx) = watch::channel(initial_cfg.clone()); + let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone()); + let mut reload_state = ReloadState::new(Some(initial_hash)); - let _ = std::fs::remove_file(path); - } + std::fs::write(&path, "[access.users\nuser = \"broken\"\n").unwrap(); + assert!(reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).is_none()); + assert_eq!( + config_tx.borrow().general.ad_tag.as_deref(), + Some(initial_tag) + ); + + write_reload_config(&path, Some(final_tag), None); + reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap(); + assert_eq!( + config_tx.borrow().general.ad_tag.as_deref(), + Some(final_tag) + ); + + let _ = std::fs::remove_file(path); +} diff --git a/src/config/hot_reload/watcher.rs b/src/config/hot_reload/watcher.rs index 8e09251..e2b3e60 100644 --- a/src/config/hot_reload/watcher.rs +++ b/src/config/hot_reload/watcher.rs @@ -123,7 +123,6 @@ fn apply_watch_manifest( } } - /// Load config, validate, diff against current, and broadcast if changed. pub(super) fn reload_config( config_path: &PathBuf, @@ -199,7 +198,6 @@ pub(super) fn reload_config( Some(next_manifest) } - /// Spawn the hot-reload watcher task. /// /// Uses `notify` (inotify on Linux) to detect file changes instantly. diff --git a/src/config/load.rs b/src/config/load.rs index 31cc8fd..96b1802 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -23,7 +23,6 @@ mod strict_keys; // Precomputed user authentication data for handshake hot paths. mod runtime_auth; // Post-deserialization validation helpers. -mod validation; mod decode; mod effective; mod pipeline; @@ -31,6 +30,7 @@ mod validate_core; mod validate_me; mod validate_runtime; mod validate_server; +mod validation; use self::includes::{hash_rendered_snapshot, normalize_config_path, preprocess_includes}; use self::normalize::{ @@ -41,8 +41,8 @@ use self::normalize::{ pub(crate) use self::runtime_auth::UserAuthSnapshot; use self::strict_keys::handle_unknown_config_keys; use self::validation::{ - normalize_upstream_family_policy, validate_listener_runtime_profiles, - validate_logging_config, validate_network_cfg, validate_upstreams, + normalize_upstream_family_policy, validate_listener_runtime_profiles, validate_logging_config, + validate_network_cfg, validate_upstreams, }; const MAX_ME_WRITER_CMD_CHANNEL_CAPACITY: usize = 16_384; diff --git a/src/config/load/decode.rs b/src/config/load/decode.rs index ac3d53d..d67ddd6 100644 --- a/src/config/load/decode.rs +++ b/src/config/load/decode.rs @@ -50,8 +50,7 @@ pub(super) fn decode_source_graph(graph: ConfigSourceGraph) -> Result Result Result Result Result<()> { } let mut exclusive_mask = HashMap::with_capacity(config.censorship.exclusive_mask.len()); - let mut exclusive_mask_targets = - HashMap::with_capacity(config.censorship.exclusive_mask.len()); + let mut exclusive_mask_targets = HashMap::with_capacity(config.censorship.exclusive_mask.len()); for (domain, target) in std::mem::take(&mut config.censorship.exclusive_mask) { let domain = normalize_domain_to_ascii(&domain, "censorship.exclusive_mask domain")?; - let target = - normalize_exclusive_mask_target(&target, "censorship.exclusive_mask target")?; + let target = normalize_exclusive_mask_target(&target, "censorship.exclusive_mask target")?; let Some((host, port)) = parse_exclusive_mask_target(&target) else { return Err(ProxyError::Config(format!( "Invalid censorship.exclusive_mask target for '{}': '{}'. Expected host:port with port > 0", diff --git a/src/config/load/validate_core.rs b/src/config/load/validate_core.rs index bc00d34..50c30e8 100644 --- a/src/config/load/validate_core.rs +++ b/src/config/load/validate_core.rs @@ -207,8 +207,7 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { )); } - if config.timeouts.relay_client_idle_hard_secs < config.timeouts.relay_client_idle_soft_secs - { + if config.timeouts.relay_client_idle_hard_secs < config.timeouts.relay_client_idle_soft_secs { return Err(ProxyError::Config( "timeouts.relay_client_idle_hard_secs must be >= timeouts.relay_client_idle_soft_secs" .to_string(), diff --git a/src/config/load/validate_me.rs b/src/config/load/validate_me.rs index 2da68a4..8c24fa6 100644 --- a/src/config/load/validate_me.rs +++ b/src/config/load/validate_me.rs @@ -168,8 +168,7 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { } if config.general.me_route_backpressure_base_timeout_ms > 5000 { return Err(ProxyError::Config( - "general.me_route_backpressure_base_timeout_ms must be within [1, 5000]" - .to_string(), + "general.me_route_backpressure_base_timeout_ms must be within [1, 5000]".to_string(), )); } @@ -182,15 +181,13 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { } if config.general.me_route_backpressure_high_timeout_ms > 5000 { return Err(ProxyError::Config( - "general.me_route_backpressure_high_timeout_ms must be within [1, 5000]" - .to_string(), + "general.me_route_backpressure_high_timeout_ms must be within [1, 5000]".to_string(), )); } if !(1..=100).contains(&config.general.me_route_backpressure_high_watermark_pct) { return Err(ProxyError::Config( - "general.me_route_backpressure_high_watermark_pct must be within [1, 100]" - .to_string(), + "general.me_route_backpressure_high_watermark_pct must be within [1, 100]".to_string(), )); } diff --git a/src/config/load/validate_runtime.rs b/src/config/load/validate_runtime.rs index d6969e0..ae05757 100644 --- a/src/config/load/validate_runtime.rs +++ b/src/config/load/validate_runtime.rs @@ -94,8 +94,7 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { )); } - if !(4096..=16 * 1024 * 1024) - .contains(&config.general.me_d2c_frame_buf_shrink_threshold_bytes) + if !(4096..=16 * 1024 * 1024).contains(&config.general.me_d2c_frame_buf_shrink_threshold_bytes) { return Err(ProxyError::Config( "general.me_d2c_frame_buf_shrink_threshold_bytes must be within [4096, 16777216]" @@ -105,15 +104,13 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { if !(4096..=1024 * 1024).contains(&config.general.direct_relay_copy_buf_c2s_bytes) { return Err(ProxyError::Config( - "general.direct_relay_copy_buf_c2s_bytes must be within [4096, 1048576]" - .to_string(), + "general.direct_relay_copy_buf_c2s_bytes must be within [4096, 1048576]".to_string(), )); } if !(8192..=2 * 1024 * 1024).contains(&config.general.direct_relay_copy_buf_s2c_bytes) { return Err(ProxyError::Config( - "general.direct_relay_copy_buf_s2c_bytes must be within [8192, 2097152]" - .to_string(), + "general.direct_relay_copy_buf_s2c_bytes must be within [8192, 2097152]".to_string(), )); } @@ -177,8 +174,7 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { || config.general.me_pool_drain_soft_evict_budget_per_core > 64 { return Err(ProxyError::Config( - "general.me_pool_drain_soft_evict_budget_per_core must be within [1, 64]" - .to_string(), + "general.me_pool_drain_soft_evict_budget_per_core must be within [1, 64]".to_string(), )); } diff --git a/src/config/load/validate_server.rs b/src/config/load/validate_server.rs index a131435..1e89b00 100644 --- a/src/config/load/validate_server.rs +++ b/src/config/load/validate_server.rs @@ -1,8 +1,7 @@ use super::*; pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { - if !(1..=MAX_API_REQUEST_BODY_LIMIT_BYTES) - .contains(&config.server.api.request_body_limit_bytes) + if !(1..=MAX_API_REQUEST_BODY_LIMIT_BYTES).contains(&config.server.api.request_body_limit_bytes) { return Err(ProxyError::Config( "server.api.request_body_limit_bytes must be within [1, 1048576]".to_string(), diff --git a/src/config/mod.rs b/src/config/mod.rs index b50cbaa..f073d51 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -5,6 +5,6 @@ pub mod hot_reload; mod load; mod types; -pub(crate) use load::{ConfigSourceGraph, LoadedConfig}; pub use load::ProxyConfig; +pub(crate) use load::{ConfigSourceGraph, LoadedConfig}; pub use types::*; diff --git a/src/config/tests/load_basic_tests/api_tests.rs b/src/config/tests/load_basic_tests/api_tests.rs index cb83cc1..b22ab96 100644 --- a/src/config/tests/load_basic_tests/api_tests.rs +++ b/src/config/tests/load_basic_tests/api_tests.rs @@ -22,7 +22,6 @@ fn api_minimal_runtime_cache_ttl_out_of_range_is_rejected() { let _ = std::fs::remove_file(path); } - #[test] fn api_runtime_edge_cache_ttl_out_of_range_is_rejected() { let toml = r#" @@ -88,4 +87,3 @@ fn api_runtime_edge_events_capacity_out_of_range_is_rejected() { assert!(err.contains("server.api.runtime_edge_events_capacity must be within [16, 4096]")); let _ = std::fs::remove_file(path); } - diff --git a/src/config/tests/load_basic_tests/conntrack_tests.rs b/src/config/tests/load_basic_tests/conntrack_tests.rs index f50641a..e8fb3e3 100644 --- a/src/config/tests/load_basic_tests/conntrack_tests.rs +++ b/src/config/tests/load_basic_tests/conntrack_tests.rs @@ -173,4 +173,3 @@ fn force_close_bumped_when_below_drain_ttl() { assert_eq!(cfg.general.me_reinit_drain_timeout_secs, 90); let _ = std::fs::remove_file(path); } - diff --git a/src/config/tests/load_basic_tests/defaults_access_tests.rs b/src/config/tests/load_basic_tests/defaults_access_tests.rs index a716de5..7d3d8de 100644 --- a/src/config/tests/load_basic_tests/defaults_access_tests.rs +++ b/src/config/tests/load_basic_tests/defaults_access_tests.rs @@ -476,4 +476,3 @@ fn impl_defaults_are_sourced_from_default_helpers() { default_user_max_tcp_conns_global_each() ); } - diff --git a/src/config/tests/load_basic_tests/legacy_policy_tests.rs b/src/config/tests/load_basic_tests/legacy_policy_tests.rs index 8a3dff1..2864672 100644 --- a/src/config/tests/load_basic_tests/legacy_policy_tests.rs +++ b/src/config/tests/load_basic_tests/legacy_policy_tests.rs @@ -422,4 +422,3 @@ fn update_every_zero_is_rejected() { assert!(err.contains("general.update_every must be > 0")); let _ = std::fs::remove_file(path); } - diff --git a/src/config/tests/load_basic_tests/me_route_tests.rs b/src/config/tests/load_basic_tests/me_route_tests.rs index e628b73..7b4afcf 100644 --- a/src/config/tests/load_basic_tests/me_route_tests.rs +++ b/src/config/tests/load_basic_tests/me_route_tests.rs @@ -334,4 +334,3 @@ fn me_pool_min_fresh_ratio_out_of_range_is_rejected() { assert!(err.contains("general.me_pool_min_fresh_ratio must be within [0.0, 1.0]")); let _ = std::fs::remove_file(path); } - diff --git a/src/config/tests/load_basic_tests/me_startup_tests.rs b/src/config/tests/load_basic_tests/me_startup_tests.rs index a60aec3..ae8e7fe 100644 --- a/src/config/tests/load_basic_tests/me_startup_tests.rs +++ b/src/config/tests/load_basic_tests/me_startup_tests.rs @@ -302,4 +302,3 @@ fn rpc_proxy_req_every_zero_and_valid_range_are_accepted() { assert_eq!(cfg_valid.general.rpc_proxy_req_every, 40); let _ = std::fs::remove_file(path_valid); } - diff --git a/src/config/tests/load_basic_tests/synlimit_mss_tests.rs b/src/config/tests/load_basic_tests/synlimit_mss_tests.rs index bfbcf4e..f896e4c 100644 --- a/src/config/tests/load_basic_tests/synlimit_mss_tests.rs +++ b/src/config/tests/load_basic_tests/synlimit_mss_tests.rs @@ -117,7 +117,6 @@ fn synlimit_synfix_zero_values_are_rejected() { } } - #[test] fn client_mss_presets_and_listener_override_are_resolved() { let toml = r#" @@ -366,4 +365,3 @@ fn listener_client_mss_invalid_preset_is_rejected() { assert!(err.contains("must be \"\", extreme-low, tspu, 2in8")); let _ = std::fs::remove_file(path); } - diff --git a/src/config/tests/load_basic_tests/tls_fetch_tests.rs b/src/config/tests/load_basic_tests/tls_fetch_tests.rs index 5b8338d..6fda831 100644 --- a/src/config/tests/load_basic_tests/tls_fetch_tests.rs +++ b/src/config/tests/load_basic_tests/tls_fetch_tests.rs @@ -183,4 +183,3 @@ fn valid_ad_tag_is_preserved_during_load() { ); let _ = std::fs::remove_file(path); } - diff --git a/src/config/types.rs b/src/config/types.rs index 756e342..204242a 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -28,14 +28,23 @@ pub use access::{AccessConfig, CidrRateLimitKey, RateLimitBps}; #[allow(unused_imports)] pub(crate) use access::{CidrAutoTemplate, CidrAutoTemplateFamily}; pub use api::{ApiConfig, ApiGrayAction}; -pub use censorship::{AntiCensorshipConfig, ExclusiveMaskTarget, TlsFetchConfig, TlsFetchProfile, UnknownSniAction}; +pub use censorship::{ + AntiCensorshipConfig, ExclusiveMaskTarget, TlsFetchConfig, TlsFetchProfile, UnknownSniAction, +}; pub use general::GeneralConfig; pub use links::{LinksConfig, ShowLink}; -pub use logging::{LogLevel, LoggingConfig, LoggingDestination, LogRotation}; +pub use logging::{LogLevel, LogRotation, LoggingConfig, LoggingDestination}; pub use network::{NetworkConfig, ProxyModes, UpstreamConfig, UpstreamType}; -pub use policies::{MeBindStaleMode, MeFloorMode, MeRouteNoWriterMode, MeSocksKdfPolicy, MeTelemetryLevel, MeWriterPickMode, RstOnCloseMode, TelemetryConfig, UserMaxUniqueIpsMode}; +pub use policies::{ + MeBindStaleMode, MeFloorMode, MeRouteNoWriterMode, MeSocksKdfPolicy, MeTelemetryLevel, + MeWriterPickMode, RstOnCloseMode, TelemetryConfig, UserMaxUniqueIpsMode, +}; #[allow(unused_imports)] -pub use server::{CLIENT_MSS_2IN8, CLIENT_MSS_EXTREME_LOW, CLIENT_MSS_MAX, CLIENT_MSS_MIN, CLIENT_MSS_TSPU, ConntrackBackend, ConntrackControlConfig, ConntrackMode, ConntrackPressureProfile, ListenerConfig, ServerConfig, SynLimitMode, TimeoutsConfig}; +pub use server::{ + CLIENT_MSS_2IN8, CLIENT_MSS_EXTREME_LOW, CLIENT_MSS_MAX, CLIENT_MSS_MIN, CLIENT_MSS_TSPU, + ConntrackBackend, ConntrackControlConfig, ConntrackMode, ConntrackPressureProfile, + ListenerConfig, ServerConfig, SynLimitMode, TimeoutsConfig, +}; fn default_quota_state_path() -> PathBuf { PathBuf::from("telemt.limit.json") diff --git a/src/config/types/links.rs b/src/config/types/links.rs index e0095d0..1d8ac04 100644 --- a/src/config/types/links.rs +++ b/src/config/types/links.rs @@ -28,7 +28,6 @@ impl Default for LinksConfig { } } - /// In TOML, this can be: /// - `show_link = "*"` — show links for all users /// - `show_link = ["a", "b"]` — show links for specific users diff --git a/src/config/types/policies.rs b/src/config/types/policies.rs index 6dad3b0..dde198d 100644 --- a/src/config/types/policies.rs +++ b/src/config/types/policies.rs @@ -227,4 +227,3 @@ impl Default for TelemetryConfig { } } } - diff --git a/src/maestro/bootstrap.rs b/src/maestro/bootstrap.rs index d4a2a80..9fe8a58 100644 --- a/src/maestro/bootstrap.rs +++ b/src/maestro/bootstrap.rs @@ -6,9 +6,7 @@ use tracing::{info, warn}; use tracing_subscriber::{EnvFilter, fmt, prelude::*, reload as tracing_reload}; use crate::config::{LogLevel, ProxyConfig}; -use crate::startup::{ - COMPONENT_CONFIG_LOAD, COMPONENT_TRACING_INIT, StartupTracker, -}; +use crate::startup::{COMPONENT_CONFIG_LOAD, COMPONENT_TRACING_INIT, StartupTracker}; use super::helpers::{ parse_cli, print_maestro_line, resolve_runtime_base_dir, resolve_runtime_config_path, diff --git a/src/maestro/listeners/accept.rs b/src/maestro/listeners/accept.rs index e1cb6e7..08621b8 100644 --- a/src/maestro/listeners/accept.rs +++ b/src/maestro/listeners/accept.rs @@ -251,8 +251,9 @@ impl ListenerSlot { pub(super) async fn stop(&mut self) -> Result<(), String> { self.cancellation.cancel(); if let Some(task) = self.task.take() { - task.await - .map_err(|error_value| format!("listener {} task failed: {error_value}", self.spec.addr))?; + task.await.map_err(|error_value| { + format!("listener {} task failed: {error_value}", self.spec.addr) + })?; } Ok(()) } diff --git a/src/maestro/listeners/bind.rs b/src/maestro/listeners/bind.rs index ff999dd..4b2ba93 100644 --- a/src/maestro/listeners/bind.rs +++ b/src/maestro/listeners/bind.rs @@ -10,8 +10,8 @@ use tracing::{error, info, warn}; use crate::config::ProxyConfig; use crate::startup::{COMPONENT_LISTENERS_BIND, StartupTracker}; -use crate::transport::socket::{activate_listener_socket, bind_listener_socket}; use crate::transport::find_listener_processes; +use crate::transport::socket::{activate_listener_socket, bind_listener_socket}; use super::plan::{ListenerBindSpec, listener_bind_plan}; use crate::maestro::helpers::print_proxy_links; @@ -88,9 +88,7 @@ fn log_bind_error(addr: SocketAddr, reuse_allow: bool, error_value: &std::io::Er } } -pub(super) fn prepare_listener( - spec: ListenerBindSpec, -) -> std::io::Result { +pub(super) fn prepare_listener(spec: ListenerBindSpec) -> std::io::Result { match bind_listener_socket(spec.addr, &spec.options) { Ok(socket) => Ok(PreparedTcpListener { socket, spec }), Err(error_value) => { @@ -165,7 +163,12 @@ fn print_configured_links( if config.general.links.show.is_empty() || config.general.links.public_host.is_none() { return; } - let host = config.general.links.public_host.as_deref().unwrap_or_default(); + let host = config + .general + .links + .public_host + .as_deref() + .unwrap_or_default(); let port = config .general .links diff --git a/src/maestro/listeners/control.rs b/src/maestro/listeners/control.rs index dfa9028..7837bbe 100644 --- a/src/maestro/listeners/control.rs +++ b/src/maestro/listeners/control.rs @@ -8,9 +8,7 @@ use crate::config::ProxyConfig; use crate::maestro::generation::RuntimeGeneration; use super::accept::ListenerSlot; -use super::bind::{ - BoundListeners, BoundTcpListener, PreparedTcpListener, prepare_listener, -}; +use super::bind::{BoundListeners, BoundTcpListener, PreparedTcpListener, prepare_listener}; use super::plan::{ListenerBindSpec, listener_bind_plan}; #[cfg(unix)] use super::unix::UnixAcceptHandle; diff --git a/src/maestro/listeners/plan.rs b/src/maestro/listeners/plan.rs index 3f22daf..df905f9 100644 --- a/src/maestro/listeners/plan.rs +++ b/src/maestro/listeners/plan.rs @@ -15,10 +15,7 @@ pub(super) struct ListenerBindSpec { pub(super) tls_response_fragment_size: Option, } -fn listener_port_or_legacy( - listener: &crate::config::ListenerConfig, - server: &ServerConfig, -) -> u16 { +fn listener_port_or_legacy(listener: &crate::config::ListenerConfig, server: &ServerConfig) -> u16 { listener.port.unwrap_or(server.port) } diff --git a/src/maestro/listeners/unix.rs b/src/maestro/listeners/unix.rs index 5a16bbf..30f08df 100644 --- a/src/maestro/listeners/unix.rs +++ b/src/maestro/listeners/unix.rs @@ -74,8 +74,7 @@ async fn run_unix_accept_loop( }; let connection_id = connection_counter.fetch_add(1, Ordering::Relaxed); - let fake_peer = - SocketAddr::from(([127, 0, 0, 1], (connection_id % 65535) as u16)); + let fake_peer = SocketAddr::from(([127, 0, 0, 1], (connection_id % 65535) as u16)); let stats = runtime.stats.clone(); let upstream_manager = runtime.upstream_manager.clone(); let replay_checker = runtime.replay_checker.clone(); diff --git a/src/maestro/orchestrator.rs b/src/maestro/orchestrator.rs index 8798603..8f7f849 100644 --- a/src/maestro/orchestrator.rs +++ b/src/maestro/orchestrator.rs @@ -8,9 +8,7 @@ use tracing::{error, info, warn}; use crate::api; use crate::ip_tracker::UserIpTracker; use crate::network::probe::{decide_network_capabilities, log_probe_result, run_probe}; -use crate::proxy::direct_buffer_budget::{ - DirectBufferBudget, resolve_direct_buffer_hard_limit, -}; +use crate::proxy::direct_buffer_budget::{DirectBufferBudget, resolve_direct_buffer_hard_limit}; use crate::proxy::route_mode::{RelayRouteMode, RouteRuntimeController}; use crate::proxy::shared_state::ProxySharedState; use crate::startup::{COMPONENT_API_BOOTSTRAP, COMPONENT_NETWORK_PROBE}; diff --git a/src/maestro/reload.rs b/src/maestro/reload.rs index e0e6a60..8773f73 100644 --- a/src/maestro/reload.rs +++ b/src/maestro/reload.rs @@ -3,8 +3,8 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; -use serde::{Deserialize, Serialize}; use parking_lot::Mutex; +use serde::{Deserialize, Serialize}; use tokio::sync::mpsc; use crate::config::ProxyConfig; @@ -260,9 +260,9 @@ impl ReloadControl { .active_generation .load(Ordering::Acquire) .saturating_add(1); - let status = self - .status_store - .reserve(target_generation, config_revision, request.clone())?; + let status = + self.status_store + .reserve(target_generation, config_revision, request.clone())?; Ok(ReloadReservation { permit: Some(permit), status, diff --git a/src/maestro/reload_supervisor.rs b/src/maestro/reload_supervisor.rs index f446c19..8f4b24e 100644 --- a/src/maestro/reload_supervisor.rs +++ b/src/maestro/reload_supervisor.rs @@ -3,8 +3,8 @@ use std::sync::Arc; use std::time::Duration; use arc_swap::ArcSwap; -use tokio::sync::watch; use tokio::sync::Mutex; +use tokio::sync::watch; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; @@ -158,10 +158,7 @@ impl ReloadSupervisor { let old_runtime = self.active_runtime.load_full(); let resolved = resolve_reload_config(&old_runtime.config(), &command.config); self.control - .set_deferred_fields( - command.reload_id, - resolved.deferred_process_fields.clone(), - ) + .set_deferred_fields(command.reload_id, resolved.deferred_process_fields.clone()) .await; let prepared = match prepare_runtime( diff --git a/src/maestro/runtime_build.rs b/src/maestro/runtime_build.rs index b4b843e..e14d2c6 100644 --- a/src/maestro/runtime_build.rs +++ b/src/maestro/runtime_build.rs @@ -337,8 +337,8 @@ pub(crate) fn resolve_reload_config( || old.server.listen_backlog != desired.server.listen_backlog; let listener_policy_changed = listener_identity_matches && !listener_process_fields_equal(&old.server, &desired.server); - let unsupported_identity_change = !listener_identity_matches - && !listener_rebind_supported(old, desired); + let unsupported_identity_change = + !listener_identity_matches && !listener_rebind_supported(old, desired); if global_listener_policy_changed || listener_policy_changed || unsupported_identity_change { fields.push("server.listeners".to_string()); effective.server.port = old.server.port; @@ -402,9 +402,7 @@ pub(crate) fn resolve_reload_config( fields.push("general.data_path".to_string()); effective.general.data_path = old.general.data_path.clone(); } - if serde_json::to_value(&old.logging).ok() - != serde_json::to_value(&desired.logging).ok() - { + if serde_json::to_value(&old.logging).ok() != serde_json::to_value(&desired.logging).ok() { fields.push("logging".to_string()); effective.logging = old.logging.clone(); } diff --git a/src/maestro/runtime_startup.rs b/src/maestro/runtime_startup.rs index 4e6343a..e23d65a 100644 --- a/src/maestro/runtime_startup.rs +++ b/src/maestro/runtime_startup.rs @@ -11,15 +11,13 @@ use crate::conntrack_control; use crate::crypto::SecureRandom; use crate::ip_tracker::UserIpTracker; use crate::network::probe::{NetworkDecision, NetworkProbe}; -use crate::proxy::direct_buffer_budget::{ - DirectBufferBudget, run_direct_buffer_budget_controller, -}; +use crate::proxy::direct_buffer_budget::{DirectBufferBudget, run_direct_buffer_budget_controller}; use crate::proxy::route_mode::RouteRuntimeController; use crate::proxy::shared_state::ProxySharedState; use crate::startup::{ - COMPONENT_DC_CONNECTIVITY_PING, COMPONENT_ME_CONNECTIVITY_PING, - COMPONENT_ME_POOL_CONSTRUCT, COMPONENT_ME_POOL_INIT_STAGE1, COMPONENT_ME_PROXY_CONFIG_V4, - COMPONENT_ME_PROXY_CONFIG_V6, COMPONENT_ME_SECRET_FETCH, StartupMeStatus, StartupTracker, + COMPONENT_DC_CONNECTIVITY_PING, COMPONENT_ME_CONNECTIVITY_PING, COMPONENT_ME_POOL_CONSTRUCT, + COMPONENT_ME_POOL_INIT_STAGE1, COMPONENT_ME_PROXY_CONFIG_V4, COMPONENT_ME_PROXY_CONFIG_V6, + COMPONENT_ME_SECRET_FETCH, StartupMeStatus, StartupTracker, }; use crate::stats::beobachten::BeobachtenStore; use crate::stats::{ReplayChecker, Stats}; diff --git a/src/proxy/handshake.rs b/src/proxy/handshake.rs index 474b807..5e2b86d 100644 --- a/src/proxy/handshake.rs +++ b/src/proxy/handshake.rs @@ -69,13 +69,14 @@ pub(crate) use self::tls_handshake::handle_tls_handshake_with_shared_and_options #[cfg(test)] pub(crate) use self::auth_probe::{ auth_probe_fail_streak_for_testing_in_shared, auth_probe_is_throttled_for_testing_in_shared, - auth_probe_record_failure_for_testing, auth_probe_saturation_is_throttled_at_for_testing_in_shared, + auth_probe_record_failure_for_testing, + auth_probe_saturation_is_throttled_at_for_testing_in_shared, auth_probe_saturation_is_throttled_for_testing_in_shared, auth_probe_saturation_state_for_testing_in_shared, auth_probe_saturation_state_lock_for_testing_in_shared, auth_probe_state_for_testing_in_shared, - clear_auth_probe_state_for_testing_in_shared, clear_unknown_sni_warn_state_for_testing_in_shared, - clear_warned_secrets_for_testing_in_shared, should_emit_unknown_sni_warn_for_testing_in_shared, - warned_secrets_for_testing_in_shared, + clear_auth_probe_state_for_testing_in_shared, + clear_unknown_sni_warn_state_for_testing_in_shared, clear_warned_secrets_for_testing_in_shared, + should_emit_unknown_sni_warn_for_testing_in_shared, warned_secrets_for_testing_in_shared, }; const ACCESS_SECRET_BYTES: usize = 16; diff --git a/src/proxy/handshake/auth_candidates.rs b/src/proxy/handshake/auth_candidates.rs index a0ef56a..0db6d9f 100644 --- a/src/proxy/handshake/auth_candidates.rs +++ b/src/proxy/handshake/auth_candidates.rs @@ -44,7 +44,10 @@ pub(super) fn sticky_hint_get_by_ip(shared: &ProxySharedState, peer_ip: IpAddr) .map(|entry| *entry) } -pub(super) fn sticky_hint_get_by_ip_prefix(shared: &ProxySharedState, peer_ip: IpAddr) -> Option { +pub(super) fn sticky_hint_get_by_ip_prefix( + shared: &ProxySharedState, + peer_ip: IpAddr, +) -> Option { shared .handshake .sticky_user_by_ip_prefix @@ -104,7 +107,11 @@ pub(super) fn record_recent_user_success_in(shared: &ProxySharedState, user_id: ring[idx].store(user_id.saturating_add(1), Ordering::Relaxed); } -pub(super) fn mark_candidate_if_new(tried_user_ids: &mut [u32], tried_len: &mut usize, user_id: u32) -> bool { +pub(super) fn mark_candidate_if_new( + tried_user_ids: &mut [u32], + tried_len: &mut usize, + user_id: u32, +) -> bool { if tried_user_ids[..*tried_len].contains(&user_id) { return false; } @@ -222,7 +229,11 @@ pub(super) fn warn_invalid_secret_once_in( } } -pub(super) fn decode_user_secret(shared: &ProxySharedState, name: &str, secret_hex: &str) -> Option> { +pub(super) fn decode_user_secret( + shared: &ProxySharedState, + name: &str, + secret_hex: &str, +) -> Option> { match hex::decode(secret_hex) { Ok(bytes) if bytes.len() == ACCESS_SECRET_BYTES => Some(bytes), Ok(bytes) => { @@ -251,7 +262,11 @@ pub(super) fn decode_user_secret(shared: &ProxySharedState, name: &str, secret_h // over TCP (DD). Enforcing this separation prevents an attacker from using a // TLS-capable client to bypass the operator intent for the direct MTProto mode, // and vice versa. -pub(super) fn mode_enabled_for_proto(config: &ProxyConfig, proto_tag: ProtoTag, is_tls: bool) -> bool { +pub(super) fn mode_enabled_for_proto( + config: &ProxyConfig, + proto_tag: ProtoTag, + is_tls: bool, +) -> bool { match proto_tag { ProtoTag::Secure => { if is_tls { @@ -289,4 +304,3 @@ pub(super) fn decode_user_secrets_in( secrets } - diff --git a/src/proxy/handshake/auth_probe.rs b/src/proxy/handshake/auth_probe.rs index 3e9b81b..91f4e94 100644 --- a/src/proxy/handshake/auth_probe.rs +++ b/src/proxy/handshake/auth_probe.rs @@ -86,7 +86,11 @@ pub(super) fn auth_probe_scan_start_offset_in( auth_probe_eviction_offset_in(shared, peer_ip, now) % state_len } -pub(super) fn auth_probe_is_throttled_in(shared: &ProxySharedState, peer_ip: IpAddr, now: Instant) -> bool { +pub(super) fn auth_probe_is_throttled_in( + shared: &ProxySharedState, + peer_ip: IpAddr, + now: Instant, +) -> bool { let peer_ip = normalize_auth_probe_ip(peer_ip); let state = &shared.handshake.auth_probe; let Some(entry) = state.get(&peer_ip) else { @@ -135,7 +139,10 @@ pub(super) fn auth_probe_should_apply_preauth_throttle_in( auth_probe_saturation_grace_exhausted_in(shared, peer_ip, now) } -pub(super) fn auth_probe_saturation_is_throttled_in(shared: &ProxySharedState, now: Instant) -> bool { +pub(super) fn auth_probe_saturation_is_throttled_in( + shared: &ProxySharedState, + now: Instant, +) -> bool { let mut guard = shared .handshake .auth_probe_saturation @@ -198,7 +205,11 @@ pub(super) fn auth_probe_note_expensive_invalid_scan_in( auth_probe_note_saturation_in(shared, now); } -pub(super) fn auth_probe_record_failure_in(shared: &ProxySharedState, peer_ip: IpAddr, now: Instant) { +pub(super) fn auth_probe_record_failure_in( + shared: &ProxySharedState, + peer_ip: IpAddr, + now: Instant, +) { let peer_ip = normalize_auth_probe_ip(peer_ip); let state = &shared.handshake.auth_probe; auth_probe_record_failure_with_state_in(shared, state, peer_ip, now); @@ -504,5 +515,3 @@ pub(super) async fn maybe_apply_server_hello_delay(config: &ProxyConfig) { tokio::time::sleep(Duration::from_millis(delay_ms)).await; } } - - diff --git a/src/proxy/handshake/mtproto.rs b/src/proxy/handshake/mtproto.rs index ca57063..3eea0ef 100644 --- a/src/proxy/handshake/mtproto.rs +++ b/src/proxy/handshake/mtproto.rs @@ -417,4 +417,3 @@ where debug!(peer = %peer, "MTProto handshake: no matching user found"); HandshakeResult::BadClient { reader, writer } } - diff --git a/src/proxy/handshake/nonce.rs b/src/proxy/handshake/nonce.rs index 9f595d8..1c25830 100644 --- a/src/proxy/handshake/nonce.rs +++ b/src/proxy/handshake/nonce.rs @@ -96,4 +96,3 @@ pub fn encrypt_tg_nonce(nonce: &[u8; HANDSHAKE_LEN]) -> Vec { let (encrypted, _, _) = encrypt_tg_nonce_with_ciphers(nonce); encrypted } - diff --git a/src/proxy/handshake/session.rs b/src/proxy/handshake/session.rs index 2bebc79..8731372 100644 --- a/src/proxy/handshake/session.rs +++ b/src/proxy/handshake/session.rs @@ -32,4 +32,3 @@ impl Drop for HandshakeSuccess { self.enc_iv.zeroize(); } } - diff --git a/src/proxy/handshake/tls_handshake.rs b/src/proxy/handshake/tls_handshake.rs index 500946b..0e4e2bb 100644 --- a/src/proxy/handshake/tls_handshake.rs +++ b/src/proxy/handshake/tls_handshake.rs @@ -237,7 +237,6 @@ where } } - let Some(validation) = tls_validation::validate_tls_client( handshake, peer, @@ -422,4 +421,3 @@ async fn write_tls_response( writer.write_all(response).await?; writer.flush().await } - diff --git a/src/synlimit_control/mod.rs b/src/synlimit_control/mod.rs index 554f14c..7cd3243 100644 --- a/src/synlimit_control/mod.rs +++ b/src/synlimit_control/mod.rs @@ -78,9 +78,7 @@ pub(crate) async fn clear_synlimit_rules_all_backends() -> Result async fn clear_synlimit_rules_for_namespace(namespace: &SynLimitNamespace) -> Result { if !has_firewall_privileges() { - return Err( - "SYN limiter cleanup requires root or CAP_NET_ADMIN privileges".to_string(), - ); + return Err("SYN limiter cleanup requires root or CAP_NET_ADMIN privileges".to_string()); } let mut errors = Vec::new(); diff --git a/src/synlimit_control/pf.rs b/src/synlimit_control/pf.rs index c6474bd..2d54450 100644 --- a/src/synlimit_control/pf.rs +++ b/src/synlimit_control/pf.rs @@ -31,7 +31,12 @@ pub(super) async fn apply_synlimit_rules( } let script = pf_synlimit_script(targets); - run_command("pfctl", &["-a", namespace.pf_anchor.as_str(), "-f", "-"], Some(script)).await + run_command( + "pfctl", + &["-a", namespace.pf_anchor.as_str(), "-f", "-"], + Some(script), + ) + .await } async fn has_pf_anchor_hook() -> Result { @@ -98,7 +103,10 @@ mod tests { #[test] fn pf_script_uses_native_rate_limited_pass() { let mut targets = SynLimitTargets::default(); - targets.pf_v4 = vec![test_rule(Some(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7))), 443)]; + targets.pf_v4 = vec![test_rule( + Some(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7))), + 443, + )]; let script = pf_synlimit_script(&targets); assert!(script.contains( diff --git a/src/synlimit_control/pf_runtime_tests.rs b/src/synlimit_control/pf_runtime_tests.rs index 6b75b27..a74c8e0 100644 --- a/src/synlimit_control/pf_runtime_tests.rs +++ b/src/synlimit_control/pf_runtime_tests.rs @@ -29,10 +29,7 @@ fn targets(low_family: &str) -> SynLimitTargets { _ => panic!("TELEMT_PF_LOW_FAMILY must be v4 or v6"), }; SynLimitTargets { - pf_v4: vec![rule( - IpAddr::V4(Ipv4Addr::new(198, 18, 1, 1)), - v4_rate, - )], + pf_v4: vec![rule(IpAddr::V4(Ipv4Addr::new(198, 18, 1, 1)), v4_rate)], pf_v6: vec![rule( IpAddr::V6("fd00:18:1::1".parse::().unwrap()), v6_rate, diff --git a/src/transport/socket/fragmented_send_wire_tests.rs b/src/transport/socket/fragmented_send_wire_tests.rs index 744816b..acea623 100644 --- a/src/transport/socket/fragmented_send_wire_tests.rs +++ b/src/transport/socket/fragmented_send_wire_tests.rs @@ -133,26 +133,28 @@ async fn run_server(addr: SocketAddr, fragment_size: u16, fake_cert_len: usize) let replay_checker = ReplayChecker::new(128, Duration::from_secs(60)); let rng = SecureRandom::new(); let shared = ProxySharedState::new(); - let (tls_reader, mut tls_writer, user) = match - handle_tls_handshake_with_shared_and_options( - &client_hello, - read_half, - write_half, - peer, - &config, - &replay_checker, - &rng, - None, - &shared, - TlsResponseWriteOptions::tcp(raw_fd, Some(fragment_size)), - ) - .await + let (tls_reader, mut tls_writer, user) = match handle_tls_handshake_with_shared_and_options( + &client_hello, + read_half, + write_half, + peer, + &config, + &replay_checker, + &rng, + None, + &shared, + TlsResponseWriteOptions::tcp(raw_fd, Some(fragment_size)), + ) + .await { HandshakeResult::Success(result) => result, _ => panic!("wire-test FakeTLS authentication failed"), }; assert_eq!(user, "wire"); - tls_writer.write_all(&vec![0xA5; BULK_PAYLOAD_LEN]).await.unwrap(); + tls_writer + .write_all(&vec![0xA5; BULK_PAYLOAD_LEN]) + .await + .unwrap(); tls_writer.shutdown().await.unwrap(); drop(tls_reader); // SAFETY: the write half still owns the accepted socket while it is borrowed. diff --git a/src/transport/socket/tests.rs b/src/transport/socket/tests.rs index 4e9f851..5ecfeb2 100644 --- a/src/transport/socket/tests.rs +++ b/src/transport/socket/tests.rs @@ -351,9 +351,7 @@ async fn test_chunked_send_has_no_fd_growth_after_success_and_cancellation_stres send_tcp_fragmented_fd(success_fd, &[0x5A], 92) .await .unwrap_or_else(|error| { - panic!( - "success cycle {iteration} failed with MSS {success_mss_before}: {error}" - ) + panic!("success cycle {iteration} failed with MSS {success_mss_before}: {error}") }); let mut received = [0_u8; 1]; success_client.read_exact(&mut received).await.unwrap();