Compare commits

..

4 Commits

Author SHA1 Message Date
ospab df1a14d15c chore: release v0.4.3-beta.2 on beta 2026-08-03 18:50:37 +03:00
ospab d915efc715 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.
2026-08-03 18:49:57 +03:00
ospab b673219894 fix(server)!: never fall back to a direct connection when a rule says proxy
Reported: with an upstream proxy configured, Google flags the session because
the server's address and the proxy's exit address differ, and YouTube
geolocates to the server rather than the proxy exit. Both follow from the same
defect in three places — an outbound rule that says Proxy was being satisfied
by connecting DIRECTLY whenever the proxy could not be used:

  - router.rs, UdpSessionRouter::send_to: action Proxy with no UDP proxy
    established fell through to the direct socket.
  - outbound.rs, connect_target: any protocol string other than exactly
    "socks5"/"http" hit a `_ =>` arm that connected directly. A typo, or just
    "SOCKS5", silently un-proxied ALL TCP.
  - outbound.rs, connect_udp_target: non-socks5 upstream returned
    UdpProxySocket::Direct, with a comment noting HTTP cannot carry UDP —
    correct in itself, but the chosen fallback leaks.

The visible symptom is precisely what these produce: TCP goes through the
proxy while UDP (QUIC — which is what YouTube uses) leaves from the server,
so one session presents two exit addresses.

All three now fail closed. A rule asking for the proxy is never honoured by
sending in the clear: a dropped flow is visible and debuggable, a
deanonymising leak is neither. Protocol matching is also case-insensitive
now, and the SOCKS5 UDP ASSOCIATE failure warns unconditionally rather than
only under `debug`, which is why this could go unnoticed.

Behaviour change: with an HTTP upstream, or a broken SOCKS5 UDP ASSOCIATE,
UDP now fails instead of leaking. Where that is genuinely wanted, it must be
stated in the config as an explicit udp rule with action "direct".

Also carries the relay key-sync diagnostics: a 404 there means the URL is
missing the panel's secret webpath (the API is nested under it, not at /api),
which the previous bare "HTTP 404" gave no way to work out.
2026-08-03 17:57:42 +03:00
ospab a1c146aff3 fix(relay): explain the 404 — the API lives under the panel's secret webpath
A relay configured with upstream_api_url = "http://HOST:9090" fails every key
sync with a bare "API returned HTTP 404", which reads like the server is down
or the token is wrong. Neither is true: the management API is nested under the
target server's api.webpath (create_api_router mounts it at
"/{webpath}/api"), because that secret segment is what keeps 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.

Nothing said so — the config template, the wizard prompt and the shipped
example all suggested exactly the host:port form that cannot work.

  - sync_keys now reports the full URL and, for 404 specifically, states that
    the webpath must be included and what the URL should look like. 401 is
    called out separately as a token mismatch, since the two are otherwise
    indistinguishable from the log.
  - The relay config template, the shipped example and the wizard prompt now
    show the path-bearing form, and the wizard warns when the URL entered has
    no path segment rather than letting it fail later.

Docs under docs/ and the wiki are being rewritten concurrently and are left
alone here.
2026-07-31 20:00:29 +03:00
8 changed files with 514 additions and 383 deletions

View File

@ -2,5 +2,5 @@
"target_version": "0.4.3",
"branch": "beta",
"alpha_iteration": 0,
"beta_iteration": 1
"beta_iteration": 2
}

View File

@ -20,9 +20,15 @@
// Адрес следующего узла в цепочке UDP
"upstream_udp": "TARGET_SERVER_IP:50000",
// URL API конечного (целевого) сервера для синхронизации access_keys
// Должен быть доступен с этого relay-сервера (можно через SSH-туннель)
"upstream_api_url": "http://TARGET_SERVER_IP:9090",
// URL API конечного (целевого) сервера для синхронизации access_keys.
// Должен быть доступен с этого relay-сервера (можно через SSH-туннель).
//
// ВАЖНО: URL обязан включать секретный путь панели (api.webpath целевого
// сервера). Management API смонтирован ВНУТРИ этого пути именно он скрывает
// панель от сканеров, поэтому голый host:port попадает в несуществующий
// маршрут, и синхронизация падает с 404 ещё до проверки токена.
// Это тот же адрес, по которому вы открываете веб-панель.
"upstream_api_url": "http://TARGET_SERVER_IP:9090/TARGET_SERVER_WEBPATH",
// Bearer-токен для доступа к API целевого сервера
// Должен совпадать с api.token в конфиге target-сервера

