From 840b6f563a3b529a6047afb33c8fcbc702d1b3ab Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sun, 23 Aug 2026 10:56:00 +0300 Subject: [PATCH] Rustfmt --- src/config/load/runtime_web.rs | 25 +++------ src/config/load/validate_web.rs | 41 ++++++++++----- .../tests/load_basic_tests/web_tests.rs | 5 +- src/config/types/web.rs | 5 +- src/maestro/generation.rs | 2 +- src/maestro/listeners/accept.rs | 7 +-- src/maestro/listeners/control.rs | 14 +++-- src/maestro/runtime_build.rs | 3 +- src/proxy/authenticated.rs | 4 +- src/proxy/client.rs | 6 +-- src/proxy/handshake.rs | 2 +- src/proxy/handshake/auth_candidates.rs | 7 +-- src/proxy/handshake/mtproto.rs | 14 +++-- src/web/bridge.rs | 26 ++++++---- src/web/frame.rs | 13 ++--- src/web/http.rs | 43 +++++----------- src/web/http/activity.rs | 5 +- src/web/http/body.rs | 12 ++--- src/web/http/decoy.rs | 41 +++++++-------- src/web/http/request.rs | 32 ++++-------- src/web/http/tests.rs | 51 ++++++++++--------- src/web/manager.rs | 42 ++++++--------- src/web/manager/admission.rs | 22 ++------ src/web/manager/lifecycle.rs | 4 +- src/web/manager/state.rs | 16 ++---- src/web/session.rs | 12 +++-- src/web/session/backend.rs | 6 +-- src/web/session/downlink.rs | 40 ++++----------- src/web/session/lanes.rs | 33 ++++++------ src/web/session/lanes/tests.rs | 15 +++--- src/web/session/uplink.rs | 30 ++++------- 31 files changed, 232 insertions(+), 346 deletions(-) diff --git a/src/config/load/runtime_web.rs b/src/config/load/runtime_web.rs index a85175c..55f2167 100644 --- a/src/config/load/runtime_web.rs +++ b/src/config/load/runtime_web.rs @@ -47,10 +47,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(), - )?; + let capability = + derive_web_capability(&client_secret[..client_secret_len], vhost.host.as_bytes())?; if !capabilities.insert(capability) { return Err(ProxyError::Config(format!( "WEB vhost `{}` contains profiles with the same client capability", @@ -97,9 +95,8 @@ pub(super) fn rebuild(config: &mut ProxyConfig) -> Result<()> { /// Derives the Telegram Desktop WEB capability for one exact secret and host. pub(crate) fn derive_web_capability(secret: &[u8], host: &[u8]) -> Result<[u8; 32]> { - let mut mac = Hmac::::new_from_slice(secret).map_err(|_| { - ProxyError::Config("WEB capability secret must not be empty".to_string()) - })?; + let mut mac = Hmac::::new_from_slice(secret) + .map_err(|_| ProxyError::Config("WEB capability secret must not be empty".to_string()))?; mac.update(WEB_CAPABILITY_CONTEXT); mac.update(host); Ok(mac.finalize().into_bytes().into()) @@ -159,13 +156,7 @@ fn build_decoy( }) } WebDecoyConfig::StaticDirectory { directory, index } => { - let site = load_static_site( - directory, - index, - limits, - static_files, - static_bytes, - )?; + let site = load_static_site(directory, index, limits, static_files, static_bytes)?; Ok(WebRuntimeDecoy::StaticDirectory(Arc::new(site))) } } @@ -239,8 +230,7 @@ fn load_static_directory( })?; if *total_files >= limits.max_static_files { return Err(ProxyError::Config( - "WEB static entries exceed process-wide web.limits.max_static_files" - .to_string(), + "WEB static entries exceed process-wide web.limits.max_static_files".to_string(), )); } *total_files += 1; @@ -317,8 +307,7 @@ fn load_static_directory( })?; if *total_bytes > limits.max_static_bytes { return Err(ProxyError::Config( - "WEB static snapshots exceed process-wide web.limits.max_static_bytes" - .to_string(), + "WEB static snapshots exceed process-wide web.limits.max_static_bytes".to_string(), )); } let relative = path.strip_prefix(root).map_err(|_| { diff --git a/src/config/load/validate_web.rs b/src/config/load/validate_web.rs index ba202f6..d31840a 100644 --- a/src/config/load/validate_web.rs +++ b/src/config/load/validate_web.rs @@ -59,9 +59,7 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> { } validate_limits(&config.web.limits)?; - if config.web.carrier == WebCarrier::HttpsLanes - && config.web.limits.max_http_handlers < 2 - { + if config.web.carrier == WebCarrier::HttpsLanes && config.web.limits.max_http_handlers < 2 { return config_error("web.carrier=https-lanes requires web.limits.max_http_handlers >= 2"); } validate_timeouts(&config.web.timeouts)?; @@ -155,11 +153,20 @@ fn validate_limits(limits: &WebLimitsConfig) -> Result<()> { ("max_streams_per_session", limits.max_streams_per_session), ("max_streams_global", limits.max_streams_global), ("max_stream_handshakes", limits.max_stream_handshakes), - ("pending_bytes_per_session", limits.pending_bytes_per_session), + ( + "pending_bytes_per_session", + limits.pending_bytes_per_session, + ), ("pending_bytes_global", limits.pending_bytes_global), - ("pending_items_per_session", limits.pending_items_per_session), + ( + "pending_items_per_session", + limits.pending_items_per_session, + ), ("pending_items_global", limits.pending_items_global), - ("control_bytes_per_session", limits.control_bytes_per_session), + ( + "control_bytes_per_session", + limits.control_bytes_per_session, + ), ("control_bytes_global", limits.control_bytes_global), ("max_bootstraps_global", limits.max_bootstraps_global), ("max_bootstraps_per_ip", limits.max_bootstraps_per_ip), @@ -181,11 +188,16 @@ fn validate_limits(limits: &WebLimitsConfig) -> Result<()> { ("max_stream_handshakes", limits.max_stream_handshakes), ] { if value > tokio::sync::Semaphore::MAX_PERMITS { - return config_error(&format!("web.limits.{field} exceeds Tokio semaphore capacity")); + return config_error(&format!( + "web.limits.{field} exceeds Tokio semaphore capacity" + )); } } let rates = [ - ("new_bootstraps_per_minute", limits.new_bootstraps_per_minute), + ( + "new_bootstraps_per_minute", + limits.new_bootstraps_per_minute, + ), ("new_bootstraps_burst", limits.new_bootstraps_burst), ("new_sessions_per_minute", limits.new_sessions_per_minute), ("new_sessions_burst", limits.new_sessions_burst), @@ -241,7 +253,9 @@ fn validate_limits(limits: &WebLimitsConfig) -> Result<()> { let required_control_bytes_global = control_items_global .checked_mul(control_frame_cost) .ok_or_else(|| { - ProxyError::Config("web.limits global control byte reservation overflowed usize".to_string()) + ProxyError::Config( + "web.limits global control byte reservation overflowed usize".to_string(), + ) })?; if control_items_per_session >= limits.pending_items_per_session || control_items_global >= limits.pending_items_global @@ -367,10 +381,7 @@ fn validate_vhosts(config: &mut ProxyConfig) -> Result<()> { let mut hosts = HashSet::with_capacity(config.web.vhosts.len()); let mut profile_count = 0usize; for (vhost_idx, vhost) in config.web.vhosts.iter_mut().enumerate() { - vhost.host = normalize_web_host( - &vhost.host, - &format!("web.vhosts[{vhost_idx}].host"), - )?; + vhost.host = normalize_web_host(&vhost.host, &format!("web.vhosts[{vhost_idx}].host"))?; if !hosts.insert(vhost.host.clone()) { return config_error(&format!("duplicate WEB vhost host `{}`", vhost.host)); } @@ -404,7 +415,9 @@ fn validate_vhosts(config: &mut ProxyConfig) -> Result<()> { .max_streams_per_session .unwrap_or(limits.max_streams_per_session); if profile.max_sessions == Some(0) - || profile.max_sessions.is_some_and(|value| value > limits.max_sessions_global) + || profile + .max_sessions + .is_some_and(|value| value > limits.max_sessions_global) || profile.max_streams == Some(0) || profile .max_streams diff --git a/src/config/tests/load_basic_tests/web_tests.rs b/src/config/tests/load_basic_tests/web_tests.rs index e803cda..02f38aa 100644 --- a/src/config/tests/load_basic_tests/web_tests.rs +++ b/src/config/tests/load_basic_tests/web_tests.rs @@ -94,10 +94,7 @@ fn web_semaphore_limits_are_rejected_before_runtime_construction() { #[test] fn web_ipv6_decoy_uses_a_valid_http_authority() { - let ipv6 = WEB_CONFIG.replace( - "http://127.0.0.1:18081", - "http://[::1]:18081", - ); + let ipv6 = WEB_CONFIG.replace("http://127.0.0.1:18081", "http://[::1]:18081"); 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(); diff --git a/src/config/types/web.rs b/src/config/types/web.rs index d8396a5..829fdfb 100644 --- a/src/config/types/web.rs +++ b/src/config/types/web.rs @@ -359,10 +359,7 @@ pub(crate) struct WebRuntimeProfile { /// Runtime-ready ordinary-site fallback. #[derive(Debug)] pub(crate) enum WebRuntimeDecoy { - HttpUpstream { - addr: SocketAddr, - authority: String, - }, + HttpUpstream { addr: SocketAddr, authority: String }, StaticDirectory(Arc), } diff --git a/src/maestro/generation.rs b/src/maestro/generation.rs index 6af04e7..de29b7e 100644 --- a/src/maestro/generation.rs +++ b/src/maestro/generation.rs @@ -10,11 +10,11 @@ use tokio_util::task::TaskTracker; use crate::config::ProxyConfig; use crate::crypto::SecureRandom; use crate::ip_tracker::UserIpTracker; +use crate::proxy::authenticated::ClientRuntimeDeps; #[cfg(test)] use crate::proxy::route_mode::RelayRouteMode; use crate::proxy::route_mode::RouteRuntimeController; use crate::proxy::shared_state::ProxySharedState; -use crate::proxy::authenticated::ClientRuntimeDeps; use crate::stats::beobachten::BeobachtenStore; use crate::stats::{ReplayChecker, Stats}; use crate::stream::BufferPool; diff --git a/src/maestro/listeners/accept.rs b/src/maestro/listeners/accept.rs index bebb6dd..3ceae5d 100644 --- a/src/maestro/listeners/accept.rs +++ b/src/maestro/listeners/accept.rs @@ -301,12 +301,7 @@ impl ListenerSlot { ); tokio::time::timeout(connection_stop_timeout, self.connections.wait()) .await - .map_err(|_| { - format!( - "listener {} connection shutdown timed out", - self.spec.addr - ) - })?; + .map_err(|_| format!("listener {} connection shutdown timed out", self.spec.addr))?; Ok(()) } diff --git a/src/maestro/listeners/control.rs b/src/maestro/listeners/control.rs index 3bce116..c1b3470 100644 --- a/src/maestro/listeners/control.rs +++ b/src/maestro/listeners/control.rs @@ -4,16 +4,16 @@ use std::sync::Arc; use arc_swap::ArcSwap; -use crate::config::ProxyConfig; use crate::config::ListenerTransport; +use crate::config::ProxyConfig; use crate::maestro::generation::RuntimeGeneration; use super::accept::ListenerSlot; use super::bind::{BoundListeners, BoundTcpListener, PreparedTcpListener, prepare_listener}; use super::plan::{ListenerBindSpec, listener_bind_plan}; -use crate::web::manager::WebProcessRuntime; #[cfg(unix)] use super::unix::UnixAcceptHandle; +use crate::web::manager::WebProcessRuntime; /// Process-owned listener inventory and accept-task lifecycle controller. pub(crate) struct ListenerManager { @@ -54,11 +54,7 @@ impl ListenerManager { let addr = listener.spec.addr; slots.insert( addr, - ListenerSlot::start( - listener, - active_runtime.clone(), - web_runtime.clone(), - ), + ListenerSlot::start(listener, active_runtime.clone(), web_runtime.clone()), ); } #[cfg(unix)] @@ -103,7 +99,9 @@ impl ListenerManager { .map(|(addr, spec)| (*addr, spec.clone())) .collect::>(); if web_inventory_changed { - return Err("WEB listener inventory is process-owned; process restart required".to_string()); + return Err( + "WEB listener inventory is process-owned; process restart required".to_string(), + ); } let current_addresses: BTreeSet<_> = self.slots.keys().copied().collect(); let target_addresses: BTreeSet<_> = target_specs.keys().copied().collect(); diff --git a/src/maestro/runtime_build.rs b/src/maestro/runtime_build.rs index ad01d2c..204e146 100644 --- a/src/maestro/runtime_build.rs +++ b/src/maestro/runtime_build.rs @@ -406,8 +406,7 @@ pub(crate) fn resolve_reload_config( fields.push("logging".to_string()); effective.logging = old.logging.clone(); } - if serde_json::to_value(&old.web.limits).ok() - != serde_json::to_value(&desired.web.limits).ok() + if serde_json::to_value(&old.web.limits).ok() != serde_json::to_value(&desired.web.limits).ok() { fields.push("web.limits".to_string()); effective.web.limits = old.web.limits.clone(); diff --git a/src/proxy/authenticated.rs b/src/proxy/authenticated.rs index c79a5a6..60315fb 100644 --- a/src/proxy/authenticated.rs +++ b/src/proxy/authenticated.rs @@ -11,9 +11,7 @@ use crate::error::{ProxyError, Result}; use crate::ip_tracker::UserIpTracker; use crate::proxy::direct_relay::handle_via_direct_with_shared_and_conntrack; use crate::proxy::handshake::HandshakeSuccess; -use crate::proxy::middle_relay::{ - handle_via_middle_proxy, handle_via_middle_proxy_with_conntrack, -}; +use crate::proxy::middle_relay::{handle_via_middle_proxy, handle_via_middle_proxy_with_conntrack}; use crate::proxy::route_mode::{RelayRouteMode, RouteRuntimeController}; use crate::proxy::shared_state::{ConntrackClosePolicy, ProxySharedState}; use crate::stats::Stats; diff --git a/src/proxy/client.rs b/src/proxy/client.rs index 99c0b72..c5108ab 100644 --- a/src/proxy/client.rs +++ b/src/proxy/client.rs @@ -43,9 +43,7 @@ use crate::transport::{UpstreamManager, configure_client_socket, parse_proxy_pro use crate::proxy::authenticated::{ClientRuntimeDeps, run_authenticated}; #[cfg(test)] -use crate::proxy::authenticated::{ - UserConnectionReservation, acquire_user_connection_reservation, -}; +use crate::proxy::authenticated::{UserConnectionReservation, acquire_user_connection_reservation}; use crate::proxy::handshake::{ HandshakeSuccess, TlsResponseWriteOptions, handle_mtproto_handshake_with_shared, handle_tls_handshake_with_shared, handle_tls_handshake_with_shared_and_options, @@ -53,9 +51,9 @@ use crate::proxy::handshake::{ #[cfg(test)] use crate::proxy::handshake::{handle_mtproto_handshake, handle_tls_handshake}; use crate::proxy::masking::handle_bad_client_with_shared; -use crate::proxy::route_mode::RouteRuntimeController; #[cfg(test)] use crate::proxy::route_mode::RelayRouteMode; +use crate::proxy::route_mode::RouteRuntimeController; use crate::proxy::shared_state::{ConntrackClosePolicy, ProxySharedState}; fn beobachten_ttl(config: &ProxyConfig) -> Duration { diff --git a/src/proxy/handshake.rs b/src/proxy/handshake.rs index 12a54be..96cefd3 100644 --- a/src/proxy/handshake.rs +++ b/src/proxy/handshake.rs @@ -57,8 +57,8 @@ use self::tls_auth::{parse_tls_auth_material, validate_tls_secret_candidate}; pub(crate) use self::auth_probe::{AuthProbeSaturationState, AuthProbeState}; #[cfg(test)] pub use self::mtproto::handle_mtproto_handshake; -pub use self::mtproto::handle_mtproto_handshake_with_shared; pub(crate) use self::mtproto::handle_mtproto_handshake_for_web_user; +pub use self::mtproto::handle_mtproto_handshake_with_shared; #[allow(unused_imports)] pub use self::nonce::{encrypt_tg_nonce, encrypt_tg_nonce_with_ciphers, generate_tg_nonce}; pub use self::session::HandshakeSuccess; diff --git a/src/proxy/handshake/auth_candidates.rs b/src/proxy/handshake/auth_candidates.rs index 0f6261b..48954ee 100644 --- a/src/proxy/handshake/auth_candidates.rs +++ b/src/proxy/handshake/auth_candidates.rs @@ -274,12 +274,7 @@ pub(super) fn mode_enabled_for_proto( proto_tag: ProtoTag, is_tls: bool, ) -> bool { - mode_enabled_for_proto_with_policy( - config, - proto_tag, - is_tls, - MtprotoModePolicy::Configured, - ) + mode_enabled_for_proto_with_policy(config, proto_tag, is_tls, MtprotoModePolicy::Configured) } fn mode_enabled_for_proto_with_policy( diff --git a/src/proxy/handshake/mtproto.rs b/src/proxy/handshake/mtproto.rs index 70fea18..da9b351 100644 --- a/src/proxy/handshake/mtproto.rs +++ b/src/proxy/handshake/mtproto.rs @@ -203,15 +203,23 @@ where } let mut matched = exact_user_id.is_some_and(|user_id| try_user_id!(user_id)); - if exact_user.is_none() && let Some(user_id) = sticky_ip_hint { + if exact_user.is_none() + && let Some(user_id) = sticky_ip_hint + { matched = try_user_id!(user_id); } - if exact_user.is_none() && !matched && let Some(user_id) = preferred_user_id { + if exact_user.is_none() + && !matched + && let Some(user_id) = preferred_user_id + { matched = try_user_id!(user_id); } - if exact_user.is_none() && !matched && let Some(user_id) = sticky_prefix_hint { + if exact_user.is_none() + && !matched + && let Some(user_id) = sticky_prefix_hint + { matched = try_user_id!(user_id); } diff --git a/src/web/bridge.rs b/src/web/bridge.rs index a57bfe1..4418325 100644 --- a/src/web/bridge.rs +++ b/src/web/bridge.rs @@ -313,16 +313,22 @@ mod tests { assert!(page.body.contains("X-Lane-ID")); assert!(page.body.contains("const when=Date.parse(header)")); assert!(page.body.contains("},{once:false});")); - assert!(page.body.contains("if(!sessionToken)throw new Error('missing session token')")); + assert!( + page.body + .contains("if(!sessionToken)throw new Error('missing session token')") + ); assert!(!page.body.contains("welcomeBytes")); - assert!(page - .body - .contains("for(const value of splitFrames(data))if(value.id!==lane.id)")); - assert!(page - .body - .contains("let frames;try{frames=splitFrames(value)}catch(error){fail();return}")); - assert!(page - .content_security_policy - .contains("frame-ancestors http://127.0.0.1:*")); + assert!( + page.body + .contains("for(const value of splitFrames(data))if(value.id!==lane.id)") + ); + assert!( + page.body + .contains("let frames;try{frames=splitFrames(value)}catch(error){fail();return}") + ); + assert!( + page.content_security_policy + .contains("frame-ancestors http://127.0.0.1:*") + ); } } diff --git a/src/web/frame.rs b/src/web/frame.rs index f02f757..7a0743b 100644 --- a/src/web/frame.rs +++ b/src/web/frame.rs @@ -98,15 +98,10 @@ pub(crate) fn parse_all<'a>( return Err(FrameError::Incomplete); } let frame_type = FrameType::parse(remaining[0]).ok_or(FrameError::UnknownType)?; - let stream_id = u32::from(remaining[1]) << 16 - | u32::from(remaining[2]) << 8 - | u32::from(remaining[3]); - let payload_len = u32::from_be_bytes([ - remaining[4], - remaining[5], - remaining[6], - remaining[7], - ]) as usize; + let stream_id = + u32::from(remaining[1]) << 16 | u32::from(remaining[2]) << 8 | u32::from(remaining[3]); + let payload_len = + u32::from_be_bytes([remaining[4], remaining[5], remaining[6], remaining[7]]) as usize; if payload_len > limits.max_frame_payload_bytes { return Err(FrameError::PayloadLimit); } diff --git a/src/web/http.rs b/src/web/http.rs index 3af4865..47efad9 100644 --- a/src/web/http.rs +++ b/src/web/http.rs @@ -34,9 +34,9 @@ mod request; #[cfg(test)] mod tests; -use decoy::serve_decoy; use activity::{ActivityBody, RequestActivity}; use body::{CollectBodyError, CollectedBody, collect_body}; +use decoy::serve_decoy; use request::{ bearer_token_hash, binary_content_type, bridge_candidate, canonical_request_host, canonical_u64_header, client_ip, match_profile, @@ -84,10 +84,7 @@ pub(crate) async fn serve_connection( } else { service_unavailable() }; - let response = response.map(|body| { - ActivityBody::new(body, activity) - .boxed_unsync() - }); + let response = response.map(|body| ActivityBody::new(body, activity).boxed_unsync()); Ok::<_, Infallible>(response) } }); @@ -174,12 +171,7 @@ async fn handle_root( let Some(profile) = profile.filter(|_| canonical && request.method() == Method::GET) else { return serve_decoy(request, vhost, false, &runtime).await; }; - let Some(client_ip) = client_ip( - &request, - peer, - client_ip_source, - trusted_proxy_cidrs, - ) else { + let Some(client_ip) = client_ip(&request, peer, client_ip_source, trusted_proxy_cidrs) else { strip_query(&mut request); return serve_decoy(request, vhost, true, &runtime).await; }; @@ -242,21 +234,14 @@ async fn handle_api( if request.uri().query().is_some() || request.headers().contains_key(header::COOKIE) { return serve_decoy(request, vhost, true, &runtime).await; } - let Some(client_ip) = client_ip( - &request, - peer, - client_ip_source, - trusted_proxy_cidrs, - ) else { + let Some(client_ip) = client_ip(&request, peer, client_ip_source, trusted_proxy_cidrs) else { return serve_decoy(request, vhost, true, &runtime).await; }; let Some(token_hash) = bearer_token_hash(&request) else { return serve_decoy(request, vhost, true, &runtime).await; }; match request.uri().path() { - "/api/v1/session" => { - handle_session(request, runtime, vhost, token_hash, client_ip).await - } + "/api/v1/session" => handle_session(request, runtime, vhost, token_hash, client_ip).await, "/api/v1/up" => handle_up(request, runtime, vhost, token_hash).await, "/api/v1/down" => handle_down(request, runtime, vhost, token_hash).await, _ => serve_decoy(request, vhost, true, &runtime).await, @@ -356,7 +341,12 @@ async fn handle_up( let Some(lane_id) = carrier_lane(&request, session.carrier()) else { return serve_decoy(request, vhost, true, &runtime).await; }; - let limit = runtime.active_generation().config().web.limits.max_body_bytes; + let limit = runtime + .active_generation() + .config() + .web + .limits + .max_body_bytes; let CollectedBody { request, body, @@ -510,10 +500,7 @@ fn bad_gateway() -> HttpResponse { } fn generic_not_found() -> HttpResponse { - full_response( - StatusCode::NOT_FOUND, - Bytes::from_static(b"not found\n"), - ) + full_response(StatusCode::NOT_FOUND, Bytes::from_static(b"not found\n")) } fn full_response(status: StatusCode, body: Bytes) -> HttpResponse { @@ -523,11 +510,7 @@ fn full_response(status: StatusCode, body: Bytes) -> HttpResponse { .boxed_unsync(); let mut response = Response::new(body); *response.status_mut() = status; - insert_header( - &mut response, - header::CONTENT_LENGTH, - &length.to_string(), - ); + insert_header(&mut response, header::CONTENT_LENGTH, &length.to_string()); response } diff --git a/src/web/http/activity.rs b/src/web/http/activity.rs index f6afa28..8d75486 100644 --- a/src/web/http/activity.rs +++ b/src/web/http/activity.rs @@ -37,10 +37,7 @@ pub(super) struct ActivityBody { impl ActivityBody { /// Binds one response body to its request activity guard. pub(super) fn new(inner: HttpBody, activity: RequestActivity) -> Self { - Self { - inner, - activity, - } + Self { inner, activity } } } diff --git a/src/web/http/body.rs b/src/web/http/body.rs index f0a72fe..c37e662 100644 --- a/src/web/http/body.rs +++ b/src/web/http/body.rs @@ -2,8 +2,8 @@ use std::time::Duration; use bytes::Bytes; use http_body_util::{BodyExt, Empty, Limited}; -use hyper::body::{Body as _, Incoming}; use hyper::Request; +use hyper::body::{Body as _, Incoming}; use crate::web::manager::WebProcessRuntime; @@ -50,14 +50,8 @@ pub(super) async fn collect_body( let Some((reader_budget, body_budget)) = runtime.try_body_budget(limit) else { return Err(CollectBodyError::Limit); }; - let body_timeout = Duration::from_secs( - runtime - .active_generation() - .config() - .web - .timeouts - .body_secs, - ); + let body_timeout = + Duration::from_secs(runtime.active_generation().config().web.timeouts.body_secs); let body = match tokio::time::timeout(body_timeout, Limited::new(body, limit).collect()).await { Ok(Ok(body)) => body.to_bytes(), _ => { diff --git a/src/web/http/decoy.rs b/src/web/http/decoy.rs index b565c2d..3d194ea 100644 --- a/src/web/http/decoy.rs +++ b/src/web/http/decoy.rs @@ -11,8 +11,7 @@ use hyper_util::rt::TokioIo; use tokio::net::TcpStream; use super::{ - BoxError, HttpBody, HttpResponse, bad_gateway, full_response, generic_not_found, - insert_header, + BoxError, HttpBody, HttpResponse, bad_gateway, full_response, generic_not_found, insert_header, }; use crate::config::{WebRuntimeDecoy, WebRuntimeVhost}; use crate::web::manager::WebProcessRuntime; @@ -128,14 +127,13 @@ fn static_entry( header::X_CONTENT_TYPE_OPTIONS, HeaderValue::from_static("nosniff"), ); - response.headers_mut().insert(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY")); + response + .headers_mut() + .insert(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY")); response } -fn resolve_static_path<'a>( - path: &str, - site: &'a crate::config::WebStaticSite, -) -> Option<&'a str> { +fn resolve_static_path<'a>(path: &str, site: &'a crate::config::WebStaticSite) -> Option<&'a str> { if !path.starts_with('/') || path.contains('\\') || path.contains("//") @@ -151,7 +149,10 @@ fn resolve_static_path<'a>( path }; if site.assets.contains_key(route) { - return site.assets.get_key_value(route).map(|(key, _)| key.as_str()); + return site + .assets + .get_key_value(route) + .map(|(key, _)| key.as_str()); } if route == "/favicon.ico" && site.assets.contains_key("/favicon.svg") { return Some("/favicon.svg"); @@ -198,23 +199,19 @@ async fn proxy_to_upstream( .max_header_bytes; let mut builder = hyper::client::conn::http1::Builder::new(); builder.max_buf_size(max_header_bytes); - let (mut sender, connection) = match tokio::time::timeout( - header_timeout, - builder.handshake(TokioIo::new(stream)), - ) - .await - { - Ok(Ok(parts)) => parts, - _ => return bad_gateway(), - }; + let (mut sender, connection) = + match tokio::time::timeout(header_timeout, builder.handshake(TokioIo::new(stream))).await { + Ok(Ok(parts)) => parts, + _ => return bad_gateway(), + }; runtime.spawn_auxiliary(async move { let _ = connection.await; }); - let mut response = match tokio::time::timeout(header_timeout, sender.send_request(request)).await - { - Ok(Ok(response)) => response, - _ => return bad_gateway(), - }; + let mut response = + match tokio::time::timeout(header_timeout, sender.send_request(request)).await { + Ok(Ok(response)) => response, + _ => return bad_gateway(), + }; remove_hop_by_hop(response.headers_mut()); response.map(|body| { body.map_err(|error| -> BoxError { Box::new(error) }) diff --git a/src/web/http/request.rs b/src/web/http/request.rs index 963aad3..669f21c 100644 --- a/src/web/http/request.rs +++ b/src/web/http/request.rs @@ -2,15 +2,13 @@ use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use base64::Engine as _; -use hyper::header; use hyper::Request; +use hyper::header; use ipnetwork::IpNetwork; use sha2::{Digest, Sha256}; use subtle::ConstantTimeEq; -use crate::config::{ - WebClientIpSource, WebRuntimeProfile, WebRuntimeVhost, -}; +use crate::config::{WebClientIpSource, WebRuntimeProfile, WebRuntimeVhost}; use crate::web::manager::TokenHash; /// Parses one lowercase canonical Host value restricted to the public HTTPS port. @@ -26,8 +24,7 @@ pub(super) fn canonical_request_host(request: &Request) -> Option<&str> { return None; } let host = value.strip_suffix(":443").unwrap_or(value); - if authority.host() != host || host.bytes().any(|byte| byte.is_ascii_uppercase()) - { + if authority.host() != host || host.bytes().any(|byte| byte.is_ascii_uppercase()) { return None; } Some(host) @@ -74,14 +71,14 @@ pub(super) fn bridge_candidate(query: Option<&str>) -> ([u8; 32], bool) { return (candidate, false); } let mut decoded = [0u8; 32]; - let Ok(decoded_len) = base64::engine::general_purpose::URL_SAFE_NO_PAD - .decode_slice(value, &mut decoded) + let Ok(decoded_len) = + base64::engine::general_purpose::URL_SAFE_NO_PAD.decode_slice(value, &mut decoded) else { return (candidate, false); }; let mut canonical = [0u8; 43]; - let Ok(encoded_len) = base64::engine::general_purpose::URL_SAFE_NO_PAD - .encode_slice(decoded, &mut canonical) + let Ok(encoded_len) = + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode_slice(decoded, &mut canonical) else { return (candidate, false); }; @@ -114,8 +111,7 @@ pub(super) fn bearer_token_hash(request: &Request) -> Option { let values = request.headers().get_all(header::AUTHORIZATION); let mut values = values.iter(); let value = values.next()?.to_str().ok()?; - if values.next().is_some() || !value.starts_with("Bearer ") || value.matches(' ').count() != 1 - { + if values.next().is_some() || !value.starts_with("Bearer ") || value.matches(' ').count() != 1 { return None; } let token = value.strip_prefix("Bearer ")?; @@ -133,7 +129,7 @@ pub(super) fn bearer_token_hash(request: &Request) -> Option { (decoded_len == decoded.len() && encoded_len == canonical.len() && bool::from(canonical.ct_eq(token.as_bytes()))) - .then(|| Sha256::digest(decoded).into()) + .then(|| Sha256::digest(decoded).into()) } /// Checks the exact carrier media type without accepting duplicate headers. @@ -146,10 +142,7 @@ pub(super) fn binary_content_type(request: &Request) -> bool { } /// Parses one canonical unsigned decimal carrier sequence header. -pub(super) fn canonical_u64_header( - request: &Request, - name: &'static str, -) -> Option { +pub(super) fn canonical_u64_header(request: &Request, name: &'static str) -> Option { let values = request.headers().get_all(name); let mut values = values.iter(); let value = values.next()?.to_str().ok()?; @@ -183,10 +176,7 @@ mod tests { .header("x-forwarded-for", "192.0.2.10") .body(()) .unwrap(); - assert_eq!( - canonical_request_host(&request), - Some("proxy.example.com") - ); + assert_eq!(canonical_request_host(&request), Some("proxy.example.com")); let trusted: [IpNetwork; 1] = ["127.0.0.1/32".parse().unwrap()]; assert_eq!( client_ip( diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs index b8d3fd7..ca874fc 100644 --- a/src/web/http/tests.rs +++ b/src/web/http/tests.rs @@ -164,10 +164,8 @@ async fn https_carrier_bootstraps_and_closes_one_session() { let session = response_header(create_headers, "x-session-token"); assert_eq!(session.len(), 43); - let replacement = test_runtime_generation( - 2, - runtime_config(capability, WebCarrier::HttpsLanes), - ); + let replacement = + test_runtime_generation(2, runtime_config(capability, WebCarrier::HttpsLanes)); active_runtime.store(Arc::clone(&replacement)); tokio::time::sleep(std::time::Duration::from_millis(1100)).await; let retry_response = request(&listener, &runtime, create_retry).await; @@ -183,10 +181,16 @@ async fn https_carrier_bootstraps_and_closes_one_session() { .into_bytes(); let next_root_response = request(&listener, &runtime, next_root).await; let (_, next_root_body) = split_response(&next_root_response); - assert!(next_root_body.windows(11).any(|value| value == b"bootstrap='")); - assert!(next_root_body - .windows(21) - .any(|value| value == b"carrier='https-lanes'")); + assert!( + next_root_body + .windows(11) + .any(|value| value == b"bootstrap='") + ); + assert!( + next_root_body + .windows(21) + .any(|value| value == b"carrier='https-lanes'") + ); let close = format!( "DELETE /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {session}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" @@ -262,10 +266,7 @@ async fn unused_bootstrap_survives_equivalent_runtime_generation_swap() { .map(|(token, _)| token) .unwrap(); - let replacement = test_runtime_generation( - 2, - runtime_config(capability, WebCarrier::Https), - ); + let replacement = test_runtime_generation(2, runtime_config(capability, WebCarrier::Https)); active_runtime.store(Arc::clone(&replacement)); let hello = frame::encode(FrameType::Hello, 0, &[1]); let mut create = format!( @@ -305,10 +306,8 @@ async fn unused_bootstrap_is_rejected_after_profile_identity_change() { .map(|(token, _)| token) .unwrap(); - let replacement = test_runtime_generation( - 2, - runtime_config(capability, WebCarrier::HttpsLanes), - ); + let replacement = + test_runtime_generation(2, runtime_config(capability, WebCarrier::HttpsLanes)); active_runtime.store(Arc::clone(&replacement)); let hello = frame::encode(FrameType::Hello, 0, &[1]); let mut create = format!( @@ -330,10 +329,7 @@ async fn unused_bootstrap_is_rejected_after_profile_identity_change() { #[tokio::test] async fn https_lanes_is_advertised_and_requires_canonical_lane_headers() { let capability = [9u8; 32]; - let generation = test_runtime_generation( - 1, - runtime_config(capability, WebCarrier::HttpsLanes), - ); + let generation = test_runtime_generation(1, runtime_config(capability, WebCarrier::HttpsLanes)); let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&generation))); let runtime = WebProcessRuntime::start(active_runtime); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -361,7 +357,10 @@ async fn https_lanes_is_advertised_and_requires_canonical_lane_headers() { create.extend_from_slice(&hello); let create_response = request(&listener, &runtime, create).await; let (create_headers, _) = split_response(&create_response); - assert_eq!(response_header(create_headers, "x-carrier-mode"), "https-lanes"); + assert_eq!( + response_header(create_headers, "x-carrier-mode"), + "https-lanes" + ); let session = response_header(create_headers, "x-session-token").to_string(); let pong = frame::encode(FrameType::Pong, 0, &[]); @@ -375,10 +374,12 @@ async fn https_lanes_is_advertised_and_requires_canonical_lane_headers() { let (uplink_headers, _) = split_response(&uplink_response); assert!(uplink_headers.starts_with(b"HTTP/1.1 204")); assert_eq!(response_header(uplink_headers, "x-up-ack"), "1"); - assert!(!std::str::from_utf8(uplink_headers) - .unwrap() - .lines() - .any(|line| line.to_ascii_lowercase().starts_with("content-length:"))); + assert!( + !std::str::from_utf8(uplink_headers) + .unwrap() + .lines() + .any(|line| line.to_ascii_lowercase().starts_with("content-length:")) + ); let mut missing_lane = format!( "POST /api/v1/up HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {session}\r\nContent-Type: application/octet-stream\r\nX-Up-Seq: 2\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", diff --git a/src/web/manager.rs b/src/web/manager.rs index 62feabe..ef30345 100644 --- a/src/web/manager.rs +++ b/src/web/manager.rs @@ -89,9 +89,7 @@ pub(crate) struct WebProcessRuntime { impl WebProcessRuntime { /// Starts one process-scoped manager using immutable allocation ceilings. - pub(crate) fn start( - active_runtime: Arc>, - ) -> Arc { + pub(crate) fn start(active_runtime: Arc>) -> Arc { let limits = active_runtime.load().config().web.limits.clone(); let runtime = Arc::new(Self { active_runtime, @@ -169,9 +167,7 @@ impl WebProcessRuntime { /// Reserves one logical stream in the inner MTProxy handshake phase. pub(crate) fn try_stream_handshake(&self) -> Option { - let permit = Arc::clone(&self.stream_handshakes) - .try_acquire_owned() - .ok(); + let permit = Arc::clone(&self.stream_handshakes).try_acquire_owned().ok(); if permit.is_none() { self.record_stream_rejected(); } @@ -236,7 +232,11 @@ impl WebProcessRuntime { let mut state = self.state.lock(); remove_expired_locked(&mut state, now); if state.closed - || state.bootstraps_per_ip.get(&client_ip).copied().unwrap_or(0) + || state + .bootstraps_per_ip + .get(&client_ip) + .copied() + .unwrap_or(0) >= self.limits.max_bootstraps_per_ip || !allow_rate( &mut state.bootstrap_rate, @@ -313,10 +313,7 @@ impl WebProcessRuntime { if !digest_matches { return Err(ManagerError::Authentication); } - let session = entry - .session - .as_ref() - .ok_or(ManagerError::Authentication)?; + let session = entry.session.as_ref().ok_or(ManagerError::Authentication)?; return Ok(CreateResult { token: entry.session_token.as_str().to_owned(), carrier: session.carrier(), @@ -448,14 +445,11 @@ impl WebProcessRuntime { let fits = if control { bytes <= self.limits.control_bytes_global && items <= control_item_reserve - && state.pending_bytes - <= self.limits.pending_bytes_global.saturating_sub(bytes) - && state.pending_items - <= self.limits.pending_items_global.saturating_sub(items) + && state.pending_bytes <= self.limits.pending_bytes_global.saturating_sub(bytes) + && state.pending_items <= self.limits.pending_items_global.saturating_sub(items) && state.pending_control_bytes <= self.limits.control_bytes_global.saturating_sub(bytes) - && state.pending_control_items - <= control_item_reserve.saturating_sub(items) + && state.pending_control_items <= control_item_reserve.saturating_sub(items) } else { let data_bytes = state .pending_bytes @@ -464,14 +458,11 @@ impl WebProcessRuntime { .pending_items .saturating_sub(state.pending_control_items); let (byte_limit, item_limit) = if downlink { - let uplink_bytes = self - .limits - .max_body_bytes - .saturating_add( - self.limits - .max_frames_per_body - .saturating_mul(crate::web::session::QUEUE_ITEM_COST), - ); + let uplink_bytes = self.limits.max_body_bytes.saturating_add( + self.limits + .max_frames_per_body + .saturating_mul(crate::web::session::QUEUE_ITEM_COST), + ); ( data_byte_limit.saturating_sub(uplink_bytes), data_item_limit.saturating_sub(self.limits.max_frames_per_body), @@ -531,5 +522,4 @@ impl WebProcessRuntime { fn record_limit_hit(&self) { self.limit_hits.fetch_add(1, Ordering::Relaxed); } - } diff --git a/src/web/manager/admission.rs b/src/web/manager/admission.rs index 61a3520..b1bc54e 100644 --- a/src/web/manager/admission.rs +++ b/src/web/manager/admission.rs @@ -2,9 +2,7 @@ use std::net::{IpAddr, SocketAddr}; use std::sync::atomic::Ordering; use std::time::Instant; -use super::state::{ - allocate_stream_port, allow_rate, decrement_map, release_stream_port, -}; +use super::state::{allocate_stream_port, allow_rate, decrement_map, release_stream_port}; use super::{ProfileKey, WebProcessRuntime}; impl WebProcessRuntime { @@ -43,10 +41,7 @@ impl WebProcessRuntime { return None; }; state.streams_live += 1; - *state - .streams_per_profile - .entry(profile_key) - .or_insert(0) += 1; + *state.streams_per_profile.entry(profile_key).or_insert(0) += 1; self.streams_opened.fetch_add(1, Ordering::Relaxed); Some(peer_port) } @@ -105,12 +100,7 @@ mod tests { let downlink_bytes = data_bytes - uplink_bytes; let downlink_items = data_items - runtime.limits.max_frames_per_body; - assert!(runtime.try_reserve_pending( - downlink_bytes, - downlink_items, - false, - true, - )); + assert!(runtime.try_reserve_pending(downlink_bytes, downlink_items, false, true,)); assert!(runtime.try_reserve_pending( uplink_bytes, runtime.limits.max_frames_per_body, @@ -120,11 +110,7 @@ mod tests { assert!(!runtime.try_reserve_pending(1, 1, false, true)); runtime.release_pending(downlink_bytes, downlink_items, false); - runtime.release_pending( - uplink_bytes, - runtime.limits.max_frames_per_body, - false, - ); + runtime.release_pending(uplink_bytes, runtime.limits.max_frames_per_body, false); runtime.shutdown().await; } } diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs index e367470..2eddf77 100644 --- a/src/web/manager/lifecycle.rs +++ b/src/web/manager/lifecycle.rs @@ -4,10 +4,8 @@ use std::time::{Duration, Instant}; use tracing::info; +use super::state::{ClosedToken, decrement_map, remove_bootstrap_locked, remove_expired_locked}; use super::{ProfileKey, TokenHash, WebProcessRuntime}; -use super::state::{ - ClosedToken, decrement_map, remove_bootstrap_locked, remove_expired_locked, -}; impl WebProcessRuntime { /// Removes one closed session and retains a bounded host-bound replay marker. diff --git a/src/web/manager/state.rs b/src/web/manager/state.rs index 8895c5d..1bdbe7f 100644 --- a/src/web/manager/state.rs +++ b/src/web/manager/state.rs @@ -135,17 +135,11 @@ pub(super) fn matching_profile( } /// Applies one token-bucket admission decision at a caller-supplied monotonic time. -pub(super) fn allow_rate( - state: &mut RateState, - now: Instant, - per_minute: u32, - burst: u32, -) -> bool { +pub(super) fn allow_rate(state: &mut RateState, now: Instant, per_minute: u32, burst: u32) -> bool { let burst = f64::from(burst); if let Some(last) = state.last { let elapsed = now.saturating_duration_since(last).as_secs_f64(); - state.tokens = - (state.tokens + elapsed * f64::from(per_minute) / 60.0).min(burst); + state.tokens = (state.tokens + elapsed * f64::from(per_minute) / 60.0).min(burst); } else { state.tokens = burst; } @@ -216,9 +210,9 @@ where /// Computes the process-wide item reserve required for session control progress. pub(super) fn control_item_reserve(limits: &WebLimitsConfig) -> usize { - limits.max_sessions_global.saturating_mul( - 16usize.saturating_add(limits.max_streams_per_session.saturating_mul(3)), - ) + limits + .max_sessions_global + .saturating_mul(16usize.saturating_add(limits.max_streams_per_session.saturating_mul(3))) } /// Allocates a non-zero source port unique among live streams for one KDF route. diff --git a/src/web/session.rs b/src/web/session.rs index 45795f0..38d5a00 100644 --- a/src/web/session.rs +++ b/src/web/session.rs @@ -314,7 +314,9 @@ impl WebSession { if !self.queue_window_locked(&mut state, stream_id, count as u32) { drop(state); self.close(); - return Poll::Ready(Err(io::Error::other("WEB session control budget exhausted"))); + return Poll::Ready(Err(io::Error::other( + "WEB session control budget exhausted", + ))); } Poll::Ready(Ok(())) } @@ -368,14 +370,14 @@ impl WebSession { /// Returns the process queue-capacity notification source while the manager lives. pub(super) fn budget_notify(&self) -> Option> { - self.manager.upgrade().map(|manager| manager.budget_notify()) + self.manager + .upgrade() + .map(|manager| manager.budget_notify()) } fn release_stream_reservation(&self, peer_port: u16) { let removed = self.state.lock().active_peer_ports.remove(&peer_port); - if removed - && let Some(manager) = self.manager.upgrade() - { + if removed && let Some(manager) = self.manager.upgrade() { manager.release_stream( self.profile_key, self.client_ip, diff --git a/src/web/session/backend.rs b/src/web/session/backend.rs index 083f39d..430298c 100644 --- a/src/web/session/backend.rs +++ b/src/web/session/backend.rs @@ -3,9 +3,9 @@ use std::sync::Arc; use std::sync::atomic::Ordering; use std::time::Duration; +use crate::proxy::shared_state::ConntrackClosePolicy; use crate::web::frame::FrameType; use crate::web::stream::WebLogicalStream; -use crate::proxy::shared_state::ConntrackClosePolicy; use super::{WebSession, inbound_queue_cost}; @@ -70,9 +70,7 @@ impl WebSession { }); (queued, reserved) }; - if reserved - && let Some(manager) = self.manager.upgrade() - { + if reserved && let Some(manager) = self.manager.upgrade() { manager.release_stream( self.profile_key, self.client_ip, diff --git a/src/web/session/downlink.rs b/src/web/session/downlink.rs index c40f86f..d07e855 100644 --- a/src/web/session/downlink.rs +++ b/src/web/session/downlink.rs @@ -3,8 +3,7 @@ use std::time::{Duration, Instant}; use bytes::{BufMut, Bytes, BytesMut}; use super::{ - DownBatch, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame, SessionState, - WebSession, + DownBatch, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame, SessionState, WebSession, }; use crate::config::WebCarrier; use crate::web::frame::{self, FrameType}; @@ -117,9 +116,8 @@ impl WebSession { .limits .pending_bytes_per_session .saturating_sub(self.limits.control_bytes_per_session); - let item_reserve = 16usize.saturating_add( - self.limits.max_streams_per_session.saturating_mul(3), - ); + let item_reserve = + 16usize.saturating_add(self.limits.max_streams_per_session.saturating_mul(3)); let data_item_limit = self .limits .pending_items_per_session @@ -146,14 +144,11 @@ impl WebSession { .pending_items .saturating_sub(state.pending_control_items); let (byte_limit, item_limit) = if class == PendingClass::Downlink { - let uplink_bytes = self - .limits - .max_body_bytes - .saturating_add( - self.limits - .max_frames_per_body - .saturating_mul(QUEUE_ITEM_COST), - ); + let uplink_bytes = self.limits.max_body_bytes.saturating_add( + self.limits + .max_frames_per_body + .saturating_mul(QUEUE_ITEM_COST), + ); ( data_byte_limit.saturating_sub(uplink_bytes), data_item_limit.saturating_sub(self.limits.max_frames_per_body), @@ -172,12 +167,7 @@ impl WebSession { let Some(manager) = self.manager.upgrade() else { return false; }; - if !manager.try_reserve_pending( - bytes, - items, - control, - class == PendingClass::Downlink, - ) { + if !manager.try_reserve_pending(bytes, items, control, class == PendingClass::Downlink) { return false; } state.pending_bytes += bytes; @@ -301,13 +291,7 @@ impl WebSession { control: bool, ) -> bool { if self.carrier() == WebCarrier::HttpsLanes { - return self.queue_lane_frame_locked( - state, - frame_type, - stream_id, - payload, - control, - ); + return self.queue_lane_frame_locked(state, frame_type, stream_id, payload, control); } let cost = frame::HEADER_BYTES + payload.len() + QUEUE_ITEM_COST; let class = if control { @@ -425,9 +409,7 @@ mod tests { use std::net::SocketAddr; use std::sync::Arc; - use crate::config::{ - WebLimitsConfig, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig, - }; + use crate::config::{WebLimitsConfig, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig}; use crate::web::manager::WebProcessRuntime; fn session() -> Arc { diff --git a/src/web/session/lanes.rs b/src/web/session/lanes.rs index cf85c02..00f2de7 100644 --- a/src/web/session/lanes.rs +++ b/src/web/session/lanes.rs @@ -7,8 +7,8 @@ use subtle::ConstantTimeEq; use super::uplink::{inbound_reservation, validate_batch}; use super::{ - CarrierLane, DownBatch, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame, - SessionState, WebSession, remember_closed, + CarrierLane, DownBatch, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame, SessionState, + WebSession, remember_closed, }; use crate::config::{WebCarrier, WebLimitsConfig}; use crate::web::frame::{self, Frame, FrameType}; @@ -32,9 +32,11 @@ impl WebSession { return Err(ManagerError::Protocol); } }; - if frames.iter().copied().any(|value| { - value.stream_id != lane_id || frame::validate_client_shape(value).is_err() - }) { + if frames + .iter() + .copied() + .any(|value| value.stream_id != lane_id || frame::validate_client_shape(value).is_err()) + { self.close(); return Err(ManagerError::Protocol); } @@ -48,13 +50,17 @@ impl WebSession { state.last_activity = Instant::now(); if !state.carrier_lanes.contains_key(&lane_id) { if lane_id != 0 - && frames.first().is_some_and(|value| value.frame_type != FrameType::Open) + && frames + .first() + .is_some_and(|value| value.frame_type != FrameType::Open) && only_late_frames(&frames) { return Ok(sequence); } if lane_id == 0 - || frames.first().is_none_or(|value| value.frame_type != FrameType::Open) + || frames + .first() + .is_none_or(|value| value.frame_type != FrameType::Open) { drop(state); self.close(); @@ -182,12 +188,7 @@ impl WebSession { }; if let Some(batch) = acknowledged { self.release_locked(&mut state, batch.data_bytes, batch.data_items, false); - self.release_locked( - &mut state, - batch.control_bytes, - batch.control_items, - true, - ); + self.release_locked(&mut state, batch.control_bytes, batch.control_items, true); if let Some(stream) = state.streams.get_mut(&lane_id) && let Some(waker) = stream.write_waker.take() { @@ -402,11 +403,7 @@ impl WebSession { } pub(super) fn remember_closed_locked(&self, state: &mut SessionState, stream_id: u32) { - let evicted = remember_closed( - state, - stream_id, - self.limits.max_tombstones_per_session, - ); + let evicted = remember_closed(state, stream_id, self.limits.max_tombstones_per_session); if self.carrier() != WebCarrier::HttpsLanes { return; } diff --git a/src/web/session/lanes/tests.rs b/src/web/session/lanes/tests.rs index 6281800..308a053 100644 --- a/src/web/session/lanes/tests.rs +++ b/src/web/session/lanes/tests.rs @@ -4,9 +4,7 @@ use std::sync::Arc; use bytes::BytesMut; use super::*; -use crate::config::{ - WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig, -}; +use crate::config::{WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig}; use crate::web::manager::WebProcessRuntime; fn session_with_limits(limits: WebLimitsConfig) -> Arc { @@ -103,15 +101,18 @@ fn tombstone_eviction_releases_lane_budget_and_accepts_late_frames() { state.carrier_lanes.insert(7, CarrierLane::new()); let encoded = frame::encode(FrameType::Close, 7, &[]); let cost = encoded.len() + QUEUE_ITEM_COST; - state.carrier_lanes.get_mut(&7).unwrap().pending_frames.push_back( - QueuedFrame { + state + .carrier_lanes + .get_mut(&7) + .unwrap() + .pending_frames + .push_back(QueuedFrame { encoded: BytesMut::from(encoded.as_ref()), frame_type: FrameType::Close, stream_id: 7, control: true, cost, - }, - ); + }); state.pending_bytes = cost; state.pending_items = 1; state.pending_control_bytes = cost; diff --git a/src/web/session/uplink.rs b/src/web/session/uplink.rs index 649c7f0..8c506bb 100644 --- a/src/web/session/uplink.rs +++ b/src/web/session/uplink.rs @@ -138,12 +138,8 @@ impl WebSession { FrameType::Open => { let Some(peer_port) = self.reserve_stream_locked(state) else { self.remember_closed_locked(state, value.stream_id); - if !self.queue_control_locked( - state, - FrameType::Close, - value.stream_id, - &[], - ) { + if !self.queue_control_locked(state, FrameType::Close, value.stream_id, &[]) + { return false; } continue; @@ -169,8 +165,8 @@ impl WebSession { bytes: Bytes::copy_from_slice(value.payload), offset: 0, }); - *unused_bytes = unused_bytes - .saturating_sub(value.payload.len() + QUEUE_ITEM_COST); + *unused_bytes = + unused_bytes.saturating_sub(value.payload.len() + QUEUE_ITEM_COST); *unused_items = unused_items.saturating_sub(1); if let Some(waker) = stream.read_waker.take() { waker.wake(); @@ -256,8 +252,8 @@ pub(super) fn validate_batch(state: &SessionState, frames: &[Frame<'_>]) -> bool } continue; } - let was_closed = state.closed_streams.contains(&value.stream_id) - || closed.contains(&value.stream_id); + let was_closed = + state.closed_streams.contains(&value.stream_id) || closed.contains(&value.stream_id); match value.frame_type { FrameType::Open => { if live.contains_key(&value.stream_id) || was_closed { @@ -308,10 +304,7 @@ pub(super) fn validate_batch(state: &SessionState, frames: &[Frame<'_>]) -> bool true } -pub(super) fn inbound_reservation( - state: &SessionState, - frames: &[Frame<'_>], -) -> (usize, usize) { +pub(super) fn inbound_reservation(state: &SessionState, frames: &[Frame<'_>]) -> (usize, usize) { let mut live = state.streams.keys().copied().collect::>(); let mut bytes = 0usize; let mut items = 0usize; @@ -338,9 +331,7 @@ mod tests { use super::*; use std::net::SocketAddr; - use crate::config::{ - WebLimitsConfig, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig, - }; + use crate::config::{WebLimitsConfig, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig}; use crate::web::manager::WebProcessRuntime; fn session() -> Arc { @@ -383,10 +374,7 @@ mod tests { let session = session(); let body = frame::encode(FrameType::Pong, 0, &[]); session.up_active.store(true, Ordering::Release); - assert_eq!( - session.process_up(1, &body), - Err(ManagerError::Concurrent) - ); + assert_eq!(session.process_up(1, &body), Err(ManagerError::Concurrent)); assert_eq!(session.state.lock().last_up_sequence, 0); session.up_active.store(false, Ordering::Release); assert_eq!(session.process_up(1, &body), Ok(1));