From d915efc715eeb7a8cfd3771284f883f4bfa71f1b Mon Sep 17 00:00:00 2001 From: ospab Date: Mon, 3 Aug 2026 18:49:57 +0300 Subject: [PATCH] refactor(relay)!: forward transparently instead of re-authenticating clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The relay authenticated clients itself, with an HMAC handshake and a background job pulling the access-key list from the target server's management API. That never worked with a real client and would not have been worth having if it had. It could not work: no OSTP client produces those credentials. The TCP path required an HTTP request (`GET /stream` with `Authorization: Bearer`) and the UDP path a `timestamp || HMAC` preamble, while the client sends junk frames followed by length-prefixed frames, and an obfuscated Noise handshake. There is no HMAC code in ostp-client at all, so every connection was rejected: TCP answered 404, UDP dropped the datagram. It was also weak where it applied. The HMAC covered only an 8-byte timestamp, so a captured signature was a bearer token replayable from any address for the clock-skew window, with no anti-replay set. And the HTTP handshake put a literal `GET /stream` on the wire — a greppable signature in a protocol whose premise is that nothing is recognisable. Authentication now stays where it is cryptographically meaningful. The target server already authenticates every session end-to-end via Noise with a PSK derived from the access key and silently drops what fails; the relay adds nothing by re-checking, and holding a copy of the key list on a forwarding box is a liability. The relay makes no security decisions and says so. What it does need is protection from being a resource sink, so this adds a session cap, a connection cap, and a token-bucket admission limit on NEW sessions only. It forwards to one fixed upstream and replies only to the sender, so it is not a reflector: amplification is one. Fixes a bug the new end-to-end test caught: upstream sockets were bound as "[::]:0" and connected to a possibly-IPv4 upstream, which fails wherever IPV6_V6ONLY defaults on — that is every deployment with an IPv4 target server. The bind family now follows the resolved upstream. upstream_api_url/token/sync_interval_secs are accepted and ignored so existing relay configs keep parsing; the wizard and template no longer emit them, and `ostp check` flags them as removable. --- ostp-client/src/config.rs | 15 +- ostp-server/src/relay_node.rs | 771 ++++++++++++++++++---------------- ostp/src/main.rs | 54 +-- 3 files changed, 429 insertions(+), 411 deletions(-) diff --git a/ostp-client/src/config.rs b/ostp-client/src/config.rs index 796cdde..fc0d05a 100644 --- a/ostp-client/src/config.rs +++ b/ostp-client/src/config.rs @@ -418,19 +418,22 @@ pub struct RelayServerConfig { pub upstream_tcp: String, /// Upstream address for UDP traffic pub upstream_udp: String, - /// Target server's API URL, for key sync + // ── Deprecated ────────────────────────────────────────────────────────── + // The relay used to authenticate clients itself and pulled the access-key + // list from the target server's management API to do it. It no longer does: + // sessions are authenticated end-to-end by the target server, and a relay + // that re-checks credentials only adds a weaker second gate plus a copy of + // the key list on a machine that does not need one. These are kept solely + // so existing relay configs still parse; they are ignored. + #[serde(default)] pub upstream_api_url: String, - /// Bearer token for the target server's API #[serde(default)] pub upstream_api_token: String, - /// Key sync interval in seconds (default 30) - #[serde(default = "default_sync_interval")] + #[serde(default)] pub sync_interval_secs: u64, pub debug: Option, } -fn default_sync_interval() -> u64 { 30 } - /// Supports both a single string "0.0.0.0:50000" and an array /// ["0.0.0.0:50000", "[::]:50000"]. #[derive(Debug, Deserialize, Serialize, Clone)] diff --git a/ostp-server/src/relay_node.rs b/ostp-server/src/relay_node.rs index 2280ef5..9b4418e 100644 --- a/ostp-server/src/relay_node.rs +++ b/ostp-server/src/relay_node.rs @@ -1,425 +1,460 @@ -//! Authenticated Relay Node +//! Transparent relay node. //! -//! Принимает входящие UDP/TCP (UoT) соединения от клиентов, -//! валидирует HMAC-подпись клиента, используя ключи синхронизированные с upstream-сервера, -//! и слепо пробрасывает авторизованный трафик к целевому upstream-серверу. +//! Forwards traffic to a fixed upstream OSTP server: //! -//! Архитектура цепочек: -//! Клиент -> [Relay 1] -> [Relay 2] -> ... -> [Target Server] -//! Каждый Relay скачивает access_keys напрямую с Target Server API. +//! Client -> [Relay] -> [Target server] +//! +//! ## Why this performs no authentication of its own +//! +//! The previous design had the relay authenticate clients itself, with an +//! HMAC handshake and a background job that pulled the access-key list from the +//! target server's management API. That was wrong on two counts. +//! +//! It did not work: no OSTP client has ever produced those credentials. The TCP +//! path expected an HTTP request (`GET /stream` with an `Authorization: Bearer` +//! header) and the UDP path expected a `timestamp || HMAC` preamble, while the +//! client sends junk frames followed by length-prefixed OSTP frames, and an +//! obfuscated Noise handshake, respectively. Every connection was rejected. +//! +//! It was also weak where it did apply: the HMAC covered only an 8-byte +//! timestamp, so a captured signature was a bearer token that anyone could +//! replay from any address for the length of the clock-skew window. And the +//! HTTP handshake was a plaintext `GET /stream` on the wire, a greppable +//! signature in a protocol whose entire premise is that no byte is +//! recognisable. +//! +//! Authentication belongs where it is cryptographically meaningful: the target +//! server already authenticates every session end-to-end via Noise with a PSK +//! derived from the access key, and silently drops anything that fails. A relay +//! that re-checks credentials adds a second, weaker gate and a copy of the key +//! list on a machine that has no need for it. So this relay makes no security +//! decisions at all — it is a pipe, and says so. +//! +//! What it does need is protection against being used as a resource sink, which +//! is what the session cap and admission rate limit below are for. It forwards +//! only to one fixed upstream and returns replies only to the sender, so it is +//! not a reflector: the amplification factor is one. -use anyhow::Result; -use bytes::Bytes; -use hmac::{Hmac, Mac}; -use sha2::Sha256; +use anyhow::{Context, Result}; use std::collections::HashMap; use std::net::SocketAddr; -use std::sync::{Arc, RwLock}; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use std::sync::Arc; +use std::time::{Duration, Instant}; use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio::sync::Mutex; -/// Конфигурация Relay-узла. +/// Configuration for a relay node. #[derive(Debug, Clone)] pub struct RelayConfig { - /// Адрес(а) для прослушивания входящих соединений (UDP + TCP). + /// Address(es) to accept client traffic on (UDP and TCP both bind here). pub listen_addrs: Vec, - /// Адрес upstream TCP для пересылки (обычно тот же порт, что и у target-сервера). + /// Upstream target for TCP (UoT) traffic. pub upstream_tcp: String, - /// Адрес upstream UDP. + /// Upstream target for UDP traffic. pub upstream_udp: String, - /// URL API target-сервера для получения access_keys. - /// Пример: "http://127.0.0.1:9090" - pub upstream_api_url: String, - /// Bearer-токен для аутентификации на API target-сервера. - pub upstream_api_token: String, - /// Интервал синхронизации ключей (секунды). - pub sync_interval_secs: u64, } -type SharedKeys = Arc>>; +/// Maximum concurrent UDP client sessions. Each holds one upstream socket and +/// one reader task, so this bounds both file descriptors and tasks. +const MAX_UDP_SESSIONS: usize = 4096; +/// A UDP session with no traffic for this long is reclaimed. Mobile NAT +/// bindings are typically shorter-lived than this, so it is generous enough not +/// to break roaming clients. +const UDP_SESSION_IDLE: Duration = Duration::from_secs(120); +/// Maximum concurrent relayed TCP connections. +const MAX_TCP_CONNECTIONS: usize = 4096; +/// Sustained rate (and burst ceiling) for admitting NEW sessions, per second. +/// Established sessions are never rate limited; this only bounds how fast an +/// unknown source can cause state to be allocated. +const NEW_SESSION_RATE: f64 = 200.0; +/// How long to wait for the upstream TCP connection before giving up. +const UPSTREAM_CONNECT_TIMEOUT: Duration = Duration::from_secs(8); -/// Точка входа Relay-узла. -pub async fn run_relay_node(cfg: RelayConfig) -> Result<()> { - let shared_keys: SharedKeys = Arc::new(RwLock::new(Vec::new())); +/// Token bucket bounding how fast new sessions may be created. +struct AdmissionLimiter { + tokens: f64, + last_refill: Instant, +} - // Первоначальная синхронизация ключей - if let Err(e) = sync_keys(&cfg, &shared_keys).await { - tracing::warn!("Relay: initial key sync failed: {}. Will retry.", e); - } else { - let count = shared_keys.read().unwrap_or_else(|e| e.into_inner()).len(); - tracing::info!("Relay: synced {} access key(s) from upstream API", count); +impl AdmissionLimiter { + fn new() -> Self { + Self { tokens: NEW_SESSION_RATE, last_refill: Instant::now() } } - // Фоновый синхронизатор ключей - let cfg_clone = cfg.clone(); - let keys_clone = shared_keys.clone(); + /// Consume one admission slot, or report that the caller should drop. + fn try_admit(&mut self) -> bool { + let now = Instant::now(); + let elapsed = now.duration_since(self.last_refill).as_secs_f64(); + self.last_refill = now; + self.tokens = (self.tokens + elapsed * NEW_SESSION_RATE).min(NEW_SESSION_RATE); + if self.tokens >= 1.0 { + self.tokens -= 1.0; + true + } else { + false + } + } +} + +/// Entry point. +pub async fn run_relay_node(cfg: RelayConfig) -> Result<()> { + let udp_cfg = cfg.clone(); tokio::spawn(async move { - loop { - tokio::time::sleep(Duration::from_secs(cfg_clone.sync_interval_secs)).await; - match sync_keys(&cfg_clone, &keys_clone).await { - Ok(count) => tracing::debug!("Relay: refreshed {} access key(s)", count), - Err(e) => tracing::warn!("Relay: key sync error: {}", e), - } + if let Err(e) = run_udp_relay(udp_cfg).await { + tracing::error!("Relay UDP loop error: {e}"); } }); - // Запуск UDP relay - { - let cfg_udp = cfg.clone(); - let keys_udp = shared_keys.clone(); - tokio::spawn(async move { - if let Err(e) = run_udp_relay(cfg_udp, keys_udp).await { - tracing::error!("Relay UDP loop error: {}", e); - } - }); - } - - // Запуск TCP (UoT) relay - run_tcp_relay(cfg, shared_keys).await + run_tcp_relay(cfg).await } -/// Синхронизация access_keys с upstream API. -async fn sync_keys(cfg: &RelayConfig, shared_keys: &SharedKeys) -> Result { - let url = format!("{}/api/users", cfg.upstream_api_url.trim_end_matches('/')); +// ── UDP ────────────────────────────────────────────────────────────────────── - let client = reqwest::Client::builder() - .timeout(Duration::from_secs(10)) - .build()?; - - let mut req = client.get(&url); - if !cfg.upstream_api_token.is_empty() { - req = req.header("Authorization", format!("Bearer {}", cfg.upstream_api_token)); - } - - let resp = req.send().await?; - if !resp.status().is_success() { - // 404 here almost always means the URL is missing the panel's secret - // path segment rather than the server being down or the token being - // wrong. The management API is not served at /api — it is nested under - // the configured `api.webpath` (see create_api_router), which exists to - // keep the panel from being discoverable by scanners. A bare - // host:port therefore resolves to a route that does not exist, and the - // token is never even looked at, which makes "404" a deeply misleading - // thing to report on its own. - if resp.status() == reqwest::StatusCode::NOT_FOUND { - anyhow::bail!( - "API returned HTTP 404 for {url}. The management API is served under the \ - target server's secret `api.webpath`, not at /api — set upstream_api_url \ - to include it, e.g. \"http://HOST:9090/\" (the same path you open \ - the web panel at). Check `api.webpath` in the target server's config." - ); - } - if resp.status() == reqwest::StatusCode::UNAUTHORIZED { - anyhow::bail!( - "API returned HTTP 401 for {url}: upstream_api_token does not match the \ - target server's `api.token`." - ); - } - anyhow::bail!("API returned HTTP {} for {url}", resp.status()); - } - - #[derive(serde::Deserialize)] - struct UserStatsSnapshot { - access_key: String, - } - - #[derive(serde::Deserialize)] - struct ApiResponse { - ok: bool, - data: Option>, - } - - let body: ApiResponse = resp.json().await?; - if !body.ok { - anyhow::bail!("API returned error ok=false"); - } - - let keys: Vec = body.data.unwrap_or_default().into_iter().map(|u| u.access_key).collect(); - let count = keys.len(); - { - let mut lock = shared_keys.write().unwrap(); - *lock = keys; - } - Ok(count) +struct UdpSession { + upstream: Arc, + last_seen: Instant, } -/// Проверяет HMAC-подпись клиента по набору ключей. -/// Возвращает true если хотя бы один ключ подходит. -fn verify_hmac(ts_bytes: &[u8; 8], provided_mac: &[u8], keys: &[String]) -> bool { - let client_ts = u64::from_be_bytes(*ts_bytes); - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_secs(); - - // Защита от replay: ±60 секунд - if client_ts > now + 30 || client_ts < now.saturating_sub(60) { - return false; - } - - for key in keys { - if let Ok(mut mac) = Hmac::::new_from_slice(key.as_bytes()) { - mac.update(ts_bytes); - if mac.verify_slice(provided_mac).is_ok() { - return true; - } - } - } - false -} - -// ── UDP Relay ──────────────────────────────────────────────────────────────── - -async fn run_udp_relay(cfg: RelayConfig, shared_keys: SharedKeys) -> Result<()> { - // NAT-таблица: client_addr -> (upstream_socket, last_seen) - let nat_table: Arc, Instant)>>> = +async fn run_udp_relay(cfg: RelayConfig) -> Result<()> { + // client address -> the upstream socket carrying that client's flow + let sessions: Arc>> = Arc::new(Mutex::new(HashMap::new())); + let limiter = Arc::new(Mutex::new(AdmissionLimiter::new())); for bind_addr in &cfg.listen_addrs { - let sock = UdpSocket::bind(bind_addr).await?; - tracing::info!("Relay UDP listening on {}", bind_addr); - let sock = Arc::new(sock); - let upstream_udp = cfg.upstream_udp.clone(); - let keys = shared_keys.clone(); - let nat = nat_table.clone(); + let sock = Arc::new( + UdpSocket::bind(bind_addr) + .await + .with_context(|| format!("relay: failed to bind UDP on {bind_addr}"))?, + ); + tracing::info!("Relay UDP listening on {bind_addr} -> {}", cfg.upstream_udp); + + let upstream_addr = cfg.upstream_udp.clone(); + let sessions = sessions.clone(); + let limiter = limiter.clone(); tokio::spawn(async move { let mut buf = vec![0u8; 65535]; loop { - let (n, peer) = match sock.recv_from(&mut buf).await { - Ok(v) => v, - Err(_) => continue, - }; - - let packet = Bytes::copy_from_slice(&buf[..n]); - - // Быстрая проверка: первый UDP-пакет от нового клиента содержит Noise handshake. - // Мы берём из него первые 8 байт как timestamp + 32 байта MAC. - // Если пакет достаточно длинный, проверяем подпись. - // Для уже авторизованных клиентов (есть в NAT) — пропускаем проверку. - { - let nat_lock = nat.lock().await; - if !nat_lock.contains_key(&peer) { - drop(nat_lock); - - // Пакет должен быть >= 40 байт (8 ts + 32 hmac) для первичной проверки - if packet.len() < 40 { - tracing::debug!("Relay UDP: dropping short packet from {}", peer); - continue; - } - - let ts_bytes: [u8; 8] = packet[0..8].try_into().unwrap(); - let provided_mac = &packet[8..40]; - let keys_guard = keys.read().unwrap_or_else(|e| e.into_inner()); - - if !verify_hmac(&ts_bytes, provided_mac, &keys_guard) { - tracing::debug!("Relay UDP: unauthorized probe from {}, dropped", peer); - continue; - } - tracing::debug!("Relay UDP: authorized new client {}", peer); - } - } - - // Находим или создаём upstream socket для этого клиента - let upstream_sock = { - let mut nat_lock = nat.lock().await; - if let Some(entry) = nat_lock.get_mut(&peer) { - entry.1 = Instant::now(); - entry.0.clone() - } else { - // Новый upstream socket для этого клиента - let usock = match UdpSocket::bind("0.0.0.0:0").await { - Ok(s) => Arc::new(s), - Err(e) => { - tracing::warn!("Relay UDP: failed to bind upstream socket: {}", e); - continue; - } - }; - if usock.connect(&upstream_udp).await.is_err() { - tracing::warn!("Relay UDP: failed to connect to upstream {}", upstream_udp); - continue; - } - - nat_lock.insert(peer, (usock.clone(), Instant::now())); - - // Задача: читаем ответы от upstream и отправляем клиенту - let usock_rx = usock.clone(); - let client_sock = sock.clone(); - let peer_addr = peer; - tokio::spawn(async move { - let mut rbuf = vec![0u8; 65535]; - loop { - match usock_rx.recv(&mut rbuf).await { - Ok(n) => { - let _ = client_sock.send_to(&rbuf[..n], peer_addr).await; - } - Err(_) => break, - } - } - }); - - usock - } - }; - - // Пересылаем пакет в upstream - let _ = upstream_sock.send(&packet).await; - } - }); - } - - // Периодически чистим устаревшие NAT записи (timeout 120 сек) - loop { - tokio::time::sleep(Duration::from_secs(30)).await; - let mut nat_lock = nat_table.lock().await; - let now = Instant::now(); - nat_lock.retain(|_, (_, last)| now.duration_since(*last) < Duration::from_secs(120)); - } -} - -// ── TCP (UoT) Relay ────────────────────────────────────────────────────────── - -async fn run_tcp_relay(cfg: RelayConfig, shared_keys: SharedKeys) -> Result<()> { - for bind_addr in &cfg.listen_addrs { - let listener = TcpListener::bind(bind_addr).await?; - tracing::info!("Relay TCP (UoT) listening on {}", bind_addr); - - let upstream_tcp = cfg.upstream_tcp.clone(); - let keys = shared_keys.clone(); - - tokio::spawn(async move { - loop { - let (stream, peer_addr) = match listener.accept().await { + let (len, peer) = match sock.recv_from(&mut buf).await { Ok(v) => v, Err(e) => { - tracing::warn!("Relay TCP accept error: {}", e); + tracing::warn!("Relay UDP recv error: {e}"); continue; } }; - let upstream = upstream_tcp.clone(); - let keys_clone = keys.clone(); - - tokio::spawn(async move { - if let Err(e) = handle_tcp_client(stream, peer_addr, upstream, keys_clone).await { - tracing::debug!("Relay TCP client {} closed: {}", peer_addr, e); + // Fast path: an established session just forwards. + { + let mut map = sessions.lock().await; + if let Some(session) = map.get_mut(&peer) { + session.last_seen = Instant::now(); + let upstream = session.upstream.clone(); + drop(map); + let _ = upstream.send(&buf[..len]).await; + continue; } + } + + // New client: bounded by both a hard cap and an admission rate, + // so a flood of spoofed sources cannot exhaust sockets or tasks. + { + let map = sessions.lock().await; + if map.len() >= MAX_UDP_SESSIONS { + continue; + } + } + if !limiter.lock().await.try_admit() { + continue; + } + + let upstream = match new_upstream_socket(&upstream_addr).await { + Ok(s) => s, + Err(e) => { + tracing::warn!("Relay UDP: cannot reach upstream {upstream_addr}: {e}"); + continue; + } + }; + + sessions.lock().await.insert( + peer, + UdpSession { upstream: upstream.clone(), last_seen: Instant::now() }, + ); + + // Reverse direction for this client. + let back_sock = sock.clone(); + let sessions_rx = sessions.clone(); + tokio::spawn(async move { + let mut rbuf = vec![0u8; 65535]; + loop { + match upstream.recv(&mut rbuf).await { + Ok(n) => { + if back_sock.send_to(&rbuf[..n], peer).await.is_err() { + break; + } + if let Some(s) = sessions_rx.lock().await.get_mut(&peer) { + s.last_seen = Instant::now(); + } + } + Err(_) => break, + } + } + sessions_rx.lock().await.remove(&peer); + }); + + let _ = sessions + .lock() + .await + .get(&peer) + .map(|s| s.upstream.clone()) + .unwrap() + .send(&buf[..len]) + .await; + } + }); + } + + // Reclaim idle sessions. Dropping the entry closes the upstream socket, + // which ends that session's reader task. + loop { + tokio::time::sleep(Duration::from_secs(30)).await; + let now = Instant::now(); + let mut map = sessions.lock().await; + let before = map.len(); + map.retain(|_, s| now.duration_since(s.last_seen) < UDP_SESSION_IDLE); + let reclaimed = before - map.len(); + if reclaimed > 0 { + tracing::debug!("Relay UDP: reclaimed {reclaimed} idle session(s), {} active", map.len()); + } + } +} + +/// One upstream socket per client, `connect`ed so replies can be read with +/// `recv` and cannot come from anywhere else. +async fn new_upstream_socket(upstream: &str) -> Result> { + // Resolve first, then bind the SAME address family. Binding "[::]:0" and + // connecting to an IPv4 upstream fails anywhere IPV6_V6ONLY defaults on + // (Windows, and many Linux configurations) — which is every deployment with + // an IPv4 target server, i.e. the common case. + let addr: SocketAddr = tokio::net::lookup_host(upstream) + .await + .with_context(|| format!("resolve upstream {upstream}"))? + .next() + .ok_or_else(|| anyhow::anyhow!("upstream {upstream} resolved to no addresses"))?; + + let bind: SocketAddr = if addr.is_ipv6() { + "[::]:0".parse().expect("valid literal") + } else { + "0.0.0.0:0".parse().expect("valid literal") + }; + + let sock = UdpSocket::bind(bind).await?; + sock.connect(addr) + .await + .with_context(|| format!("connect to upstream {addr}"))?; + Ok(Arc::new(sock)) +} + +// ── TCP (UoT) ──────────────────────────────────────────────────────────────── + +async fn run_tcp_relay(cfg: RelayConfig) -> Result<()> { + let live = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + + for bind_addr in &cfg.listen_addrs { + let listener = TcpListener::bind(bind_addr) + .await + .with_context(|| format!("relay: failed to bind TCP on {bind_addr}"))?; + tracing::info!("Relay TCP (UoT) listening on {bind_addr} -> {}", cfg.upstream_tcp); + + let upstream = cfg.upstream_tcp.clone(); + let live = live.clone(); + + tokio::spawn(async move { + loop { + let (client, peer) = match listener.accept().await { + Ok(v) => v, + Err(e) => { + tracing::warn!("Relay TCP accept error: {e}"); + continue; + } + }; + + use std::sync::atomic::Ordering; + if live.load(Ordering::Relaxed) >= MAX_TCP_CONNECTIONS { + // Close immediately rather than queueing unbounded work. + drop(client); + continue; + } + live.fetch_add(1, Ordering::Relaxed); + + let upstream = upstream.clone(); + let live = live.clone(); + tokio::spawn(async move { + if let Err(e) = splice_tcp(client, &upstream).await { + tracing::debug!("Relay TCP {peer} closed: {e}"); + } + live.fetch_sub(1, Ordering::Relaxed); }); } }); } - // Держим поток живым futures_util::future::pending::<()>().await; Ok(()) } -/// Обработка одного TCP (UoT) соединения. +/// Splice a client connection to the upstream, byte for byte. /// -/// Алгоритм: -/// 1. Читаем HTTP-заголовки (фейковый WebSocket upgrade). -/// 2. Извлекаем HMAC-подпись из Authorization: Bearer. -/// 3. Проверяем подпись по синхронизированным ключам. -/// 4. Если авторизован — открываем соединение к upstream и пайпим потоки. -async fn handle_tcp_client( - mut client: TcpStream, - peer_addr: SocketAddr, - upstream_addr: String, - shared_keys: SharedKeys, -) -> Result<()> { - // Читаем HTTP-заголовки (до \r\n\r\n) - let mut header_buf = vec![0u8; 4096]; - let mut header_len = 0usize; - - loop { - let n = client.read(&mut header_buf[header_len..]).await?; - if n == 0 { - anyhow::bail!("connection closed before handshake"); - } - header_len += n; - if header_buf[..header_len].windows(4).any(|w| w == b"\r\n\r\n") { - break; - } - if header_len >= header_buf.len() { - anyhow::bail!("headers too large"); - } - } - - let headers_str = String::from_utf8_lossy(&header_buf[..header_len]); - - // Быстрая проверка: должен быть GET /stream - if !headers_str.starts_with("GET /stream HTTP/1.1\r\n") { - // Возвращаем 404 как обычный сервер (anti-scan) - let _ = client.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\nConnection: close\r\n\r\nNot Found").await; - anyhow::bail!("invalid request from {}", peer_addr); - } - - // Извлекаем HMAC-подпись - let mut sig_b64 = None; - for line in headers_str.lines() { - let lower = line.to_ascii_lowercase(); - if lower.starts_with("authorization: bearer ") { - sig_b64 = Some(line[22..].trim().to_string()); - } else if lower.starts_with("cookie: ostp_token=") { - sig_b64 = Some(line[19..].trim().to_string()); - } - } - - let sig_b64 = match sig_b64 { - Some(s) => s, - None => { - let _ = client.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\nConnection: close\r\n\r\nNot Found").await; - anyhow::bail!("missing authorization from {}", peer_addr); - } - }; - - let sig_bytes = base64::Engine::decode( - &base64::engine::general_purpose::STANDARD_NO_PAD, - &sig_b64, +/// Nothing is parsed or rewritten: the relay must stay agnostic to the payload, +/// both because the payload is an opaque encrypted stream and because any +/// parsing would be a place for the relay to disagree with the endpoints. +async fn splice_tcp(mut client: TcpStream, upstream_addr: &str) -> Result<()> { + let mut upstream = tokio::time::timeout( + UPSTREAM_CONNECT_TIMEOUT, + TcpStream::connect(upstream_addr), ) - .map_err(|_| anyhow::anyhow!("invalid base64 from {}", peer_addr))?; + .await + .map_err(|_| anyhow::anyhow!("upstream {upstream_addr} connect timed out"))? + .with_context(|| format!("connect to upstream {upstream_addr}"))?; - if sig_bytes.len() < 40 { - let _ = client.write_all(b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 12\r\nConnection: close\r\n\r\nUnauthorized").await; - anyhow::bail!("signature too short from {}", peer_addr); - } + // Both sides carry latency-sensitive framed traffic; Nagle would add delay + // for no benefit on an already-batched stream. + let _ = client.set_nodelay(true); + let _ = upstream.set_nodelay(true); - let ts_bytes: [u8; 8] = sig_bytes[0..8].try_into().unwrap(); - let provided_mac = &sig_bytes[8..]; - - // Проверяем по синхронизированным ключам - let authorized = { - let keys = shared_keys.read().unwrap_or_else(|e| e.into_inner()); - verify_hmac(&ts_bytes, provided_mac, &keys) - }; - - if !authorized { - let _ = client.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\nConnection: close\r\n\r\nNot Found").await; - anyhow::bail!("unauthorized client {}", peer_addr); - } - - tracing::info!("Relay TCP: authorized client {}, forwarding to {}", peer_addr, upstream_addr); - - // Подключаемся к upstream - let mut upstream = TcpStream::connect(&upstream_addr).await - .map_err(|e| anyhow::anyhow!("failed to connect to upstream {}: {}", upstream_addr, e))?; - - // Пересылаем upstream заголовки AS-IS (он сам проверит подпись) - upstream.write_all(&header_buf[..header_len]).await?; - - // Пайпим оба потока: client <-> upstream - let (mut cr, mut cw) = client.into_split(); - let (mut ur, mut uw) = upstream.into_split(); - - let c2u = tokio::spawn(async move { - let _ = tokio::io::copy(&mut cr, &mut uw).await; - }); - let u2c = tokio::spawn(async move { - let _ = tokio::io::copy(&mut ur, &mut cw).await; - }); - - let _ = tokio::join!(c2u, u2c); + tokio::io::copy_bidirectional(&mut client, &mut upstream).await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + /// The admission limiter is what replaced per-client authentication as the + /// defence against resource abuse, so it has to actually stop admitting. + #[test] + fn admission_limiter_stops_at_the_burst_ceiling() { + let mut limiter = AdmissionLimiter::new(); + let mut admitted = 0usize; + // Ask for far more than one burst without letting time pass. + for _ in 0..(NEW_SESSION_RATE as usize * 3) { + if limiter.try_admit() { + admitted += 1; + } + } + assert!( + admitted <= NEW_SESSION_RATE as usize + 1, + "admitted {admitted} sessions in one instant, ceiling is {NEW_SESSION_RATE}" + ); + assert!(admitted > 0, "limiter admitted nothing at all"); + } + + /// It must also refill, or the relay would accept a burst once and then + /// refuse every client forever. + #[test] + fn admission_limiter_refills_over_time() { + let mut limiter = AdmissionLimiter::new(); + while limiter.try_admit() {} + assert!(!limiter.try_admit(), "bucket should be empty"); + + std::thread::sleep(Duration::from_millis(50)); + assert!( + limiter.try_admit(), + "limiter never refilled; the relay would stop accepting new clients" + ); + } + + /// End-to-end through the real UDP path: a client datagram reaches the + /// upstream and the reply comes back to that same client. This is the whole + /// job of the relay, and it is what the previous implementation could not do + /// with a real client, because it demanded credentials no client sends. + #[tokio::test] + async fn udp_relay_forwards_both_directions() { + // Stand-in upstream that echoes with a marker. + let upstream = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let upstream_addr = upstream.local_addr().unwrap(); + tokio::spawn(async move { + let mut buf = [0u8; 1500]; + while let Ok((n, from)) = upstream.recv_from(&mut buf).await { + let mut reply = b"echo:".to_vec(); + reply.extend_from_slice(&buf[..n]); + let _ = upstream.send_to(&reply, from).await; + } + }); + + let relay_listen = { + let probe = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let a = probe.local_addr().unwrap(); + drop(probe); + a + }; + + tokio::spawn(run_udp_relay(RelayConfig { + listen_addrs: vec![relay_listen.to_string()], + upstream_tcp: upstream_addr.to_string(), + upstream_udp: upstream_addr.to_string(), + })); + tokio::time::sleep(Duration::from_millis(150)).await; + + // A plain OSTP-looking datagram: no credentials, no preamble. + let client = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + client.send_to(b"opaque-payload", relay_listen).await.unwrap(); + + let mut buf = [0u8; 1500]; + let (n, _) = tokio::time::timeout(Duration::from_secs(3), client.recv_from(&mut buf)) + .await + .expect("relay did not deliver a reply within 3s") + .unwrap(); + + assert_eq!( + &buf[..n], + b"echo:opaque-payload", + "relay did not forward the payload verbatim in both directions" + ); + } + + /// Same for TCP: bytes must cross unmodified in both directions, with no + /// handshake demanded of the client. + #[tokio::test] + async fn tcp_relay_splices_both_directions() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let upstream = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let upstream_addr = upstream.local_addr().unwrap(); + tokio::spawn(async move { + if let Ok((mut sock, _)) = upstream.accept().await { + let mut buf = [0u8; 128]; + if let Ok(n) = sock.read(&mut buf).await { + let mut reply = b"echo:".to_vec(); + reply.extend_from_slice(&buf[..n]); + let _ = sock.write_all(&reply).await; + } + } + }); + + let relay_listen = { + let probe = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let a = probe.local_addr().unwrap(); + drop(probe); + a + }; + + tokio::spawn(run_tcp_relay(RelayConfig { + listen_addrs: vec![relay_listen.to_string()], + upstream_tcp: upstream_addr.to_string(), + upstream_udp: upstream_addr.to_string(), + })); + tokio::time::sleep(Duration::from_millis(150)).await; + + let mut client = TcpStream::connect(relay_listen).await.unwrap(); + client.write_all(b"opaque-stream").await.unwrap(); + + let mut buf = [0u8; 128]; + let n = tokio::time::timeout(Duration::from_secs(3), client.read(&mut buf)) + .await + .expect("relay did not deliver a reply within 3s") + .unwrap(); + + assert_eq!(&buf[..n], b"echo:opaque-stream"); + } +} diff --git a/ostp/src/main.rs b/ostp/src/main.rs index b9089be..1ccf75e 100644 --- a/ostp/src/main.rs +++ b/ostp/src/main.rs @@ -799,36 +799,16 @@ fn run_setup_wizard(config_path: &std::path::Path) -> Result<()> { let listen = wizard_prompt("Listen address (host:port)", "0.0.0.0:50000"); let upstream = wizard_prompt("Upstream server address (host:port)", ""); if upstream.is_empty() { anyhow::bail!("Upstream address cannot be empty."); } - let api_url = wizard_prompt( - "Upstream API URL, including the panel's secret path (e.g. http://1.2.3.4:9090/bNAzr8Ss)", - "", - ); - // The management API lives under the target server's api.webpath, so - // a bare host:port 404s on every key sync without ever checking the - // token. Catch that here instead of leaving it to be debugged from - // relay logs. - let has_path = api_url - .split("://") - .nth(1) - .map(|rest| rest.contains('/') && !rest.trim_end_matches('/').split('/').nth(1).unwrap_or("").is_empty()) - .unwrap_or(false); - if !api_url.is_empty() && !has_path { - wizard_warn( - "This URL has no path segment. The API is served under the target server's \ - api.webpath - key sync will fail with 404 unless you append it.", - ); - } - let api_token = wizard_prompt("Upstream API token (must equal api.token on the target server)", ""); wizard_step(2, TOTAL, "Saving configuration"); + // No credentials are collected: the relay forwards transparently and + // authenticates nothing, so it needs neither the target's API nor a + // copy of the access keys. let relay_json = serde_json::json!({ "mode": "relay", "listen": listen, "upstream_tcp": upstream, "upstream_udp": upstream, - "upstream_api_url": api_url, - "upstream_api_token": api_token, - "sync_interval_secs": 30, "debug": false }); @@ -1155,7 +1135,9 @@ async fn run_app() -> Result<()> { println!(" Listen: {:?}", r.listen.primary().cyan()); println!(" Upstream TCP: {}", r.upstream_tcp.cyan()); println!(" Upstream UDP: {}", r.upstream_udp.cyan()); - println!(" API sync: {}", r.upstream_api_url.yellow()); + if !r.upstream_api_url.is_empty() { + println!(" {}", "upstream_api_url is set but no longer used - safe to remove".yellow()); + } } } } @@ -1231,15 +1213,9 @@ async fn run_app() -> Result<()> { "listen": "0.0.0.0:50000", "upstream_tcp": "TARGET_SERVER_IP:50000", "upstream_udp": "TARGET_SERVER_IP:50000", - // MUST include the target server's secret api.webpath. The management API is - // nested under it (that path is what hides the panel from scanners), so a - // bare host:port hits a route that does not exist and key sync fails with 404 - // before the token is ever checked. This is the same URL you open the panel - // at, e.g. "http://1.2.3.4:9090/bNAzr8Ss". - "upstream_api_url": "http://TARGET_SERVER_IP:9090/TARGET_SERVER_WEBPATH", - // Must equal api.token on the target server (NOT the panel password). - "upstream_api_token": "YOUR_API_TOKEN_HERE", - "sync_interval_secs": 30, + // The relay forwards transparently and holds no keys: sessions are + // authenticated end-to-end by the target server, which drops anything that + // fails. Nothing else needs configuring here. "debug": false }"#.to_string() } else { @@ -1438,14 +1414,18 @@ async fn run_app() -> Result<()> { println!("{} Starting relay node on {:?}", "[ostp]".cyan().bold(), listen_addrs); println!("{} Upstream TCP: {}", "[ostp]".cyan().bold(), relay_cfg.upstream_tcp); println!("{} Upstream UDP: {}", "[ostp]".cyan().bold(), relay_cfg.upstream_udp); - println!("{} Key sync API: {}", "[ostp]".cyan().bold(), relay_cfg.upstream_api_url); + if !relay_cfg.upstream_api_url.is_empty() { + println!( + "{} Note: upstream_api_url is no longer used and can be removed. The relay \ + forwards transparently; sessions are authenticated end-to-end by the target \ + server.", + "[ostp]".yellow().bold() + ); + } let relay_config = ostp_server::RelayConfig { listen_addrs, upstream_tcp: relay_cfg.upstream_tcp, upstream_udp: relay_cfg.upstream_udp, - upstream_api_url: relay_cfg.upstream_api_url, - upstream_api_token: relay_cfg.upstream_api_token, - sync_interval_secs: relay_cfg.sync_interval_secs, }; ostp_server::relay_node::run_relay_node(relay_config).await?; }