ospab
/
ostp
mirror of https://github.com/ospab/ostp.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor(relay)!: forward transparently instead of re-authenticating clients 

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.
This commit is contained in:
ospab 2026-08-03 18:49:57 +03:00
parent b673219894
commit d915efc715
3 changed files with 429 additions and 411 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
ostp-client/src/config.rs
View File

@ -418,19 +418,22 @@ pub struct RelayServerConfig {
pub upstream_tcp: String, pub upstream_tcp: String,
/// Upstream address for UDP traffic /// Upstream address for UDP traffic
pub upstream_udp: String, 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, pub upstream_api_url: String,
/// Bearer token for the target server's API
#[serde(default)] #[serde(default)]
pub upstream_api_token: String, pub upstream_api_token: String,
/// Key sync interval in seconds (default 30) #[serde(default)]
#[serde(default = "default_sync_interval")]
pub sync_interval_secs: u64, pub sync_interval_secs: u64,
pub debug: Option<bool>, pub debug: Option<bool>,
} }
fn default_sync_interval() -> u64 { 30 }
/// Supports both a single string "0.0.0.0:50000" and an array /// Supports both a single string "0.0.0.0:50000" and an array
/// ["0.0.0.0:50000", "[::]:50000"]. /// ["0.0.0.0:50000", "[::]:50000"].
#[derive(Debug, Deserialize, Serialize, Clone)] #[derive(Debug, Deserialize, Serialize, Clone)]

709
ostp-server/src/relay_node.rs
View File

