From a16792986fbc1b12af8dcf9f6a01eca36a5a4571 Mon Sep 17 00:00:00 2001 From: ospab Date: Thu, 25 Jun 2026 20:54:13 +0300 Subject: [PATCH] Fix client stability, metrics, and compiler warnings --- README.md | 8 +-- README.ru.md | 32 ++++----- ostp-client/src/runner.rs | 55 +++++++++++----- .../src/tunnel/inbounds/local_proxy.rs | 65 +++++++++++++++++-- ostp-client/src/tunnel/inbounds/tun.rs | 5 +- ostp-client/src/tunnel/outbounds/ostp.rs | 46 +++++++------ ostp-gui/src-tauri/Cargo.lock | 6 +- ostp-gui/src-tauri/src/lib.rs | 24 +++++-- ostp-server/src/lib.rs | 51 +++++++++++---- ostp-tun/src/sys_win_api.rs | 4 +- ostp-tun/src/windows.rs | 1 - ostp/src/main.rs | 40 ++++++------ 12 files changed, 226 insertions(+), 111 deletions(-) diff --git a/README.md b/README.md index b949af8..6f14745 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,9 @@ OSTP (Ospab Stealth Transport Protocol) is an encrypted transport protocol writt | **Multiplexed Streams**| Multiple logical TCP streams over a single encrypted UDP session with per-stream flow control. | | **Session Roaming** | Connection persistence across IP changes via session ID tracking. | | **UoT Mode** | UDP-over-TCP encapsulation with length-prefixing to bypass UDP blocking. | -| **Fallback Server** | TCP proxying to a legitimate web server to resist active probing. | +| **TCP Fragmentation** | (Zapret-style) Bypasses Deep Packet Inspection (DPI) by chunking the initial TLS/Noise handshakes. | +| **Junk Packets** | Sends randomized dummy UDP packets prior to the handshake to confuse DPI flow analyzers. | +| **Adaptive Padding** | Dynamically pads handshake and data frames up to 1024 bytes to prevent packet size fingerprinting. | | **TUN Mode** | Native network stack integration (`smoltcp`) for full-system routing without external dependencies. | | **Management API** | Built-in REST API for server administration, metrics, and key generation. | | **TURN Relay** | RFC 5766 TURN support for NAT traversal. | @@ -39,18 +41,16 @@ flowchart LR Apps[Local Apps] -->|SOCKS5 / TUN| CoreC subgraph Client [Client Node] - CoreC[OSTP Client] -.->|Encrypt & Mask| NetC[Transport Layer] + CoreC[OSTP Client] -.->|Encrypt, Pad & Chunk| NetC[Transport Layer] end NetC <==>|Encrypted UDP / UoT| NetS subgraph Server [Server Node] NetS[Transport Layer] -.->|Decrypt & Auth| CoreS[OSTP Server] - NetS -->|Unauthenticated| Fallback[Fallback Server] end CoreS -->|Relay| WWW((Internet)) - Fallback -->|Forward| Web((Web / NGINX)) ``` --- diff --git a/README.ru.md b/README.ru.md index 0582885..d34cb77 100644 --- a/README.ru.md +++ b/README.ru.md @@ -19,16 +19,18 @@ OSTP (Ospab Stealth Transport Protocol) — зашифрованный тран | Возможность | Описание | |-------------|----------| -| **Маскирование трафика** | Шифрование заголовков и полезной нагрузки с помощью HMAC ключей на каждый пакет. Трафик неотличим от шума. | -| **Noise Protocol** | `Noise_NNpsk0_25519_ChaChaPoly_BLAKE2s` — аутентификация через PSK, forward secrecy. | -| **Reliable UDP (ARQ)** | Selective ACK/NACK с rate-limited ретрансмиссией, настраиваемым reorder-буфером и exponential backoff. | -| **Мультиплексирование** | Несколько логических TCP-потоков поверх одной зашифрованной UDP-сессии с per-stream flow control. | -| **Session Roaming** | Сохранение соединения при смене IP-адреса благодаря отслеживанию по идентификатору сессии (session ID). | -| **Режим UoT** | Инкапсуляция UDP внутри TCP с указанием длины пакетов для обхода блокировок неизвестного UDP-трафика. | -| **Fallback Server** | Проксирование неаутентифицированных TCP подключений на веб-сервер для защиты от активного пробинга. | -| **TUN-режим** | Полносистемная маршрутизация через встроенный сетевой стек `smoltcp` без внешних зависимостей. | -| **Management API** | Встроенный REST API для администрирования сервера, сбора метрик и генерации ключей. | -| **TURN Relay** | Поддержка RFC 5766 TURN для обхода NAT. | +| **Маскировка Трафика** | Шифрование заголовков и данных уникальными ключами для каждого пакета. Выглядит как белый шум. | +| **Noise Protocol** | `Noise_NNpsk0_25519_ChaChaPoly_BLAKE2s` — обмен ключами с forward secrecy и PSK-аутентификацией. | +| **Надёжный UDP (ARQ)** | Выборочные ACK/NACK, лимитированные повторы, настраиваемый буфер и экспоненциальный бэкофф. | +| **Мультиплексирование** | Несколько логических TCP-стримов внутри одной UDP-сессии с по-стримовым контролем потока. | +| **Смена IP (Roaming)** | Сохранение соединений при переключении сетей благодаря трекингу по ID сессии. | +| **Режим UoT** | Упаковка UDP внутрь TCP-соединения для обхода блокировок UDP-трафика. | +| **TCP Фрагментация** | (В стиле Zapret) Обходит ТСПУ/DPI за счёт нарезки стартовых пакетов Noise/TLS на мелкие куски. | +| **Мусорные Пакеты** | Закидывает анализаторы DPI рандомными пакетами до начала рукопожатия, сбивая сигнатуры. | +| **Адаптивный Паддинг** | Добивает размер пакетов до 1024 байт, предотвращая детектирование по размерам пакетов. | +| **Режим TUN** | Нативная интеграция сетевого стека (`smoltcp`) для маршрутизации всего устройства без tun2socks. | +| **Management API** | Встроенный REST API для управления сервером, генерации ключей и сбора метрик. | +| **TURN Relay** | Поддержка стандарта RFC 5766 TURN для обхода NAT. | --- @@ -36,21 +38,19 @@ OSTP (Ospab Stealth Transport Protocol) — зашифрованный тран ```mermaid flowchart LR - Apps[Приложения] -->|SOCKS5 / TUN| CoreC + Apps[Локальные приложения] -->|SOCKS5 / TUN| CoreC subgraph Client [Клиент] - CoreC[OSTP Клиент] -.->|Шифрование| NetC[Транспортный уровень] + CoreC[OSTP Client] -.->|Шифрует и маскирует| NetC[Транспортный уровень] end NetC <==>|Зашифрованный UDP / UoT| NetS subgraph Server [Сервер] - NetS[Транспортный уровень] -.->|Дешифрование| CoreS[OSTP Сервер] - NetS -->|Неавторизованные| Fallback[Fallback Сервер] + NetS[Транспортный уровень] -.->|Расшифровывает| CoreS[OSTP Server] end - CoreS -->|Проксирование| WWW((Интернет)) - Fallback -->|Перенаправление| Web((Веб-сервер / NGINX)) + CoreS -->|Релей| WWW((Интернет)) ``` --- diff --git a/ostp-client/src/runner.rs b/ostp-client/src/runner.rs index df9a8bc..09d5d50 100644 --- a/ostp-client/src/runner.rs +++ b/ostp-client/src/runner.rs @@ -44,35 +44,58 @@ pub async fn run_client_core( let mut handles = Vec::new(); let metrics_ping = metrics.clone(); - let server_ip = config.outbounds.iter().find_map(|o| { + let server_addr = config.outbounds.iter().find_map(|o| { match o { - crate::config::OutboundConfig::Ostp { server, .. } => Some(server.clone()), - crate::config::OutboundConfig::Socks { server, .. } => Some(server.clone()), + crate::config::OutboundConfig::Ostp { server, port, .. } => Some((server.clone(), *port)), + crate::config::OutboundConfig::Socks { server, port, .. } => Some((server.clone(), *port)), _ => None, } }); - - if let Some(mut server) = server_ip { - if !server.contains(':') { - server.push_str(":443"); - } + + if let Some((host, port)) = server_addr { + // Probe the REAL server port. The OSTP server listens for UoT/TCP on the + // same port as UDP, so a plain TCP connect there confirms liveness. The + // old code hardcoded ":443" — which the server never listens on — so the + // probe failed every time and wrongly latched "reconnecting" forever even + // while the tunnel was carrying traffic (the button flickered to + // disconnected and counters appeared frozen). + let server = if host.contains(':') { host } else { format!("{host}:{port}") }; let mut shutdown_rx = shutdown_rx_ext.clone(); handles.push(tokio::spawn(async move { + // Health probe: the authoritative source of "connected". Probe the + // server immediately, then every 3s. A reachable server latches + // state=2 (even before any app traffic flows), and two consecutive + // failures drop it back to 1 (reconnecting). Per-connection dials must + // NOT drive this global state or the button flickers as connections + // open and close. + let mut consecutive_fail = 0u32; loop { + let start = std::time::Instant::now(); + let ok = matches!( + tokio::time::timeout( + std::time::Duration::from_secs(2), + tokio::net::TcpStream::connect(&server), + ) + .await, + Ok(Ok(_)) + ); + if ok { + let rtt = start.elapsed().as_millis() as u32; + metrics_ping.rtt_ms.store(rtt, Ordering::Relaxed); + metrics_ping.connection_state.store(2, Ordering::Relaxed); + consecutive_fail = 0; + } else { + consecutive_fail += 1; + if consecutive_fail >= 2 { + metrics_ping.connection_state.store(1, Ordering::Relaxed); + } + } tokio::select! { _ = tokio::time::sleep(std::time::Duration::from_secs(3)) => {} _ = shutdown_rx.changed() => { if *shutdown_rx.borrow() { break; } } } - let start = std::time::Instant::now(); - if let Ok(Ok(_)) = tokio::time::timeout( - std::time::Duration::from_secs(2), - tokio::net::TcpStream::connect(&server) - ).await { - let rtt = start.elapsed().as_millis() as u32; - metrics_ping.rtt_ms.store(rtt, Ordering::Relaxed); - } } })); } diff --git a/ostp-client/src/tunnel/inbounds/local_proxy.rs b/ostp-client/src/tunnel/inbounds/local_proxy.rs index 6e2def6..00731ba 100644 --- a/ostp-client/src/tunnel/inbounds/local_proxy.rs +++ b/ostp-client/src/tunnel/inbounds/local_proxy.rs @@ -4,9 +4,58 @@ use crate::config::{ClientConfig, InboundConfig}; use crate::tunnel::router::{Router, Session}; use crate::tunnel::outbounds::OutboundManager; use tokio::net::TcpListener; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncWrite, AsyncReadExt, AsyncWriteExt}; use tokio::sync::watch; +use portable_atomic::Ordering; + +struct MetricStream { + inner: T, + metrics: Arc, +} + +impl AsyncRead for MetricStream { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let filled_before = buf.filled().len(); + let res = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf); + if let std::task::Poll::Ready(Ok(())) = &res { + let filled_after = buf.filled().len(); + if filled_after > filled_before { + // local client read from remote (this means recv from tunnel) + self.metrics.bytes_recv.fetch_add((filled_after - filled_before) as u64, Ordering::Relaxed); + } + } + res + } +} + +impl AsyncWrite for MetricStream { + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let res = std::pin::Pin::new(&mut self.inner).poll_write(cx, buf); + if let std::task::Poll::Ready(Ok(n)) = &res { + // local client write to remote (this means sent to tunnel) + self.metrics.bytes_sent.fetch_add(*n as u64, Ordering::Relaxed); + } + res + } + + fn poll_flush(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + pub async fn run_socks_inbound( _config: ClientConfig, inbound_config: InboundConfig, @@ -16,7 +65,6 @@ pub async fn run_socks_inbound( metrics: Arc, is_primary: bool, ) -> Result<()> { - use portable_atomic::Ordering; let InboundConfig::LocalProxy { tag, protocol, listen, port, set_system_proxy } = inbound_config else { return Err(anyhow!("Invalid config for LocalProxy inbound")); }; @@ -57,13 +105,14 @@ pub async fn run_socks_inbound( let proto = protocol.clone(); let inbound_tag = tag.clone(); + let metrics_clone = metrics.clone(); tokio::spawn(async move { if proto == "socks" { - if let Err(e) = handle_socks5_connection(&mut stream, &rt, &om, &inbound_tag, client_addr).await { + if let Err(e) = handle_socks5_connection(&mut stream, &rt, &om, &inbound_tag, client_addr, metrics_clone).await { tracing::debug!("SOCKS5 handling error: {}", e); } } else if proto == "http" { - if let Err(e) = handle_http_connection(&mut stream, &rt, &om, &inbound_tag, client_addr).await { + if let Err(e) = handle_http_connection(&mut stream, &rt, &om, &inbound_tag, client_addr, metrics_clone).await { tracing::debug!("HTTP proxy handling error: {}", e); } } else { @@ -84,6 +133,7 @@ async fn handle_socks5_connection( outbound_manager: &Arc, inbound_tag: &str, client_addr: std::net::SocketAddr, + metrics: Arc, ) -> Result<()> { let mut buf = [0u8; 256]; @@ -153,7 +203,8 @@ async fn handle_socks5_connection( stream.write_all(&[0x05, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]).await?; // Forward data - tokio::io::copy_bidirectional(stream, &mut remote_stream).await?; + let mut metric_remote = MetricStream { inner: remote_stream, metrics }; + tokio::io::copy_bidirectional(stream, &mut metric_remote).await?; } Err(e) => { tracing::warn!("SOCKS5 TCP dial failed to {}: {}", outbound_tag, e); @@ -171,6 +222,7 @@ async fn handle_http_connection( outbound_manager: &Arc, inbound_tag: &str, client_addr: std::net::SocketAddr, + metrics: Arc, ) -> Result<()> { // Basic HTTP CONNECT implementation let mut buf = [0u8; 4096]; @@ -231,7 +283,8 @@ async fn handle_http_connection( remote_stream.write_all(&buf[0..n]).await?; } - tokio::io::copy_bidirectional(stream, &mut remote_stream).await?; + let mut metric_remote = MetricStream { inner: remote_stream, metrics }; + tokio::io::copy_bidirectional(stream, &mut metric_remote).await?; } Err(e) => { tracing::warn!("HTTP TCP dial failed to {}: {}", outbound_tag, e); diff --git a/ostp-client/src/tunnel/inbounds/tun.rs b/ostp-client/src/tunnel/inbounds/tun.rs index c3809de..cbf5d21 100644 --- a/ostp-client/src/tunnel/inbounds/tun.rs +++ b/ostp-client/src/tunnel/inbounds/tun.rs @@ -17,11 +17,12 @@ pub async fn run_tun_inbound( ) -> Result<()> { use netstack_smoltcp::StackBuilder; - use portable_atomic::Ordering; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use futures::{StreamExt, SinkExt}; + use portable_atomic::Ordering; - let InboundConfig::Tun { tag, auto_route, mtu, fd: _fd, .. } = inbound_config else { + #[allow(unused_variables)] + let InboundConfig::Tun { tag, auto_route, mtu, fd, .. } = inbound_config else { return Err(anyhow!("Invalid config for TUN inbound")); }; diff --git a/ostp-client/src/tunnel/outbounds/ostp.rs b/ostp-client/src/tunnel/outbounds/ostp.rs index 86d0749..e735f73 100644 --- a/ostp-client/src/tunnel/outbounds/ostp.rs +++ b/ostp-client/src/tunnel/outbounds/ostp.rs @@ -122,16 +122,14 @@ pub async fn dial_tcp( } if !handshake_success { + // A single proxied connection failing must NOT mark the whole tunnel + // as disconnected — global connection_state is owned by the health + // probe in run_client_core, not by per-target dials. tracing::warn!("TCP handshake failed or protocol machine error"); - if let Some(m) = &metrics { - m.connection_state.store(0, std::sync::atomic::Ordering::Relaxed); - } return; } - if let Some(m) = &metrics { - m.connection_state.store(2, std::sync::atomic::Ordering::Relaxed); - } + // The global health probe (in runner.rs) is the only authoritative source of connection state. // Send connection request let connect_msg = ostp_core::relay::RelayMessage::Connect(format!("{}:{}", target_host_str, target_port)); @@ -268,16 +266,26 @@ pub async fn handle_udp( let config = make_initiator_config(session_id, access_key, transport_cfg); let mut machine = ProtocolMachine::new(config)?; - // Send UDP Junk Packets (Amnezia style) to break DPI heuristics + // Amnezia-style junk to break DPI heuristics — but ONLY over stream + // transports (UoT/TCP), where it rides inside the connection. Over plain + // UDP each junk is a standalone datagram of random bytes that the server + // cannot tell from a port scan: it logs every one as an "Unauthorized + // probe", wastes CPU trying every key on it, and can trip the server's + // anti-probe defenses against this very client. The server is not + // coordinated to expect/discard junk (unlike AmneziaWG's Jc/Jmin/Jmax), so + // junk-over-UDP is pure self-inflicted noise. Gate it to stream transports. use rand::Rng; - let num_junk = rand::thread_rng().gen_range(2..=5); - for _ in 0..num_junk { - let junk_len = rand::thread_rng().gen_range(100..=1000); - let mut junk = vec![0u8; junk_len]; - rand::thread_rng().fill(&mut junk[..]); - let junk_bytes = bytes::Bytes::from(junk); - let _ = transport.send(&junk_bytes).await; - tokio::time::sleep(std::time::Duration::from_millis(5)).await; + let junk_enabled = matches!(transport_cfg.r#type.as_str(), "uot" | "tcp"); + if junk_enabled { + let num_junk = rand::thread_rng().gen_range(2..=5); + for _ in 0..num_junk { + let junk_len = rand::thread_rng().gen_range(100..=1000); + let mut junk = vec![0u8; junk_len]; + rand::thread_rng().fill(&mut junk[..]); + let junk_bytes = bytes::Bytes::from(junk); + let _ = transport.send(&junk_bytes).await; + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } } // Send handshake first @@ -293,15 +301,11 @@ pub async fn handle_udp( ).await { Ok(Ok(n)) => { let _ = machine.on_event(OstpEvent::Inbound(bytes::Bytes::copy_from_slice(&buf[..n]))); - if let Some(m) = &metrics { - m.connection_state.store(2, std::sync::atomic::Ordering::Relaxed); - } } _ => { + // Per-dial timeout: do not touch global connection_state (owned by the + // health probe). Just give up on this one target connection. tracing::warn!("OSTP handshake timeout for {}:{}", server, port); - if let Some(m) = &metrics { - m.connection_state.store(0, std::sync::atomic::Ordering::Relaxed); - } return Ok(()); } } diff --git a/ostp-gui/src-tauri/Cargo.lock b/ostp-gui/src-tauri/Cargo.lock index b4f9620..f36661a 100644 --- a/ostp-gui/src-tauri/Cargo.lock +++ b/ostp-gui/src-tauri/Cargo.lock @@ -2665,7 +2665,7 @@ dependencies = [ [[package]] name = "ostp-client" -version = "0.3.17" +version = "0.3.18" dependencies = [ "anyhow", "base64 0.22.1", @@ -2700,7 +2700,7 @@ dependencies = [ [[package]] name = "ostp-core" -version = "0.3.17" +version = "0.3.18" dependencies = [ "anyhow", "byteorder", @@ -2742,7 +2742,7 @@ dependencies = [ [[package]] name = "ostp-tun" -version = "0.3.17" +version = "0.3.18" dependencies = [ "anyhow", "libc", diff --git a/ostp-gui/src-tauri/src/lib.rs b/ostp-gui/src-tauri/src/lib.rs index d9490d9..7e2c7b4 100644 --- a/ostp-gui/src-tauri/src/lib.rs +++ b/ostp-gui/src-tauri/src/lib.rs @@ -420,14 +420,26 @@ async fn stop_tunnel(state: tauri::State<'_, AppState>) -> Result async fn start_tunnel(state: tauri::State<'_, AppState>, app: tauri::AppHandle) -> Result { let mut guard = state.0.lock().await; - if let Some(ref t) = guard.tunnel { - match t { - TunnelHandle::InProcess(s) if !s.handle.is_finished() => return Ok(true), - TunnelHandle::Helper(_) => return Ok(true), - _ => {} + // If a tunnel is already running (a UI/backend desync left a stale handle, or + // the user is switching servers), tear it down before starting a fresh one — + // otherwise we'd silently keep the old connection/server. start_tunnel is only + // ever invoked on an explicit connect (the UI calls it only while it believes + // it is disconnected), so restarting here is safe. + match guard.tunnel.take() { + Some(TunnelHandle::Helper(h)) => { + let stop_cmd = serde_json::json!({ "cmd": "stop", "token": h.token }).to_string(); + let _ = h.cmd_tx.send(format!("{}\n", stop_cmd)).await; + // Let the elevated helper stop the tunnel and release the TUN adapter + // before a new helper tries to create it (avoids ostp_tun name clashes). + tokio::time::sleep(std::time::Duration::from_millis(1200)).await; } + Some(TunnelHandle::InProcess(mut s)) => { + if let Some(tx) = s.shutdown_tx.take() { let _ = tx.send(true); } + s.handle.abort(); + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), s.handle).await; + } + None => {} } - guard.tunnel = None; let path = get_config_path(); let content = std::fs::read_to_string(&path).map_err(|e| e.to_string())?; diff --git a/ostp-server/src/lib.rs b/ostp-server/src/lib.rs index 5821532..9d2437c 100644 --- a/ostp-server/src/lib.rs +++ b/ostp-server/src/lib.rs @@ -70,6 +70,13 @@ pub(crate) struct RemoteState { // ── Public API ─────────────────────────────────────────────────────────────── +pub type ConnectRequest = ( + u32, + u16, + String, + Result<(tokio::net::tcp::OwnedWriteHalf, mpsc::Sender<()>), String>, +); + pub async fn run_server( bind_addrs: Vec, server_public_ip: Option, @@ -343,6 +350,13 @@ pub async fn run_server( // Headless event logger tokio::spawn(async move { + // Rate-limit unauthorized-probe logging: a single client dial sends + // several Amnezia-style junk packets, and a real DPI sweep can send + // far more. Log the first probe of each ~30s window immediately, then + // suppress the rest and emit a count — so the log stays readable and + // a genuine probe is never fully hidden. + let mut probe_window_start: Option = None; + let mut probe_suppressed: u64 = 0; while let Some(ev) = ui_event_rx.recv().await { match ev { UiEvent::Log(msg) => { @@ -360,8 +374,23 @@ pub async fn run_server( tracing::info!("Access key created: {key}"); } UiEvent::UnauthorizedProbe { peer, bytes, reason } => { - // Make it a warn so it's always visible outside debug mode! - tracing::warn!("Unauthorized probe from {peer} ({bytes} bytes): {reason}"); + let now = std::time::Instant::now(); + let elapsed = probe_window_start + .map(|s| now.duration_since(s)) + .unwrap_or(std::time::Duration::MAX); + if elapsed >= std::time::Duration::from_secs(30) { + if probe_suppressed > 0 { + tracing::warn!( + "(+{} more unauthorized probes suppressed in the previous ~30s)", + probe_suppressed + ); + } + probe_window_start = Some(now); + probe_suppressed = 0; + tracing::warn!("Unauthorized probe from {peer} ({bytes} bytes): {reason}"); + } else { + probe_suppressed += 1; + } } UiEvent::PeerSeen { .. } => {} _ => {} @@ -402,7 +431,7 @@ async fn run_server_loop( let mut remotes: HashMap<(u32, u16), RemoteState> = HashMap::new(); let (stream_tx, mut stream_rx) = mpsc::unbounded_channel::<(u32, u16, Vec)>(); let (udp_reply_tx, mut udp_reply_rx) = mpsc::unbounded_channel::<(u32, u16, String, Vec)>(); - let (connect_tx, mut connect_rx) = mpsc::unbounded_channel::<(u32, u16, String, Result<(tokio::net::tcp::OwnedWriteHalf, mpsc::Sender<()>), String>)>(); + let (connect_tx, mut connect_rx) = mpsc::unbounded_channel::(); let tcp_map = std::sync::Arc::new(tokio::sync::RwLock::new(HashMap::new())); @@ -414,15 +443,10 @@ async fn run_server_loop( let tx = udp_tx.clone(); tokio::spawn(async move { let mut buf = vec![0_u8; 65535]; - loop { - match sock_clone.recv_from(&mut buf).await { - Ok((size, peer)) => { - let packet = Bytes::copy_from_slice(&buf[..size]); - if tx.send((packet, peer)).await.is_err() { - break; - } - } - Err(_) => break, + while let Ok((size, peer)) = sock_clone.recv_from(&mut buf).await { + let packet = Bytes::copy_from_slice(&buf[..size]); + if tx.send((packet, peer)).await.is_err() { + break; } } }); @@ -580,6 +604,7 @@ async fn run_server_loop( Ok(()) } +#[allow(clippy::too_many_arguments)] async fn handle_udp_packet( packet: Bytes, peer: std::net::SocketAddr, @@ -590,7 +615,7 @@ async fn handle_udp_packet( ui_event_tx: &mpsc::UnboundedSender, stream_tx: mpsc::UnboundedSender<(u32, u16, Vec)>, udp_reply_tx: mpsc::UnboundedSender<(u32, u16, String, Vec)>, - connect_tx: mpsc::UnboundedSender<(u32, u16, String, Result<(tokio::net::tcp::OwnedWriteHalf, mpsc::Sender<()>), String>)>, + connect_tx: mpsc::UnboundedSender, router: std::sync::Arc, peer_last_seen: &mut HashMap, peer_available: &mut HashMap, diff --git a/ostp-tun/src/sys_win_api.rs b/ostp-tun/src/sys_win_api.rs index b8234ac..b7412ad 100644 --- a/ostp-tun/src/sys_win_api.rs +++ b/ostp-tun/src/sys_win_api.rs @@ -110,7 +110,7 @@ pub fn set_dns_servers(adapter_luid: u64, dns: &str) -> Result<()> { .chain(Some(0)) .collect(); - let mut settings = DNS_INTERFACE_SETTINGS { + let settings = DNS_INTERFACE_SETTINGS { Version: 1, // DNS_INTERFACE_SETTINGS_VERSION1 Flags: 1, // DNS_SETTING_IPV4 Domain: windows::core::PWSTR::null(), @@ -124,7 +124,7 @@ pub fn set_dns_servers(adapter_luid: u64, dns: &str) -> Result<()> { }; let luid = windows::Win32::NetworkManagement::Ndis::NET_LUID_LH { Value: adapter_luid }; - let guid = GUID::zeroed(); // We can pass zeroed GUID and just use LUID? Wait, SetInterfaceDnsSettings requires GUID. + let _guid = GUID::zeroed(); // We can pass zeroed GUID and just use LUID? Wait, SetInterfaceDnsSettings requires GUID. // Actually, setting DNS via SetInterfaceDnsSettings requires the interface GUID, which we can get from ConvertInterfaceLuidToGuid. unsafe { diff --git a/ostp-tun/src/windows.rs b/ostp-tun/src/windows.rs index 9c410bb..0a97fb9 100644 --- a/ostp-tun/src/windows.rs +++ b/ostp-tun/src/windows.rs @@ -28,7 +28,6 @@ impl Drop for WindowsRouteGuard { } pub async fn create(opts: OstpTunOptions) -> Result { - const CREATE_NO_WINDOW: u32 = 0x08000000; let (phys_gw, phys_if) = windows_route::sys::get_default_ipv4_route() .ok_or_else(|| anyhow!("Cannot find physical default IPv4 route"))?; diff --git a/ostp/src/main.rs b/ostp/src/main.rs index 5c0f9e3..5b1f169 100644 --- a/ostp/src/main.rs +++ b/ostp/src/main.rs @@ -633,23 +633,23 @@ fn wizard_save_config(config_path: &std::path::Path, json_value: &serde_json::Va } match fs::write(¤t_path, serde_json::to_string_pretty(json_value)?) { - Ok(_) => { + Ok(_) if current_path.exists() => { wizard_ok(&format!("Configuration saved to {:?}", current_path)); - return Ok(current_path); + Ok(current_path) } - Err(e) => { - wizard_warn(&format!("Could not write to {:?}: {}", current_path, e)); + _ => { + wizard_warn(&format!("Could not write to {:?}", current_path)); // Attempt 2: fallback to current directory let fallback = std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")).join("config.json"); wizard_warn(&format!("Falling back to {:?}", fallback)); match fs::write(&fallback, serde_json::to_string_pretty(json_value)?) { - Ok(_) => { + Ok(_) if fallback.exists() => { wizard_ok(&format!("Configuration saved to {:?}", fallback)); - return Ok(fallback); + Ok(fallback) } - Err(e2) => { - wizard_warn(&format!("Could not write to fallback {:?}: {}", fallback, e2)); + _ => { + wizard_warn(&format!("Could not write to fallback {:?}", fallback)); anyhow::bail!("Failed to save configuration to any location."); } } @@ -777,11 +777,11 @@ fn run_setup_wizard(config_path: &std::path::Path) -> Result<()> { let _ = &sni; let server_parts: Vec<&str> = server.split(':').collect(); - let server_host = server_parts.get(0).unwrap_or(&"127.0.0.1"); + let server_host = server_parts.first().unwrap_or(&"127.0.0.1"); let server_port = server_parts.get(1).unwrap_or(&"50000").parse::().unwrap_or(50000); let socks_parts: Vec<&str> = socks_bind.split(':').collect(); - let socks_host = socks_parts.get(0).unwrap_or(&"127.0.0.1"); + let socks_host = socks_parts.first().unwrap_or(&"127.0.0.1"); let socks_port = socks_parts.get(1).unwrap_or(&"1088").parse::().unwrap_or(1088); let client_json = serde_json::json!({ @@ -884,7 +884,7 @@ fn run_setup_wizard(config_path: &std::path::Path) -> Result<()> { wizard_step(3, TOTAL, "Service registration"); // intentional: step text then daemon call below - let port_str = listen.split(':').last().unwrap_or("50000"); + let port_str = listen.split(':').next_back().unwrap_or("50000"); let port: u16 = port_str.parse().unwrap_or(50000); let server_json = serde_json::json!({ "mode": "server", @@ -925,7 +925,7 @@ fn run_setup_wizard(config_path: &std::path::Path) -> Result<()> { // Print share links let host = get_or_ask_public_ip(config_path); - let port = listen.split(':').last().unwrap_or("50000"); + let port = listen.split(':').next_back().unwrap_or("50000"); println!(); wizard_section("Share links for clients:"); for (i, key) in access_keys.iter().enumerate() { @@ -1058,7 +1058,7 @@ fn run_setup_wizard(config_path: &std::path::Path) -> Result<()> { wizard_register_windows_service(&actual_path)?; } - let port = listen.split(':').last().unwrap_or("50000"); + let port = listen.split(':').next_back().unwrap_or("50000"); println!(); wizard_section("Share links for clients:"); for (i, key) in access_keys.iter().enumerate() { @@ -1683,8 +1683,7 @@ async fn run_app() -> Result<()> { } if let Some(key) = first_key { let host = get_or_ask_public_ip(&args.config); - let mut query_params = Vec::::new(); - query_params.push("type=udp".to_string()); + let mut query_params = vec!["type=udp".to_string()]; let mut link = format!("ostp://{}@{}:50000", key, host); if !query_params.is_empty() { @@ -1786,8 +1785,7 @@ async fn run_app() -> Result<()> { } } for (idx, user) in users.iter().enumerate() { - let mut query_params = Vec::::new(); - query_params.push("type=udp".to_string()); + let mut query_params = vec!["type=udp".to_string()]; let mut link = format!("ostp://{}@{}:{}", user.key(), host, port); if !query_params.is_empty() { @@ -2433,7 +2431,7 @@ fn extract_server_listen(old: &serde_json::Value) -> (String, u16) { // Old format: "listen": "0.0.0.0:50000" if let Some(s) = old.get("listen").and_then(|v| v.as_str()) { let parts: Vec<&str> = s.split(':').collect(); - let h = parts.get(0).unwrap_or(&"0.0.0.0").to_string(); + let h = parts.first().unwrap_or(&"0.0.0.0").to_string(); let p = parts.get(1).and_then(|x| x.parse().ok()).unwrap_or(50000); return (h, p); } @@ -2511,7 +2509,7 @@ fn extract_server_api(old: &serde_json::Value) -> (String, u16, String, String, if let Some(api) = old.get("api") { let bind = api.get("bind").and_then(|v| v.as_str()).unwrap_or("127.0.0.1:9090"); let parts: Vec<&str> = bind.split(':').collect(); - let listen = parts.get(0).unwrap_or(&"127.0.0.1").to_string(); + let listen = parts.first().unwrap_or(&"127.0.0.1").to_string(); let port = parts.get(1).and_then(|p| p.parse().ok()).unwrap_or(9090); let token = api.get("token").and_then(|v| v.as_str()).unwrap_or("YOUR_SECRET_TOKEN").to_string(); let webpath = api.get("webpath").and_then(|v| v.as_str()).unwrap_or("/admin").to_string(); @@ -2581,7 +2579,7 @@ fn extract_client_server(old: &serde_json::Value) -> (String, u16, String, Strin // Old flat format let server_full = old.get("server").and_then(|v| v.as_str()).unwrap_or("YOUR_SERVER_IP:50000"); let parts: Vec<&str> = server_full.split(':').collect(); - let server = parts.get(0).unwrap_or(&"YOUR_SERVER_IP").to_string(); + let server = parts.first().unwrap_or(&"YOUR_SERVER_IP").to_string(); let port = parts.get(1).and_then(|p| p.parse().ok()).unwrap_or(50000); let key = old.get("access_key").and_then(|v| v.as_str()).unwrap_or("").to_string(); let transport = old.get("transport").and_then(|t| t.get("mode").or(t.get("type"))).and_then(|v| v.as_str()).unwrap_or("udp").to_string(); @@ -2604,7 +2602,7 @@ fn extract_client_socks(old: &serde_json::Value) -> (String, u16) { // Old flat format let bind = old.get("socks5_bind").and_then(|v| v.as_str()).unwrap_or("127.0.0.1:1088"); let parts: Vec<&str> = bind.split(':').collect(); - let listen = parts.get(0).unwrap_or(&"127.0.0.1").to_string(); + let listen = parts.first().unwrap_or(&"127.0.0.1").to_string(); let port = parts.get(1).and_then(|p| p.parse().ok()).unwrap_or(1088); (listen, port) }