From 831cd0f1a7aa17995126dec62ab8b0cf62caebf1 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Thu, 15 Oct 2026 07:05:36 +0300 Subject: [PATCH] WEB: share immutable profiles + decoys across exact host lists --- src/api/config_edit.rs | 4 + src/api/config_edit/host_list_tests.rs | 139 +++++++ src/config/hot_reload.rs | 4 + src/config/hot_reload/host_list_tests.rs | 68 ++++ src/config/load/decoy_dns.rs | 4 +- src/config/load/decoy_dns/tests.rs | 33 +- src/config/load/runtime_web.rs | 94 +++-- src/config/load/strict_keys.rs | 9 +- src/config/load/validate_web/vhosts.rs | 65 +++- .../tests/load_basic_tests/web_tests.rs | 48 ++- .../web_tests/host_list_tests.rs | 241 ++++++++++++ src/config/types.rs | 4 +- src/config/types/web.rs | 28 +- src/config/types/web/runtime.rs | 23 +- src/maestro/helpers/runtime.rs | 23 +- .../helpers/runtime/host_list_tests.rs | 73 ++++ src/web/http.rs | 8 +- src/web/http/base_path_tests.rs | 4 +- src/web/http/base_path_tests/credentials.rs | 8 +- src/web/http/base_path_tests/routing.rs | 4 +- src/web/http/decoy.rs | 2 +- src/web/http/decoy_dns_tests.rs | 4 +- src/web/http/decoy_fasttrack_tests.rs | 4 +- src/web/http/host_list_benchmarks.rs | 71 ++++ src/web/http/host_list_tests.rs | 350 ++++++++++++++++++ src/web/http/host_list_tests/sessions.rs | 211 +++++++++++ src/web/http/recovery.rs | 8 +- src/web/http/tests.rs | 40 +- src/web/manager/carrier_outcome.rs | 2 +- src/web/manager/credentials.rs | 2 +- src/web/manager/session_admission.rs | 2 +- src/web/manager/session_creation.rs | 38 +- .../manager/session_creation/replacement.rs | 12 +- src/web/manager/state.rs | 29 +- src/web/manager/state/expiry/fixtures.rs | 28 +- src/web/session/backend.rs | 8 +- src/web/session/backend_tests.rs | 34 +- src/web/session/conveyor.rs | 1 + src/web/session/conveyor/tests.rs | 28 +- src/web/session/downlink_tests.rs | 28 +- src/web/session/lane_uplink.rs | 1 + src/web/session/lanes/tests.rs | 28 +- src/web/session/negotiation/tests.rs | 28 +- src/web/session/status.rs | 8 +- src/web/session/stream_io/tests.rs | 28 +- src/web/session/uplink.rs | 8 +- src/web/session/uplink_tests.rs | 28 +- src/web/session/websocket.rs | 14 +- src/web/session/websocket/tests.rs | 64 ++-- src/web/trace/exchange.rs | 7 +- src/web/trace/types.rs | 4 +- 51 files changed, 1661 insertions(+), 343 deletions(-) create mode 100644 src/api/config_edit/host_list_tests.rs create mode 100644 src/config/hot_reload/host_list_tests.rs create mode 100644 src/config/tests/load_basic_tests/web_tests/host_list_tests.rs create mode 100644 src/maestro/helpers/runtime/host_list_tests.rs create mode 100644 src/web/http/host_list_benchmarks.rs create mode 100644 src/web/http/host_list_tests.rs create mode 100644 src/web/http/host_list_tests/sessions.rs diff --git a/src/api/config_edit.rs b/src/api/config_edit.rs index 4894c7c..6c34dbb 100644 --- a/src/api/config_edit.rs +++ b/src/api/config_edit.rs @@ -463,3 +463,7 @@ mod conveyor_tests; #[cfg(test)] #[path = "config_edit/decoy_dns_tests.rs"] mod decoy_dns_tests; + +// Exact host groups retain the source shape and process-deferred limit checks. +#[cfg(test)] +mod host_list_tests; diff --git a/src/api/config_edit/host_list_tests.rs b/src/api/config_edit/host_list_tests.rs new file mode 100644 index 0000000..90c1ab3 --- /dev/null +++ b/src/api/config_edit/host_list_tests.rs @@ -0,0 +1,139 @@ +use super::*; + +fn source() -> &'static str { + r#" +[access.users] +alice = "000102030405060708090a0b0c0d0e0f" +[web] +enabled = true +[[server.listeners]] +ip = "127.0.0.1" +port = 18080 +transport = "web" +web_trusted_proxy_cidrs = ["127.0.0.1/32"] +[[web.vhosts]] +host = "a.example.com" +public_addr = "203.0.113.10:443" +[web.vhosts.decoy] +mode = "http_upstream" +upstream = "http://127.0.0.1:18081" +[[web.vhosts.profiles]] +user = "alice" +secret_mode = "dd" +"# +} + +fn patch(hosts: &[&str]) -> Json { + serde_json::json!({"web": {"vhosts": [{ + "hosts": hosts, + "public_addr": "203.0.113.10:443", + "base_path": "MixedCase/path", + "decoy": {"mode": "http_upstream", "upstream": "http://127.0.0.1:18081"}, + "profiles": [{"user": "alice", "secret_mode": "dd"}] + }]}}) +} + +#[tokio::test] +async fn grouped_hosts_patch_preserves_source_shape_and_revision_guards() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.toml"); + std::fs::write(&path, source()).unwrap(); + let initial = ProxyConfig::load(&path).unwrap(); + let (legacy, revision) = read_managed_config(&path).await.unwrap(); + assert_eq!( + legacy["web"]["vhosts"][0]["host"].as_str(), + Some("a.example.com") + ); + assert!(legacy["web"]["vhosts"][0].get("hosts").is_none()); + let mut response = apply_patch_to_path( + &path, + &patch(&["A.Example.COM", "b.example.com"]), + Some(revision.clone()), + ) + .await + .unwrap(); + let desired = ProxyConfig::load(&path).unwrap(); + let resolved = reconcile_runtime_effect(&mut response, &initial, &desired).unwrap(); + assert!(resolved.runtime_changed); + assert!(response.runtime_reload_required); + assert!(!response.process_restart_required); + assert_eq!( + resolved + .effective + .web + .runtime + .as_ref() + .unwrap() + .profiles + .len(), + 2 + ); + let (managed, after_revision) = read_managed_config(&path).await.unwrap(); + assert_eq!(managed["web"]["vhosts"].as_array().unwrap().len(), 1); + let group = &managed["web"]["vhosts"][0]; + assert!(group.get("host").is_none()); + assert_eq!(group["hosts"][0].as_str(), Some("a.example.com")); + assert_eq!(group["hosts"][1].as_str(), Some("b.example.com")); + assert!(managed["web"].get("runtime").is_none()); + assert!(!toml::to_string(&managed).unwrap().contains("shared")); + assert!(managed.get("access").is_none()); + let before = std::fs::read(&path).unwrap(); + for hosts in [vec![], vec!["a.example.com", "A.example.com"], vec!["*"]] { + let error = apply_patch_to_path(&path, &patch(&hosts), Some(after_revision.clone())) + .await + .unwrap_err(); + assert_eq!(error.status, hyper::StatusCode::BAD_REQUEST); + assert_eq!(std::fs::read(&path).unwrap(), before); + assert_eq!(read_managed_config(&path).await.unwrap().1, after_revision); + } + let conflict = apply_patch_to_path(&path, &patch(&["c.example.com"]), Some(revision)) + .await + .unwrap_err(); + assert_eq!(conflict.code, "revision_conflict"); + assert_eq!(std::fs::read(&path).unwrap(), before); + apply_patch_to_path(&path, &patch(&["b.example.com"]), Some(after_revision)) + .await + .unwrap(); + let replaced = ProxyConfig::load(&path).unwrap(); + let runtime = replaced.web.runtime.as_ref().unwrap(); + assert_eq!(runtime.vhosts.len(), 1); + assert!(!runtime.vhosts.contains_key("a.example.com")); +} + +#[tokio::test] +async fn grouped_hosts_cannot_bypass_frozen_process_limits() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.toml"); + std::fs::write( + &path, + format!( + "{}\n[web.limits]\nmax_vhosts = 1\nmax_profiles = 1\n", + source() + ), + ) + .unwrap(); + let active = ProxyConfig::load(&path).unwrap(); + let before = std::fs::read(&path).unwrap(); + for (hosts, host_limit, profile_limit, error_field) in [ + (vec!["a.example.com", "b.example.com"], 2, 2, "max_vhosts"), + (vec!["a.example.com"], 1, 2, "max_profiles"), + ] { + let mut candidate = patch(&hosts); + candidate["web"]["limits"] = + serde_json::json!({"max_vhosts": host_limit, "max_profiles": profile_limit}); + if hosts.len() == 1 { + candidate["web"]["vhosts"][0]["profiles"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({"user": "alice", "secret_mode": "plain"})); + } + let error = prepare_patch_to_path(&path, &candidate, None) + .await + .err() + .expect("expanded config must respect active process limits"); + assert!(error.message.contains(error_field), "{error:?}"); + assert_eq!(error.status, hyper::StatusCode::BAD_REQUEST); + assert_eq!(std::fs::read(&path).unwrap(), before); + } + assert_eq!(active.web.runtime.as_ref().unwrap().vhosts.len(), 1); +} diff --git a/src/config/hot_reload.rs b/src/config/hot_reload.rs index 8cb4d78..a43f4df 100644 --- a/src/config/hot_reload.rs +++ b/src/config/hot_reload.rs @@ -76,3 +76,7 @@ mod carrier_method_tests; #[cfg(test)] #[path = "hot_reload/conveyor_tests.rs"] mod conveyor_tests; + +// Host groups are validated and published as one complete immutable snapshot. +#[cfg(test)] +mod host_list_tests; diff --git a/src/config/hot_reload/host_list_tests.rs b/src/config/hot_reload/host_list_tests.rs new file mode 100644 index 0000000..ec7dd3d --- /dev/null +++ b/src/config/hot_reload/host_list_tests.rs @@ -0,0 +1,68 @@ +use super::*; + +#[test] +fn host_list_reload_publishes_complete_tables_and_preserves_invalid_candidates() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.toml"); + let source = r#" +[access.users] +alice = "000102030405060708090a0b0c0d0e0f" +[web] +enabled = true +[[server.listeners]] +ip = "127.0.0.1" +port = 18080 +transport = "web" +web_trusted_proxy_cidrs = ["127.0.0.1/32"] +[[web.vhosts]] +host = "a.example.com" +public_addr = "203.0.113.10:443" +[web.vhosts.decoy] +mode = "http_upstream" +upstream = "http://127.0.0.1:18081" +[[web.vhosts.profiles]] +user = "alice" +secret_mode = "dd" +"#; + std::fs::write(&path, source).unwrap(); + let loaded = ProxyConfig::load_with_metadata(&path).unwrap(); + let initial = Arc::new(loaded.config); + let (config_tx, _config_rx) = watch::channel(Arc::clone(&initial)); + let (log_tx, _log_rx) = watch::channel(initial.general.log_level.clone()); + let mut state = ReloadState::new(Some(loaded.rendered_hash)); + let capability = initial.web.runtime.as_ref().unwrap().capabilities[0]; + for (selector, valid, hosts) in [ + ( + "hosts = [\"a.example.com\", \"A.Example.COM\"]", + false, + vec![], + ), + ( + "hosts = [\"b.example.com\", \"a.example.com\"]", + true, + vec!["b.example.com", "a.example.com"], + ), + ("hosts = [\"a.example.com\"]", true, vec!["a.example.com"]), + ] { + let before = config_tx.borrow().clone(); + std::fs::write(&path, source.replace("host = \"a.example.com\"", selector)).unwrap(); + reload_config(&path, &config_tx, &log_tx, None, None, &mut state); + let applied = config_tx.borrow().clone(); + if !valid { + assert!(Arc::ptr_eq(&before, &applied)); + continue; + } + let runtime = applied.web.runtime.as_ref().unwrap(); + assert_eq!(runtime.vhosts.len(), hosts.len()); + assert_eq!(runtime.profiles.len(), hosts.len()); + assert_eq!(runtime.capabilities.len(), hosts.len()); + assert_eq!(runtime.vhosts["a.example.com"].capabilities[0], capability); + for (idx, host) in hosts.iter().enumerate() { + assert_eq!(runtime.profiles[idx].host, *host); + assert_eq!( + runtime.capabilities[idx], + runtime.vhosts[*host].capabilities[0] + ); + } + } +} diff --git a/src/config/load/decoy_dns.rs b/src/config/load/decoy_dns.rs index 63604f8..65869f3 100644 --- a/src/config/load/decoy_dns.rs +++ b/src/config/load/decoy_dns.rs @@ -191,8 +191,8 @@ impl ProxyConfig { .runtime .iter() .flat_map(|runtime| runtime.vhosts.iter()) - .filter_map(|(host, vhost)| match vhost.decoy { - WebRuntimeDecoy::HttpUpstream { addr, .. } => Some((host, addr)), + .filter_map(|(host, vhost)| match vhost.decoy.as_ref() { + WebRuntimeDecoy::HttpUpstream { addr, .. } => Some((host, *addr)), WebRuntimeDecoy::StaticDirectory(_) => None, }) } diff --git a/src/config/load/decoy_dns/tests.rs b/src/config/load/decoy_dns/tests.rs index cf64177..f9054fc 100644 --- a/src/config/load/decoy_dns/tests.rs +++ b/src/config/load/decoy_dns/tests.rs @@ -45,7 +45,7 @@ async fn prepared(answers: &[&str]) -> LoadedConfig { fn endpoint(config: &ProxyConfig) -> (SocketAddr, &str) { let runtime = config.web.runtime.as_ref().unwrap(); - match &runtime.vhosts["proxy.example.com"].decoy { + match runtime.vhosts["proxy.example.com"].decoy.as_ref() { WebRuntimeDecoy::HttpUpstream { addr, authority } => (*addr, authority), _ => panic!("expected HTTP decoy"), } @@ -55,7 +55,7 @@ fn endpoint(config: &ProxyConfig) -> (SocketAddr, &str) { async fn decoy_dns_snapshot_deduplicates_normalized_origins_and_preserves_order() { let mut parsed = source(SOURCE); let mut second = parsed.config.web.vhosts[0].clone(); - second.host = "bsi.bund.de".to_string(); + second.host = Some("bsi.bund.de".to_string()); second.decoy = WebDecoyConfig::HttpUpstream { upstream: "http://example.com:18081/".to_string(), resolve: WebDecoyResolve::Startup, @@ -328,3 +328,32 @@ async fn decoy_dns_default_port_authority_and_evidence_port_fence() { ); assert!(config.rebuild_runtime_web().is_err()); } + +#[tokio::test] +async fn grouped_hosts_share_one_resolved_origin_and_endpoint_inventory() { + let mut parsed = source(&SOURCE.replace( + "host = \"proxy.example.com\"", + "hosts = [\"proxy.example.com\", \"other.example.com\"]", + )); + let calls = AtomicUsize::new(0); + prepare(&mut parsed.config, |_, _| { + calls.fetch_add(1, Ordering::SeqCst); + std::future::ready(Ok(vec!["127.0.0.2:18081".parse().unwrap()])) + }) + .await + .unwrap(); + let loaded = pipeline::finish(parsed).unwrap(); + let runtime = loaded.config.web.runtime.as_ref().unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert!(Arc::ptr_eq( + &runtime.vhosts["proxy.example.com"].decoy, + &runtime.vhosts["other.example.com"].decoy + )); + let mut reordered = loaded.config.clone(); + reordered.web.vhosts[0].hosts.as_mut().unwrap().reverse(); + reordered.rebuild_runtime_web().unwrap(); + assert!(loaded.config.web_decoy_endpoints_equal(&reordered)); + reordered.web.vhosts[0].hosts.as_mut().unwrap().pop(); + reordered.rebuild_runtime_web().unwrap(); + assert!(!loaded.config.web_decoy_endpoints_equal(&reordered)); +} diff --git a/src/config/load/runtime_web.rs b/src/config/load/runtime_web.rs index e7abf2c..a23c263 100644 --- a/src/config/load/runtime_web.rs +++ b/src/config/load/runtime_web.rs @@ -48,17 +48,27 @@ pub(super) fn rebuild(config: &mut ProxyConfig) -> Result<()> { let carrier_candidates: Arc<[WebCarrier]> = config.web.carrier_candidates().into(); for (vhost_idx, vhost) in config.web.vhosts.iter().enumerate() { - let decoy = build_decoy( + let decoy = Arc::new(build_decoy( config, vhost_idx, vhost, &config.web.limits, &mut static_files, &mut static_bytes, - )?; - let mut profiles = Vec::with_capacity(vhost.profiles.len()); - let mut capability_table = Vec::with_capacity(vhost.profiles.len()); - let mut capabilities = HashSet::with_capacity(vhost.profiles.len()); + )?); + let mut host_profiles: Vec<_> = vhost + .host + .iter() + .chain(vhost.hosts.iter().flatten()) + .map(|host| { + ( + host, + Vec::with_capacity(vhost.profiles.len()), + Vec::with_capacity(vhost.profiles.len()), + HashSet::with_capacity(vhost.profiles.len()), + ) + }) + .collect(); for profile in &vhost.profiles { let user_id = auth.user_id_by_name(&profile.user).ok_or_else(|| { ProxyError::Config(format!( @@ -71,20 +81,8 @@ pub(super) fn rebuild(config: &mut ProxyConfig) -> Result<()> { })?; let (client_secret, client_secret_len) = client_secret(auth_entry.secret, profile.secret_mode); - let capability = derive_web_capability( - &client_secret[..client_secret_len], - vhost.host.as_bytes(), - vhost.base_path.as_bytes(), - )?; let key_fingerprint = debug_key_fingerprint(&client_secret[..client_secret_len]); - if !capabilities.insert(capability) { - return Err(ProxyError::Config(format!( - "WEB vhost `{}` contains profiles with the same client capability", - vhost.host - ))); - } - let runtime_profile = Arc::new(WebRuntimeProfile { - host: vhost.host.clone(), + let shared = Arc::new(WebRuntimeProfileShared { public_addr: vhost.public_addr, user: profile.user.clone(), credential_id: auth_entry.credential_id, @@ -98,7 +96,6 @@ pub(super) fn rebuild(config: &mut ProxyConfig) -> Result<()> { .web .timeouts .carrier_negotiation_deadlines_secs, - capability, key_fingerprint, max_sessions: profile .max_sessions @@ -110,27 +107,46 @@ pub(super) fn rebuild(config: &mut ProxyConfig) -> Result<()> { .max_streams_per_session .unwrap_or(config.web.limits.max_streams_per_session), }); - capability_table.push(capability); - runtime_capabilities.push(capability); - profiles.push(Arc::clone(&runtime_profile)); - runtime_profiles.push(runtime_profile); + // Only host-bound bindings and capability indexes expand with the alias count. + for (host, profiles, capability_table, capabilities) in &mut host_profiles { + let capability = derive_web_capability( + &client_secret[..client_secret_len], + host.as_bytes(), + vhost.base_path.as_bytes(), + )?; + if !capabilities.insert(capability) { + return Err(ProxyError::Config(format!( + "WEB vhost `{host}` contains profiles with the same client capability" + ))); + } + profiles.push(Arc::new(WebRuntimeProfile { + host: (*host).clone(), + capability, + shared: Arc::clone(&shared), + })); + capability_table.push(capability); + } + } + for (host, profiles, capability_table, _) in host_profiles { + runtime_capabilities.extend(capability_table.iter().copied()); + runtime_profiles.extend(profiles.iter().cloned()); + runtime_vhosts.insert( + host.clone(), + Arc::new(WebRuntimeVhost { + host: host.clone(), + base: if vhost.base_path.is_empty() { + "/".to_string() + } else { + format!("/{}/", vhost.base_path) + }, + decoy_fasttrack_mode: config.web.decoy_fasttrack_mode, + decoy: Arc::clone(&decoy), + decoy_header_secs: config.web.timeouts.decoy_header_secs, + profiles, + capabilities: capability_table.into_boxed_slice(), + }), + ); } - runtime_vhosts.insert( - vhost.host.clone(), - Arc::new(WebRuntimeVhost { - host: vhost.host.clone(), - base: if vhost.base_path.is_empty() { - "/".to_string() - } else { - format!("/{}/", vhost.base_path) - }, - decoy_fasttrack_mode: config.web.decoy_fasttrack_mode, - decoy, - decoy_header_secs: config.web.timeouts.decoy_header_secs, - profiles, - capabilities: capability_table.into_boxed_slice(), - }), - ); } config.web.runtime = Some(Arc::new(WebRuntimeConfig { diff --git a/src/config/load/strict_keys.rs b/src/config/load/strict_keys.rs index 171896d..8d54baa 100644 --- a/src/config/load/strict_keys.rs +++ b/src/config/load/strict_keys.rs @@ -368,7 +368,14 @@ const WEB_TIMEOUTS_CONFIG_KEYS: &[&str] = &[ "decoy_resolve_secs", ]; -const WEB_VHOST_CONFIG_KEYS: &[&str] = &["host", "base_path", "public_addr", "decoy", "profiles"]; +const WEB_VHOST_CONFIG_KEYS: &[&str] = &[ + "host", + "hosts", + "base_path", + "public_addr", + "decoy", + "profiles", +]; const WEB_DECOY_CONFIG_KEYS: &[&str] = &["mode", "upstream", "directory", "index", "resolve"]; const WEB_PROFILE_CONFIG_KEYS: &[&str] = &[ "user", diff --git a/src/config/load/validate_web/vhosts.rs b/src/config/load/validate_web/vhosts.rs index d3a8527..66f3078 100644 --- a/src/config/load/validate_web/vhosts.rs +++ b/src/config/load/validate_web/vhosts.rs @@ -1,21 +1,57 @@ use super::*; +/// Bounds expanded routing tables before normalizing exact hostname selectors. pub(super) fn validate_vhosts(config: &mut ProxyConfig) -> Result<()> { let limits = &config.web.limits; - if config.web.vhosts.len() > limits.max_vhosts { - return config_error("web.vhosts exceeds web.limits.max_vhosts"); - } - let mut hosts = HashSet::with_capacity(config.web.vhosts.len()); + let mut host_count = 0usize; let mut profile_count = 0usize; + for (vhost_idx, vhost) in config.web.vhosts.iter().enumerate() { + let count = match (&vhost.host, &vhost.hosts) { + (Some(_), None) => 1, + (None, Some(hosts)) if !hosts.is_empty() => hosts.len(), + _ => { + return config_error(&format!( + "web.vhosts[{vhost_idx}] requires exactly one of host or non-empty hosts" + )); + } + }; + host_count = host_count + .checked_add(count) + .ok_or_else(|| ProxyError::Config("WEB hostname count overflowed usize".to_string()))?; + profile_count = count + .checked_mul(vhost.profiles.len()) + .and_then(|expanded| profile_count.checked_add(expanded)) + .ok_or_else(|| ProxyError::Config("WEB profile count overflowed usize".to_string()))?; + if host_count > limits.max_vhosts { + return config_error("web.vhosts exceeds web.limits.max_vhosts"); + } + if profile_count > limits.max_profiles { + return config_error("WEB profiles exceed web.limits.max_profiles"); + } + } + let mut hosts = HashSet::with_capacity(host_count); for (vhost_idx, vhost) in config.web.vhosts.iter_mut().enumerate() { - vhost.host = normalize_web_host(&vhost.host, &format!("web.vhosts[{vhost_idx}].host"))?; + let single = vhost.host.is_some(); + for (host_idx, host) in vhost + .host + .iter_mut() + .chain(vhost.hosts.iter_mut().flatten()) + .enumerate() + { + let field = if single { + format!("web.vhosts[{vhost_idx}].host") + } else { + format!("web.vhosts[{vhost_idx}].hosts[{host_idx}]") + }; + *host = normalize_web_host(host, &field)?; + if !hosts.insert(host.clone()) { + return config_error(&format!("duplicate WEB vhost host `{host}` at {field}")); + } + } validate_web_base_path( &vhost.base_path, &format!("web.vhosts[{vhost_idx}].base_path"), )?; - if !hosts.insert(vhost.host.clone()) { - return config_error(&format!("duplicate WEB vhost host `{}`", vhost.host)); - } if vhost.public_addr.port() != 443 || vhost.public_addr.ip().is_unspecified() { return config_error(&format!( "web.vhosts[{vhost_idx}].public_addr must be a concrete socket address on port 443" @@ -42,8 +78,8 @@ pub(super) fn validate_vhosts(config: &mut ProxyConfig) -> Result<()> { } if !profiles.insert((profile.user.as_str(), profile.secret_mode)) { return config_error(&format!( - "duplicate WEB profile for user `{}` in vhost `{}`", - profile.user, vhost.host + "duplicate WEB profile for user `{}` in web.vhosts[{vhost_idx}]", + profile.user )); } let max_streams = profile.max_streams.unwrap_or(limits.max_streams_global); @@ -68,14 +104,8 @@ pub(super) fn validate_vhosts(config: &mut ProxyConfig) -> Result<()> { "web.vhosts[{vhost_idx}].profiles[{profile_idx}] limits must be non-zero and within global WEB limits" )); } - profile_count = profile_count.checked_add(1).ok_or_else(|| { - ProxyError::Config("WEB profile count overflowed usize".to_string()) - })?; } } - if profile_count > limits.max_profiles { - return config_error("WEB profiles exceed web.limits.max_profiles"); - } Ok(()) } @@ -98,6 +128,7 @@ fn validate_web_base_path(value: &str, field: &str) -> Result<()> { )) } +/// Applies the client-compatible hostname rules only during configuration loading. pub(super) fn normalize_web_host(value: &str, field: &str) -> Result { let input = value.trim(); if input.is_empty() @@ -137,6 +168,7 @@ pub(super) fn normalize_web_host(value: &str, field: &str) -> Result { Ok(host) } +/// Rejects numeric suffixes that clients may interpret as address literals. pub(super) fn web_host_last_label_is_numeric(host: &str) -> bool { let label = host.rsplit('.').next().unwrap_or_default(); let digits = label @@ -148,6 +180,7 @@ pub(super) fn web_host_last_label_is_numeric(host: &str) -> bool { label.bytes().all(|byte| byte.is_ascii_digit()) } +/// Validates one group's ordinary-site origin or bounded static snapshot source. pub(super) fn validate_decoy(vhost_idx: usize, decoy: &WebDecoyConfig) -> Result<()> { match decoy { WebDecoyConfig::HttpUpstream { upstream, resolve } => { diff --git a/src/config/tests/load_basic_tests/web_tests.rs b/src/config/tests/load_basic_tests/web_tests.rs index 976b171..a61cada 100644 --- a/src/config/tests/load_basic_tests/web_tests.rs +++ b/src/config/tests/load_basic_tests/web_tests.rs @@ -11,6 +11,9 @@ mod carrier_method_tests; mod conveyor_tests; #[path = "web_tests/decoy_dns_tests.rs"] mod decoy_dns_tests; +// Exact host groups must preserve the legacy per-host routing contract. +#[path = "web_tests/host_list_tests.rs"] +mod host_list_tests; const WEB_CONFIG: &str = r#" [access.users] @@ -58,17 +61,17 @@ fn web_config_builds_canonical_runtime_snapshot() { assert_eq!(vhost.capabilities[0], vhost.profiles[0].capability); assert_eq!(runtime.capabilities.as_ref(), vhost.capabilities.as_ref()); assert_eq!(vhost.decoy_fasttrack_mode, WebDecoyFastTrackMode::Off); - assert_eq!(vhost.profiles[0].user, "alice"); - assert_eq!(vhost.profiles[0].secret_mode, WebSecretMode::Dd); - assert_eq!(vhost.profiles[0].carrier, WebCarrier::HttpsLanes); - assert_eq!(vhost.profiles[0].max_sessions, 4); - assert_eq!(vhost.profiles[0].max_streams, 64); - assert_eq!(vhost.profiles[0].max_streams_per_session, 16); - assert_eq!(vhost.profiles[0].key_fingerprint.len(), 16); - assert_ne!(vhost.profiles[0].key_fingerprint, "0001020304050607"); - assert!(!vhost.profiles[0].carrier_negotiation_enabled); + assert_eq!(vhost.profiles[0].shared.user, "alice"); + assert_eq!(vhost.profiles[0].shared.secret_mode, WebSecretMode::Dd); + assert_eq!(vhost.profiles[0].shared.carrier, WebCarrier::HttpsLanes); + assert_eq!(vhost.profiles[0].shared.max_sessions, 4); + assert_eq!(vhost.profiles[0].shared.max_streams, 64); + assert_eq!(vhost.profiles[0].shared.max_streams_per_session, 16); + assert_eq!(vhost.profiles[0].shared.key_fingerprint.len(), 16); + assert_ne!(vhost.profiles[0].shared.key_fingerprint, "0001020304050607"); + assert!(!vhost.profiles[0].shared.carrier_negotiation_enabled); assert_eq!( - vhost.profiles[0].carriers.as_ref(), + vhost.profiles[0].shared.carriers.as_ref(), [WebCarrier::HttpsLanes] ); } @@ -251,7 +254,11 @@ fn web_profile_user_labels_are_bounded_for_runtime_status() { fn web_carriers_missing_or_false_disable_negotiation() { let missing = load_config_from_temp_toml(WEB_CONFIG); assert!(!missing.web.carrier_negotiation_enabled()); - assert!(!missing.web.runtime.unwrap().profiles[0].carrier_learning); + assert!( + !missing.web.runtime.unwrap().profiles[0] + .shared + .carrier_learning + ); let disabled = WEB_CONFIG.replace( "carrier = \"https-lanes\"", @@ -259,9 +266,16 @@ fn web_carriers_missing_or_false_disable_negotiation() { ); let disabled = load_config_from_temp_toml(&disabled); assert!(!disabled.web.carrier_negotiation_enabled()); - assert!(!disabled.web.runtime.as_ref().unwrap().profiles[0].carrier_learning); + assert!( + !disabled.web.runtime.as_ref().unwrap().profiles[0] + .shared + .carrier_learning + ); assert_eq!( - disabled.web.runtime.unwrap().profiles[0].carriers.as_ref(), + disabled.web.runtime.unwrap().profiles[0] + .shared + .carriers + .as_ref(), [WebCarrier::HttpsLanes] ); } @@ -277,14 +291,14 @@ fn web_carrier_array_enables_ordered_negotiation_and_appends_fallback() { assert!(!config.web.carrier_learning); let profile = &config.web.runtime.unwrap().profiles[0]; assert_eq!( - profile.carriers.as_ref(), + profile.shared.carriers.as_ref(), [ WebCarrier::Websocket, WebCarrier::Https, WebCarrier::HttpsLanes ] ); - assert!(!profile.carrier_learning); + assert!(!profile.shared.carrier_learning); } #[test] @@ -494,7 +508,7 @@ fn web_ipv6_decoy_uses_a_valid_http_authority() { let config = load_config_from_temp_toml(&ipv6); let runtime = config.web.runtime.expect("WEB runtime snapshot"); let vhost = runtime.vhosts.get("proxy.example.com").unwrap(); - let WebRuntimeDecoy::HttpUpstream { authority, .. } = &vhost.decoy else { + let WebRuntimeDecoy::HttpUpstream { authority, .. } = vhost.decoy.as_ref() else { panic!("expected HTTP decoy"); }; assert_eq!(authority, "[::1]:18081"); @@ -509,7 +523,7 @@ fn websocket_carriers_build_runtime_profiles_with_bounded_defaults() { let configured = WEB_CONFIG.replace("https-lanes", name); let config = load_config_from_temp_toml(&configured); let profile = &config.web.runtime.unwrap().profiles[0]; - assert_eq!(profile.carrier, carrier); + assert_eq!(profile.shared.carrier, carrier); assert_eq!(config.web.limits.websocket_bytes_global, 256 * 1024 * 1024); assert_eq!(config.web.limits.websocket_admission_watermark_pct, 75); assert_eq!(config.web.limits.websocket_eviction_watermark_pct, 90); diff --git a/src/config/tests/load_basic_tests/web_tests/host_list_tests.rs b/src/config/tests/load_basic_tests/web_tests/host_list_tests.rs new file mode 100644 index 0000000..df3f83c --- /dev/null +++ b/src/config/tests/load_basic_tests/web_tests/host_list_tests.rs @@ -0,0 +1,241 @@ +use super::*; + +#[test] +fn exact_host_list_builds_distinct_host_bound_capabilities() { + let source = WEB_CONFIG.replace( + "host = \"Proxy.Example.COM\"", + "hosts = [\"Proxy.Example.COM\", \"second.example.com\"]", + ); + let config = load_config_from_temp_toml(&source); + let runtime = config.web.runtime.as_ref().unwrap(); + assert_eq!(runtime.vhosts.len(), 2); + assert_eq!(runtime.profiles.len(), 2); + assert_eq!(runtime.capabilities.len(), 2); + assert_ne!(runtime.capabilities[0], runtime.capabilities[1]); + for host in ["proxy.example.com", "second.example.com"] { + let legacy = load_config_from_temp_toml(&WEB_CONFIG.replace("Proxy.Example.COM", host)); + assert_eq!( + runtime.vhosts[host].capabilities, + legacy.web.runtime.as_ref().unwrap().vhosts[host].capabilities, + ); + } +} + +#[test] +fn grouped_profiles_share_policy_but_match_legacy_bindings() { + for mode in ["plain", "dd"] { + for base in ["", "MixedCase/path"] { + let source = WEB_CONFIG + .replace("secret_mode = \"dd\"", &format!("secret_mode = \"{mode}\"")) + .replace( + "public_addr =", + &format!("base_path = \"{base}\"\npublic_addr ="), + ); + let grouped = load_config_from_temp_toml(&source.replace( + "host = \"Proxy.Example.COM\"", + "hosts = [\"Proxy.Example.COM\", \"second.example.com\"]", + )); + let runtime = grouped.web.runtime.as_ref().unwrap(); + let first = &runtime.vhosts["proxy.example.com"]; + let second = &runtime.vhosts["second.example.com"]; + assert!(Arc::ptr_eq(&first.decoy, &second.decoy)); + assert!(Arc::ptr_eq( + &first.profiles[0].shared, + &second.profiles[0].shared + )); + for host in ["proxy.example.com", "second.example.com"] { + let legacy = load_config_from_temp_toml(&source.replace("Proxy.Example.COM", host)); + let expected = &legacy.web.runtime.as_ref().unwrap().vhosts[host]; + let actual = &runtime.vhosts[host]; + assert_eq!(actual.base, expected.base); + assert_eq!(actual.capabilities, expected.capabilities); + assert_eq!(actual.profiles[0].host, host); + let actual = &actual.profiles[0].shared; + let expected = &expected.profiles[0].shared; + assert_eq!(actual.user, expected.user); + assert_eq!(actual.credential_id, expected.credential_id); + assert_eq!(actual.key_fingerprint, expected.key_fingerprint); + assert_eq!(actual.public_addr, expected.public_addr); + assert_eq!(actual.carriers, expected.carriers); + assert_eq!(actual.max_sessions, expected.max_sessions); + assert_eq!(actual.max_streams, expected.max_streams); + assert_eq!( + actual.max_streams_per_session, + expected.max_streams_per_session + ); + } + } + } +} + +#[test] +fn host_selectors_reject_ambiguity_and_canonical_collisions() { + for selector in [ + "", + "host = \"\"", + "hosts = []", + "hosts = [\"\"]", + "host = \"a.example.com\"\nhosts = []", + "host = \"a.example.com\"\nhosts = [\"b.example.com\"]", + "hosts = [\"a.example.com\", \"A.EXAMPLE.COM\"]", + "hosts = [\"BÜCHER.example\", \"xn--bcher-kva.example\"]", + "hosts = [\"*.example.com\"]", + "hosts = [\"*\"]", + "hosts = [\"127.0.0.1\"]", + "hosts = [\"a.example.com:443\"]", + "hosts = [\"a.example.com.\"]", + "hosts = [\"a.example.com/path\"]", + ] { + let error = load_config_error_from_temp_toml( + &WEB_CONFIG.replace("host = \"Proxy.Example.COM\"", selector), + ); + assert!(error.contains("web.vhosts[0]"), "{selector}: {error}"); + } + let group = WEB_CONFIG.replace( + "host = \"Proxy.Example.COM\"", + "hosts = [\"a.example.com\", \"proxy.example.com\"]", + ); + let (_, legacy_vhost) = WEB_CONFIG.split_once("[[web.vhosts]]").unwrap(); + let error = load_config_error_from_temp_toml(&format!("{group}\n[[web.vhosts]]{legacy_vhost}")); + assert!(error.contains("duplicate WEB vhost host `proxy.example.com` at web.vhosts[1].host")); +} + +#[test] +fn host_selectors_round_trip_without_expanding_the_source() { + for selector in [ + "host = \"Proxy.Example.COM\"", + "hosts = [\"Proxy.Example.COM\"]", + "hosts = [\"Proxy.Example.COM\", \"BÜCHER.example\"]", + ] { + let config = load_config_from_temp_toml(&format!( + "[general]\nconfig_strict = true\n{}", + WEB_CONFIG.replace("host = \"Proxy.Example.COM\"", selector), + )); + let value = serde_json::to_value(&config.web).unwrap(); + let vhosts = value["vhosts"].as_array().unwrap(); + assert_eq!(vhosts.len(), 1); + if selector.starts_with("host =") { + assert_eq!(vhosts[0]["host"], "proxy.example.com"); + assert!(vhosts[0].get("hosts").is_none()); + } else { + assert!(vhosts[0].get("host").is_none()); + assert_eq!(vhosts[0]["hosts"][0], "proxy.example.com"); + if vhosts[0]["hosts"].as_array().unwrap().len() == 2 { + assert_eq!(vhosts[0]["hosts"][1], "xn--bcher-kva.example"); + } + } + assert!(value.get("runtime").is_none()); + let mut document = toml::map::Map::new(); + document.insert( + "web".to_string(), + toml::Value::try_from(&config.web).unwrap(), + ); + let prefix = WEB_CONFIG.split_once("[web]").unwrap().0; + let round_trip = load_config_from_temp_toml(&format!( + "[general]\nconfig_strict = true\n{prefix}{}", + toml::to_string(&document).unwrap(), + )); + assert_eq!(round_trip.web.vhosts, config.web.vhosts); + } +} + +#[test] +fn expanded_host_profile_limits_apply_before_runtime_preparation() { + let source = WEB_CONFIG.replace( + "host = \"Proxy.Example.COM\"", + "hosts = [\"a.example.com\", \"b.example.com\"]", + ); + let source = format!( + "{source}\n[[web.vhosts.profiles]]\nuser = \"alice\"\nsecret_mode = \"plain\"\n[web.limits]\nmax_vhosts = 2\nmax_profiles = 4\n" + ); + let config = load_config_from_temp_toml(&source); + let runtime = config.web.runtime.as_ref().unwrap(); + assert_eq!(runtime.vhosts.len(), 2); + assert_eq!(runtime.profiles.len(), 4); + assert_eq!(runtime.capabilities.len(), 4); + assert_eq!( + runtime + .profiles + .iter() + .map(|profile| profile.host.as_str()) + .collect::>(), + [ + "a.example.com", + "a.example.com", + "b.example.com", + "b.example.com" + ] + ); + assert!(!Arc::ptr_eq( + &runtime.profiles[0].shared, + &runtime.profiles[1].shared + )); + assert!( + load_config_error_from_temp_toml(&source.replace("max_vhosts = 2", "max_vhosts = 1")) + .contains("max_vhosts") + ); + assert!( + load_config_error_from_temp_toml(&source.replace("max_profiles = 4", "max_profiles = 3")) + .contains("max_profiles") + ); + let invalid_origin = source.replace("http://127.0.0.1:18081", "http://example.com:18081"); + assert!( + load_config_error_from_temp_toml( + &invalid_origin.replace("max_profiles = 4", "max_profiles = 3") + ) + .contains("max_profiles") + ); +} + +#[test] +fn grouped_static_snapshot_is_loaded_and_charged_once() { + let directory = tempfile::tempdir().unwrap(); + std::fs::write(directory.path().join("index.html"), b"hello").unwrap(); + let source = WEB_CONFIG + .replace( + "host = \"Proxy.Example.COM\"", + "hosts = [\"a.example.com\", \"b.example.com\"]", + ) + .replace( + "mode = \"http_upstream\"\nupstream = \"http://127.0.0.1:18081\"", + &format!( + "mode = \"static_directory\"\ndirectory = {:?}", + directory.path() + ), + ); + let limits = + "\n[web.limits]\nmax_static_files = 1\nmax_static_file_bytes = 5\nmax_static_bytes = 5\n"; + let config = load_config_from_temp_toml(&format!("{source}{limits}")); + let runtime = config.web.runtime.as_ref().unwrap(); + let a = &runtime.vhosts["a.example.com"]; + let b = &runtime.vhosts["b.example.com"]; + assert!(Arc::ptr_eq(&a.decoy, &b.decoy)); + let WebRuntimeDecoy::StaticDirectory(site) = a.decoy.as_ref() else { + panic!("static decoy"); + }; + assert_eq!(site.assets["/index.html"].body.as_ref(), b"hello"); + let (prefix, group) = source.split_once("[[web.vhosts]]").unwrap(); + let selector = "hosts = [\"a.example.com\", \"b.example.com\"]"; + let separate = format!( + "{prefix}[[web.vhosts]]{}\n[[web.vhosts]]{}{limits}", + group.replace(selector, "host = \"a.example.com\""), + group.replace(selector, "host = \"b.example.com\"") + ); + assert!(load_config_error_from_temp_toml(&separate).contains("max_static")); + let expanded = load_config_from_temp_toml( + &separate + .replace("max_static_files = 1", "max_static_files = 2") + .replace("max_static_bytes = 5", "max_static_bytes = 10"), + ); + let expanded = expanded.web.runtime.as_ref().unwrap(); + assert!(!Arc::ptr_eq( + &expanded.vhosts["a.example.com"].decoy, + &expanded.vhosts["b.example.com"].decoy + )); + assert!(!Arc::ptr_eq( + &expanded.profiles[0].shared, + &expanded.profiles[1].shared + )); + std::fs::write(directory.path().join("index.html"), b"longer").unwrap(); + assert!(load_config_error_from_temp_toml(&format!("{source}{limits}")).contains("max_static")); +} diff --git a/src/config/types.rs b/src/config/types.rs index 16e9764..29ebccc 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -58,8 +58,8 @@ pub use web::{ WebSecretMode, WebTimeoutsConfig, WebVhostConfig, }; pub(crate) use web::{ - WebDecoyDnsSnapshot, WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile, WebRuntimeVhost, - WebStaticAsset, WebStaticSite, + WebDecoyDnsSnapshot, WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile, + WebRuntimeProfileShared, WebRuntimeVhost, WebStaticAsset, WebStaticSite, }; pub(crate) use web_carrier::WEB_CARRIER_LEARNING_MIN_ENTRIES; #[allow(unused_imports)] diff --git a/src/config/types/web.rs b/src/config/types/web.rs index 8b4514f..5815a46 100644 --- a/src/config/types/web.rs +++ b/src/config/types/web.rs @@ -73,19 +73,23 @@ pub enum WebDecoyConfig { }, } -/// One externally visible WEB hostname and its explicit access profiles. +/// Exact WEB hostnames sharing one immutable decoy and explicit profile policy. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct WebVhostConfig { - /// Canonical lowercase ACE hostname used by Telegram Desktop. - pub host: String, + /// Legacy single hostname selector, mutually exclusive with `hosts`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub host: Option, + /// Non-empty exact hostname list, normalized to lowercase ACE during validation. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub hosts: Option>, /// Optional canonical WEB endpoint prefix without surrounding slashes. #[serde(default)] pub base_path: String, /// Stable public destination tuple used by inner relay routing and KDF metadata. pub public_addr: SocketAddr, - /// Ordinary-site fallback for this hostname. + /// Ordinary-site fallback shared by every hostname in this group. pub decoy: WebDecoyConfig, - /// Access users and exact secret modes enabled for this hostname. + /// Access profiles whose limits apply independently to each hostname. #[serde(default)] pub profiles: Vec, } @@ -192,19 +196,19 @@ pub struct WebLimitsConfig { /// Live bootstrap credential ceiling for one forwarded client address. #[serde(default = "default_web_max_bootstraps_per_ip")] pub max_bootstraps_per_ip: usize, - /// Maximum configured WEB virtual-host count. + /// Maximum concrete hostname count across all WEB groups. #[serde(default = "default_web_max_vhosts")] pub max_vhosts: usize, - /// Maximum configured WEB access-profile count across all virtual hosts. + /// Maximum expanded hostname/profile binding count across all WEB groups. #[serde(default = "default_web_max_profiles")] pub max_profiles: usize, - /// Maximum static snapshot entry count across all virtual hosts. + /// Maximum static snapshot entry count, charged once per source group. #[serde(default = "default_web_max_static_files")] pub max_static_files: usize, /// Maximum bytes read from one static snapshot file. #[serde(default = "default_web_max_static_file_bytes")] pub max_static_file_bytes: usize, - /// Maximum static snapshot bytes across all virtual hosts. + /// Maximum static snapshot bytes, charged once per source group. #[serde(default = "default_web_max_static_bytes")] pub max_static_bytes: usize, /// Maximum retained WEB debug record count. @@ -454,7 +458,7 @@ pub struct WebConfig { /// WEB lifecycle deadlines. #[serde(default)] pub timeouts: WebTimeoutsConfig, - /// Public hostnames served by WEB listeners. + /// Exact hostname groups served by WEB listeners. #[serde(default)] pub vhosts: Vec, /// Validated immutable runtime snapshot built during configuration loading. @@ -511,6 +515,6 @@ impl Default for WebConfig { // Immutable runtime WEB configuration consumed by hot paths. mod runtime; pub(crate) use runtime::{ - WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile, WebRuntimeVhost, WebStaticAsset, - WebStaticSite, + WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile, WebRuntimeProfileShared, WebRuntimeVhost, + WebStaticAsset, WebStaticSite, }; diff --git a/src/config/types/web/runtime.rs b/src/config/types/web/runtime.rs index 58b79ff..1eb9014 100644 --- a/src/config/types/web/runtime.rs +++ b/src/config/types/web/runtime.rs @@ -21,7 +21,7 @@ pub(crate) struct WebRuntimeVhost { /// Restart-frozen decoy capability-scan policy. pub(crate) decoy_fasttrack_mode: WebDecoyFastTrackMode, /// Immutable ordinary-site fallback snapshot. - pub(crate) decoy: WebRuntimeDecoy, + pub(crate) decoy: Arc, /// Upstream connect and response-head deadline. pub(crate) decoy_header_secs: u64, /// Exact capability profiles accepted by this host. @@ -35,6 +35,16 @@ pub(crate) struct WebRuntimeVhost { pub(crate) struct WebRuntimeProfile { /// Canonical host that owns this profile. pub(crate) host: String, + /// HMAC-derived capability bound to this exact host and endpoint base. + pub(crate) capability: [u8; 32], + /// Immutable policy shared within a source group, never the quota owner. + pub(crate) shared: Arc, +} + +/// Host-independent profile policy retained by all bindings in one source group. +/// Limits still apply to each binding's exact host/user/mode quota owner. +#[derive(Debug)] +pub(crate) struct WebRuntimeProfileShared { /// Stable public destination tuple supplied to relay routing. pub(crate) public_addr: SocketAddr, /// Exact access user authenticated by logical streams. @@ -53,8 +63,6 @@ pub(crate) struct WebRuntimeProfile { pub(crate) carriers: Arc<[WebCarrier]>, /// Cumulative carrier-attempt deadlines frozen when the bridge is issued. pub(crate) carrier_negotiation_deadlines_secs: [u64; 4], - /// HMAC-derived bridge capability. - pub(crate) capability: [u8; 32], /// Non-secret domain-separated client-secret fingerprint for debugging. pub(crate) key_fingerprint: String, /// Per-profile live session ceiling. @@ -68,7 +76,14 @@ pub(crate) struct WebRuntimeProfile { /// Runtime-ready ordinary-site fallback. #[derive(Debug)] pub(crate) enum WebRuntimeDecoy { - HttpUpstream { addr: SocketAddr, authority: String }, + /// Fixed private origin prepared before the generation is published. + HttpUpstream { + /// Resolved destination, independent of the request Host. + addr: SocketAddr, + /// Configured origin authority forwarded to the upstream server. + authority: String, + }, + /// Immutable files acquired once for the entire source group. StaticDirectory(Arc), } diff --git a/src/maestro/helpers/runtime.rs b/src/maestro/helpers/runtime.rs index fcd19d1..98497a3 100644 --- a/src/maestro/helpers/runtime.rs +++ b/src/maestro/helpers/runtime.rs @@ -71,30 +71,29 @@ pub(crate) fn print_web_proxy_links(config: &ProxyConfig) { .resolve_users(&config.access.users); let mut heading_printed = false; for profile in &runtime.profiles { - if !shown.iter().any(|user| user.as_str() == profile.user) { + if !shown + .iter() + .any(|user| user.as_str() == profile.shared.user) + { continue; } if !heading_printed { print_maestro_line("WEB proxy links"); heading_printed = true; } - let Some(secret) = config.access.users.get(&profile.user) else { + let Some(secret) = config.access.users.get(&profile.shared.user) else { continue; }; - let Some(vhost) = config - .web - .vhosts - .iter() - .find(|vhost| vhost.host == profile.host) - else { + let Some(vhost) = runtime.vhosts.get(&profile.host) else { continue; }; print_maestro_line(format!( "User: {} ({:?})", - profile.user, profile.secret_mode + profile.shared.user, profile.shared.secret_mode )); + let base_path = vhost.base.trim_matches('/'); if let Some(link) = - format_web_proxy_link(&profile.host, &vhost.base_path, secret, profile.secret_mode) + format_web_proxy_link(&profile.host, base_path, secret, profile.shared.secret_mode) { print_maestro_line(format!("WEB: {link}")); } @@ -474,3 +473,7 @@ mod tests { } } } + +// Execute the actual emitter in a child so missing alias links cannot hide behind formatter tests. +#[cfg(test)] +mod host_list_tests; diff --git a/src/maestro/helpers/runtime/host_list_tests.rs b/src/maestro/helpers/runtime/host_list_tests.rs new file mode 100644 index 0000000..5c42c39 --- /dev/null +++ b/src/maestro/helpers/runtime/host_list_tests.rs @@ -0,0 +1,73 @@ +use super::*; + +#[test] +fn grouped_startup_links_include_every_alias_and_selected_profile() { + const CHILD: &str = "TELEMT_F1_LINK_TEST_CHILD"; + if std::env::var_os(CHILD).is_some() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.toml"); + std::fs::write( + &path, + r#" +[general.links] +show = ["alice"] +[access.users] +alice = "000102030405060708090a0b0c0d0e0f" +bob = "101112131415161718191a1b1c1d1e1f" +[[server.listeners]] +ip = "127.0.0.1" +port = 18080 +transport = "web" +web_trusted_proxy_cidrs = ["127.0.0.1/32"] +[web] +enabled = true +[[web.vhosts]] +hosts = ["a.example.com", "b.example.com"] +base_path = "MixedCase/path" +public_addr = "203.0.113.10:443" +[web.vhosts.decoy] +mode = "http_upstream" +upstream = "http://127.0.0.1:18081" +[[web.vhosts.profiles]] +user = "alice" +secret_mode = "plain" +[[web.vhosts.profiles]] +user = "alice" +secret_mode = "dd" +[[web.vhosts.profiles]] +user = "bob" +secret_mode = "dd" +"#, + ) + .unwrap(); + print_web_proxy_links(&ProxyConfig::load(&path).unwrap()); + return; + } + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "maestro::helpers::runtime::host_list_tests::grouped_startup_links_include_every_alias_and_selected_profile", "--nocapture"]) + .env(CHILD, "1") + .output().unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert_eq!(stderr.matches("WEB: tg://webproxy?").count(), 4, "{stderr}"); + for host in ["a.example.com", "b.example.com"] { + for mode in [ + crate::config::WebSecretMode::Plain, + crate::config::WebSecretMode::Dd, + ] { + let link = format_web_proxy_link( + host, + "MixedCase/path", + "000102030405060708090a0b0c0d0e0f", + mode, + ) + .unwrap(); + assert_eq!(stderr.matches(&link).count(), 1, "{stderr}"); + } + } + assert!(!stderr.contains("User: bob")); +} diff --git a/src/web/http.rs b/src/web/http.rs index bb387c7..1b81fb6 100644 --- a/src/web/http.rs +++ b/src/web/http.rs @@ -380,10 +380,10 @@ async fn handle_root( config.web.limits.carrier_batch_bytes, config.web.limits.pending_bytes_per_session, config.web.limits.pending_items_per_session, - profile.max_streams_per_session, - profile.carrier_negotiation_enabled, - profile.carriers.len(), - profile.carrier_negotiation_deadlines_secs, + profile.shared.max_streams_per_session, + profile.shared.carrier_negotiation_enabled, + profile.shared.carriers.len(), + profile.shared.carrier_negotiation_deadlines_secs, config.web.timeouts.long_poll_secs, config.web.timeouts.bridge_request_secs, config.web.timeouts.bridge_retry_secs, diff --git a/src/web/http/base_path_tests.rs b/src/web/http/base_path_tests.rs index 3337fa8..5ddd609 100644 --- a/src/web/http/base_path_tests.rs +++ b/src/web/http/base_path_tests.rs @@ -401,10 +401,10 @@ async fn prefixed_decoy_request_keeps_its_original_path_and_query() { host: "proxy.example.com".to_string(), base: "/relay/".to_string(), decoy_fasttrack_mode: WebDecoyFastTrackMode::Off, - decoy: WebRuntimeDecoy::HttpUpstream { + decoy: Arc::new(WebRuntimeDecoy::HttpUpstream { addr: site_addr, authority: "decoy.internal".to_string(), - }, + }), decoy_header_secs: 1, profiles: vec![Arc::clone(&profile)], capabilities: vec![capability].into_boxed_slice(), diff --git a/src/web/http/base_path_tests/credentials.rs b/src/web/http/base_path_tests/credentials.rs index 45be870..f480bc6 100644 --- a/src/web/http/base_path_tests/credentials.rs +++ b/src/web/http/base_path_tests/credentials.rs @@ -66,10 +66,10 @@ async fn every_authentic_credential_placement_stays_out_of_the_upstream() { host: "proxy.example.com".to_string(), base: "/relay/".to_string(), decoy_fasttrack_mode: WebDecoyFastTrackMode::Off, - decoy: WebRuntimeDecoy::HttpUpstream { + decoy: Arc::new(WebRuntimeDecoy::HttpUpstream { addr: site_addr, authority: "decoy.internal".to_string(), - }, + }), decoy_header_secs: 1, profiles: vec![profile.clone()], capabilities: vec![capability].into_boxed_slice(), @@ -81,10 +81,10 @@ async fn every_authentic_credential_placement_stays_out_of_the_upstream() { host: "other.example.com".to_string(), base: "/other/".to_string(), decoy_fasttrack_mode: WebDecoyFastTrackMode::Off, - decoy: WebRuntimeDecoy::HttpUpstream { + decoy: Arc::new(WebRuntimeDecoy::HttpUpstream { addr: site_addr, authority: "decoy.internal".to_string(), - }, + }), decoy_header_secs: 1, profiles: Vec::new(), capabilities: Vec::new().into_boxed_slice(), diff --git a/src/web/http/base_path_tests/routing.rs b/src/web/http/base_path_tests/routing.rs index a0253b9..d923b6f 100644 --- a/src/web/http/base_path_tests/routing.rs +++ b/src/web/http/base_path_tests/routing.rs @@ -198,10 +198,10 @@ async fn decoy_forwarding_preserves_every_reference_request_target() { host: previous.host.clone(), base: previous.base.clone(), decoy_fasttrack_mode: previous.decoy_fasttrack_mode, - decoy: WebRuntimeDecoy::HttpUpstream { + decoy: Arc::new(WebRuntimeDecoy::HttpUpstream { addr: site_addr, authority: "decoy.internal".to_string(), - }, + }), decoy_header_secs: 1, profiles: previous.profiles.clone(), capabilities: previous.capabilities.clone(), diff --git a/src/web/http/decoy.rs b/src/web/http/decoy.rs index b87ed0b..a306d2a 100644 --- a/src/web/http/decoy.rs +++ b/src/web/http/decoy.rs @@ -46,7 +46,7 @@ where .boxed_unsync() }; let request = Request::from_parts(parts, body); - match &vhost.decoy { + match vhost.decoy.as_ref() { WebRuntimeDecoy::StaticDirectory(site) => { let mut response = serve_static(request, site); if sanitize_transport { diff --git a/src/web/http/decoy_dns_tests.rs b/src/web/http/decoy_dns_tests.rs index fa1178f..7b49a36 100644 --- a/src/web/http/decoy_dns_tests.rs +++ b/src/web/http/decoy_dns_tests.rs @@ -16,10 +16,10 @@ async fn decoy_dns_pinned_socket_keeps_configured_hostname_in_actual_http_reques let snapshot = Arc::get_mut(config.web.runtime.as_mut().unwrap()).unwrap(); let vhost = Arc::get_mut(snapshot.vhosts.get_mut("proxy.example.com").unwrap()).unwrap(); // A public hostname deliberately paired with a loopback socket proves no request-time lookup. - vhost.decoy = WebRuntimeDecoy::HttpUpstream { + vhost.decoy = Arc::new(WebRuntimeDecoy::HttpUpstream { addr: origin.local_addr().unwrap(), authority: "bsi.bund.de:8080".to_string(), - }; + }); let generation = test_runtime_generation(1, config); let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(generation.clone()))); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/src/web/http/decoy_fasttrack_tests.rs b/src/web/http/decoy_fasttrack_tests.rs index 038eaa5..3a5a66f 100644 --- a/src/web/http/decoy_fasttrack_tests.rs +++ b/src/web/http/decoy_fasttrack_tests.rs @@ -67,10 +67,10 @@ async fn observe_http_origin( let mut config = runtime_config_with_fasttrack(capability, WebCarrier::Https, mode); let runtime_config = Arc::get_mut(config.web.runtime.as_mut().unwrap()).unwrap(); let vhost = Arc::get_mut(runtime_config.vhosts.get_mut("proxy.example.com").unwrap()).unwrap(); - vhost.decoy = WebRuntimeDecoy::HttpUpstream { + vhost.decoy = Arc::new(WebRuntimeDecoy::HttpUpstream { addr: origin.local_addr().unwrap(), authority: "decoy.example".to_string(), - }; + }); let generation = test_runtime_generation(1, config); let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&generation))); let runtime = WebProcessRuntime::start(active_runtime); diff --git a/src/web/http/host_list_benchmarks.rs b/src/web/http/host_list_benchmarks.rs new file mode 100644 index 0000000..47245af --- /dev/null +++ b/src/web/http/host_list_benchmarks.rs @@ -0,0 +1,71 @@ +use super::*; +use std::hint::black_box; +use std::time::Instant; + +fn memory() -> (u64, u64) { + let smaps = std::fs::read_to_string("/proc/self/smaps_rollup").unwrap(); + let value = |name: &str| { + smaps + .lines() + .find_map(|line| line.strip_prefix(name)) + .unwrap() + .split_whitespace() + .next() + .unwrap() + .parse::() + .unwrap() + }; + (value("Rss:"), value("Anonymous:")) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "manual matched before/after source-snapshot measurement"] +async fn matched_host_group_probe() { + let path = std::env::var("TELEMT_F1_PROBE_CONFIG").unwrap(); + let before = memory(); + let started = Instant::now(); + let mut config = ProxyConfig::load(std::path::Path::new(&path)).unwrap(); + let load_us = started.elapsed().as_micros(); + black_box(&config); + let after = memory(); + let snapshot = config.web.runtime.as_ref().unwrap(); + let host_count = snapshot.vhosts.len(); + let profile_count = snapshot.profiles.len(); + let profile = &snapshot.profiles[0]; + let host = profile.host.clone(); + let cap = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(profile.capability); + let base = snapshot.vhosts[&host].base.clone(); + config.web.limits.max_bootstraps_per_ip = 512; + let generation = test_runtime_generation(1, config); + generation.proxy_shared.apply_user_config( + &generation.config().access.users, + &std::collections::HashMap::new(), + ); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(generation.clone()))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let raw = format!("GET {base}?bridge={cap} HTTP/1.1\r\nHost: {host}\r\nX-Forwarded-For: 192.0.2.10\r\nConnection: close\r\n\r\n").into_bytes(); + let mut samples = Vec::new(); + for index in 0..110 { + let started = Instant::now(); + let response = request(&listener, &runtime, raw.clone()).await; + let elapsed = started.elapsed().as_micros(); + assert!(response.starts_with(b"HTTP/1.1 200")); + if index >= 10 { + samples.push(elapsed); + } + } + samples.sort_unstable(); + println!( + "F1_PROBE {}", + serde_json::json!({ + "load_us": load_us, "rss_kib": after.0, "anonymous_kib": after.1, + "rss_delta_kib": after.0 as i64 - before.0 as i64, + "anonymous_delta_kib": after.1 as i64 - before.1 as i64, + "hosts": host_count, "profiles": profile_count, + "bridge_p50_us": samples[50], "bridge_p95_us": samples[95] + }) + ); + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} diff --git a/src/web/http/host_list_tests.rs b/src/web/http/host_list_tests.rs new file mode 100644 index 0000000..6cbb14c --- /dev/null +++ b/src/web/http/host_list_tests.rs @@ -0,0 +1,350 @@ +use super::*; +use crate::maestro::generation::RuntimeGeneration; +use crate::web::manager::{BridgeDiagnosticEvent, CarrierRequest, ManagerError, TokenHash}; +use sha2::{Digest, Sha256}; +use std::collections::HashMap; + +// Admission and generation identity use the same parsed configuration as HTTP routing. +#[path = "host_list_tests/sessions.rs"] +mod sessions; + +const HOSTS: [&str; 2] = ["a.example.com", "b.example.com"]; +const SELECTOR: &str = "hosts = [\"a.example.com\", \"b.example.com\"]"; + +fn source(selector: &str, origin: std::net::SocketAddr) -> String { + format!( + r#" +[access.users] +alice = "000102030405060708090a0b0c0d0e0f" +[[server.listeners]] +ip = "127.0.0.1" +port = 18080 +transport = "web" +web_trusted_proxy_cidrs = ["127.0.0.1/32"] +[web] +enabled = true +carrier = "https" +[[web.vhosts]] +{selector} +base_path = "relay" +public_addr = "203.0.113.10:443" +[web.vhosts.decoy] +mode = "http_upstream" +upstream = "http://{origin}" +[[web.vhosts.profiles]] +user = "alice" +secret_mode = "plain" +max_sessions = 1 +max_streams = 1 +max_streams_per_session = 1 +"# + ) +} + +fn generation(id: u64, text: &str) -> Arc { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.toml"); + std::fs::write(&path, text).unwrap(); + let generation = test_runtime_generation(id, ProxyConfig::load(&path).unwrap()); + generation + .proxy_shared + .apply_user_config(&generation.config().access.users, &HashMap::new()); + generation +} + +fn profile(runtime: &WebProcessRuntime, host: &str) -> Arc { + runtime + .active_generation() + .config() + .web + .runtime + .as_ref() + .unwrap() + .vhosts[host] + .profiles[0] + .clone() +} + +fn token_hash(token: &str) -> TokenHash { + Sha256::digest( + base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(token) + .unwrap(), + ) + .into() +} + +fn raw_request(method: &str, path: &str, host: &str, headers: &str, body: &[u8]) -> Vec { + let mut raw = format!( + "{method} {path} HTTP/1.1\r\nHost: {host}\r\nX-Forwarded-For: 192.0.2.10\r\n{headers}Content-Length: {}\r\nConnection: close\r\n\r\n", body.len() + ).into_bytes(); + raw.extend_from_slice(body); + raw +} + +fn bootstrap(runtime: &Arc, host: &str) -> String { + runtime + .issue_bootstrap(profile(runtime, host), "192.0.2.10".parse().unwrap()) + .unwrap() + .token +} + +fn create( + runtime: &Arc, + host: &str, + bootstrap: &str, +) -> Result { + runtime + .create_session( + token_hash(bootstrap), + host, + "192.0.2.10".parse().unwrap(), + &frame::encode(FrameType::Hello, 0, &[1]), + CarrierRequest::legacy([0; 32]), + false, + ) + .map(|created| created.token) +} + +fn assert_private(response: &[u8]) { + let (headers, body) = split_response(response); + assert!( + headers.starts_with(b"HTTP/1.1 404"), + "{}", + String::from_utf8_lossy(response) + ); + assert_eq!(response_header(headers, "cache-control"), "no-store"); + assert_eq!(body, b"not found\n"); +} + +async fn stop(runtime: Arc, generations: &[Arc]) { + runtime.shutdown().await; + for generation in generations { + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} + +#[tokio::test] +async fn grouped_hosts_contain_credentials_in_every_fasttrack_mode() { + for mode in ["off", "shadow", "enforce"] { + let site = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let text = source(SELECTOR, site.local_addr().unwrap()).replace( + "enabled = true", + &format!("enabled = true\ndecoy_fasttrack_mode = \"{mode}\""), + ); + let generation = generation(1, &text); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(generation.clone()))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + for (owner, other) in [(HOSTS[0], HOSTS[1]), (HOSTS[1], HOSTS[0])] { + let capability = base64::engine::general_purpose::URL_SAFE_NO_PAD + .encode(profile(&runtime, owner).capability); + let response = request( + &listener, + &runtime, + raw_request( + "GET", + &format!("/relay/?bridge={capability}"), + owner, + "", + &[], + ), + ) + .await; + assert!(response.starts_with(b"HTTP/1.1 200")); + let body = std::str::from_utf8(split_response(&response).1).unwrap(); + let bootstrap = body + .split_once("bootstrap=\"") + .unwrap() + .1 + .split_once('"') + .unwrap() + .0 + .to_string(); + let hello = frame::encode(FrameType::Hello, 0, &[1]); + let auth = format!( + "Authorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\n" + ); + assert_private( + &request( + &listener, + &runtime, + raw_request("POST", "/relay/api/v1/session", other, &auth, &hello), + ) + .await, + ); + let response = request( + &listener, + &runtime, + raw_request("POST", "/relay/api/v1/session", owner, &auth, &hello), + ) + .await; + assert!(response.starts_with(b"HTTP/1.1 200")); + let session = + response_header(split_response(&response).0, "x-session-token").to_string(); + assert!(matches!( + runtime.get_session(token_hash(&session), other), + Err(ManagerError::Authentication) + )); + assert_eq!( + runtime.close_token(token_hash(&session), other, None), + Err(ManagerError::Authentication) + ); + assert!( + runtime + .bootstrap_trace_identity(token_hash(&bootstrap), other) + .is_none() + ); + assert_eq!( + runtime.claim_bridge_diagnostic( + token_hash(&bootstrap), + other, + BridgeDiagnosticEvent::RuntimeStarted + ), + Err(ManagerError::Authentication) + ); + for secret in [&capability, &bootstrap, &session] { + let encoded = secret + .bytes() + .map(|byte| format!("%{byte:02X}")) + .collect::(); + for path in [ + format!("/relay/?bridge={secret}"), + format!("/wrong/{secret}"), + format!("/wrong/{encoded}"), + format!("/wrong?opaque={secret}"), + ] { + assert_private( + &request( + &listener, + &runtime, + raw_request("GET", &path, other, "", &[]), + ) + .await, + ); + } + let head = request( + &listener, + &runtime, + raw_request( + "HEAD", + "/relay/", + other, + &format!("X-Opaque: {secret}\r\n"), + &[], + ), + ) + .await; + assert!(head.starts_with(b"HTTP/1.1 404")); + assert_eq!( + response_header(split_response(&head).0, "cache-control"), + "no-store" + ); + } + for (method, path, headers, body) in [ + ( + "POST", + "/relay/api/v1/up", + "X-Up-Seq: 0\r\n", + hello.as_ref(), + ), + ( + "POST", + "/relay/api/v1/down", + "X-Down-Cursor: 0\r\n", + &[][..], + ), + ("DELETE", "/relay/api/v1/session", "", &[][..]), + ( + "POST", + "/relay/api/v1/diagnostic", + "Content-Type: application/json\r\n", + b"{}".as_ref(), + ), + ] { + let auth = format!("Authorization: Bearer {session}\r\n{headers}"); + assert_private( + &request( + &listener, + &runtime, + raw_request(method, path, other, &auth, body), + ) + .await, + ); + } + } + for host in ["unknown.example.com", "A.EXAMPLE.COM", "a.example.com:444"] { + let response = + request(&listener, &runtime, raw_request("GET", "/", host, "", &[])).await; + assert!(response.starts_with(b"HTTP/1.1 404")); + } + assert!( + tokio::time::timeout(std::time::Duration::from_millis(30), site.accept()) + .await + .is_err() + ); + // A positive control ensures the origin would observe an actual fallback request. + let site_task = tokio::spawn(async move { + let (mut socket, _) = site.accept().await.unwrap(); + let mut head = [0; 2048]; + let read = socket.read(&mut head).await.unwrap(); + assert!( + std::str::from_utf8(&head[..read]) + .unwrap() + .contains("/ordinary") + ); + socket + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\nsite") + .await + .unwrap(); + }); + let ordinary = request( + &listener, + &runtime, + raw_request("GET", "/ordinary", HOSTS[0], "", &[]), + ) + .await; + assert_eq!(split_response(&ordinary).1, b"site"); + site_task.await.unwrap(); + stop(runtime, &[generation]).await; + } +} + +#[tokio::test] +async fn alias_recovery_cannot_retire_a_foreign_session() { + let generation = generation(1, &source(SELECTOR, "127.0.0.1:18081".parse().unwrap())); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(generation.clone()))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let a = create(&runtime, HOSTS[0], &bootstrap(&runtime, HOSTS[0])).unwrap(); + let cap = base64::engine::general_purpose::URL_SAFE_NO_PAD + .encode(profile(&runtime, HOSTS[1]).capability); + let response = request( + &listener, + &runtime, + raw_request( + "GET", + &format!("/relay/?bridge={cap}"), + HOSTS[1], + &format!( + "Accept: application/vnd.telemt.web-recovery+json\r\nAuthorization: Bearer {a}\r\n" + ), + &[], + ), + ) + .await; + assert!( + response.starts_with(b"HTTP/1.1 200"), + "{}", + String::from_utf8_lossy(&response) + ); + let value: serde_json::Value = serde_json::from_slice(split_response(&response).1).unwrap(); + let recovered = value["bootstrap"].as_str().unwrap(); + assert_eq!( + create(&runtime, HOSTS[0], recovered), + Err(ManagerError::Authentication) + ); + let b = create(&runtime, HOSTS[1], recovered).unwrap(); + assert!(runtime.get_session(token_hash(&a), HOSTS[0]).is_ok()); + assert!(runtime.get_session(token_hash(&b), HOSTS[1]).is_ok()); + stop(runtime, &[generation]).await; +} diff --git a/src/web/http/host_list_tests/sessions.rs b/src/web/http/host_list_tests/sessions.rs new file mode 100644 index 0000000..336dac1 --- /dev/null +++ b/src/web/http/host_list_tests/sessions.rs @@ -0,0 +1,211 @@ +use super::*; +use tokio::sync::Barrier; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn per_host_quotas_survive_concurrent_creation_and_generation_swaps() { + let source = source(SELECTOR, "127.0.0.1:18081".parse().unwrap()); + let initial = generation(1, &source); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(initial.clone()))); + let mut generations = vec![initial]; + for id in 2..=9 { + let selector = if id % 2 == 0 { + "hosts = [\"b.example.com\", \"a.example.com\"]" + } else { + SELECTOR + }; + generations.push(generation(id, &source.replace(SELECTOR, selector))); + } + let barrier = Arc::new(Barrier::new(34)); + let mut tasks = Vec::new(); + for host in HOSTS { + for _ in 0..16 { + let token = bootstrap(&runtime, host); + let runtime = runtime.clone(); + let barrier = barrier.clone(); + tasks.push(tokio::spawn(async move { + barrier.wait().await; + (host, create(&runtime, host, &token)) + })); + } + } + let reloader = { + let runtime = runtime.clone(); + let barrier = barrier.clone(); + let generations = generations[1..].to_vec(); + tokio::spawn(async move { + barrier.wait().await; + for generation in generations { + runtime.activate_generation(generation); + tokio::task::yield_now().await; + } + }) + }; + barrier.wait().await; + let mut accepted = Vec::new(); + for task in tasks { + let (host, result) = task.await.unwrap(); + match result { + Ok(token) => accepted.push((host, token)), + Err(error) => assert_eq!(error, ManagerError::Limit), + } + } + reloader.await.unwrap(); + assert_eq!(accepted.len(), 2); + for host in HOSTS { + assert_eq!( + accepted.iter().filter(|(owner, _)| *owner == host).count(), + 1 + ); + } + let sessions: Vec<_> = accepted + .iter() + .map(|(host, token)| runtime.get_session(token_hash(token), host).unwrap()) + .collect(); + assert_ne!(sessions[0].profile_key(), sessions[1].profile_key()); + assert_eq!( + serde_json::to_value(runtime.try_status()).unwrap()["manager"]["profiles"], + 2 + ); + + let barrier = Arc::new(Barrier::new(33)); + let mut tasks = Vec::new(); + let ip = "192.0.2.10".parse().unwrap(); + let addr = "203.0.113.10:443".parse().unwrap(); + for session in &sessions { + let key = session.profile_key(); + for _ in 0..16 { + let runtime = runtime.clone(); + let barrier = barrier.clone(); + tasks.push(tokio::spawn(async move { + barrier.wait().await; + (key, runtime.try_acquire_stream(key, 1, ip, addr)) + })); + } + } + barrier.wait().await; + let mut ports = Vec::new(); + for task in tasks { + let (key, result) = task.await.unwrap(); + match result { + Ok(port) => ports.push((key, port)), + Err(error) => assert_eq!(error, ManagerError::Limit), + } + } + assert_eq!(ports.len(), 2); + assert_ne!(ports[0].0, ports[1].0); + assert_eq!( + serde_json::to_value(runtime.try_status()).unwrap()["streams"]["profiles"], + 2 + ); + for (key, port) in ports { + runtime.release_stream(key, ip, addr, port); + } + assert_eq!( + serde_json::to_value(runtime.try_status()).unwrap()["streams"]["profiles"], + 0 + ); + stop(runtime, &generations).await; +} + +#[tokio::test] +async fn regrouping_preserves_live_quota_and_bootstrap_when_limits_change() { + let grouped = source(SELECTOR, "127.0.0.1:18081".parse().unwrap()); + let (prefix, group) = grouped.split_once("[[web.vhosts]]").unwrap(); + let separate = format!( + "{prefix}[[web.vhosts]]{}\n[[web.vhosts]]{}", + group.replace(SELECTOR, "host = \"a.example.com\""), + group.replace(SELECTOR, "host = \"b.example.com\"") + ); + let initial = generation(1, &separate); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(initial.clone()))); + let a = create(&runtime, HOSTS[0], &bootstrap(&runtime, HOSTS[0])).unwrap(); + let key = runtime + .get_session(token_hash(&a), HOSTS[0]) + .unwrap() + .profile_key(); + let unused_b = bootstrap(&runtime, HOSTS[1]); + let next = generation( + 2, + &grouped + .replace(SELECTOR, "hosts = [\"b.example.com\", \"a.example.com\"]") + .replace("max_sessions = 1", "max_sessions = 2"), + ); + runtime.activate_generation(next.clone()); + let b = create(&runtime, HOSTS[1], &unused_b).unwrap(); + assert_eq!( + runtime + .get_session(token_hash(&a), HOSTS[0]) + .unwrap() + .profile_key(), + key + ); + let second_a = create(&runtime, HOSTS[0], &bootstrap(&runtime, HOSTS[0])).unwrap(); + assert_eq!( + runtime + .get_session(token_hash(&second_a), HOSTS[0]) + .unwrap() + .profile_key(), + key + ); + assert_eq!( + create(&runtime, HOSTS[0], &bootstrap(&runtime, HOSTS[0])), + Err(ManagerError::Limit) + ); + + let stale_b = bootstrap(&runtime, HOSTS[1]); + let removed = generation( + 3, + &grouped + .replace(SELECTOR, "hosts = [\"a.example.com\"]") + .replace("max_sessions = 1", "max_sessions = 2"), + ); + runtime.activate_generation(removed.clone()); + assert_eq!( + create(&runtime, HOSTS[1], &stale_b), + Err(ManagerError::Authentication) + ); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let response = request( + &listener, + &runtime, + raw_request("GET", "/relay/", HOSTS[1], "", &[]), + ) + .await; + assert!(response.starts_with(b"HTTP/1.1 404")); + assert!(runtime.get_session(token_hash(&a), HOSTS[0]).is_ok()); + assert!(runtime.get_session(token_hash(&b), HOSTS[1]).is_ok()); + assert_eq!( + create(&runtime, HOSTS[0], &bootstrap(&runtime, HOSTS[0])), + Err(ManagerError::Limit) + ); + stop(runtime, &[initial, next, removed]).await; +} + +#[tokio::test] +async fn grouped_profiles_keep_process_and_ip_session_limits() { + for global in [1, 2] { + let text = format!( + "{}\n[web.limits]\nmax_sessions_global = {global}\nmax_sessions_per_ip = 1\n", + source(SELECTOR, "127.0.0.1:18081".parse().unwrap()) + ); + let generation = generation(1, &text); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(generation.clone()))); + create(&runtime, HOSTS[0], &bootstrap(&runtime, HOSTS[0])).unwrap(); + let b = bootstrap(&runtime, HOSTS[1]); + assert_eq!(create(&runtime, HOSTS[1], &b), Err(ManagerError::Limit)); + let result = runtime.create_session( + token_hash(&b), + HOSTS[1], + "192.0.2.11".parse().unwrap(), + &frame::encode(FrameType::Hello, 0, &[1]), + CarrierRequest::legacy([0; 32]), + false, + ); + if global == 1 { + assert!(matches!(result, Err(ManagerError::Limit))); + } else { + assert!(result.is_ok()); + } + stop(runtime, &[generation]).await; + } +} diff --git a/src/web/http/recovery.rs b/src/web/http/recovery.rs index d0cbe4f..f140ca4 100644 --- a/src/web/http/recovery.rs +++ b/src/web/http/recovery.rs @@ -83,7 +83,7 @@ pub(super) fn response( carrier_batch_bytes: limits.carrier_batch_bytes, pending_bytes_per_session: limits.pending_bytes_per_session, pending_items_per_session: limits.pending_items_per_session, - max_streams_per_session: profile.max_streams_per_session, + max_streams_per_session: profile.shared.max_streams_per_session, }, timeouts: RecoveryTimeouts { long_poll_secs: timeouts.long_poll_secs, @@ -94,9 +94,9 @@ pub(super) fn response( reconnect_grace_secs: timeouts.reconnect_grace_secs, }, negotiation: RecoveryNegotiation { - enabled: profile.carrier_negotiation_enabled, - candidate_count: profile.carriers.len(), - deadlines_secs: profile.carrier_negotiation_deadlines_secs, + enabled: profile.shared.carrier_negotiation_enabled, + candidate_count: profile.shared.carriers.len(), + deadlines_secs: profile.shared.carrier_negotiation_deadlines_secs, carrier_probe_coalesce_ms: timeouts.carrier_probe_coalesce_ms, }, }; diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs index 76abb0a..1a0d6a6 100644 --- a/src/web/http/tests.rs +++ b/src/web/http/tests.rs @@ -127,20 +127,22 @@ fn runtime_config_with_carriers_and_deadlines( ) -> ProxyConfig { let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: "203.0.113.10:443".parse().unwrap(), - user: "alice".to_string(), - secret_mode: WebSecretMode::Plain, - carrier, - carrier_negotiation_enabled, - carrier_learning, - carriers: Arc::clone(&carriers), - carrier_negotiation_deadlines_secs, capability, - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 4, - max_streams: 16, - max_streams_per_session: 4, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: "203.0.113.10:443".parse().unwrap(), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier, + carrier_negotiation_enabled, + carrier_learning, + carriers: Arc::clone(&carriers), + carrier_negotiation_deadlines_secs, + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 4, + max_streams: 16, + max_streams_per_session: 4, + }), }); let mut assets = BTreeMap::new(); assets.insert( @@ -159,7 +161,7 @@ fn runtime_config_with_carriers_and_deadlines( host: "proxy.example.com".to_string(), base: base.to_string(), decoy_fasttrack_mode: WebDecoyFastTrackMode::Off, - decoy: WebRuntimeDecoy::StaticDirectory(Arc::clone(&site)), + decoy: Arc::new(WebRuntimeDecoy::StaticDirectory(Arc::clone(&site))), decoy_header_secs: 1, profiles: vec![Arc::clone(&profile)], capabilities: vec![capability].into_boxed_slice(), @@ -172,7 +174,7 @@ fn runtime_config_with_carriers_and_deadlines( host: "other.example.com".to_string(), base: "/".to_string(), decoy_fasttrack_mode: WebDecoyFastTrackMode::Off, - decoy: WebRuntimeDecoy::StaticDirectory(site), + decoy: Arc::new(WebRuntimeDecoy::StaticDirectory(site)), decoy_header_secs: 1, profiles: Vec::new(), capabilities: Vec::new().into_boxed_slice(), @@ -547,3 +549,11 @@ async fn https_lanes_is_advertised_and_requires_canonical_lane_headers() { generation.stop_sessions().await; generation.stop_background_tasks().await; } + +// Exercise loaded host groups through real HTTP and process-owned admission. +#[path = "host_list_tests.rs"] +mod host_list_tests; + +// Manual matched measurements keep timing assertions out of deterministic regressions. +#[path = "host_list_benchmarks.rs"] +mod host_list_benchmarks; diff --git a/src/web/manager/carrier_outcome.rs b/src/web/manager/carrier_outcome.rs index 7c78ec3..cf50cbe 100644 --- a/src/web/manager/carrier_outcome.rs +++ b/src/web/manager/carrier_outcome.rs @@ -57,7 +57,7 @@ impl WebProcessRuntime { carrier: session.carrier(), attempt: entry.carrier_attempt, candidate_count: u8::try_from(entry.carrier_candidates.len()).unwrap_or(4), - deadline_secs: entry.profile.carrier_negotiation_deadlines_secs[3], + deadline_secs: entry.profile.shared.carrier_negotiation_deadlines_secs[3], state: if entry.carrier_phase == CarrierChainPhase::Provisional && session.is_carrier_committed() { diff --git a/src/web/manager/credentials.rs b/src/web/manager/credentials.rs index db6bbf0..ba0668a 100644 --- a/src/web/manager/credentials.rs +++ b/src/web/manager/credentials.rs @@ -136,7 +136,7 @@ impl WebProcessRuntime { }; let Some(mut user_publication) = generation .proxy_shared - .claim_authenticated_user(&profile.user, profile.credential_id) + .claim_authenticated_user(&profile.shared.user, profile.shared.credential_id) else { self.telemetry .record_rejection(WebRejectionReason::UserDisabled); diff --git a/src/web/manager/session_admission.rs b/src/web/manager/session_admission.rs index 4efc8f5..555ae98 100644 --- a/src/web/manager/session_admission.rs +++ b/src/web/manager/session_admission.rs @@ -23,7 +23,7 @@ pub(super) fn admit_initial( .get(&profile_key) .copied() .unwrap_or(0) - >= profile.max_sessions + >= profile.shared.max_sessions { runtime.record_limit_hit(); runtime diff --git a/src/web/manager/session_creation.rs b/src/web/manager/session_creation.rs index 9b29373..336c310 100644 --- a/src/web/manager/session_creation.rs +++ b/src/web/manager/session_creation.rs @@ -132,7 +132,7 @@ impl WebProcessRuntime { candidate_count: automatic .then(|| u8::try_from(entry.carrier_candidates.len()).unwrap_or(4)), deadline_secs: automatic - .then_some(entry.profile.carrier_negotiation_deadlines_secs[3]), + .then_some(entry.profile.shared.carrier_negotiation_deadlines_secs[3]), carrier_state: automatic.then_some(carrier_state), }; let identity = session.trace_identity(); @@ -180,7 +180,7 @@ impl WebProcessRuntime { if entry.carrier_started_at.is_some_and(|started| { now.saturating_duration_since(started) >= Duration::from_secs( - entry.profile.carrier_negotiation_deadlines_secs[deadline_index], + entry.profile.shared.carrier_negotiation_deadlines_secs[deadline_index], ) }) { return Err(ManagerError::Protocol); @@ -225,14 +225,18 @@ impl WebProcessRuntime { .runtime .as_ref() .and_then(|runtime| matching_profile(runtime, &issued_profile)) - .filter(|profile| generation.proxy_shared.is_user_enabled(&profile.user)) + .filter(|profile| { + generation + .proxy_shared + .is_user_enabled(&profile.shared.user) + }) .ok_or(ManagerError::Authentication)?; let profile_key = profile_key(&profile); - if carrier_request.is_automatic() && !profile.carrier_negotiation_enabled { + if carrier_request.is_automatic() && !profile.shared.carrier_negotiation_enabled { return Err(ManagerError::Protocol); } let capability_selection = - carrier_request.uses_capabilities() && profile.carrier_negotiation_enabled; + carrier_request.uses_capabilities() && profile.shared.carrier_negotiation_enabled; let learning_policy = ( config.web.carrier_negotiation_enabled() && config.web.carrier_learning, config.web.carrier_negotiation_aggressiveness, @@ -240,7 +244,7 @@ impl WebProcessRuntime { Duration::from_secs(config.web.timeouts.carrier_health_secs), ); let (candidates, scores, learning_epoch, learning_disposition) = - if capability_selection && profile.carrier_learning && learning_policy.0 { + if capability_selection && profile.shared.carrier_learning && learning_policy.0 { let learning = self.learning.lock(); match learning.epoch_for_policy( generation.id, @@ -252,7 +256,7 @@ impl WebProcessRuntime { super::learning::CarrierLearningEpoch::Ready(epoch) => { let (candidates, scores) = learning.rank( now, - &profile.carriers, + &profile.shared.carriers, carrier_request, profile_key, client_ip, @@ -266,13 +270,13 @@ impl WebProcessRuntime { (candidates, scores, Some(epoch), disposition) } super::learning::CarrierLearningEpoch::Pending => ( - supported_candidates(&profile.carriers, carrier_request), + supported_candidates(&profile.shared.carriers, carrier_request), [0; 4], None, WebCarrierSelectionDisposition::PolicyPending, ), super::learning::CarrierLearningEpoch::Exhausted => ( - supported_candidates(&profile.carriers, carrier_request), + supported_candidates(&profile.shared.carriers, carrier_request), [0; 4], None, WebCarrierSelectionDisposition::EpochExhausted, @@ -280,7 +284,7 @@ impl WebProcessRuntime { } } else if capability_selection { ( - supported_candidates(&profile.carriers, carrier_request), + supported_candidates(&profile.shared.carriers, carrier_request), [0; 4], None, if learning_policy.0 { @@ -290,12 +294,12 @@ impl WebProcessRuntime { }, ) } else if carrier_request.uses_capabilities() - && !carrier_request.supports(profile.carrier) + && !carrier_request.supports(profile.shared.carrier) { return Err(ManagerError::Protocol); } else { ( - vec![profile.carrier], + vec![profile.shared.carrier], [0; 4], None, WebCarrierSelectionDisposition::ProfileDisabled, @@ -307,7 +311,7 @@ impl WebProcessRuntime { let _operator_admission = self.try_operator_admission()?; let Some(mut user_publication) = generation .proxy_shared - .claim_authenticated_user(&profile.user, profile.credential_id) + .claim_authenticated_user(&profile.shared.user, profile.shared.credential_id) else { return Err(ManagerError::Closed); }; @@ -328,9 +332,9 @@ impl WebProcessRuntime { .record_rejection(crate::web::telemetry::WebRejectionReason::SessionCapacity); return Err(ManagerError::Limit); }; - let carrier_deadline_at = carrier_request - .is_automatic() - .then_some(now + Duration::from_secs(profile.carrier_negotiation_deadlines_secs[3])); + let carrier_deadline_at = carrier_request.is_automatic().then_some( + now + Duration::from_secs(profile.shared.carrier_negotiation_deadlines_secs[3]), + ); let learning_context = learning_epoch.map(|epoch| CarrierLearningContext { profile_key, client_ip, @@ -412,7 +416,7 @@ impl WebProcessRuntime { candidate_count: carrier_request.is_automatic().then_some(candidate_count), deadline_secs: carrier_request .is_automatic() - .then_some(profile.carrier_negotiation_deadlines_secs[3]), + .then_some(profile.shared.carrier_negotiation_deadlines_secs[3]), carrier_state: carrier_request .is_automatic() .then_some(CarrierChainPhase::Provisional.as_str()), diff --git a/src/web/manager/session_creation/replacement.rs b/src/web/manager/session_creation/replacement.rs index fac719c..96a5575 100644 --- a/src/web/manager/session_creation/replacement.rs +++ b/src/web/manager/session_creation/replacement.rs @@ -45,10 +45,10 @@ impl WebProcessRuntime { self.cancel_replacement(bootstrap_hash, &replacement.old_session); return Err(ManagerError::Closed); } - let Some(mut user_publication) = generation - .proxy_shared - .claim_authenticated_user(&replacement.profile.user, replacement.profile.credential_id) - else { + let Some(mut user_publication) = generation.proxy_shared.claim_authenticated_user( + &replacement.profile.shared.user, + replacement.profile.shared.credential_id, + ) else { drop(state); self.cancel_replacement(bootstrap_hash, &replacement.old_session); return Err(ManagerError::Closed); @@ -75,7 +75,7 @@ impl WebProcessRuntime { self.cancel_replacement(bootstrap_hash, &replacement.old_session); return Err(ManagerError::Limit); }; - let learning_context = (replacement.profile.carrier_learning + let learning_context = (replacement.profile.shared.carrier_learning && replacement.learning_epoch != 0) .then_some(CarrierLearningContext { profile_key: replacement.profile_key, @@ -155,7 +155,7 @@ impl WebProcessRuntime { carrier: replacement.carrier, attempt: Some(replacement.attempt), candidate_count: Some(u8::try_from(entry.carrier_candidates.len()).unwrap_or(4)), - deadline_secs: Some(entry.profile.carrier_negotiation_deadlines_secs[3]), + deadline_secs: Some(entry.profile.shared.carrier_negotiation_deadlines_secs[3]), carrier_state: Some(CarrierChainPhase::Provisional.as_str()), }; let predecessor_session_id = entry.predecessor_session_id; diff --git a/src/web/manager/state.rs b/src/web/manager/state.rs index 509de4b..baaf757 100644 --- a/src/web/manager/state.rs +++ b/src/web/manager/state.rs @@ -275,9 +275,9 @@ pub(super) fn profile_key(profile: &WebRuntimeProfile) -> ProfileKey { digest.update(WEB_PROFILE_OWNER_CONTEXT); digest.update((profile.host.len() as u64).to_be_bytes()); digest.update(profile.host.as_bytes()); - digest.update((profile.user.len() as u64).to_be_bytes()); - digest.update(profile.user.as_bytes()); - digest.update([match profile.secret_mode { + digest.update((profile.shared.user.len() as u64).to_be_bytes()); + digest.update(profile.shared.user.as_bytes()); + digest.update([match profile.shared.secret_mode { crate::config::WebSecretMode::Plain => 0, crate::config::WebSecretMode::Dd => 1, }]); @@ -294,18 +294,19 @@ pub(super) fn matching_profile( .iter() .find(|profile| { profile.host == expected.host - && profile.public_addr == expected.public_addr - && profile.user == expected.user - && profile.credential_id == expected.credential_id - && profile.secret_mode == expected.secret_mode - && profile.carrier == expected.carrier - && profile.carrier_negotiation_enabled == expected.carrier_negotiation_enabled - && profile.carrier_learning == expected.carrier_learning - && profile.carriers == expected.carriers - && profile.carrier_negotiation_deadlines_secs - == expected.carrier_negotiation_deadlines_secs + && profile.shared.public_addr == expected.shared.public_addr + && profile.shared.user == expected.shared.user + && profile.shared.credential_id == expected.shared.credential_id + && profile.shared.secret_mode == expected.shared.secret_mode + && profile.shared.carrier == expected.shared.carrier + && profile.shared.carrier_negotiation_enabled + == expected.shared.carrier_negotiation_enabled + && profile.shared.carrier_learning == expected.shared.carrier_learning + && profile.shared.carriers == expected.shared.carriers + && profile.shared.carrier_negotiation_deadlines_secs + == expected.shared.carrier_negotiation_deadlines_secs && profile.capability == expected.capability - && profile.key_fingerprint == expected.key_fingerprint + && profile.shared.key_fingerprint == expected.shared.key_fingerprint }) .cloned() } diff --git a/src/web/manager/state/expiry/fixtures.rs b/src/web/manager/state/expiry/fixtures.rs index dbcd9e5..257e4f9 100644 --- a/src/web/manager/state/expiry/fixtures.rs +++ b/src/web/manager/state/expiry/fixtures.rs @@ -19,20 +19,22 @@ impl Fixture { authority, profile: Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: "203.0.113.10:443".parse().unwrap(), - user: "alice".to_string(), - secret_mode: WebSecretMode::Plain, - carrier: WebCarrier::Https, - carrier_negotiation_enabled: false, - carrier_learning: true, - carriers: Arc::from([WebCarrier::Https]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [7; 32], - credential_id: credential_id_from_hex(secret).unwrap(), - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 1024, - max_streams: 1024, - max_streams_per_session: 1, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: "203.0.113.10:443".parse().unwrap(), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier: WebCarrier::Https, + carrier_negotiation_enabled: false, + carrier_learning: true, + carriers: Arc::from([WebCarrier::Https]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: credential_id_from_hex(secret).unwrap(), + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 1024, + max_streams: 1024, + max_streams_per_session: 1, + }), }), } } diff --git a/src/web/session/backend.rs b/src/web/session/backend.rs index 9042f43..d22cc7d 100644 --- a/src/web/session/backend.rs +++ b/src/web/session/backend.rs @@ -168,7 +168,7 @@ impl WebSession { manager.release_stream( self.profile_key, self.client_ip, - self.profile.public_addr, + self.profile.shared.public_addr, peer_port, ); } @@ -311,8 +311,8 @@ async fn run_stream( peer, &deps.config, &replay_checker, - &session.profile.user, - session.profile.secret_mode, + &session.profile.shared.user, + session.profile.shared.secret_mode, &deps.shared, ) .await, @@ -373,7 +373,7 @@ async fn run_stream( writer, success, deps, - session.profile.public_addr, + session.profile.shared.public_addr, peer, ConntrackClosePolicy::Suppress, ) diff --git a/src/web/session/backend_tests.rs b/src/web/session/backend_tests.rs index 238aace..3dcc855 100644 --- a/src/web/session/backend_tests.rs +++ b/src/web/session/backend_tests.rs @@ -66,20 +66,22 @@ fn test_runtime_with_dc( ) -> TestRuntime { let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: "203.0.113.10:443".parse().unwrap(), - user: "default".to_string(), - secret_mode: WebSecretMode::Plain, - carrier, - carrier_negotiation_enabled: false, - carrier_learning: true, - carriers: Arc::from([carrier]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [7; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 4, - max_streams: 16, - max_streams_per_session: 4, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: "203.0.113.10:443".parse().unwrap(), + user: "default".to_string(), + secret_mode: WebSecretMode::Plain, + carrier, + carrier_negotiation_enabled: false, + carrier_learning: true, + carriers: Arc::from([carrier]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 4, + max_streams: 16, + max_streams_per_session: 4, + }), }); let mut config = ProxyConfig::default(); config.web.enabled = true; @@ -415,16 +417,16 @@ async fn cancellation_while_waiting_for_data_releases_stream_ownership() { .manager .try_acquire_stream( runtime.session.profile_key, - runtime.session.profile.max_streams, + runtime.session.profile.shared.max_streams, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, ) .unwrap(); assert_eq!(peer_port, 1); runtime.manager.release_stream( runtime.session.profile_key, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, peer_port, ); runtime.shutdown().await; diff --git a/src/web/session/conveyor.rs b/src/web/session/conveyor.rs index 451deee..9080fec 100644 --- a/src/web/session/conveyor.rs +++ b/src/web/session/conveyor.rs @@ -202,6 +202,7 @@ impl WebSession { if existing.is_none() { let limit = self .profile + .shared .max_streams_per_session .saturating_add(self.limits.max_tombstones_per_session) .saturating_add(1); diff --git a/src/web/session/conveyor/tests.rs b/src/web/session/conveyor/tests.rs index f279b57..4aaba31 100644 --- a/src/web/session/conveyor/tests.rs +++ b/src/web/session/conveyor/tests.rs @@ -21,20 +21,22 @@ fn fixture( WebProcessRuntime::start(Arc::new(ArcSwap::from(test_runtime_generation(1, config)))); let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".into(), - public_addr: "203.0.113.10:443".parse().unwrap(), - user: "default".into(), - secret_mode: WebSecretMode::Plain, - carrier, - carrier_negotiation_enabled: automatic, - carrier_learning: false, - carriers: Arc::from([carrier]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [0; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".into(), - max_sessions: 2, - max_streams: 2, - max_streams_per_session: 2, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: "203.0.113.10:443".parse().unwrap(), + user: "default".into(), + secret_mode: WebSecretMode::Plain, + carrier, + carrier_negotiation_enabled: automatic, + carrier_learning: false, + carriers: Arc::from([carrier]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".into(), + max_sessions: 2, + max_streams: 2, + max_streams_per_session: 2, + }), }); let session = WebSession::new( Arc::downgrade(&manager), diff --git a/src/web/session/downlink_tests.rs b/src/web/session/downlink_tests.rs index fa04c22..d1ab801 100644 --- a/src/web/session/downlink_tests.rs +++ b/src/web/session/downlink_tests.rs @@ -32,20 +32,22 @@ fn session() -> (Arc, Arc) { let manager = WebProcessRuntime::start(Arc::new(ArcSwap::from(generation))); let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), - user: "alice".to_string(), - secret_mode: WebSecretMode::Plain, - carrier: WebCarrier::Https, - carrier_negotiation_enabled: false, - carrier_learning: false, - carriers: Arc::from([WebCarrier::Https]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [0; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 1, - max_streams: 1, - max_streams_per_session: 1, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier: WebCarrier::Https, + carrier_negotiation_enabled: false, + carrier_learning: false, + carriers: Arc::from([WebCarrier::Https]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 1, + max_streams: 1, + max_streams_per_session: 1, + }), }); let timeouts = WebTimeoutsConfig { long_poll_secs: 1, diff --git a/src/web/session/lane_uplink.rs b/src/web/session/lane_uplink.rs index c3b1d52..42f3d16 100644 --- a/src/web/session/lane_uplink.rs +++ b/src/web/session/lane_uplink.rs @@ -101,6 +101,7 @@ impl WebSession { } let lane_limit = self .profile + .shared .max_streams_per_session .saturating_add(self.limits.max_tombstones_per_session) .saturating_add(1); diff --git a/src/web/session/lanes/tests.rs b/src/web/session/lanes/tests.rs index 06d568b..645724d 100644 --- a/src/web/session/lanes/tests.rs +++ b/src/web/session/lanes/tests.rs @@ -47,20 +47,22 @@ fn new_session_with_automatic( ) -> Arc { let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), - user: "alice".to_string(), - secret_mode: WebSecretMode::Plain, - carrier: WebCarrier::HttpsLanes, - carrier_negotiation_enabled: false, - carrier_learning: true, - carriers: Arc::from([WebCarrier::HttpsLanes]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [0; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 1, - max_streams: 2, - max_streams_per_session: 2, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier: WebCarrier::HttpsLanes, + carrier_negotiation_enabled: false, + carrier_learning: true, + carriers: Arc::from([WebCarrier::HttpsLanes]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 1, + max_streams: 2, + max_streams_per_session: 2, + }), }); WebSession::new( manager, diff --git a/src/web/session/negotiation/tests.rs b/src/web/session/negotiation/tests.rs index 2dae145..7033b26 100644 --- a/src/web/session/negotiation/tests.rs +++ b/src/web/session/negotiation/tests.rs @@ -11,20 +11,22 @@ use crate::web::session::{SessionCloseOutcome, SessionCloseReason}; fn session(carrier: WebCarrier, deadline: Instant) -> Arc { let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), - user: "alice".to_string(), - secret_mode: WebSecretMode::Plain, - carrier, - carrier_negotiation_enabled: true, - carrier_learning: false, - carriers: Arc::from([carrier]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [0; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 1, - max_streams: 1, - max_streams_per_session: 1, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier, + carrier_negotiation_enabled: true, + carrier_learning: false, + carriers: Arc::from([carrier]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 1, + max_streams: 1, + max_streams_per_session: 1, + }), }); WebSession::new( std::sync::Weak::::new(), diff --git a/src/web/session/status.rs b/src/web/session/status.rs index 325c9a3..5537acc 100644 --- a/src/web/session/status.rs +++ b/src/web/session/status.rs @@ -90,8 +90,8 @@ impl WebSession { trace_session_id: self.trace_session_id, client_ip: self.client_ip, host: self.profile.host.clone(), - user: self.profile.user.clone(), - key_id: self.profile.key_fingerprint.clone(), + user: self.profile.shared.user.clone(), + key_id: self.profile.shared.key_fingerprint.clone(), carrier: self.selected_carrier, attempt: self.carrier_attempt, client_class: self.carrier_class.as_str(), @@ -144,12 +144,12 @@ impl WebSession { /// Returns the configured non-secret user label. pub(crate) fn profile_user(&self) -> &str { - &self.profile.user + &self.profile.shared.user } /// Returns the configured non-secret key fingerprint. pub(crate) fn key_id(&self) -> &str { - &self.profile.key_fingerprint + &self.profile.shared.key_fingerprint } } diff --git a/src/web/session/stream_io/tests.rs b/src/web/session/stream_io/tests.rs index 0e231e7..a745ab8 100644 --- a/src/web/session/stream_io/tests.rs +++ b/src/web/session/stream_io/tests.rs @@ -134,20 +134,22 @@ fn probe_waker(probe: Arc) -> Waker { fn session() -> Arc { let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), - user: "alice".to_string(), - secret_mode: WebSecretMode::Plain, - carrier: WebCarrier::Https, - carrier_negotiation_enabled: false, - carrier_learning: false, - carriers: Arc::from([WebCarrier::Https]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [0; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 1, - max_streams: 1, - max_streams_per_session: 1, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier: WebCarrier::Https, + carrier_negotiation_enabled: false, + carrier_learning: false, + carriers: Arc::from([WebCarrier::Https]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 1, + max_streams: 1, + max_streams_per_session: 1, + }), }); WebSession::new( Weak::::new(), diff --git a/src/web/session/uplink.rs b/src/web/session/uplink.rs index 8757a4f..75821f7 100644 --- a/src/web/session/uplink.rs +++ b/src/web/session/uplink.rs @@ -342,7 +342,7 @@ impl WebSession { effects: &mut DeferredSessionEffects, ) -> Option { let manager = self.manager.upgrade()?; - if state.active_peer_ports.len() >= self.profile.max_streams_per_session { + if state.active_peer_ports.len() >= self.profile.shared.max_streams_per_session { manager.record_stream_rejected_reason( crate::web::telemetry::WebRejectionReason::StreamSessionCapacity, ); @@ -350,9 +350,9 @@ impl WebSession { } let (peer_port, notify) = manager.try_acquire_stream_quiet( self.profile_key, - self.profile.max_streams, + self.profile.shared.max_streams, self.client_ip, - self.profile.public_addr, + self.profile.shared.public_addr, ); if let Some(notify) = notify { effects.notify(notify); @@ -364,7 +364,7 @@ impl WebSession { if let Some(notify) = manager.release_stream_quiet( self.profile_key, self.client_ip, - self.profile.public_addr, + self.profile.shared.public_addr, peer_port, ) { effects.notify(notify); diff --git a/src/web/session/uplink_tests.rs b/src/web/session/uplink_tests.rs index 4f29894..cc813bf 100644 --- a/src/web/session/uplink_tests.rs +++ b/src/web/session/uplink_tests.rs @@ -16,20 +16,22 @@ fn session() -> Arc { fn session_with_automatic(automatic: bool) -> Arc { let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), - user: "alice".to_string(), - secret_mode: WebSecretMode::Plain, - carrier: WebCarrier::Https, - carrier_negotiation_enabled: false, - carrier_learning: true, - carriers: Arc::from([WebCarrier::Https]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [0; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 1, - max_streams: 1, - max_streams_per_session: 1, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier: WebCarrier::Https, + carrier_negotiation_enabled: false, + carrier_learning: true, + carriers: Arc::from([WebCarrier::Https]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 1, + max_streams: 1, + max_streams_per_session: 1, + }), }); WebSession::new( std::sync::Weak::::new(), diff --git a/src/web/session/websocket.rs b/src/web/session/websocket.rs index cd36ab3..4ddbef4 100644 --- a/src/web/session/websocket.rs +++ b/src/web/session/websocket.rs @@ -76,7 +76,7 @@ impl WebSession { if state.closed || self.cancel.is_cancelled() { return Err(ManagerError::Closed); } - if state.active_peer_ports.len() >= self.profile.max_streams_per_session + if state.active_peer_ports.len() >= self.profile.shared.max_streams_per_session || state.streams.contains_key(&lane_id) || state.closing_streams.contains_key(&lane_id) || state.closed_streams.contains(&lane_id) @@ -89,9 +89,9 @@ impl WebSession { }; let (peer_port, notify) = manager.try_acquire_stream_quiet( self.profile_key, - self.profile.max_streams, + self.profile.shared.max_streams, self.client_ip, - self.profile.public_addr, + self.profile.shared.public_addr, ); if let Some(notify) = notify { effects.notify(notify); @@ -108,7 +108,7 @@ impl WebSession { if let Some(notify) = manager.release_stream_quiet( self.profile_key, self.client_ip, - self.profile.public_addr, + self.profile.shared.public_addr, peer_port, ) { effects.notify(notify); @@ -122,7 +122,7 @@ impl WebSession { if let Some(notify) = manager.release_stream_quiet( self.profile_key, self.client_ip, - self.profile.public_addr, + self.profile.shared.public_addr, peer_port, ) { effects.notify(notify); @@ -149,7 +149,7 @@ impl WebSession { if let Some(notify) = manager.release_stream_quiet( self.profile_key, self.client_ip, - self.profile.public_addr, + self.profile.shared.public_addr, peer_port, ) { effects.notify(notify); @@ -395,7 +395,7 @@ impl WebSession { manager.release_stream( self.profile_key, self.client_ip, - self.profile.public_addr, + self.profile.shared.public_addr, claim.peer_port, ); } diff --git a/src/web/session/websocket/tests.rs b/src/web/session/websocket/tests.rs index f914249..74386a2 100644 --- a/src/web/session/websocket/tests.rs +++ b/src/web/session/websocket/tests.rs @@ -48,20 +48,22 @@ impl TestRuntime { fn runtime(admission: bool) -> TestRuntime { let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), - public_addr: "203.0.113.10:443".parse().unwrap(), - user: "default".to_string(), - secret_mode: WebSecretMode::Plain, - carrier: WebCarrier::WebsocketLanes, - carrier_negotiation_enabled: false, - carrier_learning: true, - carriers: Arc::from([WebCarrier::WebsocketLanes]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], capability: [7; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 2, - max_streams: 1, - max_streams_per_session: 1, + shared: Arc::new(crate::config::WebRuntimeProfileShared { + public_addr: "203.0.113.10:443".parse().unwrap(), + user: "default".to_string(), + secret_mode: WebSecretMode::Plain, + carrier: WebCarrier::WebsocketLanes, + carrier_negotiation_enabled: false, + carrier_learning: true, + carriers: Arc::from([WebCarrier::WebsocketLanes]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 2, + max_streams: 1, + max_streams_per_session: 1, + }), }); let mut config = ProxyConfig::default(); config.web.enabled = true; @@ -121,7 +123,7 @@ fn detach_stale_lane(runtime: &TestRuntime, reservation: &WebSocketLaneReservati runtime.manager.release_stream( runtime.session.profile_key, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, claim.peer_port, ); } @@ -171,9 +173,9 @@ async fn rejected_open_retains_stream_quota_until_lane_socket_teardown() { .manager .try_acquire_stream( runtime.session.profile_key, - runtime.session.profile.max_streams, + runtime.session.profile.shared.max_streams, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, ) .is_err() ); @@ -183,15 +185,15 @@ async fn rejected_open_retains_stream_quota_until_lane_socket_teardown() { .manager .try_acquire_stream( runtime.session.profile_key, - runtime.session.profile.max_streams, + runtime.session.profile.shared.max_streams, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, ) .unwrap(); runtime.manager.release_stream( runtime.session.profile_key, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, peer_port, ); runtime.shutdown().await; @@ -213,15 +215,15 @@ async fn closed_session_releases_bound_lane_quota_on_reservation_drop() { .manager .try_acquire_stream( runtime.session.profile_key, - runtime.session.profile.max_streams, + runtime.session.profile.shared.max_streams, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, ) .unwrap(); runtime.manager.release_stream( runtime.session.profile_key, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, peer_port, ); runtime.shutdown().await; @@ -254,15 +256,15 @@ async fn closed_session_releases_transferred_rejected_lane_quota() { .manager .try_acquire_stream( runtime.session.profile_key, - runtime.session.profile.max_streams, + runtime.session.profile.shared.max_streams, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, ) .unwrap(); runtime.manager.release_stream( runtime.session.profile_key, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, peer_port, ); runtime.shutdown().await; @@ -295,9 +297,9 @@ async fn closed_session_keeps_stream_owned_quota_until_task_completion() { .manager .try_acquire_stream( runtime.session.profile_key, - runtime.session.profile.max_streams, + runtime.session.profile.shared.max_streams, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, ) .is_err() ); @@ -306,15 +308,15 @@ async fn closed_session_keeps_stream_owned_quota_until_task_completion() { .manager .try_acquire_stream( runtime.session.profile_key, - runtime.session.profile.max_streams, + runtime.session.profile.shared.max_streams, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, ) .unwrap(); runtime.manager.release_stream( runtime.session.profile_key, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, peer_port, ); runtime.shutdown().await; @@ -475,7 +477,7 @@ async fn exact_lane_teardown_wakes_stream_after_releasing_session_lock() { runtime.manager.release_stream( runtime.session.profile_key, runtime.session.client_ip, - runtime.session.profile.public_addr, + runtime.session.profile.shared.public_addr, claim.peer_port, ); drop(reservation); diff --git a/src/web/trace/exchange.rs b/src/web/trace/exchange.rs index bc1a576..0da8a5b 100644 --- a/src/web/trace/exchange.rs +++ b/src/web/trace/exchange.rs @@ -150,17 +150,18 @@ impl HttpTraceExchange { /// Binds non-secret profile and process session identity. pub(crate) fn bind_profile(&self, profile: &WebRuntimeProfile, session_id: u64) { let dynamic = profile + .shared .user .len() - .saturating_add(profile.key_fingerprint.len()); + .saturating_add(profile.shared.key_fingerprint.len()); let mut state = self.state.lock(); if state.phase != ExchangePhase::Open { return; } state.identity.session_id = Some(session_id); if self.reserve_locked(&mut state, dynamic) { - state.identity.user = Some(profile.user.clone()); - state.identity.key_fingerprint = Some(profile.key_fingerprint.clone()); + state.identity.user = Some(profile.shared.user.clone()); + state.identity.key_fingerprint = Some(profile.shared.key_fingerprint.clone()); } } diff --git a/src/web/trace/types.rs b/src/web/trace/types.rs index e320fca..9ed5ef0 100644 --- a/src/web/trace/types.rs +++ b/src/web/trace/types.rs @@ -155,8 +155,8 @@ impl TraceIdentity { ) -> Self { Self { session_id, - user: Some(profile.user.clone()), - key_fingerprint: Some(profile.key_fingerprint.clone()), + user: Some(profile.shared.user.clone()), + key_fingerprint: Some(profile.shared.key_fingerprint.clone()), } } }