@ -1,425 +1,460 @@
//! Authenticated Relay Node //! Transparent relay node.
//! //!
//! Принимает входящие UDP/TCP (UoT) соединения от клиентов, //! Forwards traffic to a fixed upstream OSTP server:
//! валидирует HMAC-подпись клиента, используя ключи синхронизированные с upstream-сервера,
//! и слепо пробрасывает авторизованный трафик к целевому upstream-серверу.
//! //!
//! Архитектура цепочек: //! Client -> [Relay] -> [Target server]
//! Клиент -> [Relay 1] -> [Relay 2] -> ... -> [Target Server] //!
//! Каждый Relay скачивает access_keys напрямую с Target Server API. //! ## 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 anyhow::{Context, Result};
use bytes::Bytes;
use hmac::{Hmac, Mac};
use sha2::Sha256;
use std::collections::HashMap; use std::collections::HashMap;
use std::net::SocketAddr; use std::net::SocketAddr;
use std::sync::{Arc, RwLock}; use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio::net::{TcpListener, TcpStream, UdpSocket};
use tokio::sync::Mutex; use tokio::sync::Mutex;
/// Конфигурация Relay-узла. /// Configuration for a relay node.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct RelayConfig { pub struct RelayConfig {
/// Адрес(а) для прослушивания входящих соединений (UDP + TCP). /// Address(es) to accept client traffic on (UDP and TCP both bind here).
pub listen_addrs: Vec<String>, pub listen_addrs: Vec<String>,
/// Адрес upstream TCP для пересылки (обычно тот же порт, что и у target-сервера). /// Upstream target for TCP (UoT) traffic.
pub upstream_tcp: String, pub upstream_tcp: String,
/// Адрес upstream UDP. /// Upstream target for UDP traffic.
pub upstream_udp: String, 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<RwLock<Vec<String>>>; /// 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-узла. /// Token bucket bounding how fast new sessions may be created.
pub async fn run_relay_node(cfg: RelayConfig) -> Result<()> { struct AdmissionLimiter {
let shared_keys: SharedKeys = Arc::new(RwLock::new(Vec::new())); tokens: f64,
last_refill: Instant,
}
// Первоначальная синхронизация ключей impl AdmissionLimiter {
if let Err(e) = sync_keys(&cfg, &shared_keys).await { fn new() -> Self {
tracing::warn!("Relay: initial key sync failed: {}. Will retry.", e); Self { tokens: NEW_SESSION_RATE, last_refill: Instant::now() }
}
/// 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 { } 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);
}
// Фоновый синхронизатор ключей
let cfg_clone = cfg.clone();
let keys_clone = shared_keys.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),
}
}
});
// Запуск 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
}
/// Синхронизация access_keys с upstream API.
async fn sync_keys(cfg: &RelayConfig, shared_keys: &SharedKeys) -> Result<usize> {
let url = format!("{}/api/users", cfg.upstream_api_url.trim_end_matches('/'));
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/<webpath>\" (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<Vec<UserStatsSnapshot>>,
}
let body: ApiResponse = resp.json().await?;
if !body.ok {
anyhow::bail!("API returned error ok=false");
}
let keys: Vec<String> = 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)
}
/// Проверяет 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::<Sha256>::new_from_slice(key.as_bytes()) {
mac.update(ts_bytes);
if mac.verify_slice(provided_mac).is_ok() {
return true;
}
}
}
false false
}
}
} }
// ── UDP Relay ──────────────────────────────────────────────────────────────── /// Entry point.
pub async fn run_relay_node(cfg: RelayConfig) -> Result<()> {
let udp_cfg = cfg.clone();
tokio::spawn(async move {
if let Err(e) = run_udp_relay(udp_cfg).await {
tracing::error!("Relay UDP loop error: {e}");
}
});
async fn run_udp_relay(cfg: RelayConfig, shared_keys: SharedKeys) -> Result<()> { run_tcp_relay(cfg).await
// NAT-таблица: client_addr -> (upstream_socket, last_seen) }
let nat_table: Arc<Mutex<HashMap<SocketAddr, (Arc<UdpSocket>, Instant)>>> =
// ── UDP ──────────────────────────────────────────────────────────────────────
struct UdpSession {
upstream: Arc<UdpSocket>,
last_seen: Instant,
}
async fn run_udp_relay(cfg: RelayConfig) -> Result<()> {
// client address -> the upstream socket carrying that client's flow
let sessions: Arc<Mutex<HashMap<SocketAddr, UdpSession>>> =
Arc::new(Mutex::new(HashMap::new())); Arc::new(Mutex::new(HashMap::new()));
let limiter = Arc::new(Mutex::new(AdmissionLimiter::new()));
for bind_addr in &cfg.listen_addrs { for bind_addr in &cfg.listen_addrs {
let sock = UdpSocket::bind(bind_addr).await?; let sock = Arc::new(
tracing::info!("Relay UDP listening on {}", bind_addr); UdpSocket::bind(bind_addr)
let sock = Arc::new(sock); .await
let upstream_udp = cfg.upstream_udp.clone(); .with_context(|| format!("relay: failed to bind UDP on {bind_addr}"))?,
let keys = shared_keys.clone(); );
let nat = nat_table.clone(); 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 { tokio::spawn(async move {
let mut buf = vec![0u8; 65535]; let mut buf = vec![0u8; 65535];
loop { loop {
let (n, peer) = match sock.recv_from(&mut buf).await { let (len, peer) = match sock.recv_from(&mut buf).await {
Ok(v) => v, 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) => { Err(e) => {
tracing::warn!("Relay UDP: failed to bind upstream socket: {}", e); tracing::warn!("Relay UDP recv error: {e}");
continue; continue;
} }
}; };
if usock.connect(&upstream_udp).await.is_err() {
tracing::warn!("Relay UDP: failed to connect to upstream {}", upstream_udp); // 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; continue;
} }
nat_lock.insert(peer, (usock.clone(), Instant::now())); 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;
}
};
// Задача: читаем ответы от upstream и отправляем клиенту sessions.lock().await.insert(
let usock_rx = usock.clone(); peer,
let client_sock = sock.clone(); UdpSession { upstream: upstream.clone(), last_seen: Instant::now() },
let peer_addr = peer; );
// Reverse direction for this client.
let back_sock = sock.clone();
let sessions_rx = sessions.clone();
tokio::spawn(async move { tokio::spawn(async move {
let mut rbuf = vec![0u8; 65535]; let mut rbuf = vec![0u8; 65535];
loop { loop {
match usock_rx.recv(&mut rbuf).await { match upstream.recv(&mut rbuf).await {
Ok(n) => { Ok(n) => {
let _ = client_sock.send_to(&rbuf[..n], peer_addr).await; 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, Err(_) => break,
} }
} }
sessions_rx.lock().await.remove(&peer);
}); });
usock let _ = sessions
} .lock()
}; .await
.get(&peer)
// Пересылаем пакет в upstream .map(|s| s.upstream.clone())
let _ = upstream_sock.send(&packet).await; .unwrap()
.send(&buf[..len])
.await;
} }
}); });
} }
// Периодически чистим устаревшие NAT записи (timeout 120 сек) // Reclaim idle sessions. Dropping the entry closes the upstream socket,
// which ends that session's reader task.
loop { loop {
tokio::time::sleep(Duration::from_secs(30)).await; tokio::time::sleep(Duration::from_secs(30)).await;
let mut nat_lock = nat_table.lock().await;
let now = Instant::now(); let now = Instant::now();
nat_lock.retain(|_, (_, last)| now.duration_since(*last) < Duration::from_secs(120)); 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());
}
} }
} }
// ── TCP (UoT) Relay ────────────────────────────────────────────────────────── /// 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<Arc<UdpSocket>> {
// 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));
async fn run_tcp_relay(cfg: RelayConfig, shared_keys: SharedKeys) -> Result<()> {
for bind_addr in &cfg.listen_addrs { for bind_addr in &cfg.listen_addrs {
let listener = TcpListener::bind(bind_addr).await?; let listener = TcpListener::bind(bind_addr)
tracing::info!("Relay TCP (UoT) listening on {}", 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_tcp = cfg.upstream_tcp.clone(); let upstream = cfg.upstream_tcp.clone();
let keys = shared_keys.clone(); let live = live.clone();
tokio::spawn(async move { tokio::spawn(async move {
loop { loop {
let (stream, peer_addr) = match listener.accept().await { let (client, peer) = match listener.accept().await {
Ok(v) => v, Ok(v) => v,
Err(e) => { Err(e) => {
tracing::warn!("Relay TCP accept error: {}", e); tracing::warn!("Relay TCP accept error: {e}");
continue; continue;
} }
}; };
let upstream = upstream_tcp.clone(); use std::sync::atomic::Ordering;
let keys_clone = keys.clone(); 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 { tokio::spawn(async move {
if let Err(e) = handle_tcp_client(stream, peer_addr, upstream, keys_clone).await { if let Err(e) = splice_tcp(client, &upstream).await {
tracing::debug!("Relay TCP client {} closed: {}", peer_addr, e); tracing::debug!("Relay TCP {peer} closed: {e}");
} }
live.fetch_sub(1, Ordering::Relaxed);
}); });
} }
}); });
} }
// Держим поток живым
futures_util::future::pending::<()>().await; futures_util::future::pending::<()>().await;
Ok(()) Ok(())
} }
/// Обработка одного TCP (UoT) соединения. /// Splice a client connection to the upstream, byte for byte.
/// ///
/// Алгоритм: /// Nothing is parsed or rewritten: the relay must stay agnostic to the payload,
/// 1. Читаем HTTP-заголовки (фейковый WebSocket upgrade). /// both because the payload is an opaque encrypted stream and because any
/// 2. Извлекаем HMAC-подпись из Authorization: Bearer. /// parsing would be a place for the relay to disagree with the endpoints.
/// 3. Проверяем подпись по синхронизированным ключам. async fn splice_tcp(mut client: TcpStream, upstream_addr: &str) -> Result<()> {
/// 4. Если авторизован — открываем соединение к upstream и пайпим потоки. let mut upstream = tokio::time::timeout(
async fn handle_tcp_client( UPSTREAM_CONNECT_TIMEOUT,
mut client: TcpStream, TcpStream::connect(upstream_addr),
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,
) )
.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 { // Both sides carry latency-sensitive framed traffic; Nagle would add delay
let _ = client.write_all(b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 12\r\nConnection: close\r\n\r\nUnauthorized").await; // for no benefit on an already-batched stream.
anyhow::bail!("signature too short from {}", peer_addr); let _ = client.set_nodelay(true);
} let _ = upstream.set_nodelay(true);
let ts_bytes: [u8; 8] = sig_bytes[0..8].try_into().unwrap(); tokio::io::copy_bidirectional(&mut client, &mut upstream).await?;
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);
Ok(()) 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");
}
}