View File

@ -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<bool>,
}
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)]

View File

@ -16,7 +16,7 @@ publish_to: 'none' # Remove this line if you wish to publish to pub.dev
# https://developer.apple.com/library/archive/documentation/General/Reference/InfoPlistKeyReference/Articles/CoreFoundationKeys.html
# In Windows, build-name is used as the major, minor, and patch parts
# of the product and file versions while build-number is used as the build suffix.
version: 0.4.3+26
version: 0.4.3+27
environment:
sdk: ^3.11.4

View File

@ -48,10 +48,23 @@ pub async fn connect_target(
}
if action == OutboundAction::Proxy {
let proxy_addr = format!("{}:{}", outbound.address, outbound.port);
return match outbound.protocol.as_str() {
// Case-insensitive: a config saying "SOCKS5" means the same thing
// as "socks5", and silently treating it as unknown is a trap.
return match outbound.protocol.to_ascii_lowercase().as_str() {
"socks5" => connect_via_socks5(&proxy_addr, target).await,
"http" => connect_via_http(&proxy_addr, target).await,
_ => connect_direct(target, connect_timeout).await,
// FAIL CLOSED. This used to fall through to a direct
// connection, so any unrecognised protocol string — a typo,
// a case difference, an empty value — silently sent ALL TCP
// straight out of the server while the operator believed it
// was proxied. Combined with the same bug on the UDP path,
// that is how one session ends up presenting two different
// exit addresses to the remote site.
other => Err(anyhow::anyhow!(
"outbound.protocol is \"{other}\", which is not a supported proxy type \
(expected \"socks5\" or \"http\"); refusing to connect to {target} \
directly, because the rules asked for the proxy"
)),
};
}
}
@ -370,10 +383,22 @@ pub async fn connect_udp_target(
}
if action == OutboundAction::Proxy {
let proxy_addr = format!("{}:{}", outbound.address, outbound.port);
if outbound.protocol == "socks5" {
if outbound.protocol.eq_ignore_ascii_case("socks5") {
return connect_udp_via_socks5(&proxy_addr, server_udp).await;
}
// HTTP CONNECT does not support UDP. Fallback to direct.
// FAIL CLOSED. HTTP CONNECT genuinely cannot carry UDP — but the
// answer to that is not to send the datagrams in the clear. The
// previous "fallback to direct" honoured a Proxy rule by
// egressing from the server's own address, so with an HTTP
// upstream every UDP flow (QUIC, DNS) leaked while TCP stayed
// proxied, presenting two exit IPs to the same remote site.
return Err(anyhow::anyhow!(
"outbound rules route UDP to {target} through the proxy, but the upstream \
protocol is \"{}\", which cannot carry UDP. Refusing to send directly. \
Use a socks5 upstream, or add an explicit udp rule with action \"direct\" \
or \"block\" so the intent is recorded in the config.",
outbound.protocol
));
}
}
}

View File

