mirror of https://github.com/ospab/ostp.git
Compare commits
No commits in common. "master" and "v0.4.2" have entirely different histories.
|
|
@ -477,19 +477,12 @@ jobs:
|
|||
working-directory: ostp-gui
|
||||
run: |
|
||||
npm install
|
||||
# TUN mode shells out to this helper, elevated via pkexec. Only the
|
||||
# Windows job used to build it, so the Linux package shipped without
|
||||
# it and TUN could never start.
|
||||
cargo build -p ostp-tun-helper --release --target ${{ matrix.target }} --manifest-path ../Cargo.toml
|
||||
npx tauri build --no-bundle --target ${{ matrix.target }}
|
||||
|
||||
- name: Package Portable Tarball
|
||||
run: |
|
||||
set -euo pipefail
|
||||
mkdir ostp-linux-gui-${{ matrix.arch }}
|
||||
cp ostp-gui/src-tauri/target/${{ matrix.target }}/release/ostp-gui ostp-linux-gui-${{ matrix.arch }}/
|
||||
# The GUI looks for the helper next to its own executable first.
|
||||
cp target/${{ matrix.target }}/release/ostp-tun-helper ostp-linux-gui-${{ matrix.arch }}/
|
||||
tar -czf ostp-linux-gui-${{ matrix.arch }}.tar.gz ostp-linux-gui-${{ matrix.arch }}
|
||||
|
||||
- name: Upload to GitHub Release
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"target_version": "0.4.4",
|
||||
"target_version": "0.4.2",
|
||||
"branch": "master",
|
||||
"alpha_iteration": 0,
|
||||
"beta_iteration": 0
|
||||
"beta_iteration": 5
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1386,7 +1386,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
|
|||
|
||||
[[package]]
|
||||
name = "ostp"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
|
|
@ -1409,7 +1409,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-client"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
|
|
@ -1440,7 +1440,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-core"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"bytes",
|
||||
|
|
@ -1474,7 +1474,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-server"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
|
|
@ -1507,7 +1507,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-tun"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"libc",
|
||||
|
|
@ -1519,7 +1519,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-tun-helper"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"chrono",
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ resolver = "2"
|
|||
[workspace.package]
|
||||
edition = "2021"
|
||||
license = "AGPL-3.0"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
|
||||
[workspace.dependencies]
|
||||
anyhow = "1.0"
|
||||
|
|
|
|||
|
|
@ -20,15 +20,9 @@
|
|||
// Адрес следующего узла в цепочке — UDP
|
||||
"upstream_udp": "TARGET_SERVER_IP:50000",
|
||||
|
||||
// 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",
|
||||
// URL API конечного (целевого) сервера для синхронизации access_keys
|
||||
// Должен быть доступен с этого relay-сервера (можно через SSH-туннель)
|
||||
"upstream_api_url": "http://TARGET_SERVER_IP:9090",
|
||||
|
||||
// Bearer-токен для доступа к API целевого сервера
|
||||
// Должен совпадать с api.token в конфиге target-сервера
|
||||
|
|
|
|||
|
|
@ -23,12 +23,6 @@ use crate::tunnel::{ProxyEvent, ProxyToClientMsg};
|
|||
/// candidate address is tried.
|
||||
const UOT_CONNECT_TIMEOUT: Duration = Duration::from_secs(4);
|
||||
|
||||
/// How long to keep retrying a resume-triggered reconnect before handing the
|
||||
/// problem back to the ordinary stall path. That path is what releases the
|
||||
/// system proxy, so this is really a bound on how long the machine may be left
|
||||
/// with no working internet at all after waking.
|
||||
const RESUME_RECONNECT_GIVE_UP: Duration = Duration::from_secs(45);
|
||||
|
||||
static SOCKET_PROTECTOR: std::sync::OnceLock<Box<dyn Fn(i32) -> bool + Send + Sync>> = std::sync::OnceLock::new();
|
||||
|
||||
pub fn set_socket_protector<F>(f: F)
|
||||
|
|
@ -143,21 +137,6 @@ pub struct Bridge {
|
|||
last_rtt_ms: f64,
|
||||
last_sample_at: Instant,
|
||||
last_valid_recv: Instant,
|
||||
/// Set when a suspend/resume is detected, cleared once a reconnect actually
|
||||
/// succeeds. Waking is precisely when the network is least likely to be
|
||||
/// ready — Wi-Fi has not reassociated yet — so a single attempt fired
|
||||
/// milliseconds after resume usually fails, and a one-shot forced reconnect
|
||||
/// then fell back to the ordinary 25s stall heuristic. That heuristic keys
|
||||
/// off a monotonic clock which does not advance while the machine is
|
||||
/// asleep, so it could take a further 25s of real uptime to fire, or not
|
||||
/// fire at all. Retrying until success removes the dependency on either.
|
||||
forced_reconnect_pending: bool,
|
||||
last_forced_reconnect_try: Instant,
|
||||
/// Wall-clock start of the current resume-reconnect campaign, used to bound
|
||||
/// it. Wall clock rather than Instant because the monotonic clock does not
|
||||
/// advance across suspend on Windows, so it cannot measure anything that
|
||||
/// begins at wake.
|
||||
forced_reconnect_started: Option<SystemTime>,
|
||||
}
|
||||
|
||||
impl Bridge {
|
||||
|
|
@ -194,9 +173,6 @@ impl Bridge {
|
|||
last_rtt_ms: 0.0,
|
||||
last_sample_at: Instant::now(),
|
||||
last_valid_recv: Instant::now(),
|
||||
forced_reconnect_pending: false,
|
||||
last_forced_reconnect_try: Instant::now(),
|
||||
forced_reconnect_started: None,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -279,64 +255,7 @@ impl Bridge {
|
|||
let _ = tx.send(UiEvent::Log(format!(
|
||||
"Resumed after ~{}s suspend — forcing clean reconnect", wall_gap.as_secs()
|
||||
))).await;
|
||||
self.forced_reconnect_pending = true;
|
||||
self.forced_reconnect_started = Some(SystemTime::now());
|
||||
self.last_forced_reconnect_try = Instant::now() - Duration::from_secs(60);
|
||||
}
|
||||
|
||||
// Give up if resume reconnects keep failing. Retrying forever
|
||||
// looks harmless but is not: the system proxy stays pointed at
|
||||
// our local listener the whole time, so the machine has NO
|
||||
// working internet — not merely no tunnel — while the UI sits
|
||||
// on "connecting". Handing the retry to the ordinary keepalive
|
||||
// path restores the proxy through its hard-timeout branch,
|
||||
// which force=true deliberately skips.
|
||||
//
|
||||
// Measured on the wall clock: Instant does not advance across
|
||||
// suspend on Windows (QPC stops), so a monotonic deadline can
|
||||
// not bound anything that starts at wake.
|
||||
if self.forced_reconnect_pending {
|
||||
let pending_for = self
|
||||
.forced_reconnect_started
|
||||
.and_then(|t| t.elapsed().ok())
|
||||
.unwrap_or_default();
|
||||
if pending_for > RESUME_RECONNECT_GIVE_UP {
|
||||
self.forced_reconnect_pending = false;
|
||||
self.forced_reconnect_started = None;
|
||||
let _ = tx.send(UiEvent::Log(format!(
|
||||
"Reconnect after suspend failed for {}s — releasing the system \
|
||||
proxy so normal traffic works; will keep retrying in the \
|
||||
background",
|
||||
pending_for.as_secs()
|
||||
))).await;
|
||||
// Make the ordinary stall path fire on the next
|
||||
// keepalive tick: it is the one that tears the proxy
|
||||
// back down (or, with kill switch on, deliberately
|
||||
// keeps blocking).
|
||||
self.last_valid_recv = Instant::now()
|
||||
.checked_sub(Duration::from_secs(3600))
|
||||
.unwrap_or_else(Instant::now);
|
||||
}
|
||||
}
|
||||
|
||||
// Keep retrying a resume-triggered reconnect until one lands.
|
||||
// The first attempt fires within half a second of waking, when
|
||||
// the NIC is typically still reassociating, so treating it as
|
||||
// one-shot left the tunnel dead until some other timer noticed.
|
||||
if self.running
|
||||
&& self.forced_reconnect_pending
|
||||
&& self.last_forced_reconnect_try.elapsed() >= Duration::from_secs(3)
|
||||
{
|
||||
self.last_forced_reconnect_try = Instant::now();
|
||||
self.handle_keepalive(true, &mut sessions_opt, &mut udp_rx_opt, &mut proxy_guard, &mut stream_map, &tx, &proxy_tx, &mut proxy_rx).await;
|
||||
// handle_keepalive refreshes last_valid_recv only when a
|
||||
// session was actually established, so this is a real
|
||||
// success check rather than "we tried".
|
||||
if self.last_valid_recv.elapsed() < Duration::from_secs(3) {
|
||||
self.forced_reconnect_pending = false;
|
||||
self.forced_reconnect_started = None;
|
||||
let _ = tx.send(UiEvent::Log("Reconnected after suspend".into())).await;
|
||||
}
|
||||
}
|
||||
if self.running {
|
||||
self.emit_metrics(&tx).await;
|
||||
|
|
@ -353,20 +272,7 @@ impl Bridge {
|
|||
}
|
||||
}
|
||||
proxy_ev = proxy_rx.recv(), if self.running && sessions_opt.as_ref().map(|s| {
|
||||
// Upper bound matches MAX_CWND_PACKETS in ostp-core's congestion
|
||||
// controller. The old 16384 ceiling let ~20 MB sit in flight,
|
||||
// which on a mobile uplink is minutes of buffered queue rather
|
||||
// than throughput — the app kept handing over data long after
|
||||
// the path had stopped draining it.
|
||||
// Two independent gates. cwnd bounds how much may be in
|
||||
// flight; pacing bounds how FAST it is released. Without the
|
||||
// second, a full window goes out back-to-back and lands in
|
||||
// the bottleneck's buffer as standing queue rather than
|
||||
// throughput — the thing that produced multi-second RTT.
|
||||
s.iter().any(|ses| {
|
||||
ses.machine.in_flight_count() < ses.machine.cwnd_packets().clamp(16, 1024)
|
||||
&& ses.machine.can_pace_packet()
|
||||
})
|
||||
s.iter().any(|ses| ses.machine.in_flight_count() < ses.machine.cwnd_packets().clamp(16, 16384))
|
||||
}).unwrap_or(true) => {
|
||||
self.handle_proxy_event(proxy_ev, &mut sessions_opt, &mut stream_map, &tx, &proxy_tx).await;
|
||||
}
|
||||
|
|
@ -1020,21 +926,7 @@ impl Bridge {
|
|||
Ok(addrs) => addrs.collect(),
|
||||
Err(e) => return Err(anyhow::anyhow!("failed to resolve server address {}: {}", self.server_addr, e)),
|
||||
};
|
||||
// IPv4 first. Addresses are tried strictly in order, each burning its
|
||||
// full retry budget before the next is touched, so this ordering decides
|
||||
// how long a bad family stalls the whole connect. Mobile carriers
|
||||
// routinely hand out IPv6 with no working route and BLACKHOLE it rather
|
||||
// than rejecting, so every IPv6 candidate costs the full timeout budget
|
||||
// — with several AAAA records the working IPv4 address was not reached
|
||||
// for tens of seconds. (The same ordering bug was already fixed on the
|
||||
// server's outbound path and in the UoT connect.)
|
||||
resolved_addrs.sort_by_key(|addr| if addr.is_ipv6() { 1 } else { 0 });
|
||||
|
||||
// NAT64 is a fallback for IPv6-only networks. Retrying it per failing
|
||||
// address multiplied an already-long connect: each attempt re-runs a DNS
|
||||
// lookup and another full round of handshake retries, for a path that
|
||||
// either works for the whole network or for none of it.
|
||||
let mut nat64_attempted = false;
|
||||
resolved_addrs.sort_by_key(|addr| if addr.is_ipv6() { 0 } else { 1 });
|
||||
|
||||
let mut last_err = anyhow::anyhow!("no IP addresses resolved for {}", self.server_addr);
|
||||
|
||||
|
|
@ -1047,8 +939,7 @@ impl Bridge {
|
|||
let socket = match self.try_connect_transport(target_ip, port).await {
|
||||
Ok(sock) => sock,
|
||||
Err(e) => {
|
||||
if let (std::net::IpAddr::V4(ipv4), false) = (target_ip, nat64_attempted) {
|
||||
nat64_attempted = true;
|
||||
if let std::net::IpAddr::V4(ipv4) = target_ip {
|
||||
tx.send(UiEvent::Log(format!("Direct IPv4 connection failed: {}. Trying NAT64 fallback...", e))).await.ok();
|
||||
let nat64_ipv6 = synthesize_nat64(ipv4).await;
|
||||
match self.try_connect_transport(std::net::IpAddr::V6(nat64_ipv6), port).await {
|
||||
|
|
@ -1129,8 +1020,7 @@ impl Bridge {
|
|||
let (final_socket, size) = if success {
|
||||
(socket, size)
|
||||
} else {
|
||||
if let (std::net::IpAddr::V4(ipv4), false) = (target_ip, nat64_attempted) {
|
||||
nat64_attempted = true;
|
||||
if let std::net::IpAddr::V4(ipv4) = target_ip {
|
||||
tx.send(UiEvent::Log("Direct IPv4 handshake timed out. Trying NAT64 fallback...".to_string())).await.ok();
|
||||
let nat64_ipv6 = synthesize_nat64(ipv4).await;
|
||||
match self.try_connect_transport(std::net::IpAddr::V6(nat64_ipv6), port).await {
|
||||
|
|
@ -1374,19 +1264,8 @@ fn next_profile(current: TrafficProfile) -> TrafficProfile {
|
|||
}
|
||||
|
||||
async fn synthesize_nat64(ip: std::net::Ipv4Addr) -> std::net::Ipv6Addr {
|
||||
// Well-known prefix (RFC 6052), used if discovery doesn't answer in time.
|
||||
let mut prefix = [0x00, 0x64, 0xff, 0x9b, 0, 0, 0, 0, 0, 0, 0, 0];
|
||||
// Bound the discovery lookup. This runs on exactly the networks that are
|
||||
// already misbehaving, where the resolver can hang for tens of seconds
|
||||
// before giving up — unbounded, it was a large part of why connecting over
|
||||
// a broken mobile network took minutes. Falling back to the well-known
|
||||
// prefix is strictly better than waiting.
|
||||
let discovery = tokio::time::timeout(
|
||||
Duration::from_secs(2),
|
||||
tokio::net::lookup_host("ipv4only.arpa:80"),
|
||||
)
|
||||
.await;
|
||||
if let Ok(Ok(addrs)) = discovery {
|
||||
if let Ok(addrs) = tokio::net::lookup_host("ipv4only.arpa:80").await {
|
||||
for addr in addrs {
|
||||
if let std::net::SocketAddr::V6(v6) = addr {
|
||||
let octets = v6.ip().octets();
|
||||
|
|
|
|||
|
|
@ -418,22 +418,19 @@ pub struct RelayServerConfig {
|
|||
pub upstream_tcp: String,
|
||||
/// Upstream address for UDP traffic
|
||||
pub upstream_udp: String,
|
||||
// ── 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)]
|
||||
/// Target server's API URL, for key sync
|
||||
pub upstream_api_url: String,
|
||||
/// Bearer token for the target server's API
|
||||
#[serde(default)]
|
||||
pub upstream_api_token: String,
|
||||
#[serde(default)]
|
||||
/// Key sync interval in seconds (default 30)
|
||||
#[serde(default = "default_sync_interval")]
|
||||
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)]
|
||||
|
|
|
|||
|
|
@ -39,9 +39,6 @@ pub struct CongestionController {
|
|||
loss_count: u32,
|
||||
/// Pacing rate: bytes per second
|
||||
pacing_rate: u64,
|
||||
/// Token-bucket allowance for pacing, in bytes.
|
||||
pacing_tokens: f64,
|
||||
pacing_last_refill: Instant,
|
||||
/// MTU estimate (used for cwnd → packet count conversion)
|
||||
mtu: u64,
|
||||
/// Min RTT expiry: re-probe after 10 seconds
|
||||
|
|
@ -68,20 +65,6 @@ const MIN_CWND_PACKETS: u64 = 2;
|
|||
/// Min RTT expiry window (after which we re-probe)
|
||||
const MIN_RTT_EXPIRY: Duration = Duration::from_secs(10);
|
||||
/// Minimum RTO (RFC 6298: 1s in TCP; we use 50ms since we own the protocol)
|
||||
/// Absolute ceiling on the congestion window, in packets. At a ~1200-byte MTU
|
||||
/// this is roughly 1.2 MB in flight — already far above the bandwidth-delay
|
||||
/// product of any link this protocol realistically runs over, so anything
|
||||
/// beyond it is standing queue, not throughput. The client previously allowed
|
||||
/// up to 16384 packets (~20 MB), which on a mobile uplink is minutes of buffer.
|
||||
const MAX_CWND_PACKETS: u64 = 1024;
|
||||
/// SRTT/min_rtt ratio at which slow start stops. Doubling is what fills a deep
|
||||
/// buffer fastest, so growth must end when the queue starts building rather
|
||||
/// than waiting for a loss that a deep buffer may never produce.
|
||||
const RTT_INFLATION_EXIT_SLOW_START: f64 = 2.0;
|
||||
/// SRTT/min_rtt ratio treated as a standing queue that must be actively drained.
|
||||
const RTT_INFLATION_BACKOFF: f64 = 4.0;
|
||||
/// How much pacing allowance may accumulate, expressed as time-at-rate.
|
||||
const PACING_BURST: Duration = Duration::from_millis(10);
|
||||
const RTO_MIN: Duration = Duration::from_millis(50);
|
||||
/// Maximum RTO
|
||||
const RTO_MAX: Duration = Duration::from_secs(16);
|
||||
|
|
@ -130,50 +113,9 @@ impl CongestionController {
|
|||
min_rtt_stamp: now,
|
||||
slow_start_losses: 0,
|
||||
slow_start_loss_window_start: now,
|
||||
pacing_tokens: (INITIAL_CWND_PACKETS * mtu) as f64,
|
||||
pacing_last_refill: now,
|
||||
}
|
||||
}
|
||||
|
||||
/// Bytes of pacing allowance available right now, without consuming any.
|
||||
///
|
||||
/// Read-only so the send path can use it as an admission check before it
|
||||
/// commits to building a datagram.
|
||||
pub fn pacing_available(&self) -> f64 {
|
||||
let elapsed = self.pacing_last_refill.elapsed().as_secs_f64();
|
||||
(self.pacing_tokens + elapsed * self.pacing_rate as f64).min(self.pacing_burst())
|
||||
}
|
||||
|
||||
/// Whether at least one full-size packet may be released right now.
|
||||
pub fn can_pace_packet(&self) -> bool {
|
||||
self.pacing_available() >= self.mtu as f64
|
||||
}
|
||||
|
||||
/// Ceiling on accumulated allowance.
|
||||
///
|
||||
/// Pacing intervals here are fractions of a millisecond, so releasing
|
||||
/// strictly one packet at a time would need a sub-millisecond timer per
|
||||
/// packet. Instead we allow a short burst — the same trade every real
|
||||
/// pacing implementation makes — sized so the loop's existing ~10ms wakeups
|
||||
/// can still saturate the configured rate, with a small floor so a
|
||||
/// cold/low estimate can never wedge sending entirely.
|
||||
fn pacing_burst(&self) -> f64 {
|
||||
let by_rate = self.pacing_rate as f64 * PACING_BURST.as_secs_f64();
|
||||
by_rate.max((self.mtu * 4) as f64)
|
||||
}
|
||||
|
||||
/// Refill from elapsed time and deduct `bytes`. Called on the real send
|
||||
/// path; allowance is permitted to go negative so an oversized packet still
|
||||
/// pays for itself rather than being released for free.
|
||||
fn consume_pacing(&mut self, bytes: u64) {
|
||||
let now = Instant::now();
|
||||
let elapsed = now.duration_since(self.pacing_last_refill).as_secs_f64();
|
||||
self.pacing_last_refill = now;
|
||||
self.pacing_tokens =
|
||||
(self.pacing_tokens + elapsed * self.pacing_rate as f64).min(self.pacing_burst())
|
||||
- bytes as f64;
|
||||
}
|
||||
|
||||
/// Returns the current congestion window in bytes.
|
||||
pub fn cwnd(&self) -> u64 {
|
||||
self.cwnd
|
||||
|
|
@ -225,11 +167,6 @@ impl CongestionController {
|
|||
/// Record that we sent `bytes` of data.
|
||||
pub fn on_send(&mut self, bytes: u64) {
|
||||
self.bytes_in_flight = self.bytes_in_flight.saturating_add(bytes);
|
||||
// Charge the pacing bucket here rather than at the admission check, so
|
||||
// every byte that actually reaches the wire is paid for exactly once —
|
||||
// including retransmits, which are precisely what must not be allowed
|
||||
// to bypass the rate limit and pile into an already-full queue.
|
||||
self.consume_pacing(bytes);
|
||||
}
|
||||
|
||||
/// Record that `bytes` were acknowledged but WITHOUT a usable RTT sample
|
||||
|
|
@ -261,46 +198,9 @@ impl CongestionController {
|
|||
|
||||
/// Congestion-window growth shared by both ACK paths (slow start / probe).
|
||||
fn grow_window(&mut self, bytes: u64) {
|
||||
// ── Delay-based congestion signal ────────────────────────────────────
|
||||
// A loss-only controller is blind on a deeply-buffered path, and mobile
|
||||
// carrier buffers are very deep: they absorb a burst instead of dropping
|
||||
// it, so no loss is ever signalled and cwnd keeps growing. The queue —
|
||||
// not the link — is what grows, and the standing delay it adds shows up
|
||||
// as RTT inflating far above the path's floor. Left unchecked this is a
|
||||
// positive feedback loop: bigger queue -> larger RTT samples -> larger
|
||||
// SRTT -> larger RTO -> retransmits pile on -> bigger queue, which is
|
||||
// how a session ends up reporting multi-second (even multi-minute) RTT
|
||||
// and stalls video until the buffer finally drains or the user
|
||||
// reconnects. Treat sustained RTT inflation as congestion in its own
|
||||
// right, exactly as it is.
|
||||
let inflation = if self.rtt_initialized && !self.min_rtt.is_zero() {
|
||||
self.srtt.as_secs_f64() / self.min_rtt.as_secs_f64()
|
||||
} else {
|
||||
1.0
|
||||
};
|
||||
|
||||
if inflation >= RTT_INFLATION_BACKOFF {
|
||||
// Standing queue is severe — actively drain it.
|
||||
self.cwnd = (self.cwnd / 2).max(MIN_CWND_PACKETS * self.mtu);
|
||||
self.ssthresh = self.cwnd;
|
||||
self.phase = Phase::ProbeBandwidth;
|
||||
tracing::debug!(cwnd = self.cwnd, inflation, "congestion: draining standing queue");
|
||||
self.clamp_cwnd();
|
||||
return;
|
||||
}
|
||||
|
||||
// State machine
|
||||
match self.phase {
|
||||
Phase::SlowStart => {
|
||||
// Exponential doubling is what fills a deep buffer fastest, so
|
||||
// leave slow start as soon as the queue starts to build rather
|
||||
// than waiting for the loss that may never come.
|
||||
if inflation >= RTT_INFLATION_EXIT_SLOW_START {
|
||||
self.ssthresh = self.cwnd;
|
||||
self.phase = Phase::ProbeBandwidth;
|
||||
tracing::debug!(cwnd = self.cwnd, inflation, "congestion: RTT inflation ended slow start");
|
||||
self.clamp_cwnd();
|
||||
return;
|
||||
}
|
||||
// Exponential growth: increase cwnd by acked bytes (doubles per RTT)
|
||||
self.cwnd = self.cwnd.saturating_add(bytes);
|
||||
if self.cwnd >= self.ssthresh {
|
||||
|
|
@ -313,21 +213,6 @@ impl CongestionController {
|
|||
self.cwnd = self.cwnd.saturating_add(bytes * self.mtu / self.cwnd.max(1));
|
||||
}
|
||||
}
|
||||
|
||||
self.clamp_cwnd();
|
||||
}
|
||||
|
||||
/// Hard ceiling on the congestion window.
|
||||
///
|
||||
/// Independent of any estimate: no real path this protocol runs over has a
|
||||
/// bandwidth-delay product anywhere near this, so a window above it is
|
||||
/// buffered queue rather than data in transit. Without it, slow start on a
|
||||
/// buffer that never drops could grow the window into the tens of megabytes.
|
||||
fn clamp_cwnd(&mut self) {
|
||||
let ceiling = MAX_CWND_PACKETS.saturating_mul(self.mtu);
|
||||
if self.cwnd > ceiling {
|
||||
self.cwnd = ceiling;
|
||||
}
|
||||
}
|
||||
|
||||
/// Record a loss event.
|
||||
|
|
@ -447,94 +332,6 @@ mod tests {
|
|||
assert!(cc.cwnd() < initial);
|
||||
}
|
||||
|
||||
/// The bufferbloat case: a deep buffer absorbs everything, so NOTHING is
|
||||
/// ever lost, but the standing queue inflates RTT. A loss-only controller
|
||||
/// grows cwnd forever here — which is how a session ends up reporting
|
||||
/// multi-second RTT and stalling video.
|
||||
#[test]
|
||||
fn test_rtt_inflation_halts_growth_without_any_loss() {
|
||||
let mut cc = CongestionController::new(1200);
|
||||
|
||||
// Establish a low path floor; this becomes min_rtt.
|
||||
for _ in 0..4 {
|
||||
cc.on_send(1200);
|
||||
cc.on_ack(1200, Duration::from_millis(20));
|
||||
}
|
||||
let cwnd_before = cc.cwnd();
|
||||
|
||||
// Queue builds: RTT climbs far above the floor, still zero loss.
|
||||
for _ in 0..20 {
|
||||
cc.on_send(1200);
|
||||
cc.on_ack(1200, Duration::from_millis(400));
|
||||
}
|
||||
|
||||
assert!(
|
||||
cc.cwnd() <= cwnd_before,
|
||||
"cwnd kept growing while the queue was inflating RTT ({} -> {})",
|
||||
cwnd_before,
|
||||
cc.cwnd()
|
||||
);
|
||||
}
|
||||
|
||||
/// Pacing must actually bound the release rate: draining the bucket has to
|
||||
/// deny the next packet. Without this the congestion window alone decides,
|
||||
/// and a whole window leaves back-to-back.
|
||||
#[test]
|
||||
fn test_pacing_bucket_denies_once_drained() {
|
||||
let mut cc = CongestionController::new(1200);
|
||||
assert!(cc.can_pace_packet(), "a fresh controller must allow sending");
|
||||
|
||||
// Spend well beyond one burst allowance.
|
||||
let burst_bytes = cc.pacing_available();
|
||||
let mut spent = 0.0;
|
||||
while spent <= burst_bytes + 1200.0 {
|
||||
cc.on_send(1200);
|
||||
spent += 1200.0;
|
||||
}
|
||||
|
||||
assert!(
|
||||
!cc.can_pace_packet(),
|
||||
"pacing allowed unbounded sending: {} bytes still available after spending {}",
|
||||
cc.pacing_available(),
|
||||
spent
|
||||
);
|
||||
}
|
||||
|
||||
/// The allowance must refill over time, or sending would stall permanently
|
||||
/// once the first burst is spent.
|
||||
#[test]
|
||||
fn test_pacing_bucket_refills_over_time() {
|
||||
let mut cc = CongestionController::new(1200);
|
||||
while cc.can_pace_packet() {
|
||||
cc.on_send(1200);
|
||||
}
|
||||
assert!(!cc.can_pace_packet());
|
||||
|
||||
std::thread::sleep(Duration::from_millis(25));
|
||||
assert!(
|
||||
cc.can_pace_packet(),
|
||||
"pacing bucket never refilled; sending would be stuck forever"
|
||||
);
|
||||
}
|
||||
|
||||
/// cwnd must never exceed the absolute ceiling, however long slow start
|
||||
/// runs unopposed — above it the window is buffered queue, not throughput.
|
||||
#[test]
|
||||
fn test_cwnd_never_exceeds_absolute_ceiling() {
|
||||
let mut cc = CongestionController::new(1200);
|
||||
// Constant RTT: no inflation signal, so only the hard cap can stop this.
|
||||
for _ in 0..5000 {
|
||||
cc.on_send(1200);
|
||||
cc.on_ack(1200, Duration::from_millis(30));
|
||||
}
|
||||
assert!(
|
||||
cc.cwnd() <= MAX_CWND_PACKETS * 1200,
|
||||
"cwnd {} exceeded the {}-packet ceiling",
|
||||
cc.cwnd(),
|
||||
MAX_CWND_PACKETS
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_isolated_slow_start_loss_does_not_exit_slow_start() {
|
||||
// A single dropped packet (wireless noise, a brief handover blip) is
|
||||
|
|
|
|||
|
|
@ -4,11 +4,6 @@ use thiserror::Error;
|
|||
use std::collections::{BTreeMap, VecDeque};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// Upper bound on a single frame's retransmit timer, after exponential backoff
|
||||
/// is applied to the adaptive RTO. Past this the session is dead from the
|
||||
/// user's point of view, and waiting longer only delays recovery.
|
||||
const MAX_EFFECTIVE_RTO: Duration = Duration::from_secs(8);
|
||||
|
||||
use crate::congestion::CongestionController;
|
||||
use crate::crypto::{NoiseRole, NoiseSession, SessionCipher};
|
||||
use crate::framing::{AdaptivePadder, FrameHeader, FrameKind, FramedPacket, PaddingStrategy};
|
||||
|
|
@ -188,16 +183,6 @@ impl ProtocolMachine {
|
|||
self.cc.cwnd_packets() as usize
|
||||
}
|
||||
|
||||
/// Whether the pacing bucket currently allows releasing another packet.
|
||||
///
|
||||
/// The congestion window bounds how much may be UNACKNOWLEDGED; it says
|
||||
/// nothing about how fast that window is emptied onto the wire. Sending a
|
||||
/// whole window back-to-back is what drives a deep buffer into standing
|
||||
/// queue, so admission is gated on both.
|
||||
pub fn can_pace_packet(&self) -> bool {
|
||||
self.cc.can_pace_packet()
|
||||
}
|
||||
|
||||
pub fn on_send(&mut self, bytes: u64) {
|
||||
self.cc.on_send(bytes);
|
||||
}
|
||||
|
|
@ -690,15 +675,8 @@ impl ProtocolMachine {
|
|||
break;
|
||||
}
|
||||
|
||||
// Exponential backoff, but bounded in absolute terms. base_rto is
|
||||
// itself adaptive and can reach RTO_MAX (16s) on a congested path;
|
||||
// multiplying that by the 64x backoff cap yields a frame that sits
|
||||
// unretransmitted for ~17 MINUTES, long past the point where the
|
||||
// session is simply dead to the user. Cap the product so backoff
|
||||
// stays a backoff rather than an outage.
|
||||
let backoff_factor = 1u64 << (frame.retries as u64).min(6);
|
||||
let effective_rto = Duration::from_millis(base_rto_ms.saturating_mul(backoff_factor))
|
||||
.min(MAX_EFFECTIVE_RTO);
|
||||
let effective_rto = Duration::from_millis(base_rto_ms.saturating_mul(backoff_factor));
|
||||
|
||||
if now.duration_since(frame.last_sent) >= effective_rto {
|
||||
// Only burn the retry counter and reset the RTO timer when the
|
||||
|
|
|
|||
|
|
@ -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.4+31
|
||||
version: 0.4.2+25
|
||||
|
||||
environment:
|
||||
sdk: ^3.11.4
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"name": "ostp-gui",
|
||||
"private": true,
|
||||
"version": "0.4.4",
|
||||
"version": "0.4.2",
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"tauri": "tauri",
|
||||
|
|
|
|||
|
|
@ -2665,7 +2665,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-client"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64 0.22.1",
|
||||
|
|
@ -2696,7 +2696,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-core"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"bytes",
|
||||
|
|
@ -2713,7 +2713,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-gui"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"json_comments",
|
||||
|
|
@ -2733,7 +2733,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ostp-tun"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"libc",
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
[package]
|
||||
name = "ostp-gui"
|
||||
version = "0.4.4"
|
||||
version = "0.4.2"
|
||||
description = "OSTP desktop GUI"
|
||||
authors = ["ospab"]
|
||||
edition = "2021"
|
||||
|
|
|
|||
|
|
@ -204,34 +204,19 @@ fn get_wintun_install_path() -> String {
|
|||
String::new()
|
||||
}
|
||||
|
||||
/// A `Command` for a console program, with the console window suppressed.
|
||||
///
|
||||
/// The GUI is a windowed-subsystem binary, so every console child it spawns
|
||||
/// pops up a console window for as long as that child runs. With `reg`,
|
||||
/// `tasklist` and `schtasks` all being invoked from here, that surfaced as
|
||||
/// windows flashing on screen — worst while polling for the scheduled task,
|
||||
/// which could spawn twenty of them in a row.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn quiet_command(program: &str) -> std::process::Command {
|
||||
use std::os::windows::process::CommandExt;
|
||||
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
|
||||
let mut cmd = std::process::Command::new(program);
|
||||
cmd.creation_flags(CREATE_NO_WINDOW);
|
||||
cmd
|
||||
}
|
||||
|
||||
/// Sets or removes the app from Windows startup (HKCU\...\Run).
|
||||
#[tauri::command]
|
||||
fn set_autostart(enable: bool) -> Result<(), String> {
|
||||
#[cfg(target_os = "windows")]
|
||||
{
|
||||
use std::process::Command;
|
||||
let key = r"HKCU\Software\Microsoft\Windows\CurrentVersion\Run";
|
||||
let app_name = "OSTP";
|
||||
if enable {
|
||||
let exe = std::env::current_exe()
|
||||
.map_err(|e| format!("Cannot get exe path: {}", e))?;
|
||||
let exe_str = format!("\"{}\"", exe.to_string_lossy());
|
||||
let out = quiet_command("reg")
|
||||
let out = Command::new("reg")
|
||||
.args(["add", key, "/v", app_name, "/t", "REG_SZ", "/d", &exe_str, "/f"])
|
||||
.output()
|
||||
.map_err(|e| format!("reg add failed: {}", e))?;
|
||||
|
|
@ -239,71 +224,28 @@ fn set_autostart(enable: bool) -> Result<(), String> {
|
|||
return Err(String::from_utf8_lossy(&out.stderr).to_string());
|
||||
}
|
||||
} else {
|
||||
let _ = quiet_command("reg")
|
||||
let _ = Command::new("reg")
|
||||
.args(["delete", key, "/v", app_name, "/f"])
|
||||
.output();
|
||||
}
|
||||
}
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
// XDG autostart: desktop environments launch every .desktop file in
|
||||
// ~/.config/autostart on login. This is the portable equivalent of the
|
||||
// HKCU Run key above and needs no elevation.
|
||||
let path = linux_autostart_path().ok_or("Cannot determine the autostart directory")?;
|
||||
if enable {
|
||||
let exe = std::env::current_exe().map_err(|e| format!("Cannot get exe path: {}", e))?;
|
||||
if let Some(dir) = path.parent() {
|
||||
std::fs::create_dir_all(dir)
|
||||
.map_err(|e| format!("Cannot create {}: {}", dir.display(), e))?;
|
||||
}
|
||||
let entry = format!(
|
||||
"[Desktop Entry]\n\
|
||||
Type=Application\n\
|
||||
Name=OSTP\n\
|
||||
Exec=\"{}\"\n\
|
||||
Terminal=false\n\
|
||||
X-GNOME-Autostart-enabled=true\n",
|
||||
exe.display()
|
||||
);
|
||||
std::fs::write(&path, entry)
|
||||
.map_err(|e| format!("Cannot write {}: {}", path.display(), e))?;
|
||||
} else if path.exists() {
|
||||
std::fs::remove_file(&path)
|
||||
.map_err(|e| format!("Cannot remove {}: {}", path.display(), e))?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Path of the XDG autostart entry, honouring XDG_CONFIG_HOME.
|
||||
#[cfg(target_os = "linux")]
|
||||
fn linux_autostart_path() -> Option<PathBuf> {
|
||||
let base = std::env::var_os("XDG_CONFIG_HOME")
|
||||
.map(PathBuf::from)
|
||||
.filter(|p| !p.as_os_str().is_empty())
|
||||
.or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".config")))?;
|
||||
Some(base.join("autostart").join("ostp.desktop"))
|
||||
}
|
||||
|
||||
/// Checks if the app is currently in Windows startup.
|
||||
#[tauri::command]
|
||||
fn get_autostart() -> bool {
|
||||
#[cfg(target_os = "windows")]
|
||||
{
|
||||
use std::process::Command;
|
||||
let key = r"HKCU\Software\Microsoft\Windows\CurrentVersion\Run";
|
||||
let out = quiet_command("reg")
|
||||
let out = Command::new("reg")
|
||||
.args(["query", key, "/v", "OSTP"])
|
||||
.output();
|
||||
if let Ok(o) = out {
|
||||
return o.status.success();
|
||||
}
|
||||
}
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
if let Some(path) = linux_autostart_path() {
|
||||
return path.exists();
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
|
|
@ -312,7 +254,8 @@ fn get_autostart() -> bool {
|
|||
fn list_running_processes() -> Vec<String> {
|
||||
#[cfg(target_os = "windows")]
|
||||
{
|
||||
if let Ok(out) = quiet_command("tasklist")
|
||||
use std::process::Command;
|
||||
if let Ok(out) = Command::new("tasklist")
|
||||
.args(["/FO", "CSV", "/NH"])
|
||||
.output()
|
||||
{
|
||||
|
|
@ -682,18 +625,13 @@ async fn start_tun_via_helper(
|
|||
raw: &ClientConfigRaw,
|
||||
app: tauri::AppHandle,
|
||||
) -> Result<bool, String> {
|
||||
// TUN goes through a privileged helper. Elevation is implemented for
|
||||
// Windows (UAC) and Linux (polkit/pkexec); anywhere else launch_as_admin
|
||||
// reports that plainly rather than letting this fail later as a confusing
|
||||
// missing-file error.
|
||||
let port = {
|
||||
let listener = std::net::TcpListener::bind("127.0.0.1:0").map_err(|e| format!("Bind error: {}", e))?;
|
||||
listener.local_addr().unwrap().port()
|
||||
};
|
||||
|
||||
let auth_token = rand::random::<u64>().to_string();
|
||||
let helper_exe = find_helper_exe()
|
||||
.ok_or_else(|| format!("{HELPER_EXE_NAME} not found next to the app or in target/."))?;
|
||||
let helper_exe = find_helper_exe().ok_or_else(|| "ostp-tun-helper.exe not found.".to_string())?;
|
||||
launch_as_admin(&helper_exe, &auth_token, port).map_err(|e| format!("Failed to launch helper: {}", e))?;
|
||||
tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
|
||||
|
||||
|
|
@ -767,32 +705,21 @@ struct HelperPipeState {
|
|||
error_msg: Option<String>,
|
||||
}
|
||||
|
||||
/// Executable name of the TUN helper for the current platform.
|
||||
///
|
||||
/// The ".exe" suffix was hardcoded, so on Linux every lookup below searched for
|
||||
/// a file that cannot exist and the GUI reported the helper as missing on a
|
||||
/// platform where it ships without an extension.
|
||||
const HELPER_EXE_NAME: &str = if cfg!(windows) {
|
||||
"ostp-tun-helper.exe"
|
||||
} else {
|
||||
"ostp-tun-helper"
|
||||
};
|
||||
|
||||
fn find_helper_exe() -> Option<PathBuf> {
|
||||
if let Ok(exe) = std::env::current_exe() {
|
||||
if let Some(dir) = exe.parent() {
|
||||
// 1. Release/Production adjacent
|
||||
let candidate = dir.join(HELPER_EXE_NAME);
|
||||
let candidate = dir.join("ostp-tun-helper.exe");
|
||||
if candidate.exists() { return Some(candidate); }
|
||||
|
||||
|
||||
// 2. Tauri target directory fallback
|
||||
// e.g. from ostp-gui/src-tauri/target/debug/deps/
|
||||
let mut parent = dir;
|
||||
while let Some(p) = parent.parent() {
|
||||
if p.file_name().map(|n| n == "target").unwrap_or(false) {
|
||||
let deb = p.join("debug").join(HELPER_EXE_NAME);
|
||||
let deb = p.join("debug").join("ostp-tun-helper.exe");
|
||||
if deb.exists() { return Some(deb); }
|
||||
let rel = p.join("release").join(HELPER_EXE_NAME);
|
||||
let rel = p.join("release").join("ostp-tun-helper.exe");
|
||||
if rel.exists() { return Some(rel); }
|
||||
}
|
||||
parent = p;
|
||||
|
|
@ -802,13 +729,13 @@ fn find_helper_exe() -> Option<PathBuf> {
|
|||
// 3. Current working directory target fallback
|
||||
let cwd = std::env::current_dir().unwrap_or_default();
|
||||
let candidates = [
|
||||
cwd.join(HELPER_EXE_NAME),
|
||||
cwd.join("target").join("debug").join(HELPER_EXE_NAME),
|
||||
cwd.join("target").join("release").join(HELPER_EXE_NAME),
|
||||
cwd.join("..").join("target").join("debug").join(HELPER_EXE_NAME),
|
||||
cwd.join("..").join("target").join("release").join(HELPER_EXE_NAME),
|
||||
cwd.join("..").join("..").join("target").join("debug").join(HELPER_EXE_NAME),
|
||||
cwd.join("..").join("..").join("target").join("release").join(HELPER_EXE_NAME),
|
||||
cwd.join("ostp-tun-helper.exe"),
|
||||
cwd.join("target").join("debug").join("ostp-tun-helper.exe"),
|
||||
cwd.join("target").join("release").join("ostp-tun-helper.exe"),
|
||||
cwd.join("..").join("target").join("debug").join("ostp-tun-helper.exe"),
|
||||
cwd.join("..").join("target").join("release").join("ostp-tun-helper.exe"),
|
||||
cwd.join("..").join("..").join("target").join("debug").join("ostp-tun-helper.exe"),
|
||||
cwd.join("..").join("..").join("target").join("release").join("ostp-tun-helper.exe"),
|
||||
];
|
||||
for path in &candidates {
|
||||
if path.exists() { return Some(path.clone()); }
|
||||
|
|
@ -816,272 +743,8 @@ fn find_helper_exe() -> Option<PathBuf> {
|
|||
None
|
||||
}
|
||||
|
||||
/// Name of the Scheduled Task that runs the helper elevated without a prompt.
|
||||
#[cfg(target_os = "windows")]
|
||||
const HELPER_TASK_NAME: &str = "OSTP TUN Helper";
|
||||
|
||||
/// Fixed path the GUI writes launch parameters to, and the task's command line
|
||||
/// reads them from.
|
||||
///
|
||||
/// A Scheduled Task stores a FIXED command line, so the per-launch port and
|
||||
/// token cannot travel as arguments. The file lives under the user's own
|
||||
/// LOCALAPPDATA: the helper runs elevated but as the SAME user, so this keeps
|
||||
/// the token inside the trust boundary it already had — no other user can read
|
||||
/// it, which would not be true of a shared location.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn helper_args_file() -> PathBuf {
|
||||
let base = std::env::var_os("LOCALAPPDATA")
|
||||
.map(PathBuf::from)
|
||||
.unwrap_or_else(std::env::temp_dir);
|
||||
base.join("OSTP").join("helper-args.json")
|
||||
}
|
||||
|
||||
/// Minimal XML text escaping for the values interpolated into the task
|
||||
/// definition. Paths and usernames are attacker-irrelevant here but can easily
|
||||
/// contain `&`, which would otherwise produce invalid XML and a confusing
|
||||
/// schtasks parse failure.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn xml_escape(s: &str) -> String {
|
||||
s.replace('&', "&")
|
||||
.replace('<', "<")
|
||||
.replace('>', ">")
|
||||
.replace('"', """)
|
||||
.replace('\'', "'")
|
||||
}
|
||||
|
||||
/// Reverse of [`xml_escape`]. `&` must be undone last or `&lt;` would
|
||||
/// come back as `<`.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn xml_unescape(s: &str) -> String {
|
||||
s.replace(""", "\"")
|
||||
.replace("'", "'")
|
||||
.replace("<", "<")
|
||||
.replace(">", ">")
|
||||
.replace("&", "&")
|
||||
}
|
||||
|
||||
/// The exe path currently baked into the registered task, if any.
|
||||
///
|
||||
/// Queried as XML rather than `/FO LIST /V`: the list format's field labels are
|
||||
/// localized (on a Russian Windows "Task To Run" is "Задача для запуска"),
|
||||
/// whereas XML tag names are fixed. schtasks writes UTF-16LE with a BOM here,
|
||||
/// but tolerate UTF-8 in case that ever changes.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn helper_task_command() -> Option<String> {
|
||||
let out = quiet_command("schtasks")
|
||||
.args(["/Query", "/TN", HELPER_TASK_NAME, "/XML"])
|
||||
.output()
|
||||
.ok()?;
|
||||
if !out.status.success() {
|
||||
return None;
|
||||
}
|
||||
|
||||
let text = if out.stdout.starts_with(&[0xFF, 0xFE]) {
|
||||
let units: Vec<u16> = out.stdout[2..]
|
||||
.chunks_exact(2)
|
||||
.map(|c| u16::from_le_bytes([c[0], c[1]]))
|
||||
.collect();
|
||||
String::from_utf16_lossy(&units)
|
||||
} else {
|
||||
String::from_utf8_lossy(&out.stdout).into_owned()
|
||||
};
|
||||
|
||||
let start = text.find("<Command>")? + "<Command>".len();
|
||||
let end = text[start..].find("</Command>")? + start;
|
||||
Some(xml_unescape(text[start..end].trim()))
|
||||
}
|
||||
|
||||
/// Whether a task is registered AND still points at the exe we are about to run.
|
||||
///
|
||||
/// The path matters as much as the name. A task registered by a dev build (or
|
||||
/// by an install that has since moved) keeps its original `<Command>`, and
|
||||
/// `schtasks /Run` reports success merely for *accepting* the request — a task
|
||||
/// whose exe no longer exists fails asynchronously and silently. Trusting the
|
||||
/// name alone therefore bought a 60-second "Timeout connecting to helper" on
|
||||
/// every single connect, permanently, until the task was deleted by hand.
|
||||
/// Re-registering costs one consent prompt and fixes it for good.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn helper_task_matches(exe: &std::path::Path) -> bool {
|
||||
let Some(registered) = helper_task_command() else {
|
||||
return false;
|
||||
};
|
||||
let registered = registered.trim().trim_matches('"');
|
||||
|
||||
// Canonicalize both sides when possible so `..`, short 8.3 names and
|
||||
// casing differences do not read as a mismatch. A missing file cannot be
|
||||
// canonicalized — which is itself a mismatch worth re-registering over.
|
||||
match (
|
||||
std::fs::canonicalize(registered),
|
||||
std::fs::canonicalize(exe),
|
||||
) {
|
||||
(Ok(a), Ok(b)) => a == b,
|
||||
_ => registered.eq_ignore_ascii_case(&exe.display().to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Register the Scheduled Task. This is the ONLY step that needs elevation, and
|
||||
/// it happens once per machine; every later tunnel start reuses the task.
|
||||
///
|
||||
/// RunLevel=HIGHEST makes the task run elevated, and because a task launch is
|
||||
/// not an elevation request, Windows shows no consent dialog for it.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn install_helper_task(exe: &std::path::Path) -> anyhow::Result<()> {
|
||||
let args_file = helper_args_file();
|
||||
if let Some(dir) = args_file.parent() {
|
||||
std::fs::create_dir_all(dir)?;
|
||||
}
|
||||
|
||||
// Register from an XML definition rather than /TR. The command line would
|
||||
// otherwise need the exe path and the args path quoted INSIDE an already
|
||||
// quoted /TR value, escaped again through ShellExecuteW — a notoriously
|
||||
// brittle chain when either path contains a space, which both of these do
|
||||
// by default (Program Files, and usernames with spaces). XML also lets the
|
||||
// battery and time-limit settings below be stated explicitly.
|
||||
let user = format!(
|
||||
"{}\\{}",
|
||||
std::env::var("USERDOMAIN").unwrap_or_else(|_| "%COMPUTERNAME%".into()),
|
||||
std::env::var("USERNAME").unwrap_or_default()
|
||||
);
|
||||
let xml = format!(
|
||||
r#"<?xml version="1.0" encoding="UTF-16"?>
|
||||
<Task version="1.2" xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
|
||||
<RegistrationInfo>
|
||||
<Description>Runs the OSTP TUN helper elevated so enabling the tunnel does not prompt for consent every time.</Description>
|
||||
</RegistrationInfo>
|
||||
<Principals>
|
||||
<Principal id="Author">
|
||||
<UserId>{user}</UserId>
|
||||
<LogonType>InteractiveToken</LogonType>
|
||||
<RunLevel>HighestAvailable</RunLevel>
|
||||
</Principal>
|
||||
</Principals>
|
||||
<Settings>
|
||||
<MultipleInstancesPolicy>Parallel</MultipleInstancesPolicy>
|
||||
<DisallowStartIfOnBatteries>false</DisallowStartIfOnBatteries>
|
||||
<StopIfGoingOnBatteries>false</StopIfGoingOnBatteries>
|
||||
<StartWhenAvailable>false</StartWhenAvailable>
|
||||
<RunOnlyIfNetworkAvailable>false</RunOnlyIfNetworkAvailable>
|
||||
<ExecutionTimeLimit>PT0S</ExecutionTimeLimit>
|
||||
<Enabled>true</Enabled>
|
||||
<Hidden>false</Hidden>
|
||||
<AllowHardTerminate>true</AllowHardTerminate>
|
||||
</Settings>
|
||||
<Actions Context="Author">
|
||||
<Exec>
|
||||
<Command>{exe}</Command>
|
||||
<Arguments>--args-file "{args}"</Arguments>
|
||||
</Exec>
|
||||
</Actions>
|
||||
</Task>
|
||||
"#,
|
||||
user = xml_escape(&user),
|
||||
exe = xml_escape(&exe.display().to_string()),
|
||||
args = xml_escape(&args_file.display().to_string()),
|
||||
);
|
||||
|
||||
// schtasks /Create /XML expects UTF-16LE with a BOM.
|
||||
let xml_path = std::env::temp_dir().join(format!("ostp_task_{}.xml", rand::random::<u32>()));
|
||||
let mut utf16: Vec<u8> = vec![0xFF, 0xFE];
|
||||
for unit in xml.encode_utf16() {
|
||||
utf16.extend_from_slice(&unit.to_le_bytes());
|
||||
}
|
||||
std::fs::write(&xml_path, &utf16)?;
|
||||
|
||||
// Registering a HighestAvailable task is itself privileged: this is the one
|
||||
// prompt, and it happens once per machine.
|
||||
//
|
||||
// Elevate through PowerShell's Start-Process -Wait rather than
|
||||
// ShellExecuteW. ShellExecuteW returns as soon as the elevated process is
|
||||
// LAUNCHED, so the XML below was being deleted while schtasks was still
|
||||
// starting up — registration then failed, leaving the user with a consent
|
||||
// prompt that accomplished nothing, followed by a second prompt from the
|
||||
// fallback path. -Wait makes the deletion safe and lets the exit code be
|
||||
// checked instead of guessed at by polling.
|
||||
//
|
||||
// ArgumentList takes an array, so the task name and XML path never need
|
||||
// quoting or escaping through a command line, only PowerShell's own
|
||||
// single-quote doubling.
|
||||
let ps = format!(
|
||||
"$p = Start-Process -FilePath 'schtasks.exe' -Verb RunAs -Wait -PassThru \
|
||||
-WindowStyle Hidden -ArgumentList @('/Create','/TN','{}','/XML','{}','/F'); \
|
||||
exit $p.ExitCode",
|
||||
ps_quote(HELPER_TASK_NAME),
|
||||
ps_quote(&xml_path.display().to_string()),
|
||||
);
|
||||
|
||||
let status = quiet_command("powershell")
|
||||
.args(["-NoProfile", "-NonInteractive", "-WindowStyle", "Hidden", "-Command", &ps])
|
||||
.status();
|
||||
|
||||
// schtasks has exited by now, so this is safe.
|
||||
let _ = std::fs::remove_file(&xml_path);
|
||||
|
||||
match status {
|
||||
Ok(s) if s.success() => {}
|
||||
Ok(s) => anyhow::bail!(
|
||||
"registering the scheduled task failed (exit code {:?}). A declined consent prompt \
|
||||
reports 1223.",
|
||||
s.code()
|
||||
),
|
||||
Err(e) => anyhow::bail!("could not run powershell to register the task: {e}"),
|
||||
}
|
||||
|
||||
if helper_task_matches(exe) {
|
||||
Ok(())
|
||||
} else {
|
||||
anyhow::bail!("schtasks reported success but the task does not point at {}", exe.display())
|
||||
}
|
||||
}
|
||||
|
||||
/// Escape a value for embedding in a PowerShell single-quoted string.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn ps_quote(s: &str) -> String {
|
||||
s.replace('\'', "''")
|
||||
}
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
fn launch_as_admin(exe: &std::path::PathBuf, token: &str, port: u16) -> anyhow::Result<()> {
|
||||
// Preferred path: hand the parameters over in a file and trigger the
|
||||
// pre-registered task, which runs elevated with no prompt. Falls back to a
|
||||
// direct elevated launch when the task is absent (first ever run, or the
|
||||
// user removed it) — and that first run is also where the task gets created,
|
||||
// so the prompt appears once rather than on every connect.
|
||||
let args_file = helper_args_file();
|
||||
if let Some(dir) = args_file.parent() {
|
||||
let _ = std::fs::create_dir_all(dir);
|
||||
}
|
||||
let payload = serde_json::json!({ "port": port, "token": token });
|
||||
let wrote_args = std::fs::write(&args_file, payload.to_string()).is_ok();
|
||||
|
||||
if wrote_args {
|
||||
if !helper_task_matches(exe) {
|
||||
if let Err(e) = install_helper_task(exe) {
|
||||
eprintln!("[OSTP] could not register the helper task ({e}); falling back to a direct elevated launch");
|
||||
}
|
||||
}
|
||||
if helper_task_matches(exe) {
|
||||
let run = quiet_command("schtasks")
|
||||
.args(["/Run", "/TN", HELPER_TASK_NAME])
|
||||
.output();
|
||||
match run {
|
||||
Ok(o) if o.status.success() => return Ok(()),
|
||||
Ok(o) => eprintln!(
|
||||
"[OSTP] schtasks /Run failed: {}",
|
||||
String::from_utf8_lossy(&o.stderr).trim()
|
||||
),
|
||||
Err(e) => eprintln!("[OSTP] schtasks /Run could not start: {e}"),
|
||||
}
|
||||
}
|
||||
// Falling through: remove the file so a stale token is not left behind.
|
||||
let _ = std::fs::remove_file(&args_file);
|
||||
}
|
||||
|
||||
launch_as_admin_direct(exe, token, port)
|
||||
}
|
||||
|
||||
/// The original one-prompt-per-launch path, kept as the fallback.
|
||||
#[cfg(target_os = "windows")]
|
||||
fn launch_as_admin_direct(exe: &std::path::PathBuf, token: &str, port: u16) -> anyhow::Result<()> {
|
||||
use std::ffi::OsStr;
|
||||
use std::os::windows::ffi::OsStrExt;
|
||||
use std::ptr::null_mut;
|
||||
|
|
@ -1134,50 +797,8 @@ fn launch_as_admin_direct(exe: &std::path::PathBuf, token: &str, port: u16) -> a
|
|||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
fn launch_as_admin(exe: &PathBuf, token: &str, port: u16) -> Result<()> {
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::process::Command;
|
||||
|
||||
// Same shape as the Windows path: the token goes through a file rather than
|
||||
// argv, so it never shows up in the process list.
|
||||
let token_file = std::env::temp_dir().join(format!("ostp_auth_{}.tmp", rand::random::<u32>()));
|
||||
std::fs::write(&token_file, token)?;
|
||||
// Unlike Windows, /tmp is world-readable here, and this token authenticates
|
||||
// control of the privileged tunnel helper — restrict it to the owner.
|
||||
let _ = std::fs::set_permissions(&token_file, std::fs::Permissions::from_mode(0o600));
|
||||
|
||||
// pkexec is polkit's front-end: in a desktop session it raises a graphical
|
||||
// authentication dialog. sudo is not an option from a GUI process, which has
|
||||
// no terminal to prompt on.
|
||||
match Command::new("pkexec")
|
||||
.arg(exe)
|
||||
.arg("--port")
|
||||
.arg(port.to_string())
|
||||
.arg("--token-file")
|
||||
.arg(&token_file)
|
||||
.spawn()
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
|
||||
let _ = std::fs::remove_file(&token_file);
|
||||
anyhow::bail!(
|
||||
"pkexec was not found, so the TUN helper cannot be granted the privileges it \
|
||||
needs. Install polkit (package \"policykit-1\" on Debian/Ubuntu, \"polkit\" on \
|
||||
Fedora/Arch), or use proxy mode, which needs no elevation."
|
||||
)
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = std::fs::remove_file(&token_file);
|
||||
Err(e.into())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(any(target_os = "windows", target_os = "linux")))]
|
||||
fn launch_as_admin(_exe: &PathBuf, _token: &str, _port: u16) -> Result<()> {
|
||||
anyhow::bail!("TUN mode needs a privileged helper, which is implemented on Windows and Linux only. Use proxy mode on this platform.");
|
||||
}
|
||||
#[cfg(not(target_os = "windows"))]
|
||||
fn launch_as_admin(_exe: &PathBuf, _token: &str, _port: u16) -> Result<()> { anyhow::bail!("Windows only."); }
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
fn show_error_dialog(msg: &str) {
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"$schema": "https://schema.tauri.app/config/2",
|
||||
"productName": "ostp-gui",
|
||||
"version": "0.4.4",
|
||||
"version": "0.4.2",
|
||||
"identifier": "com.ospab.ostp",
|
||||
"build": {
|
||||
"frontendDist": "../src"
|
||||
|
|
@ -11,11 +11,9 @@
|
|||
"windows": [
|
||||
{
|
||||
"title": "OSTP",
|
||||
"width": 400,
|
||||
"height": 720,
|
||||
"minWidth": 360,
|
||||
"minHeight": 560,
|
||||
"resizable": true
|
||||
"width": 360,
|
||||
"height": 680,
|
||||
"resizable": false
|
||||
}
|
||||
],
|
||||
"security": {
|
||||
|
|
|
|||
|
|
@ -660,13 +660,6 @@ function loadSettingsIntoForm() {
|
|||
updateClientVisibility();
|
||||
}
|
||||
|
||||
// Last values actually pushed to the OS / backend, so repeated saves that did
|
||||
// not change them stay free. Undefined until the first save, which is correct:
|
||||
// the first one should apply.
|
||||
let lastAppliedAutostart;
|
||||
let lastAppliedTunnelConfig;
|
||||
let hotReloadTimer;
|
||||
|
||||
function collectAndSaveSettings() {
|
||||
const s = {
|
||||
tun: inTun.checked,
|
||||
|
|
@ -693,41 +686,19 @@ function collectAndSaveSettings() {
|
|||
fragChunk: parseInt(inFragChunk.value) || 2,
|
||||
fragSleep: !isNaN(parseInt(inFragSleep.value)) ? parseInt(inFragSleep.value) : 2,
|
||||
};
|
||||
// Cheap and local: safe to run on every debounced keystroke.
|
||||
saveClientSettings(s);
|
||||
updateClientVisibility();
|
||||
|
||||
// Everything below talks to the OS or restarts the tunnel. Running it per
|
||||
// keystroke is what made typing in the exclusion fields lag by seconds: the
|
||||
// 400ms debounce fires during natural pauses in typing, and each firing hit
|
||||
// the Windows registry and then tore down and rebuilt the tunnel.
|
||||
// Set autostart
|
||||
invoke('set_autostart', { enable: s.launchStartup }).catch(() => {});
|
||||
|
||||
// Only touch autostart when it actually changed — this is a registry write.
|
||||
if (s.launchStartup !== lastAppliedAutostart) {
|
||||
lastAppliedAutostart = s.launchStartup;
|
||||
invoke('set_autostart', { enable: s.launchStartup }).catch(() => {});
|
||||
}
|
||||
|
||||
// Hot-reload the tunnel only when something it actually reads has changed,
|
||||
// and on a much longer debounce: a reload is disruptive, so it should land
|
||||
// once the user has stopped editing rather than between keystrokes.
|
||||
// Hot-reload exclusions if connected
|
||||
if (appState === 'connected') {
|
||||
const tunnelRelevant = JSON.stringify([
|
||||
s.tun, s.killSwitch, s.mux, s.muxSessions, s.mtu, s.dns, s.socks,
|
||||
s.exDomains, s.exIps, s.exProcs, s.junkEnabled, s.junkPcMin, s.junkPcMax,
|
||||
s.junkPsMin, s.junkPsMax, s.tcpFrag, s.fragChunk, s.fragSleep,
|
||||
]);
|
||||
if (tunnelRelevant !== lastAppliedTunnelConfig) {
|
||||
clearTimeout(hotReloadTimer);
|
||||
hotReloadTimer = setTimeout(() => {
|
||||
lastAppliedTunnelConfig = tunnelRelevant;
|
||||
const cfg = buildConfig();
|
||||
if (cfg) {
|
||||
invoke('save_config', { jsonContent: JSON.stringify(cfg, null, 2) })
|
||||
.then(() => invoke('reload_tunnel'))
|
||||
.catch(() => {});
|
||||
}
|
||||
}, 1500);
|
||||
const cfg = buildConfig();
|
||||
if (cfg) {
|
||||
invoke('save_config', { jsonContent: JSON.stringify(cfg, null, 2) })
|
||||
.then(() => invoke('reload_tunnel'))
|
||||
.catch(() => {});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -99,13 +99,6 @@ a { text-decoration: none; }
|
|||
.app-root {
|
||||
position: relative;
|
||||
width: 100%;
|
||||
/* The window is resizable so users on desktops where the toolkit does not
|
||||
apply our DPI scaling (WebKitGTK on HiDPI Linux renders the configured
|
||||
size as raw pixels, giving a postage-stamp window) can size it themselves.
|
||||
Capping and centring the column keeps the intended narrow layout instead of
|
||||
stretching controls across a wide window. */
|
||||
max-width: 460px;
|
||||
margin: 0 auto;
|
||||
height: 100%;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
|
|
|
|||
|
|
@ -263,19 +263,8 @@ impl Dispatcher {
|
|||
self.peer_machines
|
||||
.iter()
|
||||
.map(|(&sid, ps)| {
|
||||
// Ceiling matches MAX_CWND_PACKETS in ostp-core. The old 16384
|
||||
// allowed ~20 MB outstanding toward one client — on a mobile
|
||||
// downlink that is standing queue, not throughput, and it is the
|
||||
// download direction that carries video.
|
||||
let cwnd = (ps.machine.cwnd_packets() as i64).clamp(16, 1024);
|
||||
let cwnd = (ps.machine.cwnd_packets() as i64).clamp(16, 16384);
|
||||
let in_flight = ps.machine.in_flight_count() as i64;
|
||||
// Pacing gates the RATE, cwnd only the outstanding amount. With
|
||||
// the pacing bucket empty, report no headroom so the relay
|
||||
// reader pauses instead of handing over another chunk that would
|
||||
// leave back-to-back.
|
||||
if !ps.machine.can_pace_packet() {
|
||||
return (sid, 0);
|
||||
}
|
||||
(sid, cwnd - in_flight)
|
||||
})
|
||||
.collect()
|
||||
|
|
|
|||
|
|
@ -48,23 +48,10 @@ pub async fn connect_target(
|
|||
}
|
||||
if action == OutboundAction::Proxy {
|
||||
let proxy_addr = format!("{}:{}", outbound.address, outbound.port);
|
||||
// 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() {
|
||||
return match outbound.protocol.as_str() {
|
||||
"socks5" => connect_via_socks5(&proxy_addr, target).await,
|
||||
"http" => connect_via_http(&proxy_addr, target).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"
|
||||
)),
|
||||
_ => connect_direct(target, connect_timeout).await,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -383,22 +370,10 @@ pub async fn connect_udp_target(
|
|||
}
|
||||
if action == OutboundAction::Proxy {
|
||||
let proxy_addr = format!("{}:{}", outbound.address, outbound.port);
|
||||
if outbound.protocol.eq_ignore_ascii_case("socks5") {
|
||||
if outbound.protocol == "socks5" {
|
||||
return connect_udp_via_socks5(&proxy_addr, server_udp).await;
|
||||
}
|
||||
// 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
|
||||
));
|
||||
// HTTP CONNECT does not support UDP. Fallback to direct.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,460 +1,403 @@
|
|||
//! Transparent relay node.
|
||||
//! Authenticated Relay Node
|
||||
//!
|
||||
//! Forwards traffic to a fixed upstream OSTP server:
|
||||
//! Принимает входящие UDP/TCP (UoT) соединения от клиентов,
|
||||
//! валидирует HMAC-подпись клиента, используя ключи синхронизированные с upstream-сервера,
|
||||
//! и слепо пробрасывает авторизованный трафик к целевому upstream-серверу.
|
||||
//!
|
||||
//! 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.
|
||||
//! Архитектура цепочек:
|
||||
//! Клиент -> [Relay 1] -> [Relay 2] -> ... -> [Target Server]
|
||||
//! Каждый Relay скачивает access_keys напрямую с Target Server API.
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use anyhow::Result;
|
||||
use bytes::Bytes;
|
||||
use hmac::{Hmac, Mac};
|
||||
use sha2::Sha256;
|
||||
use std::collections::HashMap;
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::{TcpListener, TcpStream, UdpSocket};
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
/// Configuration for a relay node.
|
||||
/// Конфигурация Relay-узла.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct RelayConfig {
|
||||
/// Address(es) to accept client traffic on (UDP and TCP both bind here).
|
||||
/// Адрес(а) для прослушивания входящих соединений (UDP + TCP).
|
||||
pub listen_addrs: Vec<String>,
|
||||
/// Upstream target for TCP (UoT) traffic.
|
||||
/// Адрес upstream TCP для пересылки (обычно тот же порт, что и у target-сервера).
|
||||
pub upstream_tcp: String,
|
||||
/// Upstream target for UDP traffic.
|
||||
/// Адрес upstream UDP.
|
||||
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,
|
||||
}
|
||||
|
||||
/// 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);
|
||||
type SharedKeys = Arc<RwLock<Vec<String>>>;
|
||||
|
||||
/// Token bucket bounding how fast new sessions may be created.
|
||||
struct AdmissionLimiter {
|
||||
tokens: f64,
|
||||
last_refill: Instant,
|
||||
}
|
||||
|
||||
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 {
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Entry point.
|
||||
/// Точка входа Relay-узла.
|
||||
pub async fn run_relay_node(cfg: RelayConfig) -> Result<()> {
|
||||
let udp_cfg = cfg.clone();
|
||||
let shared_keys: SharedKeys = Arc::new(RwLock::new(Vec::new()));
|
||||
|
||||
// Первоначальная синхронизация ключей
|
||||
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);
|
||||
}
|
||||
|
||||
// Фоновый синхронизатор ключей
|
||||
let cfg_clone = cfg.clone();
|
||||
let keys_clone = shared_keys.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = run_udp_relay(udp_cfg).await {
|
||||
tracing::error!("Relay UDP loop error: {e}");
|
||||
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),
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
run_tcp_relay(cfg).await
|
||||
// Запуск 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
|
||||
}
|
||||
|
||||
// ── UDP ──────────────────────────────────────────────────────────────────────
|
||||
/// Синхронизация 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('/'));
|
||||
|
||||
struct UdpSession {
|
||||
upstream: Arc<UdpSocket>,
|
||||
last_seen: Instant,
|
||||
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)
|
||||
}
|
||||
|
||||
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>>> =
|
||||
/// Проверяет 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 ────────────────────────────────────────────────────────────────
|
||||
|
||||
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)>>> =
|
||||
Arc::new(Mutex::new(HashMap::new()));
|
||||
let limiter = Arc::new(Mutex::new(AdmissionLimiter::new()));
|
||||
|
||||
for bind_addr in &cfg.listen_addrs {
|
||||
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();
|
||||
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();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut buf = vec![0u8; 65535];
|
||||
loop {
|
||||
let (len, peer) = match sock.recv_from(&mut buf).await {
|
||||
let (n, peer) = match sock.recv_from(&mut buf).await {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
tracing::warn!("Relay UDP recv error: {e}");
|
||||
continue;
|
||||
}
|
||||
Err(_) => continue,
|
||||
};
|
||||
|
||||
// Fast path: an established session just forwards.
|
||||
let packet = Bytes::copy_from_slice(&buf[..n]);
|
||||
|
||||
// Быстрая проверка: первый UDP-пакет от нового клиента содержит Noise handshake.
|
||||
// Мы берём из него первые 8 байт как timestamp + 32 байта MAC.
|
||||
// Если пакет достаточно длинный, проверяем подпись.
|
||||
// Для уже авторизованных клиентов (есть в NAT) — пропускаем проверку.
|
||||
{
|
||||
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;
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
// 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;
|
||||
}
|
||||
// Находим или создаём 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;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
};
|
||||
nat_lock.insert(peer, (usock.clone(), Instant::now()));
|
||||
|
||||
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();
|
||||
// Задача: читаем ответы от 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,
|
||||
}
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
sessions_rx.lock().await.remove(&peer);
|
||||
});
|
||||
});
|
||||
|
||||
let _ = sessions
|
||||
.lock()
|
||||
.await
|
||||
.get(&peer)
|
||||
.map(|s| s.upstream.clone())
|
||||
.unwrap()
|
||||
.send(&buf[..len])
|
||||
.await;
|
||||
usock
|
||||
}
|
||||
};
|
||||
|
||||
// Пересылаем пакет в upstream
|
||||
let _ = upstream_sock.send(&packet).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Reclaim idle sessions. Dropping the entry closes the upstream socket,
|
||||
// which ends that session's reader task.
|
||||
// Периодически чистим устаревшие NAT записи (timeout 120 сек)
|
||||
loop {
|
||||
tokio::time::sleep(Duration::from_secs(30)).await;
|
||||
let mut nat_lock = nat_table.lock().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());
|
||||
}
|
||||
nat_lock.retain(|_, (_, last)| now.duration_since(*last) < Duration::from_secs(120));
|
||||
}
|
||||
}
|
||||
|
||||
/// 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));
|
||||
// ── 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
|
||||
.with_context(|| format!("relay: failed to bind TCP on {bind_addr}"))?;
|
||||
tracing::info!("Relay TCP (UoT) listening on {bind_addr} -> {}", cfg.upstream_tcp);
|
||||
let listener = TcpListener::bind(bind_addr).await?;
|
||||
tracing::info!("Relay TCP (UoT) listening on {}", bind_addr);
|
||||
|
||||
let upstream = cfg.upstream_tcp.clone();
|
||||
let live = live.clone();
|
||||
let upstream_tcp = cfg.upstream_tcp.clone();
|
||||
let keys = shared_keys.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let (client, peer) = match listener.accept().await {
|
||||
let (stream, peer_addr) = match listener.accept().await {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
tracing::warn!("Relay TCP accept error: {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_tcp.clone();
|
||||
let keys_clone = keys.clone();
|
||||
|
||||
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}");
|
||||
if let Err(e) = handle_tcp_client(stream, peer_addr, upstream, keys_clone).await {
|
||||
tracing::debug!("Relay TCP client {} closed: {}", peer_addr, e);
|
||||
}
|
||||
live.fetch_sub(1, Ordering::Relaxed);
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Держим поток живым
|
||||
futures_util::future::pending::<()>().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Splice a client connection to the upstream, byte for byte.
|
||||
/// Обработка одного TCP (UoT) соединения.
|
||||
///
|
||||
/// 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),
|
||||
/// Алгоритм:
|
||||
/// 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,
|
||||
)
|
||||
.await
|
||||
.map_err(|_| anyhow::anyhow!("upstream {upstream_addr} connect timed out"))?
|
||||
.with_context(|| format!("connect to upstream {upstream_addr}"))?;
|
||||
.map_err(|_| anyhow::anyhow!("invalid base64 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);
|
||||
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);
|
||||
}
|
||||
|
||||
tokio::io::copy_bidirectional(&mut client, &mut upstream).await?;
|
||||
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);
|
||||
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");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -47,29 +47,12 @@ impl Router {
|
|||
|
||||
let mut proxy = None;
|
||||
if let Some(ref c) = cfg {
|
||||
if c.enabled {
|
||||
if c.protocol == "socks5" {
|
||||
let proxy_addr = format!("{}:{}", c.address, c.port);
|
||||
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
|
||||
);
|
||||
if c.enabled && 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");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -104,28 +87,9 @@ impl UdpSessionRouter {
|
|||
return Err(anyhow::anyhow!("blocked by outbound udp rule: {}", target));
|
||||
}
|
||||
if action == crate::outbound::OutboundAction::Proxy {
|
||||
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"
|
||||
)),
|
||||
};
|
||||
if let Some(p) = &self.proxy {
|
||||
return p.send_to(data, target).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,21 +4,6 @@
|
|||
// or launched via ShellExecuteW("runas").
|
||||
|
||||
fn main() {
|
||||
// Key off the TARGET, not the host. In a build script `cfg(windows)`
|
||||
// describes the machine doing the building, so cross-compiling the helper
|
||||
// from Windows to Linux took this branch and failed with "Can only compile
|
||||
// resource file when target_env is gnu or msvc". CARGO_CFG_TARGET_OS is the
|
||||
// target being built for, which is what actually decides whether a Windows
|
||||
// manifest belongs in the binary.
|
||||
let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap_or_default();
|
||||
if target_os != "windows" {
|
||||
return;
|
||||
}
|
||||
// Second gate, on the HOST: winres is declared under
|
||||
// [target.'cfg(windows)'.build-dependencies], and build-dependencies are
|
||||
// resolved against the host triple, so the crate simply does not exist when
|
||||
// building on Linux. Referencing it unconditionally would fail to compile
|
||||
// there even though the target check above already passed.
|
||||
#[cfg(windows)]
|
||||
{
|
||||
let mut res = winres::WindowsResource::new();
|
||||
|
|
|
|||
|
|
@ -24,14 +24,6 @@ fn log_to_file(msg: &str) {
|
|||
|
||||
|
||||
|
||||
/// Launch parameters handed over in a file rather than on the command line.
|
||||
/// See the `--args-file` handling in `main` for why.
|
||||
#[derive(Deserialize)]
|
||||
struct HelperArgs {
|
||||
port: u16,
|
||||
token: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(tag = "cmd", rename_all = "lowercase")]
|
||||
enum GuiCmd {
|
||||
|
|
@ -84,28 +76,6 @@ async fn main() -> Result<()> {
|
|||
let _ = std::fs::remove_file(path); // securely delete after reading
|
||||
}
|
||||
}
|
||||
// Both port and token from one file. A Scheduled Task stores a FIXED
|
||||
// command line, so anything that varies per launch cannot be passed as
|
||||
// an argument — the GUI writes this file immediately before triggering
|
||||
// the task instead. That indirection is what lets the task be created
|
||||
// once (a single UAC prompt) and reused for every later connect without
|
||||
// prompting again.
|
||||
if args[i] == "--args-file" && i + 1 < args.len() {
|
||||
let path = &args[i + 1];
|
||||
match std::fs::read_to_string(path) {
|
||||
Ok(content) => {
|
||||
let _ = std::fs::remove_file(path); // single use
|
||||
match serde_json::from_str::<HelperArgs>(&content) {
|
||||
Ok(parsed) => {
|
||||
port = parsed.port;
|
||||
expected_token = parsed.token;
|
||||
}
|
||||
Err(e) => log_to_file(&format!("Failed to parse --args-file: {e}")),
|
||||
}
|
||||
}
|
||||
Err(e) => log_to_file(&format!("Failed to read --args-file {path}: {e}")),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
log_to_file("Helper started (TCP mode)");
|
||||
|
|
|
|||
|
|
@ -28,12 +28,6 @@ enum Commands {
|
|||
Init {
|
||||
mode: String,
|
||||
},
|
||||
/// Hash a password for the web panel's `api.password_hash` config field
|
||||
#[command(name = "hash-password", alias = "hp")]
|
||||
HashPassword {
|
||||
/// The password to hash. Omit to be prompted (keeps it out of shell history).
|
||||
password: Option<String>,
|
||||
},
|
||||
/// Generate a new secure access key
|
||||
#[command(name = "gk", alias = "generate-key")]
|
||||
GenerateKey {
|
||||
|
|
@ -799,16 +793,18 @@ 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
|
||||
});
|
||||
|
||||
|
|
@ -924,38 +920,6 @@ async fn run_app() -> Result<()> {
|
|||
match cmd {
|
||||
Commands::Setup { init } => { args.setup = true; args.init = init; }
|
||||
Commands::Init { mode } => { args.init = Some(mode); }
|
||||
Commands::HashPassword { password } => {
|
||||
// The panel stores only a hash, and until now nothing in the CLI
|
||||
// could produce one: `ostp init server` writes password_hash: ""
|
||||
// and the only generator lived inside the Unix-only Server+Panel
|
||||
// wizard branch, leaving no supported way to set up API auth on a
|
||||
// plain server.
|
||||
let password = match password {
|
||||
Some(p) => p,
|
||||
None => {
|
||||
print!("Password: ");
|
||||
use std::io::Write as _;
|
||||
std::io::stdout().flush().ok();
|
||||
let mut buf = String::new();
|
||||
std::io::stdin().read_line(&mut buf)?;
|
||||
buf.trim_end_matches(['\r', '\n']).to_string()
|
||||
}
|
||||
};
|
||||
if password.is_empty() {
|
||||
anyhow::bail!("password must not be empty");
|
||||
}
|
||||
// Must match api.rs's handle_login byte for byte.
|
||||
let hash = format!(
|
||||
"{:x}",
|
||||
<sha2::Sha256 as sha2::Digest>::digest(password.as_bytes())
|
||||
);
|
||||
println!();
|
||||
println!("Add this to the \"api\" section of your config:");
|
||||
println!();
|
||||
println!(" \"password_hash\": \"{hash}\"");
|
||||
println!();
|
||||
return Ok(());
|
||||
}
|
||||
Commands::GenerateKey { format, count } => { args.generate_key = true; args.format = format; args.count = count; }
|
||||
Commands::Links => { args.links = true; }
|
||||
Commands::Check => { args.check = true; }
|
||||
|
|
@ -1135,9 +1099,7 @@ 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());
|
||||
if !r.upstream_api_url.is_empty() {
|
||||
println!(" {}", "upstream_api_url is set but no longer used - safe to remove".yellow());
|
||||
}
|
||||
println!(" API sync: {}", r.upstream_api_url.yellow());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1213,9 +1175,9 @@ async fn run_app() -> Result<()> {
|
|||
"listen": "0.0.0.0:50000",
|
||||
"upstream_tcp": "TARGET_SERVER_IP:50000",
|
||||
"upstream_udp": "TARGET_SERVER_IP:50000",
|
||||
// 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.
|
||||
"upstream_api_url": "http://TARGET_SERVER_IP:9090",
|
||||
"upstream_api_token": "YOUR_API_TOKEN_HERE",
|
||||
"sync_interval_secs": 30,
|
||||
"debug": false
|
||||
}"#.to_string()
|
||||
} else {
|
||||
|
|
@ -1414,18 +1376,14 @@ 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);
|
||||
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()
|
||||
);
|
||||
}
|
||||
println!("{} Key sync API: {}", "[ostp]".cyan().bold(), relay_cfg.upstream_api_url);
|
||||
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?;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue