diff --git a/src/api/config_edit/tests.rs b/src/api/config_edit/tests.rs index fb2f9a4..e5dbcf7 100644 --- a/src/api/config_edit/tests.rs +++ b/src/api/config_edit/tests.rs @@ -1,5 +1,54 @@ use super::*; +#[tokio::test] +async fn carrier_method_api_defaults_and_patches_are_hot() { + let (path, _directory) = temp_config("[web]\nenabled = false\n"); + let (value, _) = read_managed_config(&path).await.unwrap(); + assert_eq!(value["web"]["carrier_method"].as_str(), Some("post")); + for token in ["put", "post"] { + let active = ProxyConfig::load(&path).unwrap(); + let patch = serde_json::json!({"web": {"carrier_method": token}}); + let mut response = apply_patch_to_path(&path, &patch, None).await.unwrap(); + let desired = ProxyConfig::load(&path).unwrap(); + reconcile_runtime_effect(&mut response, &active, &desired).unwrap(); + assert!(!response.restart_required); + assert!(response.runtime_reload_required); + assert!(!response.process_restart_required); + assert!(response.deferred_process_fields.is_empty()); + let (value, revision) = read_managed_config(&path).await.unwrap(); + assert_eq!(value["web"]["carrier_method"].as_str(), Some(token)); + assert_eq!(revision, response.revision); + let written = tokio::fs::read_to_string(&path).await.unwrap(); + assert!(written.contains(&format!("carrier_method = \"{token}\""))); + } +} + +#[tokio::test] +async fn carrier_method_api_rejects_invalid_values_without_writing() { + let (path, _directory) = temp_config("[web]\ncarrier_method = \"put\"\n"); + let original = tokio::fs::read(&path).await.unwrap(); + let revision = crate::api::config_store::current_revision(&path) + .await + .unwrap(); + for value in [ + serde_json::json!("PUT"), + serde_json::json!("patch"), + serde_json::json!(true), + serde_json::json!(42), + ] { + let patch = serde_json::json!({"web": {"carrier_method": value}}); + let error = apply_patch_to_path(&path, &patch, None).await.unwrap_err(); + assert_eq!(error.status, hyper::StatusCode::BAD_REQUEST); + assert_eq!(tokio::fs::read(&path).await.unwrap(), original); + assert_eq!( + crate::api::config_store::current_revision(&path) + .await + .unwrap(), + revision + ); + } +} + #[test] fn json_object_converts_to_toml_table() { let j: Json = serde_json::json!({"censorship": {"tls_domain": "a.com"}, "default_dc": 2}); diff --git a/src/config/hot_reload.rs b/src/config/hot_reload.rs index b01ac54..101eac0 100644 --- a/src/config/hot_reload.rs +++ b/src/config/hot_reload.rs @@ -17,6 +17,7 @@ //! | `network` | `dns_overrides` | Applied immediately | //! | `access` | All user/quota fields | Effective immediately | //! | `web` | Carrier, timing, and debug policy | Applied to newly issued sessions | +//! | `web` | `carrier_method` | Applied to newly rendered bridge pages | //! Fields that require re-binding sockets (`server.listeners`, legacy //! `server.port`, `censorship.*`, `network.*`, `use_middle_proxy`) are **not** //! applied; a warning is emitted. SYN limiter rules are process-owned and are @@ -67,3 +68,7 @@ use watcher::{ReloadState, reload_config}; mod base_path_tests; #[cfg(test)] mod tests; +// Carrier method reloads preserve page-owned requests and process-owned limits. +#[cfg(test)] +#[path = "hot_reload/carrier_method_tests.rs"] +mod carrier_method_tests; diff --git a/src/config/hot_reload/carrier_method_tests.rs b/src/config/hot_reload/carrier_method_tests.rs new file mode 100644 index 0000000..0683060 --- /dev/null +++ b/src/config/hot_reload/carrier_method_tests.rs @@ -0,0 +1,50 @@ +use super::*; + +use crate::config::WebCarrierMethod; + +#[test] +fn carrier_method_reload_is_hot_and_preserves_process_limits() { + let mut active = ProxyConfig::default(); + for method in [WebCarrierMethod::Put, WebCarrierMethod::Post] { + let mut desired = active.clone(); + desired.web.carrier_method = method; + assert_eq!(classify_config_changes(&active, &desired).changed, ["web"]); + assert!(!classify_config_changes(&active, &desired).restart_required); + desired.web.limits.max_http_connections += 1; + let applied = overlay_hot_fields(&active, &desired); + assert_eq!(applied.web.carrier_method, method); + assert_eq!( + applied.web.limits.max_http_connections, + active.web.limits.max_http_connections + ); + active = applied; + } +} + +#[test] +fn carrier_method_reload_publishes_both_directions_and_keeps_last_good_value() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.toml"); + let source = |method| format!("[web]\ncarrier_method = \"{method}\"\n"); + std::fs::write(&path, source("post")).unwrap(); + let initial = Arc::new(ProxyConfig::load(&path).unwrap()); + let initial_hash = ProxyConfig::load_with_metadata(&path) + .unwrap() + .rendered_hash; + let (config_tx, _config_rx) = watch::channel(Arc::clone(&initial)); + let (log_tx, _log_rx) = watch::channel(initial.general.log_level.clone()); + let mut reload_state = ReloadState::new(Some(initial_hash)); + for (token, method) in [ + ("put", WebCarrierMethod::Put), + ("post", WebCarrierMethod::Post), + ] { + std::fs::write(&path, source(token)).unwrap(); + reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap(); + let applied = config_tx.borrow().clone(); + assert_eq!(applied.web.carrier_method, method); + std::fs::write(&path, source("PATCH")).unwrap(); + reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state); + let unchanged = config_tx.borrow().clone(); + assert!(Arc::ptr_eq(&unchanged, &applied)); + } +} diff --git a/src/config/load/strict_keys.rs b/src/config/load/strict_keys.rs index 3040518..6b5fa66 100644 --- a/src/config/load/strict_keys.rs +++ b/src/config/load/strict_keys.rs @@ -263,6 +263,7 @@ const LISTENER_CONFIG_KEYS: &[&str] = &[ const WEB_CONFIG_KEYS: &[&str] = &[ "enabled", "carrier", + "carrier_method", "carriers", "carrier_learning", "carrier_negotiation_aggressiveness", diff --git a/src/config/tests/load_basic_tests/web_tests.rs b/src/config/tests/load_basic_tests/web_tests.rs index b0a071d..1191676 100644 --- a/src/config/tests/load_basic_tests/web_tests.rs +++ b/src/config/tests/load_basic_tests/web_tests.rs @@ -3,6 +3,10 @@ use super::*; #[path = "web_tests/base_path_tests.rs"] mod base_path_tests; +// Method selection must not alter capability identity or accepted config literals. +#[path = "web_tests/carrier_method_tests.rs"] +mod carrier_method_tests; + const WEB_CONFIG: &str = r#" [access.users] alice = "000102030405060708090a0b0c0d0e0f" diff --git a/src/config/tests/load_basic_tests/web_tests/carrier_method_tests.rs b/src/config/tests/load_basic_tests/web_tests/carrier_method_tests.rs new file mode 100644 index 0000000..bc75dd6 --- /dev/null +++ b/src/config/tests/load_basic_tests/web_tests/carrier_method_tests.rs @@ -0,0 +1,48 @@ +use super::*; + +#[test] +fn carrier_method_defaults_and_roundtrips_without_changing_capabilities() { + let default = load_config_from_temp_toml(WEB_CONFIG); + assert_eq!(default.web.carrier_method, WebCarrierMethod::Post); + let capabilities = default.web.runtime.as_ref().unwrap().capabilities.clone(); + for (token, method) in [ + ("post", WebCarrierMethod::Post), + ("put", WebCarrierMethod::Put), + ] { + let configured = WEB_CONFIG.replace( + "carrier = \"https-lanes\"", + &format!("carrier = \"https-lanes\"\ncarrier_method = \"{token}\""), + ); + let source = format!("[general]\nconfig_strict = true\n{configured}"); + let config = load_config_from_temp_toml(&source); + assert_eq!(config.web.carrier_method, method); + assert_eq!( + config.web.runtime.as_ref().unwrap().capabilities, + capabilities + ); + let json = serde_json::to_value(&config.web).unwrap(); + assert_eq!(json["carrier_method"], token); + let decoded: WebConfig = serde_json::from_value(json).unwrap(); + assert_eq!(decoded.carrier_method, method); + let serialized = toml::to_string(&config.web).unwrap(); + let decoded: WebConfig = toml::from_str(&serialized).unwrap(); + assert_eq!(decoded.carrier_method, method); + } +} + +#[test] +fn carrier_method_rejects_unknown_tokens_types_and_aliases() { + for value in ["\"POST\"", "\"PUT\"", "\"patch\"", "true", "42", "[]"] { + let source = WEB_CONFIG.replace( + "carrier = \"https-lanes\"", + &format!("carrier = \"https-lanes\"\ncarrier_method = {value}"), + ); + assert!(load_config_error_from_temp_toml(&source).contains("carrier_method")); + } + let configured = WEB_CONFIG.replace( + "carrier = \"https-lanes\"", + "carrier = \"https-lanes\"\nhttp_method = \"put\"", + ); + let source = format!("[general]\nconfig_strict = true\n{configured}"); + assert!(load_config_error_from_temp_toml(&source).contains("http_method")); +} diff --git a/src/config/types.rs b/src/config/types.rs index e769ef4..c4973ea 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -63,7 +63,7 @@ pub(crate) use web::{ }; pub(crate) use web_carrier::WEB_CARRIER_LEARNING_MIN_ENTRIES; #[allow(unused_imports)] -pub use web_carrier::{WebCarrier, WebCarriers}; +pub use web_carrier::{WebCarrier, WebCarrierMethod, WebCarriers}; pub(crate) use web_debug::web_debug_fits_limits; pub use web_debug::{WebDebugBodyCapture, WebDebugConfig}; diff --git a/src/config/types/web.rs b/src/config/types/web.rs index 3f88ce5..4f8cb5a 100644 --- a/src/config/types/web.rs +++ b/src/config/types/web.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use bytes::Bytes; use serde::{Deserialize, Serialize}; -use super::web_carrier::{WebCarrier, WebCarriers}; +use super::web_carrier::{WebCarrier, WebCarrierMethod, WebCarriers}; use super::web_debug::WebDebugConfig; // Serialized WEB defaults remain separate from the runtime data model. @@ -413,6 +413,9 @@ pub struct WebConfig { /// Sole carrier when negotiation is disabled and final fallback when enabled. #[serde(default)] pub carrier: WebCarrier, + /// Page-owned HTTP method for HTTPS uplink and downlink, including retries. + #[serde(default)] + pub carrier_method: WebCarrierMethod, /// Ordered carriers considered by server-side negotiation before the fallback carrier. #[serde(default)] pub carriers: WebCarriers, @@ -471,6 +474,7 @@ impl Default for WebConfig { Self { enabled: false, carrier: WebCarrier::default(), + carrier_method: WebCarrierMethod::default(), carriers: WebCarriers::default(), carrier_learning: default_web_carrier_learning(), carrier_negotiation_aggressiveness: WebCarrierNegotiationAggressiveness::default(), diff --git a/src/config/types/web_carrier.rs b/src/config/types/web_carrier.rs index 3dcee56..bdb45ce 100644 --- a/src/config/types/web_carrier.rs +++ b/src/config/types/web_carrier.rs @@ -3,6 +3,27 @@ use serde::{Deserialize, Serialize}; /// Minimum restart-owned entries required for one complete learning sample. pub(crate) const WEB_CARRIER_LEARNING_MIN_ENTRIES: usize = 3; +/// HTTP method emitted by newly rendered HTTPS carrier bridges. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum WebCarrierMethod { + /// Preserve the original bridge request method. + #[default] + Post, + /// Allow idempotent-method connection reuse in Linux WebKitGTK. + Put, +} + +impl WebCarrierMethod { + /// Returns the HTTP token embedded in the page-owned carrier policy. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Post => "POST", + Self::Put => "PUT", + } + } +} + /// Carrier selected for one newly issued WEB relay session. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] diff --git a/src/web/bridge.rs b/src/web/bridge.rs index 1281b00..5a57389 100644 --- a/src/web/bridge.rs +++ b/src/web/bridge.rs @@ -1,5 +1,6 @@ use base64::Engine as _; +use crate::config::WebCarrierMethod; use crate::crypto::SecureRandom; /// Browser security policy for the transient Telegram Desktop bridge page. @@ -34,6 +35,7 @@ pub(crate) fn render( reconnect_grace_secs: u64, carrier_probe_coalesce_ms: u64, bridge_diagnostics_enabled: bool, + carrier_method: WebCarrierMethod, rng: &SecureRandom, ) -> BridgePage { let mut nonce = [0u8; 18]; @@ -124,6 +126,7 @@ pub(crate) fn render( .replace("__HOST__", host) .replace("__BASE_PREFIX__", base_prefix) .replace("__BOOTSTRAP__", bootstrap) + .replace("__CARRIER_METHOD__", carrier_method.as_str()) .replace("__BATCH_LIMIT__", &batch_limit.to_string()) .replace("__QUEUE_LIMIT__", &queue_limit.to_string()) .replace("__QUEUE_ITEMS__", &queue_items.to_string()) diff --git a/src/web/bridge/runtime.js b/src/web/bridge/runtime.js index 78753ca..88991eb 100644 --- a/src/web/bridge/runtime.js +++ b/src/web/bridge/runtime.js @@ -7,6 +7,8 @@ const responseBody=globalThis.TelemtBridgeResponse;if(!responseBody)throw new Er const requestSupport=globalThis.TelemtBridgeRequest;if(!requestSupport)throw new Error('missing request runtime'); const bufferSupport=globalThis.TelemtBridgeBuffers;if(!bufferSupport)throw new Error('missing buffer runtime'); const recoverySupport=globalThis.TelemtBridgeRecovery;if(!recoverySupport)throw new Error('missing recovery runtime'); +// Keep the method page-owned so recovery and config rollback cannot change frozen retries. +const carrierMethod='__CARRIER_METHOD__'; let negotiationEnabled=__NEGOTIATION_ENABLED__,candidateCount=__CANDIDATE_COUNT__,candidateDeadlines=[__CARRIER_DEADLINES__]; let longPollMs=__LONG_POLL_SECS__*1000,bridgeRequestMs=__BRIDGE_REQUEST_SECS__*1000,bridgeRetryMs=__BRIDGE_RETRY_SECS__*1000; let bridgeRecoveryMs=__BRIDGE_RECOVERY_SECS__*1000,websocketOpenMs=__WEBSOCKET_OPEN_SECS__*1000,reconnectGraceMs=__RECONNECT_GRACE_SECS__*1000; @@ -226,7 +228,7 @@ async function createSession(epoch){ async function probeHttp(probe,laneID,epoch){ try{ const headers={'X-Up-Seq':'1'},token=sessionToken,controller=attemptController,body=probe.data;if(laneID!==null)headers['X-Lane-ID']=String(laneID); - const response=await request('/api/v1/up',options('POST',token,body,headers,controller.signal)); + const response=await request('/api/v1/up',options(carrierMethod,token,body,headers,controller.signal)); if(closed||epoch!==attemptEpoch)return if(response.status!==204){advanceCarrier('http',epoch);return} if(response.headers.get('X-Up-Ack')!=='1'){advanceCarrier('protocol',epoch);return} @@ -263,14 +265,14 @@ async function runUp(){ lease=takeBatch(upPending,null);upLease=lease;lease.controller=new AbortController();const sequence=String(upSequence),token=sessionToken; for(;;){ try{ - const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence},lease.controller.signal),null,1); + const response=await request('/api/v1/up',options(carrierMethod,token,lease.body,{'X-Up-Seq':sequence},lease.controller.signal),null,1); if(response.status!==204)throw failure('http','uplink rejected'); if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','uplink acknowledgement rejected'); break; }catch(error){ let replayed=false; const recovered=await recoverTransport(error,async(signal,remaining)=>{ - const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence},signal),remaining,2); + const response=await request('/api/v1/up',options(carrierMethod,token,lease.body,{'X-Up-Seq':sequence},signal),remaining,2); if(response.status!==204)throw failure('http','uplink replay rejected'); if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','uplink replay acknowledgement rejected'); replayed=true; @@ -343,7 +345,7 @@ async function poll(){ const token=sessionToken,cursor=downCursor; try{ pollController=new AbortController(); - const response=await request('/api/v1/down',options('POST',token,null,{'X-Down-Cursor':cursor},pollController.signal),null,1); + const response=await request('/api/v1/down',options(carrierMethod,token,null,{'X-Down-Cursor':cursor},pollController.signal),null,1); if(closed||sessionToken!==token)return; if(response.status===204){status('connected');continue} if(response.status!==200)throw failure('http','downlink rejected'); @@ -354,7 +356,7 @@ async function poll(){ }catch(error){ if(closed)return; const recovered=await recoverTransport(error,async(signal,remaining)=>{ - const response=await request('/api/v1/down',options('POST',token,null,{'X-Down-Cursor':cursor},signal),remaining,2); + const response=await request('/api/v1/down',options(carrierMethod,token,null,{'X-Down-Cursor':cursor},signal),remaining,2); if(response.status===204)return; if(response.status!==200||!response.body.byteLength||!response.headers.get('X-Down-Cursor'))throw failure('http','downlink replay rejected'); }); @@ -422,14 +424,14 @@ async function runLaneUp(lane){ const sequence=String(lane.sequence),laneID=String(lane.id),token=sessionToken; for(;;){ try{ - const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},lease.controller.signal),null,1); + const response=await request('/api/v1/up',options(carrierMethod,token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},lease.controller.signal),null,1); if(response.status!==204)throw failure('http','lane uplink rejected'); if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','lane uplink acknowledgement rejected'); break; }catch(error){ let replayed=false; const recovered=await recoverTransport(error,async(signal,remaining)=>{ - const response=await request('/api/v1/up',options('POST',token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},signal),remaining,2); + const response=await request('/api/v1/up',options(carrierMethod,token,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},signal),remaining,2); if(response.status!==204)throw failure('http','lane uplink replay rejected'); if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','lane uplink replay acknowledgement rejected'); replayed=true; @@ -451,7 +453,7 @@ async function pollLane(lane){ while(!closed&&sessionToken&&lanes.get(lane.id)===lane){ const controller=new AbortController(),laneID=String(lane.id),token=sessionToken,cursor=lane.cursor;lane.controller=controller; failedToken=token;failedCursor=cursor;failedLaneID=laneID; - const response=await request('/api/v1/down',options('POST',token,null,{'X-Down-Cursor':cursor,'X-Lane-ID':laneID},controller.signal),null,1); + const response=await request('/api/v1/down',options(carrierMethod,token,null,{'X-Down-Cursor':cursor,'X-Lane-ID':laneID},controller.signal),null,1); if(closed||sessionToken!==token||lanes.get(lane.id)!==lane)return; if(response.status===204){ if(response.headers.get('X-Lane-Closed')==='1'){finishLane(lane,false);return} @@ -467,7 +469,7 @@ async function pollLane(lane){ }catch(error){ if(!closed&&lanes.get(lane.id)===lane){ const recovered=await recoverTransport(error,async(signal,remaining)=>{ - const response=await request('/api/v1/down',options('POST',failedToken,null,{'X-Down-Cursor':failedCursor,'X-Lane-ID':failedLaneID},signal),remaining,2); + const response=await request('/api/v1/down',options(carrierMethod,failedToken,null,{'X-Down-Cursor':failedCursor,'X-Lane-ID':failedLaneID},signal),remaining,2); if(response.status===204)return; if(response.status!==200||!response.body.byteLength||!response.headers.get('X-Down-Cursor'))throw failure('http','lane downlink replay rejected'); }); diff --git a/src/web/bridge/tests.rs b/src/web/bridge/tests.rs index 7775d0d..9342b0a 100644 --- a/src/web/bridge/tests.rs +++ b/src/web/bridge/tests.rs @@ -1,5 +1,73 @@ use super::*; +#[test] +fn carrier_method_is_page_owned_and_used_by_every_https_request() { + for method in [WebCarrierMethod::Post, WebCarrierMethod::Put] { + let page = render( + "proxy.example.com", + "/telegram/web/", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + 2 * 1024 * 1024, + 32 * 1024 * 1024, + 16 * 1024, + 1024, + true, + 4, + [3, 5, 8, 12], + 25, + 10, + 90, + 15, + 15, + 120, + 0, + true, + method, + &SecureRandom::new(), + ); + assert!(!page.body.contains("__")); + assert!( + page.body + .contains(&format!("const carrierMethod='{}';", method.as_str())) + ); + assert_eq!(page.body.matches("carrierMethod=").count(), 1); + assert_eq!(page.body.matches("options(carrierMethod,").count(), 9); + let requests: Vec<_> = page + .body + .lines() + .filter(|line| { + line.contains("request('/api/v1/up'") || line.contains("request('/api/v1/down'") + }) + .collect(); + assert_eq!(requests.len(), 9); + assert!( + requests + .iter() + .all(|line| line.contains("options(carrierMethod,")) + ); + assert_eq!(page.body.matches("options('POST',bootstrap,").count(), 2); + assert!( + page.body + .contains("fetch(relayBase+'/api/v1/diagnostic',{method:'POST'") + ); + assert!( + page.body + .contains("options('DELETE',token,null,headers,undefined,true)") + ); + assert!( + page.body + .contains("method:'GET',signal:requestController.signal") + ); + assert!( + page.body + .contains("exactKeys(value,['v','bootstrap','limits','timeouts','negotiation'])") + ); + assert!(!page.body.contains("policy.carrier_method")); + assert!(page.body.contains("port.postMessage({t:'status',state})")); + assert!(!page.body.contains("port.postMessage({t:'status',state,")); + } +} + fn render_page(bootstrap: &str, candidate_count: usize) -> BridgePage { render( "proxy.example.com", @@ -20,6 +88,7 @@ fn render_page(bootstrap: &str, candidate_count: usize) -> BridgePage { 120, 0, false, + WebCarrierMethod::Post, &SecureRandom::new(), ) } @@ -44,6 +113,7 @@ fn render_diagnostic_page(bootstrap: &str) -> BridgePage { 120, 0, true, + WebCarrierMethod::Post, &SecureRandom::new(), ) } @@ -95,6 +165,7 @@ fn rendered_page_resolves_carriers_against_the_exact_base_path() { 120, 0, true, + WebCarrierMethod::Post, &SecureRandom::new(), ); @@ -148,6 +219,7 @@ fn rendered_page_embeds_the_configured_bridge_timing_policy() { 119, 4, false, + WebCarrierMethod::Post, &SecureRandom::new(), ); @@ -201,6 +273,7 @@ fn disabled_negotiation_does_not_arm_a_carrier_deadline() { 120, 0, false, + WebCarrierMethod::Post, &SecureRandom::new(), ); assert!(page.body.contains( diff --git a/src/web/http.rs b/src/web/http.rs index 31a0181..9a24d5a 100644 --- a/src/web/http.rs +++ b/src/web/http.rs @@ -44,9 +44,9 @@ mod recovery; mod response; // Session creation and replacement negotiation remain separate from request routing. mod session; -// RFC 6455 upgrade validation and carrier drivers remain isolated from HTTP routing. #[cfg(test)] mod tests; +// RFC 6455 upgrade validation and carrier drivers remain isolated from HTTP routing. mod websocket; // Enabled-debug integration coverage remains separate from carrier behavior tests. #[cfg(test)] @@ -389,6 +389,7 @@ async fn handle_root( config.web.timeouts.reconnect_grace_secs, config.web.timeouts.carrier_probe_coalesce_ms, config.web.debug.bridge_diagnostics_enabled(), + config.web.carrier_method, &generation.rng, ); let mut response = full_response(StatusCode::OK, Bytes::from(page.body)); @@ -456,14 +457,13 @@ async fn handle_api( _ => serve_decoy(request, vhost, true, &runtime).await, } } - async fn handle_up( request: Request, runtime: Arc, vhost: Arc, token_hash: crate::web::manager::TokenHash, ) -> HttpResponse { - if request.method() != Method::POST || !binary_content_type(&request) { + if !matches!(*request.method(), Method::POST | Method::PUT) || !binary_content_type(&request) { return serve_decoy(request, vhost, true, &runtime).await; } let Some(sequence) = canonical_u64_header(&request, "x-up-seq").filter(|value| *value != 0) diff --git a/src/web/http/carrier_method_tests.rs b/src/web/http/carrier_method_tests.rs new file mode 100644 index 0000000..4ec4d62 --- /dev/null +++ b/src/web/http/carrier_method_tests.rs @@ -0,0 +1,480 @@ +use super::super::session_policy_tests::{open_keepalive, read_http_response}; +use super::super::*; +use crate::config::WebCarrierMethod; + +fn carrier_request(method: &str, path: &str, token: &str, headers: &str, body: &[u8]) -> Vec { + let mut bytes = format!( + "{method} {path} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.60\r\nAuthorization: Bearer {token}\r\n{headers}Content-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ) + .into_bytes(); + bytes.extend_from_slice(body); + bytes +} + +async fn bridge_page( + listener: &TcpListener, + runtime: &Arc, + capability: [u8; 32], + base: &str, +) -> Vec { + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability); + request( + listener, + runtime, + format!( + "GET {base}?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.60\r\nConnection: close\r\n\r\n" + ) + .into_bytes(), + ) + .await +} + +fn bootstrap_from(page: &[u8]) -> &str { + let (headers, body) = split_response(page); + assert!(headers.starts_with(b"HTTP/1.1 200")); + std::str::from_utf8(body) + .unwrap() + .split_once("bootstrap=\"") + .unwrap() + .1 + .split_once('"') + .unwrap() + .0 +} + +async fn create_session( + listener: &TcpListener, + runtime: &Arc, + bootstrap: &str, + base: &str, +) -> String { + let hello = frame::encode(FrameType::Hello, 0, &[1]); + let bytes = carrier_request( + "POST", + &format!("{base}api/v1/session"), + bootstrap, + "Content-Type: application/octet-stream\r\n", + &hello, + ); + let response = request(listener, runtime, bytes).await; + let (headers, _) = split_response(&response); + assert!(headers.starts_with(b"HTTP/1.1 200")); + response_header(headers, "x-session-token").to_string() +} + +async fn session_token( + listener: &TcpListener, + runtime: &Arc, + capability: [u8; 32], + base: &str, +) -> String { + let page = bridge_page(listener, runtime, capability, base).await; + create_session(listener, runtime, bootstrap_from(&page), base).await +} + +async fn assert_put_accepted(path: &str) { + for (carrier, base, method) in [WebCarrier::Https, WebCarrier::HttpsLanes] + .into_iter() + .flat_map(|carrier| { + ["/", "/telegram/web/"].into_iter().flat_map(move |base| { + [WebCarrierMethod::Post, WebCarrierMethod::Put] + .map(|method| (carrier, base, method)) + }) + }) + { + let capability = [60; 32]; + let mut config = runtime_config_with_base(capability, carrier, base); + config.web.carrier_method = method; + config.web.timeouts.long_poll_secs = 0; + let generation = test_runtime_generation(1, config); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let token = session_token(&listener, &runtime, capability, base).await; + let lane = if carrier.uses_lanes() { + "X-Lane-ID: 0\r\n" + } else { + "" + }; + let pong = frame::encode(FrameType::Pong, 0, &[]); + let (headers, body) = if path.ends_with("/up") { + ( + format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}"), + pong.as_ref(), + ) + } else { + (format!("X-Down-Cursor: 0\r\n{lane}"), &[][..]) + }; + let path = format!("{base}{}", path.trim_start_matches('/')); + let bytes = carrier_request("PUT", &path, &token, &headers, body); + let response = request(&listener, &runtime, bytes).await; + assert!( + response.starts_with(b"HTTP/1.1 204"), + "{carrier:?}: {}", + String::from_utf8_lossy(&response) + ); + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} + +#[tokio::test] +async fn put_uplink_is_accepted_for_both_https_carriers() { + assert_put_accepted("/api/v1/up").await; +} + +#[tokio::test] +async fn put_downlink_is_accepted_for_both_https_carriers() { + assert_put_accepted("/api/v1/down").await; +} + +#[tokio::test] +async fn carrier_method_mixed_retries_share_sequence_and_cursor_state() { + for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] { + for method in [WebCarrierMethod::Post, WebCarrierMethod::Put] { + let capability = [61; 32]; + let mut config = runtime_config(capability, carrier); + config.web.carrier_method = method; + config.web.timeouts.long_poll_secs = 0; + // Exhausted admission queues a deterministic CLOSE without starting a relay task. + config.web.limits.max_streams_global = 0; + let generation = test_runtime_generation(1, config); + let runtime = + WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let token = session_token(&listener, &runtime, capability, "/").await; + for (index, stream_id) in [7, 8].into_iter().enumerate() { + let lane = if carrier.uses_lanes() { + format!("X-Lane-ID: {stream_id}\r\n") + } else { + String::new() + }; + let sequence = if carrier.uses_lanes() { 1 } else { index + 1 }; + let cursor = if carrier.uses_lanes() { 0 } else { index }; + let open = frame::encode(FrameType::Open, stream_id, &[]); + let up_headers = format!( + "Content-Type: application/octet-stream\r\nX-Up-Seq: {sequence}\r\n{lane}" + ); + let methods = if index == 0 { + ["POST", "PUT"] + } else { + ["PUT", "POST"] + }; + for verb in methods { + let bytes = carrier_request(verb, "/api/v1/up", &token, &up_headers, &open); + let response = request(&listener, &runtime, bytes).await; + let (headers, body) = split_response(&response); + assert!(headers.starts_with(b"HTTP/1.1 204")); + assert_eq!(response_header(headers, "x-up-ack"), sequence.to_string()); + assert!(body.is_empty()); + } + for verb in methods { + let headers = format!("X-Down-Cursor: {cursor}\r\n{lane}"); + let bytes = carrier_request(verb, "/api/v1/down", &token, &headers, &[]); + let response = request(&listener, &runtime, bytes).await; + let (headers, body) = split_response(&response); + assert!(headers.starts_with(b"HTTP/1.1 200")); + assert_eq!( + response_header(headers, "x-down-cursor"), + (cursor + 1).to_string() + ); + assert_eq!(body, frame::encode(FrameType::Close, stream_id, &[])); + } + let headers = format!("X-Down-Cursor: {}\r\n{lane}", cursor + 1); + let bytes = carrier_request("PUT", "/api/v1/down", &token, &headers, &[]); + let response = request(&listener, &runtime, bytes).await; + assert!(response.starts_with(b"HTTP/1.1 204")); + } + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } + } +} + +#[tokio::test] +async fn carrier_method_changed_body_retry_keeps_protocol_failure() { + for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] { + let capability = [62; 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 token = session_token(&listener, &runtime, capability, "/").await; + let lane = if carrier.uses_lanes() { + "X-Lane-ID: 0\r\n" + } else { + "" + }; + let headers = format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}"); + let pong = frame::encode(FrameType::Pong, 0, &[]); + let bytes = carrier_request("POST", "/api/v1/up", &token, &headers, &pong); + let response = request(&listener, &runtime, bytes).await; + assert!(response.starts_with(b"HTTP/1.1 204")); + let mut changed = pong.to_vec(); + changed.extend_from_slice(&pong); + let bytes = carrier_request("PUT", "/api/v1/up", &token, &headers, &changed); + let response = request(&listener, &runtime, bytes).await; + assert_private_decoy(&response); + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} + +fn assert_private_decoy(response: &[u8]) { + let (headers, body) = split_response(response); + assert!(headers.starts_with(b"HTTP/1.1 404")); + assert_eq!(response_header(headers, "cache-control"), "no-store"); + assert_eq!(body, b"not found\n"); + let headers = std::str::from_utf8(headers).unwrap().to_ascii_lowercase(); + for name in ["x-up-ack:", "x-down-cursor:", "x-session-token:", "allow:"] { + assert!(!headers.contains(name)); + } +} + +#[tokio::test] +async fn carrier_method_put_preserves_authenticated_request_shape_checks() { + for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] { + let capability = [63; 32]; + let base = "/telegram/web/"; + let generation = + test_runtime_generation(1, runtime_config_with_base(capability, carrier, base)); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let token = session_token(&listener, &runtime, capability, base).await; + let pong = frame::encode(FrameType::Pong, 0, &[]); + let lane = if carrier.uses_lanes() { + "X-Lane-ID: 0\r\n" + } else { + "" + }; + let up = format!("{base}api/v1/up"); + let down = format!("{base}api/v1/down"); + let up_headers = format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}"); + let down_headers = format!("X-Down-Cursor: 0\r\n{lane}"); + let mut invalid = Vec::new(); + for media in [ + "", + "Content-Type: text/plain\r\n", + "Content-Type: application/octet-stream\r\nContent-Type: application/octet-stream\r\n", + ] { + let headers = format!("{media}X-Up-Seq: 1\r\n{lane}"); + invalid.push(carrier_request("PUT", &up, &token, &headers, &pong)); + } + for sequence in ["0", "01", ""] { + let sequence = if sequence.is_empty() { + String::new() + } else { + format!("X-Up-Seq: {sequence}\r\n") + }; + let headers = format!("Content-Type: application/octet-stream\r\n{sequence}{lane}"); + invalid.push(carrier_request("PUT", &up, &token, &headers, &pong)); + } + let invalid_lane = if carrier.uses_lanes() { + "" + } else { + "X-Lane-ID: 0\r\n" + }; + let headers = + format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{invalid_lane}"); + invalid.push(carrier_request("PUT", &up, &token, &headers, &pong)); + let headers = format!("{down_headers}Content-Type: application/octet-stream\r\n"); + invalid.push(carrier_request("PUT", &down, &token, &headers, &[])); + invalid.push(carrier_request("PUT", &down, &token, &down_headers, &[0])); + let headers = format!("X-Down-Cursor: 01\r\n{lane}"); + invalid.push(carrier_request("PUT", &down, &token, &headers, &[])); + for verb in ["GET", "PATCH"] { + invalid.push(carrier_request(verb, &up, &token, &up_headers, &pong)); + invalid.push(carrier_request(verb, &down, &token, &down_headers, &[])); + } + for path in [ + "/api/v1/up", + "/telegram/web/api/v1/up/", + "/telegram/web/api/v1/up?q=1", + "/telegram/web/api/v1//up", + "/telegram/web/api/v1/%75p", + ] { + invalid.push(carrier_request("PUT", path, &token, &up_headers, &pong)); + } + let wrong_host = String::from_utf8(carrier_request("PUT", &up, &token, &up_headers, &pong)) + .unwrap() + .replace("Host: proxy.example.com", "Host: other.example.com") + .into_bytes(); + invalid.push(wrong_host); + let path = format!("{base}api/v1/session"); + invalid.push(carrier_request("PUT", &path, &token, &up_headers, &pong)); + for bytes in invalid { + let response = request(&listener, &runtime, bytes).await; + assert_private_decoy(&response); + } + // Rejected shapes must not consume the first sequence or close the valid session. + let bytes = carrier_request("PUT", &up, &token, &up_headers, &pong); + let response = request(&listener, &runtime, bytes).await; + assert!(response.starts_with(b"HTTP/1.1 204")); + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} + +#[tokio::test] +async fn carrier_method_post_and_put_reuse_one_private_http_connection() { + for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] { + let capability = [64; 32]; + let mut config = runtime_config(capability, carrier); + config.web.timeouts.long_poll_secs = 0; + config.web.limits.max_streams_global = 0; + let generation = test_runtime_generation(1, config); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let token = session_token(&listener, &runtime, capability, "/").await; + let (mut client, cancellation, task) = open_keepalive(&listener, &runtime).await; + let lane = if carrier.uses_lanes() { + "X-Lane-ID: 7\r\n" + } else { + "" + }; + let open = frame::encode(FrameType::Open, 7, &[]); + let close = frame::encode(FrameType::Close, 7, &[]); + for verb in ["POST", "PUT", "PUT", "POST"] { + for (path, headers, body, ack_header) in [ + ( + "/api/v1/up", + format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}"), + open.as_ref(), + "x-up-ack", + ), + ( + "/api/v1/down", + format!("X-Down-Cursor: 0\r\n{lane}"), + &[][..], + "x-down-cursor", + ), + ] { + let bytes = carrier_request(verb, path, &token, &headers, body); + let bytes = String::from_utf8(bytes) + .unwrap() + .replace("Connection: close\r\n", "Connection: keep-alive\r\n") + .into_bytes(); + client.write_all(&bytes).await.unwrap(); + let response = tokio::time::timeout( + std::time::Duration::from_secs(5), + read_http_response(&mut client), + ) + .await + .unwrap(); + let (headers, body) = split_response(&response); + assert_eq!(response_header(headers, ack_header), "1"); + assert!( + !std::str::from_utf8(headers) + .unwrap() + .to_ascii_lowercase() + .contains("connection: close") + ); + if path.ends_with("/up") { + assert!(headers.starts_with(b"HTTP/1.1 204")); + assert!(body.is_empty()); + } else { + assert!(headers.starts_with(b"HTTP/1.1 200")); + assert_eq!( + response_header(headers, "content-length"), + close.len().to_string() + ); + assert_eq!(body, close); + } + } + } + cancellation.cancel(); + task.await.unwrap(); + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} + +#[tokio::test] +async fn carrier_method_reload_preserves_old_pages_bootstraps_sessions_and_recovery() { + for carrier in [WebCarrier::Https, WebCarrier::HttpsLanes] { + let capability = [65; 32]; + let mut config = runtime_config(capability, carrier); + config.web.limits.max_bootstraps_per_ip = 6; + let generation = test_runtime_generation(1, config.clone()); + 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 post_page = bridge_page(&listener, &runtime, capability, "/").await; + let old_session = + create_session(&listener, &runtime, bootstrap_from(&post_page), "/").await; + let unused_post_page = bridge_page(&listener, &runtime, capability, "/").await; + config.web.carrier_method = WebCarrierMethod::Put; + let put_generation = test_runtime_generation(2, config.clone()); + active_runtime.store(Arc::clone(&put_generation)); + let post_session = + create_session(&listener, &runtime, bootstrap_from(&unused_post_page), "/").await; + let put_page = bridge_page(&listener, &runtime, capability, "/").await; + assert!(String::from_utf8_lossy(&put_page).contains("const carrierMethod='PUT';")); + config.web.carrier_method = WebCarrierMethod::Post; + let rollback_generation = test_runtime_generation(3, config); + active_runtime.store(Arc::clone(&rollback_generation)); + let put_session = create_session(&listener, &runtime, bootstrap_from(&put_page), "/").await; + let rollback_page = bridge_page(&listener, &runtime, capability, "/").await; + assert!(String::from_utf8_lossy(&rollback_page).contains("const carrierMethod='POST';")); + assert!(String::from_utf8_lossy(&post_page).contains("const carrierMethod='POST';")); + let lane = if carrier.uses_lanes() { + "X-Lane-ID: 0\r\n" + } else { + "" + }; + let headers = format!("Content-Type: application/octet-stream\r\nX-Up-Seq: 1\r\n{lane}"); + let pong = frame::encode(FrameType::Pong, 0, &[]); + for token in [&old_session, &post_session, &put_session] { + for verb in ["PUT", "POST"] { + let bytes = carrier_request(verb, "/api/v1/up", token, &headers, &pong); + let response = request(&listener, &runtime, bytes).await; + assert!(response.starts_with(b"HTTP/1.1 204")); + } + } + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability); + let bytes = carrier_request( + "GET", + &format!("/?bridge={encoded}"), + &put_session, + "Accept: application/vnd.telemt.web-recovery+json\r\n", + &[], + ); + let recovery = request(&listener, &runtime, bytes).await; + let (recovery_headers, body) = split_response(&recovery); + assert!(recovery_headers.starts_with(b"HTTP/1.1 200")); + assert_eq!( + response_header(recovery_headers, "content-type"), + "application/vnd.telemt.web-recovery+json" + ); + let document: serde_json::Value = serde_json::from_slice(body).unwrap(); + let keys: std::collections::BTreeSet<_> = document + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect(); + assert_eq!( + keys, + ["v", "bootstrap", "limits", "timeouts", "negotiation"] + .into_iter() + .collect() + ); + let bootstrap = document["bootstrap"].as_str().unwrap(); + let recovered = create_session(&listener, &runtime, bootstrap, "/").await; + let bytes = carrier_request("PUT", "/api/v1/up", &recovered, &headers, &pong); + let response = request(&listener, &runtime, bytes).await; + assert!(response.starts_with(b"HTTP/1.1 204")); + let bytes = carrier_request("PUT", "/api/v1/up", &put_session, &headers, &pong); + let retired = request(&listener, &runtime, bytes).await; + assert_private_decoy(&retired); + runtime.shutdown().await; + for generation in [generation, put_generation, rollback_generation] { + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } + } +} diff --git a/src/web/http/down.rs b/src/web/http/down.rs index 8156efe..9f188fe 100644 --- a/src/web/http/down.rs +++ b/src/web/http/down.rs @@ -22,7 +22,7 @@ pub(super) async fn handle_down( vhost: Arc, token_hash: TokenHash, ) -> HttpResponse { - if request.method() != hyper::Method::POST + if !matches!(*request.method(), hyper::Method::POST | hyper::Method::PUT) || request.headers().contains_key(header::CONTENT_TYPE) { return serve_decoy(request, vhost, true, &runtime).await; diff --git a/src/web/http/legacy_tests.rs b/src/web/http/legacy_tests.rs index 9eb4fbf..4cef3a0 100644 --- a/src/web/http/legacy_tests.rs +++ b/src/web/http/legacy_tests.rs @@ -1,5 +1,9 @@ use super::*; +// HTTP method compatibility keeps platform workarounds outside session state. +#[path = "carrier_method_tests.rs"] +mod carrier_method_tests; + #[tokio::test] async fn windows_restricted_webview_empty_cookie_preserves_the_carrier_flow() { for (index, carrier) in [WebCarrier::Https, WebCarrier::HttpsLanes] diff --git a/src/web/http/session_policy_tests.rs b/src/web/http/session_policy_tests.rs index 5d52b6e..68e8c0b 100644 --- a/src/web/http/session_policy_tests.rs +++ b/src/web/http/session_policy_tests.rs @@ -1,6 +1,7 @@ use super::*; -async fn open_keepalive( +/// Opens one reusable private HTTP connection for integration scenarios. +pub(super) async fn open_keepalive( listener: &TcpListener, runtime: &Arc, ) -> (TcpStream, CancellationToken, tokio::task::JoinHandle<()>) { @@ -21,7 +22,8 @@ async fn open_keepalive( (client.unwrap(), cancellation, task) } -async fn read_http_response(client: &mut TcpStream) -> Vec { +/// Reads one length-delimited response without waiting for connection closure. +pub(super) async fn read_http_response(client: &mut TcpStream) -> Vec { let mut response = Vec::new(); while !response.ends_with(b"\r\n\r\n") { assert!(response.len() < 16 * 1024);