@ -1,403 +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<String>,
/// Адрес 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<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-узла.
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);
impl AdmissionLimiter {
fn new() -> Self {
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 {
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() {
anyhow::bail!("API returned HTTP {}", 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
}
}
}
// ── 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<()> {
// NAT-таблица: client_addr -> (upstream_socket, last_seen)
let nat_table: Arc<Mutex<HashMap<SocketAddr, (Arc<UdpSocket>, Instant)>>> =
run_tcp_relay(cfg).await
}
// ── 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()));
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 {
let (len, 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);
tracing::warn!("Relay UDP recv error: {e}");
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;
}
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 и отправляем клиенту
let usock_rx = usock.clone();
let client_sock = sock.clone();
let peer_addr = peer;
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 usock_rx.recv(&mut rbuf).await {
match upstream.recv(&mut rbuf).await {
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,
}
}
sessions_rx.lock().await.remove(&peer);
});
usock
}
};
// Пересылаем пакет в upstream
let _ = upstream_sock.send(&packet).await;
let _ = sessions
.lock()
.await
.get(&peer)
.map(|s| s.upstream.clone())
.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 {
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));
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 {
let listener = TcpListener::bind(bind_addr).await?;
tracing::info!("Relay TCP (UoT) listening on {}", bind_addr);
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_tcp = cfg.upstream_tcp.clone();
let keys = shared_keys.clone();
let upstream = cfg.upstream_tcp.clone();
let live = live.clone();
tokio::spawn(async move {
loop {
let (stream, peer_addr) = match listener.accept().await {
let (client, peer) = match listener.accept().await {
Ok(v) => v,
Err(e) => {
tracing::warn!("Relay TCP accept error: {}", e);
tracing::warn!("Relay TCP accept error: {e}");
continue;
}
};
let upstream = upstream_tcp.clone();
let keys_clone = keys.clone();
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) = handle_tcp_client(stream, peer_addr, upstream, keys_clone).await {
tracing::debug!("Relay TCP client {} closed: {}", peer_addr, e);
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");
}
}

View File

@ -47,12 +47,29 @@ impl Router {
let mut proxy = None;
if let Some(ref c) = cfg {
if c.enabled && c.protocol == "socks5" {
if c.enabled {
if c.protocol == "socks5" {
let proxy_addr = format!("{}:{}", c.address, c.port);
if let Ok(p) = crate::outbound::connect_udp_via_socks5(&proxy_addr, server_udp.clone()).await {
proxy = Some(Arc::new(p));
} else if self.debug {
tracing::warn!("Failed to establish SOCKS5 UDP Associate");
match crate::outbound::connect_udp_via_socks5(&proxy_addr, server_udp.clone()).await {
Ok(p) => proxy = Some(Arc::new(p)),
// Warn unconditionally, not only under `debug`. Every UDP
// flow the rules want proxied is now dropped instead of
// sent, so an operator who cannot see this has a session
// where TCP works and UDP silently does not.
Err(e) => tracing::warn!(
"SOCKS5 UDP ASSOCIATE to {proxy_addr} failed: {e}. UDP that the \
outbound rules route through the proxy will be DROPPED (it is not \
sent directly, which would expose this server's address)."
),
}
} else {
tracing::warn!(
"Upstream proxy protocol is '{}', which cannot carry UDP. UDP matching \
a Proxy rule will be DROPPED. Use a socks5 upstream for UDP, or add an \
explicit udp rule with action \"direct\" or \"block\" to make the \
intent explicit.",
c.protocol
);
}
}
}
@ -87,9 +104,28 @@ impl UdpSessionRouter {
return Err(anyhow::anyhow!("blocked by outbound udp rule: {}", target));
}
if action == crate::outbound::OutboundAction::Proxy {
if let Some(p) = &self.proxy {
return p.send_to(data, target).await;
}
return match &self.proxy {
Some(p) => p.send_to(data, target).await,
// FAIL CLOSED. This used to fall through to the direct
// socket, so whenever the UDP proxy was unavailable —
// the SOCKS5 UDP ASSOCIATE failed, or the upstream is an
// HTTP proxy, which cannot carry UDP at all — every UDP
// datagram silently egressed from the server's own
// address while TCP still went through the proxy. The
// session then had two different exit IPs, which is what
// Google flags and why YouTube (QUIC, i.e. UDP/443)
// geolocated to the server instead of the proxy exit.
//
// A rule that says "proxy" must never be satisfied by
// sending in the clear: a dropped datagram is visible and
// debuggable, a deanonymising leak is neither.
None => Err(anyhow::anyhow!(
"outbound rule requires the proxy for UDP to {target}, but no UDP \
proxy is available (SOCKS5 UDP ASSOCIATE failed, or the upstream \
is an HTTP proxy, which cannot carry UDP) - dropping rather than \
leaking the server's own address"
)),
};
}
}
}

View File

@ -799,18 +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 server API URL (e.g. http://1.2.3.4:9090)", "");
let api_token = wizard_prompt("Upstream API token (leave blank if none)", "");
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
});
@ -1137,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());
}
}
}
}
@ -1213,9 +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",
"upstream_api_url": "http://TARGET_SERVER_IP:9090",
"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 {
@ -1414,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?;
}