diff --git a/src/web/http/negotiation_tests.rs b/src/web/http/negotiation_tests.rs index 0c9df21..47fa1b3 100644 --- a/src/web/http/negotiation_tests.rs +++ b/src/web/http/negotiation_tests.rs @@ -5,6 +5,8 @@ use std::time::Duration; use sha2::{Digest, Sha256}; const CAPABILITIES: &str = "https,https-lanes,websocket,websocket-lanes"; +const NATIVE_USER_AGENT_HEADER: &str = + "User-Agent: Telegram/3951 CFNetwork/3896.100.1.2.1 Darwin/27.0.0\r\n"; fn issue_bootstrap(runtime: &Arc, client_ip: &str) -> String { let profile = runtime @@ -27,6 +29,16 @@ fn create_request( hello: &[u8], attempt: Option, failure: Option<&str>, +) -> Vec { + create_request_with_headers(bootstrap, hello, attempt, failure, "") +} + +fn create_request_with_headers( + bootstrap: &str, + hello: &[u8], + attempt: Option, + failure: Option<&str>, + extra_headers: &str, ) -> Vec { let negotiation = attempt.map_or_else(String::new, |attempt| { let failure = failure @@ -37,7 +49,7 @@ fn create_request( ) }); let mut request = 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\n{negotiation}Content-Length: {}\r\nConnection: close\r\n\r\n", + "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\n{negotiation}{extra_headers}Content-Length: {}\r\nConnection: close\r\n\r\n", hello.len() ) .into_bytes(); @@ -60,6 +72,17 @@ fn token_hash(token: &str) -> crate::web::manager::TokenHash { Sha256::digest(raw).into() } +fn assert_no_negotiation_headers(headers: &[u8]) { + for header in [ + "x-carrier-attempt", + "x-carrier-candidate-count", + "x-carrier-deadline", + "x-carrier-state", + ] { + assert!(optional_response_header(headers, header).is_none()); + } +} + #[tokio::test] async fn absent_carriers_reject_negotiation_and_preserve_legacy_creation() { let capability = [41; 32]; @@ -94,6 +117,121 @@ async fn absent_carriers_reject_negotiation_and_preserve_legacy_creation() { generation.stop_background_tasks().await; } +#[tokio::test] +async fn metadata_free_native_client_can_use_each_fixed_carrier() { + for (index, carrier) in WebCarrier::ALL.into_iter().enumerate() { + let capability = [50 + index as u8; 32]; + let generation = test_runtime_generation(1, runtime_config(capability, carrier)); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let bootstrap = issue_bootstrap(&runtime, "192.0.2.10"); + let hello = frame::encode(FrameType::Hello, 0, &[1]); + + let response = request( + &listener, + &runtime, + create_request_with_headers( + &bootstrap, + &hello, + None, + None, + NATIVE_USER_AGENT_HEADER, + ), + ) + .await; + let (headers, _) = split_response(&response); + assert!(headers.starts_with(b"HTTP/1.1 200")); + assert_eq!(response_header(headers, "x-carrier-mode"), carrier.as_str()); + assert_no_negotiation_headers(headers); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} + +#[tokio::test] +async fn metadata_free_native_client_uses_fallback_when_candidates_are_enabled() { + let capability = [55; 32]; + let generation = test_runtime_generation( + 1, + negotiation_runtime_config( + capability, + WebCarrier::HttpsLanes, + false, + Arc::from([WebCarrier::Websocket, WebCarrier::HttpsLanes]), + ), + ); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let bootstrap = issue_bootstrap(&runtime, "192.0.2.10"); + let hello = frame::encode(FrameType::Hello, 0, &[1]); + + let response = request( + &listener, + &runtime, + create_request_with_headers( + &bootstrap, + &hello, + None, + None, + NATIVE_USER_AGENT_HEADER, + ), + ) + .await; + let (headers, _) = split_response(&response); + assert!(headers.starts_with(b"HTTP/1.1 200")); + assert_eq!(response_header(headers, "x-carrier-mode"), "https-lanes"); + assert_no_negotiation_headers(headers); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} + +#[tokio::test] +async fn explicit_native_capabilities_participate_in_automatic_selection() { + let capability = [56; 32]; + let generation = test_runtime_generation( + 1, + negotiation_runtime_config( + capability, + WebCarrier::Https, + false, + Arc::from([WebCarrier::WebsocketLanes, WebCarrier::Https]), + ), + ); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let bootstrap = issue_bootstrap(&runtime, "192.0.2.10"); + let hello = frame::encode(FrameType::Hello, 0, &[1]); + + let response = request( + &listener, + &runtime, + create_request_with_headers( + &bootstrap, + &hello, + Some(1), + None, + NATIVE_USER_AGENT_HEADER, + ), + ) + .await; + let (headers, _) = split_response(&response); + assert!(headers.starts_with(b"HTTP/1.1 200")); + assert_eq!( + response_header(headers, "x-carrier-mode"), + "websocket-lanes" + ); + assert_eq!(response_header(headers, "x-carrier-attempt"), "1"); + assert_eq!(response_header(headers, "x-carrier-candidate-count"), "2"); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} + #[tokio::test] async fn negotiation_replays_replaces_and_freezes_after_carrier_commit() { let capability = [42; 32]; @@ -225,6 +363,65 @@ async fn negotiation_replays_replaces_and_freezes_after_carrier_commit() { generation.stop_background_tasks().await; } +#[tokio::test] +async fn timed_out_attempt_replays_before_successor_own_deadline() { + let capability = [57; 32]; + let generation = test_runtime_generation( + 1, + negotiation_runtime_config_with_deadlines( + capability, + WebCarrier::HttpsLanes, + false, + Arc::from([WebCarrier::Https, WebCarrier::HttpsLanes]), + [1, 5, 8, 12], + ), + ); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let bootstrap = issue_bootstrap(&runtime, "192.0.2.10"); + let hello = frame::encode(FrameType::Hello, 0, &[1]); + let first_request = create_request(&bootstrap, &hello, Some(1), None); + + let first = request(&listener, &runtime, first_request.clone()).await; + let (first_headers, _) = split_response(&first); + assert!(first_headers.starts_with(b"HTTP/1.1 200")); + assert_eq!(response_header(first_headers, "x-carrier-mode"), "https"); + let first_token = response_header(first_headers, "x-session-token").to_string(); + + tokio::time::sleep(Duration::from_millis(1_100)).await; + + let replay = request(&listener, &runtime, first_request).await; + let (replay_headers, _) = split_response(&replay); + assert!(replay_headers.starts_with(b"HTTP/1.1 200")); + assert_eq!( + response_header(replay_headers, "x-session-token"), + first_token + ); + assert_eq!(response_header(replay_headers, "x-carrier-state"), "provisional"); + + let second = request( + &listener, + &runtime, + create_request(&bootstrap, &hello, Some(2), Some("timeout")), + ) + .await; + let (second_headers, _) = split_response(&second); + assert!(second_headers.starts_with(b"HTTP/1.1 200")); + assert_eq!( + response_header(second_headers, "x-carrier-mode"), + "https-lanes" + ); + assert_eq!(response_header(second_headers, "x-carrier-attempt"), "2"); + assert_ne!( + response_header(second_headers, "x-session-token"), + first_token + ); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} + #[tokio::test] async fn https_lane_downlink_can_arrive_before_its_uplink_open() { let capability = [43; 32]; diff --git a/src/web/http/request.rs b/src/web/http/request.rs index 75d34f6..239dea5 100644 --- a/src/web/http/request.rs +++ b/src/web/http/request.rs @@ -151,11 +151,7 @@ pub(super) fn carrier_request(request: &Request, host: &str) -> Option(request: &Request, host: &str) -> Option ProxyConfig { runtime_config_with_carriers(capability, carrier, false, true, Arc::from([carrier])) } @@ -41,6 +43,41 @@ fn runtime_config_with_carriers( carrier_negotiation_enabled: bool, carrier_learning: bool, carriers: Arc<[WebCarrier]>, +) -> ProxyConfig { + runtime_config_with_carriers_and_deadlines( + capability, + carrier, + carrier_negotiation_enabled, + carrier_learning, + carriers, + TEST_CARRIER_DEADLINES_SECS, + ) +} + +pub(super) fn negotiation_runtime_config_with_deadlines( + capability: [u8; 32], + carrier: WebCarrier, + carrier_learning: bool, + carriers: Arc<[WebCarrier]>, + carrier_negotiation_deadlines_secs: [u64; 4], +) -> ProxyConfig { + runtime_config_with_carriers_and_deadlines( + capability, + carrier, + true, + carrier_learning, + carriers, + carrier_negotiation_deadlines_secs, + ) +} + +fn runtime_config_with_carriers_and_deadlines( + capability: [u8; 32], + carrier: WebCarrier, + carrier_negotiation_enabled: bool, + carrier_learning: bool, + carriers: Arc<[WebCarrier]>, + carrier_negotiation_deadlines_secs: [u64; 4], ) -> ProxyConfig { let profile = Arc::new(WebRuntimeProfile { host: "proxy.example.com".to_string(), @@ -51,7 +88,7 @@ fn runtime_config_with_carriers( carrier_negotiation_enabled, carrier_learning, carriers: Arc::clone(&carriers), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + carrier_negotiation_deadlines_secs, capability, key_fingerprint: "0000000000000000".to_string(), max_sessions: 4, @@ -97,6 +134,8 @@ fn runtime_config_with_carriers( WebCarriers::Disabled }; config.web.carrier_learning = carrier_learning; + config.web.timeouts.carrier_negotiation_deadlines_secs = + carrier_negotiation_deadlines_secs; config.web.limits.max_bootstraps_per_ip = 1; config.web.timeouts.shutdown_secs = 1; config.web.runtime = Some(Arc::new(WebRuntimeConfig { diff --git a/src/web/manager/negotiation.rs b/src/web/manager/negotiation.rs index be6f57d..0267eca 100644 --- a/src/web/manager/negotiation.rs +++ b/src/web/manager/negotiation.rs @@ -11,7 +11,7 @@ pub(crate) enum CarrierClientClass { Bridge, /// Strict same-origin browser metadata survived while the marker did not. BrowserHint, - /// A native iOS client that supports only the serialized HTTPS carrier. + /// A native iOS client classified for diagnostics and learning only. Ios, } @@ -67,7 +67,7 @@ impl CarrierFailure { } } -/// Fixed carrier capability set sent by the generated bridge. +/// Validated carrier capability set sent by a negotiation-capable client. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) struct CarrierCapabilities(u8); @@ -77,11 +77,6 @@ impl CarrierCapabilities { Self(0b1111) } - /// Returns the only carrier implemented by the native iOS client. - pub(crate) const fn ios() -> Self { - Self(0b0001) - } - /// Builds a set from a validated bit representation. pub(crate) const fn from_bits(bits: u8) -> Option { if bits != 0 && bits & !0b1111 == 0 { @@ -105,7 +100,6 @@ pub(crate) struct CarrierRequest { attempt: Option, failure: Option, user_agent_hash: [u8; 32], - initial_only: bool, } impl CarrierRequest { @@ -117,19 +111,17 @@ impl CarrierRequest { attempt: None, failure: None, user_agent_hash, - initial_only: false, } } - /// Constructs a known fixed-capability client without retry negotiation. + /// Constructs a metadata-free native client without inferring capabilities. pub(crate) const fn ios(user_agent_hash: [u8; 32]) -> Self { Self { class: CarrierClientClass::Ios, - capabilities: Some(CarrierCapabilities::ios()), + capabilities: None, attempt: None, failure: None, user_agent_hash, - initial_only: true, } } @@ -147,13 +139,12 @@ impl CarrierRequest { attempt: Some(attempt), failure, user_agent_hash, - initial_only: false, } } /// Returns whether this request participates in server-side negotiation. pub(crate) const fn is_automatic(self) -> bool { - self.capabilities.is_some() && !self.initial_only + self.capabilities.is_some() } /// Returns whether server capability filtering applies to this request. @@ -194,7 +185,6 @@ impl CarrierRequest { self.class == other.class && self.capabilities_bits() == other.capabilities_bits() && self.user_agent_hash == other.user_agent_hash - && self.initial_only == other.initial_only } /// Checks the complete idempotent identity of one exact attempt request. @@ -209,6 +199,21 @@ impl CarrierRequest { } } +/// Returns the cumulative deadline slot assigned to one carrier attempt. +pub(super) const fn carrier_attempt_deadline_index( + candidate_count: u8, + attempt: u8, +) -> Option { + if candidate_count == 0 || candidate_count > 4 || attempt == 0 || attempt > candidate_count { + return None; + } + if attempt == candidate_count { + Some(3) + } else { + Some((attempt - 1) as usize) + } +} + /// Secret-independent evidence owner frozen into an automatic session. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) struct CarrierLearningContext { @@ -225,3 +230,29 @@ pub(crate) struct CarrierLearningContext { /// Whether the authoritative client address is safe to use as learning evidence. pub(crate) ip_learning_eligible: bool, } + +#[cfg(test)] +mod tests { + use super::carrier_attempt_deadline_index; + + #[test] + fn final_candidate_uses_the_final_cumulative_deadline_slot() { + assert_eq!(carrier_attempt_deadline_index(1, 1), Some(3)); + assert_eq!(carrier_attempt_deadline_index(2, 1), Some(0)); + assert_eq!(carrier_attempt_deadline_index(2, 2), Some(3)); + assert_eq!(carrier_attempt_deadline_index(3, 1), Some(0)); + assert_eq!(carrier_attempt_deadline_index(3, 2), Some(1)); + assert_eq!(carrier_attempt_deadline_index(3, 3), Some(3)); + assert_eq!(carrier_attempt_deadline_index(4, 1), Some(0)); + assert_eq!(carrier_attempt_deadline_index(4, 2), Some(1)); + assert_eq!(carrier_attempt_deadline_index(4, 3), Some(2)); + assert_eq!(carrier_attempt_deadline_index(4, 4), Some(3)); + } + + #[test] + fn invalid_candidate_or_attempt_counts_have_no_deadline_slot() { + for (candidate_count, attempt) in [(0, 1), (5, 1), (1, 0), (1, 2), (3, 4)] { + assert_eq!(carrier_attempt_deadline_index(candidate_count, attempt), None); + } + } +} diff --git a/src/web/manager/session_creation.rs b/src/web/manager/session_creation.rs index 0786385..24cdb03 100644 --- a/src/web/manager/session_creation.rs +++ b/src/web/manager/session_creation.rs @@ -11,6 +11,7 @@ use super::state::{ CarrierChainPhase, decrement_map, matching_profile, new_unique_token, profile_key, remember_closed_token_locked, remove_expired_locked, }; +use super::negotiation::carrier_attempt_deadline_index; use super::session_admission::admit_initial; use super::{ CarrierLearningContext, CarrierRequest, CreateResult, ManagerError, TokenHash, WebProcessRuntime, @@ -154,7 +155,10 @@ impl WebProcessRuntime { else { return Err(ManagerError::Protocol); }; - let deadline_index = usize::from(next_attempt.saturating_sub(2)); + let candidate_count = u8::try_from(entry.carrier_candidates.len()) + .map_err(|_| ManagerError::Protocol)?; + let deadline_index = carrier_attempt_deadline_index(candidate_count, next_attempt) + .ok_or(ManagerError::Protocol)?; if entry.carrier_started_at.is_some_and(|started| { now.saturating_duration_since(started) >= Duration::from_secs(