From 030466603f2b90f6f8b2833c0355202930b2fdd9 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sun, 23 Aug 2026 10:22:13 +0300 Subject: [PATCH] WEB Carrier behavior aligned w/ reference Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com> --- src/config/types/server.rs | 2 +- src/web/bridge.rs | 35 ++++++--- src/web/http/request.rs | 60 +++++++++++++--- src/web/http/tests.rs | 135 ++++++++++++++++++++++++++++++++--- src/web/manager.rs | 27 ++----- src/web/manager/lifecycle.rs | 12 ---- src/web/manager/state.rs | 4 +- 7 files changed, 211 insertions(+), 64 deletions(-) diff --git a/src/config/types/server.rs b/src/config/types/server.rs index f333d76..6eb74a4 100644 --- a/src/config/types/server.rs +++ b/src/config/types/server.rs @@ -90,7 +90,7 @@ pub enum ListenerTransport { #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] #[serde(rename_all = "snake_case")] pub enum WebClientIpSource { - /// Require exactly one canonical IP in `X-Forwarded-For`. + /// Use one parseable `X-Forwarded-For` address or the trusted direct peer. #[default] XForwardedFor, } diff --git a/src/web/bridge.rs b/src/web/bridge.rs index e640ef4..a57bfe1 100644 --- a/src/web/bridge.rs +++ b/src/web/bridge.rs @@ -114,8 +114,13 @@ function joinPending(values,lane){ return {body:joined.buffer,total,count}; } function retryAfterMs(response){ - const value=Number(response.headers.get('Retry-After')); - return Number.isFinite(value)&&value>=0?Math.min(value*1000,30000):0; + const header=response.headers.get('Retry-After'); + if(!header)return 0; + const seconds=Number(header); + if(Number.isFinite(seconds)&&seconds>=0)return Math.min(seconds*1000,30000); + const when=Date.parse(header); + if(Number.isFinite(when)){const delta=when-Date.now();return delta>0?Math.min(delta,30000):0} + return 0; } async function request(path,makeOptions){ let delay=250,attempt=0;const deadline=Date.now()+90000; @@ -142,13 +147,11 @@ async function createSession(first){ try{ status('connecting'); const response=await request('/api/v1/session',()=>options('POST',bootstrap,first)); - if(response.status!==200||response.headers.get('X-Carrier-Mode')!==carrier)throw new Error('session rejected'); + if(response.status!==200||response.headers.get('X-Carrier-Mode')!==carrier)throw new Error('session creation rejected'); sessionToken=response.headers.get('X-Session-Token')||'';downCursor=response.headers.get('X-Down-Cursor')||'0'; - if(!/^[A-Za-z0-9_-]{43}$/.test(sessionToken)||downCursor!=='0')throw new Error('invalid session metadata'); + if(!sessionToken)throw new Error('missing session token'); if(closed){deleteSession();return} const welcome=await response.arrayBuffer(); - const welcomeBytes=new Uint8Array(welcome); - if(welcomeBytes.length!==8||welcomeBytes[0]!==17||welcomeBytes.slice(1).some(value=>value!==0))throw new Error('invalid welcome'); port.postMessage(welcome,[welcome]);status('connected'); if(carrier==='https-lanes')ensureLane(0); for(const data of pending.splice(0)){release(data.byteLength,1,null);queueCarrier(data)} @@ -234,7 +237,8 @@ async function pollLane(lane){ } if(response.status!==200)throw new Error('lane downlink rejected'); const next=response.headers.get('X-Down-Cursor')||'',data=await response.arrayBuffer(); - if(!next||!data.byteLength||splitFrames(data).some(value=>value.id!==lane.id))throw new Error('invalid lane downlink response'); + if(!next||!data.byteLength)throw new Error('invalid lane downlink response'); + for(const value of splitFrames(data))if(value.id!==lane.id)throw new Error('cross-lane frame'); if(closed)return; port.postMessage({t:'traffic',up:0,down:data.byteLength});port.postMessage(data,[data]);lane.cursor=next;status('connected'); } @@ -268,11 +272,14 @@ addEventListener('message',event=>{ let source;try{source=new URL(event.origin)}catch(error){return} if(source.protocol!=='http:'||source.hostname!=='127.0.0.1'||!source.port||source.origin!==event.origin)return; activatePort(event.ports[0]); -}); +},{once:false}); const androidBridge=globalThis.TelegramWebProxy; if(!initialized&&androidNonce&&androidBridge&&typeof androidBridge.postMessage==='function'){ const androidPort={onmessage:null,start(){},close(){androidBridge.onmessage=null},postMessage(value){ - if(value instanceof ArrayBuffer){for(const item of splitFrames(value))androidBridge.postMessage(item.data)}else androidBridge.postMessage(JSON.stringify(value)); + if(value instanceof ArrayBuffer){ + let frames;try{frames=splitFrames(value)}catch(error){fail();return} + for(const frame of frames)androidBridge.postMessage(frame.data); + }else androidBridge.postMessage(JSON.stringify(value)); }}; androidBridge.onmessage=event=>{let data=event.data;if(typeof data==='string'){try{data=JSON.parse(data)}catch(error){return}}if(androidPort.onmessage)androidPort.onmessage({data})}; activatePort(androidPort);androidBridge.postMessage(JSON.stringify({t:'tproxy-android-init',v:1,nonce:androidNonce})); @@ -304,6 +311,16 @@ mod tests { assert!(page.body.contains("X-Up-Seq")); assert!(page.body.contains("carrier='https-lanes'")); 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("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:*")); diff --git a/src/web/http/request.rs b/src/web/http/request.rs index 8a8a442..963aad3 100644 --- a/src/web/http/request.rs +++ b/src/web/http/request.rs @@ -33,7 +33,7 @@ pub(super) fn canonical_request_host(request: &Request) -> Option<&str> { Some(host) } -/// Accepts one canonical forwarded client address from an explicitly trusted peer. +/// Accepts one forwarded client address or the direct address of a trusted peer. pub(super) fn client_ip( request: &Request, peer: SocketAddr, @@ -51,16 +51,17 @@ pub(super) fn client_ip( }; let values = request.headers().get_all(header_name); let mut values = values.iter(); - let value = values.next()?.to_str().ok()?; - if values.next().is_some() - || value.is_empty() - || value.trim() != value - || value.contains(',') - { + let Some(value) = values.next() else { + return Some(peer.ip()); + }; + let value = value.to_str().ok()?; + if values.next().is_some() || value.trim() != value || value.contains(',') { return None; } - let ip = value.parse::().ok()?; - (ip.to_string() == value).then_some(ip) + if value.is_empty() { + return Some(peer.ip()); + } + value.parse::().ok() } /// Decodes an exact canonical bridge query without allocating credential strings. @@ -176,7 +177,7 @@ mod tests { } #[test] - fn host_and_forwarded_identity_require_canonical_single_values() { + fn host_is_canonical_and_forwarded_identity_is_single_parseable_ip() { let request = Request::builder() .header(header::HOST, "proxy.example.com:443") .header("x-forwarded-for", "192.0.2.10") @@ -197,6 +198,45 @@ mod tests { Some("192.0.2.10".parse().unwrap()) ); + let expanded_ipv6 = Request::builder() + .header("x-forwarded-for", "2001:0db8:0:0:0:0:0:10") + .body(()) + .unwrap(); + assert_eq!( + client_ip( + &expanded_ipv6, + "127.0.0.1:40000".parse().unwrap(), + WebClientIpSource::XForwardedFor, + &trusted, + ), + Some("2001:db8::10".parse().unwrap()) + ); + + let without_forwarded_address = Request::builder().body(()).unwrap(); + assert_eq!( + client_ip( + &without_forwarded_address, + "127.0.0.1:40000".parse().unwrap(), + WebClientIpSource::XForwardedFor, + &trusted, + ), + Some("127.0.0.1".parse().unwrap()) + ); + + let empty_forwarded_address = Request::builder() + .header("x-forwarded-for", "") + .body(()) + .unwrap(); + assert_eq!( + client_ip( + &empty_forwarded_address, + "127.0.0.1:40000".parse().unwrap(), + WebClientIpSource::XForwardedFor, + &trusted, + ), + Some("127.0.0.1".parse().unwrap()) + ); + let uppercase = Request::builder() .header(header::HOST, "Proxy.Example.com") .body(()) diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs index 1b093fd..b8d3fd7 100644 --- a/src/web/http/tests.rs +++ b/src/web/http/tests.rs @@ -122,15 +122,6 @@ async fn https_carrier_bootstraps_and_closes_one_session() { let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime)); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability); - let wrong_family = format!( - "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 2001:db8::10\r\nConnection: close\r\n\r\n" - ) - .into_bytes(); - let wrong_family_response = request(&listener, &runtime, wrong_family).await; - let (_, wrong_family_body) = split_response(&wrong_family_response); - assert!(!wrong_family_body - .windows(11) - .any(|value| value == b"bootstrap='")); let root = format!( "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nConnection: close\r\n\r\n" ) @@ -214,6 +205,128 @@ async fn https_carrier_bootstraps_and_closes_one_session() { replacement.stop_background_tasks().await; } +#[tokio::test] +async fn bootstrap_survives_client_address_family_change() { + let capability = [8u8; 32]; + let generation = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https)); + 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(); + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability); + let root = format!( + "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 2001:db8::10\r\nConnection: close\r\n\r\n" + ) + .into_bytes(); + let root_response = request(&listener, &runtime, root).await; + let (_, root_body) = split_response(&root_response); + let root_body = std::str::from_utf8(root_body).unwrap(); + let bootstrap = root_body + .split_once("bootstrap='") + .and_then(|(_, suffix)| suffix.split_once('\'')) + .map(|(token, _)| token) + .unwrap(); + + let hello = frame::encode(FrameType::Hello, 0, &[1]); + let mut create = format!( + "POST /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + hello.len() + ) + .into_bytes(); + create.extend_from_slice(&hello); + let create_response = request(&listener, &runtime, create).await; + assert!(create_response.starts_with(b"HTTP/1.1 200")); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} + +#[tokio::test] +async fn unused_bootstrap_survives_equivalent_runtime_generation_swap() { + let capability = [10u8; 32]; + let generation = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https)); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&generation))); + let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime)); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability); + let root = format!( + "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nConnection: close\r\n\r\n" + ) + .into_bytes(); + let root_response = request(&listener, &runtime, root).await; + let (_, root_body) = split_response(&root_response); + let root_body = std::str::from_utf8(root_body).unwrap(); + let bootstrap = root_body + .split_once("bootstrap='") + .and_then(|(_, suffix)| suffix.split_once('\'')) + .map(|(token, _)| token) + .unwrap(); + + 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!( + "POST /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + hello.len() + ) + .into_bytes(); + create.extend_from_slice(&hello); + let create_response = request(&listener, &runtime, create).await; + assert!(create_response.starts_with(b"HTTP/1.1 200")); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + replacement.stop_sessions().await; + replacement.stop_background_tasks().await; +} + +#[tokio::test] +async fn unused_bootstrap_is_rejected_after_profile_identity_change() { + let capability = [11u8; 32]; + let generation = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https)); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&generation))); + let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime)); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability); + let root = format!( + "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nConnection: close\r\n\r\n" + ) + .into_bytes(); + let root_response = request(&listener, &runtime, root).await; + let (_, root_body) = split_response(&root_response); + let root_body = std::str::from_utf8(root_body).unwrap(); + let bootstrap = root_body + .split_once("bootstrap='") + .and_then(|(_, suffix)| suffix.split_once('\'')) + .map(|(token, _)| token) + .unwrap(); + + 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!( + "POST /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + hello.len() + ) + .into_bytes(); + create.extend_from_slice(&hello); + let create_response = request(&listener, &runtime, create).await; + assert!(!create_response.starts_with(b"HTTP/1.1 200")); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + replacement.stop_sessions().await; + replacement.stop_background_tasks().await; +} + #[tokio::test] async fn https_lanes_is_advertised_and_requires_canonical_lane_headers() { let capability = [9u8; 32]; @@ -262,6 +375,10 @@ 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:"))); 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 51f114f..62feabe 100644 --- a/src/web/manager.rs +++ b/src/web/manager.rs @@ -215,7 +215,7 @@ impl WebProcessRuntime { Some((reader, body)) } - /// Issues a one-use bootstrap credential for the active generation. + /// Issues a one-use bootstrap credential for an active compatible profile. pub(crate) fn issue_bootstrap( &self, profile: Arc, @@ -229,10 +229,7 @@ impl WebProcessRuntime { .as_ref() .and_then(|runtime| matching_profile(runtime, &profile)) .ok_or(ManagerError::Authentication)?; - if !config.web.enabled - || profile.public_addr.is_ipv4() != client_ip.is_ipv4() - || !generation.proxy_shared.is_user_enabled(&profile.user) - { + if !config.web.enabled || !generation.proxy_shared.is_user_enabled(&profile.user) { return Err(ManagerError::Closed); } let now = Instant::now(); @@ -264,7 +261,6 @@ impl WebProcessRuntime { state.bootstraps.insert( hash, Bootstrap { - generation_id: generation.id, expires_at: now + Duration::from_secs(config.web.timeouts.bootstrap_lifetime_secs), issued_at: now, issuance_ip: client_ip, @@ -281,15 +277,12 @@ impl WebProcessRuntime { /// Checks whether a bootstrap token is live before reading a request body. pub(crate) fn has_bootstrap(&self, hash: TokenHash, host: &str) -> bool { - let generation_id = self.active_runtime.load().id; let now = Instant::now(); let state = self.state.lock(); - state.bootstraps.get(&hash).is_some_and(|entry| { - entry.profile.host == host - && now <= entry.expires_at - && (entry.generation_id == generation_id - || entry.used && entry.session.is_some()) - }) + state + .bootstraps + .get(&hash) + .is_some_and(|entry| entry.profile.host == host && now <= entry.expires_at) } /// Creates a session exactly once or replays the original successful result. @@ -329,9 +322,6 @@ impl WebProcessRuntime { carrier: session.carrier(), }); } - if entry.generation_id != generation.id { - return Err(ManagerError::Authentication); - } if state.closed || !config.web.enabled { return Err(ManagerError::Closed); } @@ -340,10 +330,7 @@ impl WebProcessRuntime { .runtime .as_ref() .and_then(|runtime| matching_profile(runtime, &entry.profile)) - .filter(|profile| { - profile.public_addr.is_ipv4() == client_ip.is_ipv4() - && generation.proxy_shared.is_user_enabled(&profile.user) - }) + .filter(|profile| generation.proxy_shared.is_user_enabled(&profile.user)) .ok_or(ManagerError::Authentication)?; let profile_key = profile_key(&profile); if state.sessions.len() >= self.limits.max_sessions_global diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs index 8c365c9..e367470 100644 --- a/src/web/manager/lifecycle.rs +++ b/src/web/manager/lifecycle.rs @@ -124,22 +124,10 @@ impl WebProcessRuntime { /// Expires credentials and closes idle sessions without holding locks across callbacks. pub(super) fn cleanup(&self) { - let generation_id = self.active_runtime.load().id; let now = Instant::now(); let sessions = { let mut state = self.state.lock(); remove_expired_locked(&mut state, now); - let stale_bootstraps = state - .bootstraps - .iter() - .filter_map(|(hash, bootstrap)| { - (bootstrap.generation_id != generation_id && !bootstrap.used) - .then_some(*hash) - }) - .collect::>(); - for hash in stale_bootstraps { - remove_bootstrap_locked(&mut state, hash); - } state.sessions.values().cloned().collect::>() }; for session in sessions.into_iter().filter(|session| session.is_idle(now)) { diff --git a/src/web/manager/state.rs b/src/web/manager/state.rs index 28a47a5..8895c5d 100644 --- a/src/web/manager/state.rs +++ b/src/web/manager/state.rs @@ -14,13 +14,11 @@ use crate::web::session::WebSession; /// One issued bootstrap and optional idempotent session-creation replay state. pub(super) struct Bootstrap { - /// Generation that issued the bootstrap. - pub(super) generation_id: u64, /// Credential and replay-state expiry deadline. pub(super) expires_at: Instant, /// Stable ordering point used for bounded eviction. pub(super) issued_at: Instant, - /// Forwarded client address that owns this credential. + /// Issuing address charged for the unused-bootstrap quota. pub(super) issuance_ip: IpAddr, /// Immutable profile selected during capability validation. pub(super) profile: Arc,