54
ostp/src/main.rs
View File

@ -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 listen = wizard_prompt("Listen address (host:port)", "0.0.0.0:50000");
let upstream = wizard_prompt("Upstream server address (host:port)", ""); let upstream = wizard_prompt("Upstream server address (host:port)", "");
if upstream.is_empty() { anyhow::bail!("Upstream address cannot be empty."); } 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"); 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!({ let relay_json = serde_json::json!({
"mode": "relay", "mode": "relay",
"listen": listen, "listen": listen,
"upstream_tcp": upstream, "upstream_tcp": upstream,
"upstream_udp": upstream, "upstream_udp": upstream,
"upstream_api_url": api_url,
"upstream_api_token": api_token,
"sync_interval_secs": 30,
"debug": false "debug": false
}); });
@ -1155,7 +1135,9 @@ async fn run_app() -> Result<()> {
println!(" Listen: {:?}", r.listen.primary().cyan()); println!(" Listen: {:?}", r.listen.primary().cyan());
println!(" Upstream TCP: {}", r.upstream_tcp.cyan()); println!(" Upstream TCP: {}", r.upstream_tcp.cyan());
println!(" Upstream UDP: {}", r.upstream_udp.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", "listen": "0.0.0.0:50000",
"upstream_tcp": "TARGET_SERVER_IP:50000", "upstream_tcp": "TARGET_SERVER_IP:50000",
"upstream_udp": "TARGET_SERVER_IP:50000", "upstream_udp": "TARGET_SERVER_IP:50000",
// MUST include the target server's secret api.webpath. The management API is // The relay forwards transparently and holds no keys: sessions are
// nested under it (that path is what hides the panel from scanners), so a // authenticated end-to-end by the target server, which drops anything that
// bare host:port hits a route that does not exist and key sync fails with 404 // fails. Nothing else needs configuring here.
// 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,
"debug": false "debug": false
}"#.to_string() }"#.to_string()
} else { } else {
@ -1438,14 +1414,18 @@ async fn run_app() -> Result<()> {
println!("{} Starting relay node on {:?}", "[ostp]".cyan().bold(), listen_addrs); println!("{} Starting relay node on {:?}", "[ostp]".cyan().bold(), listen_addrs);
println!("{} Upstream TCP: {}", "[ostp]".cyan().bold(), relay_cfg.upstream_tcp); println!("{} Upstream TCP: {}", "[ostp]".cyan().bold(), relay_cfg.upstream_tcp);
println!("{} Upstream UDP: {}", "[ostp]".cyan().bold(), relay_cfg.upstream_udp); 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 { let relay_config = ostp_server::RelayConfig {
listen_addrs, listen_addrs,
upstream_tcp: relay_cfg.upstream_tcp, upstream_tcp: relay_cfg.upstream_tcp,
upstream_udp: relay_cfg.upstream_udp, 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?; ostp_server::relay_node::run_relay_node(relay_config).await?;
} }