astelm/telemt
1
0
Fork 0
mirror of https://github.com/telemt/telemt.git synced 2026-04-26 23:14:10 +03:00
Code Issues Packages Projects Releases Wiki Activity

Add Shadowsocks upstream support

Browse Source
This commit is contained in:
Maxim Myalin
2026-03-18 12:36:44 +03:00
parent cb87b2eac3
commit def42f0baa
20 changed files with 1476 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/transport/middle_proxy/ping.rs
View File

@@ -7,6 +7,7 @@ use tokio::net::UdpSocket;
use crate::config::{UpstreamConfig, UpstreamType};
use crate::crypto::SecureRandom;
use crate::error::ProxyError;
use crate::transport::shadowsocks::sanitize_shadowsocks_url;
use crate::transport::{UpstreamEgressInfo, UpstreamRouteKind};
use super::MePool;
@@ -40,7 +41,11 @@ pub fn format_sample_line(sample: &MePingSample) -> String {
let sign = if sample.dc >= 0 { "+" } else { "-" };
let addr = format!("{}:{}", sample.addr.ip(), sample.addr.port());
match (sample.connect_ms, sample.handshake_ms.as_ref(), sample.error.as_ref()) {
match (
sample.connect_ms,
sample.handshake_ms.as_ref(),
sample.error.as_ref(),
) {
(Some(conn), Some(hs), None) => format!(
" {sign} {addr}\tPing: {:.0} ms / RPC: {:.0} ms / OK",
conn, hs
@@ -121,6 +126,7 @@ fn route_from_egress(egress: Option<UpstreamEgressInfo>) -> Option<String> {
None => route,
})
}
UpstreamRouteKind::Shadowsocks => Some("shadowsocks".to_string()),
}
}
@@ -232,6 +238,9 @@ pub async fn format_me_route(
}
UpstreamType::Socks4 { address, .. } => format!("socks4://{address}"),
UpstreamType::Socks5 { address, .. } => format!("socks5://{address}"),
UpstreamType::Shadowsocks { url, .. } => sanitize_shadowsocks_url(url)
.map(|address| format!("shadowsocks://{address}"))
.unwrap_or_else(|_| "shadowsocks://invalid".to_string()),
};
}
@@ -254,6 +263,12 @@ pub async fn format_me_route(
if has_socks5 {
kinds.push("socks5");
}
if enabled_upstreams
.iter()
.any(|u| matches!(u.upstream_type, UpstreamType::Shadowsocks { .. }))
{
kinds.push("shadowsocks");
}
format!("mixed upstreams ({})", kinds.join(", "))
}
@@ -335,7 +350,10 @@ pub async fn run_me_ping(pool: &Arc<MePool>, rng: &SecureRandom) -> Vec<MePingRe
Ok((stream, conn_rtt, upstream_egress)) => {
connect_ms = Some(conn_rtt);
route = route_from_egress(upstream_egress);
match pool.handshake_only(stream, addr, upstream_egress, rng).await {
match pool
.handshake_only(stream, addr, upstream_egress, rng)
.await
{
Ok(hs) => {
handshake_ms = Some(hs.handshake_ms);
// drop halves to close

6
src/transport/mod.rs
View File

@@ -2,6 +2,7 @@
pub mod pool;
pub mod proxy_protocol;
pub mod shadowsocks;
pub mod socket;
pub mod socks;
pub mod upstream;
@@ -14,5 +15,8 @@ pub use socket::*;
#[allow(unused_imports)]
pub use socks::*;
#[allow(unused_imports)]
pub use upstream::{DcPingResult, StartupPingResult, UpstreamEgressInfo, UpstreamManager, UpstreamRouteKind};
pub use upstream::{
DcPingResult, StartupPingResult, UpstreamEgressInfo, UpstreamManager, UpstreamRouteKind,
UpstreamStream,
};
pub mod middle_proxy;

60
src/transport/shadowsocks.rs Normal file
View File

@@ -0,0 +1,60 @@
use std::net::{IpAddr, SocketAddr};
use std::time::Duration;
use shadowsocks::{
ProxyClientStream,
config::{ServerConfig, ServerType},
context::Context,
net::ConnectOpts,
};
use crate::error::{ProxyError, Result};
pub(crate) type ShadowsocksStream = ProxyClientStream<shadowsocks::net::TcpStream>;
fn parse_server_config(url: &str, connect_timeout: Duration) -> Result<ServerConfig> {
let mut config = ServerConfig::from_url(url)
.map_err(|error| ProxyError::Config(format!("invalid shadowsocks url: {error}")))?;
if config.plugin().is_some() {
return Err(ProxyError::Config(
"shadowsocks plugins are not supported".to_string(),
));
}
config.set_timeout(connect_timeout);
Ok(config)
}
pub(crate) fn sanitize_shadowsocks_url(url: &str) -> Result<String> {
Ok(parse_server_config(url, Duration::from_secs(1))?
.addr()
.to_string())
}
fn connect_opts_for_interface(interface: &Option<String>) -> ConnectOpts {
let mut opts = ConnectOpts::default();
if let Some(interface) = interface {
if let Ok(ip) = interface.parse::<IpAddr>() {
opts.bind_local_addr = Some(SocketAddr::new(ip, 0));
} else {
opts.bind_interface = Some(interface.clone());
}
}
opts
}
pub(crate) async fn connect_shadowsocks(
url: &str,
interface: &Option<String>,
target: SocketAddr,
connect_timeout: Duration,
) -> Result<ShadowsocksStream> {
let config = parse_server_config(url, connect_timeout)?;
let context = Context::new_shared(ServerType::Local);
let opts = connect_opts_for_interface(interface);
ProxyClientStream::connect_with_opts(context, &config, target, &opts)
.await
.map_err(ProxyError::Io)
}

468
src/transport/upstream.rs
View File

@@ -4,22 +4,28 @@
#![allow(deprecated)]
use rand::Rng;
use std::collections::{BTreeSet, HashMap};
use std::net::{SocketAddr, IpAddr};
use std::net::{IpAddr, SocketAddr};
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio::net::TcpStream;
use tokio::sync::RwLock;
use tokio::time::Instant;
use rand::Rng;
use tracing::{debug, warn, info, trace};
use tracing::{debug, info, trace, warn};
use crate::config::{UpstreamConfig, UpstreamType};
use crate::error::{Result, ProxyError};
use crate::error::{ProxyError, Result};
use crate::network::dns_overrides::{resolve_socket_addr, split_host_port};
use crate::protocol::constants::{TG_DATACENTERS_V4, TG_DATACENTERS_V6, TG_DATACENTER_PORT};
use crate::protocol::constants::{TG_DATACENTER_PORT, TG_DATACENTERS_V4, TG_DATACENTERS_V6};
use crate::stats::Stats;
use crate::transport::shadowsocks::{
ShadowsocksStream, connect_shadowsocks, sanitize_shadowsocks_url,
};
use crate::transport::socket::{create_outgoing_socket_bound, resolve_interface_ip};
use crate::transport::socks::{connect_socks4, connect_socks5};
@@ -47,7 +53,10 @@ struct LatencyEma {
impl LatencyEma {
const fn new(alpha: f64) -> Self {
Self { value_ms: None, alpha }
Self {
value_ms: None,
alpha,
}
}
fn update(&mut self, sample_ms: f64) {
@@ -131,11 +140,17 @@ impl UpstreamState {
return Some(ms);
}
let (sum, count) = self.dc_latency.iter()
let (sum, count) = self
.dc_latency
.iter()
.filter_map(|l| l.get())
.fold((0.0, 0u32), |(s, c), v| (s + v, c + 1));
if count > 0 { Some(sum / count as f64) } else { None }
if count > 0 {
Some(sum / count as f64)
} else {
None
}
}
}
@@ -158,11 +173,78 @@ pub struct StartupPingResult {
pub both_available: bool,
}
pub enum UpstreamStream {
Tcp(TcpStream),
Shadowsocks(Box<ShadowsocksStream>),
}
impl std::fmt::Debug for UpstreamStream {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Tcp(_) => f.write_str("UpstreamStream::Tcp(..)"),
Self::Shadowsocks(_) => f.write_str("UpstreamStream::Shadowsocks(..)"),
}
}
}
impl UpstreamStream {
pub fn into_tcp(self) -> Result<TcpStream> {
match self {
Self::Tcp(stream) => Ok(stream),
Self::Shadowsocks(_) => Err(ProxyError::Config(
"shadowsocks upstreams are not supported when general.use_middle_proxy = true"
.to_string(),
)),
}
}
}
impl AsyncRead for UpstreamStream {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
match self.get_mut() {
Self::Tcp(stream) => Pin::new(stream).poll_read(cx, buf),
Self::Shadowsocks(stream) => Pin::new(stream.as_mut()).poll_read(cx, buf),
}
}
}
impl AsyncWrite for UpstreamStream {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
match self.get_mut() {
Self::Tcp(stream) => Pin::new(stream).poll_write(cx, buf),
Self::Shadowsocks(stream) => Pin::new(stream.as_mut()).poll_write(cx, buf),
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
match self.get_mut() {
Self::Tcp(stream) => Pin::new(stream).poll_flush(cx),
Self::Shadowsocks(stream) => Pin::new(stream.as_mut()).poll_flush(cx),
}
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
match self.get_mut() {
Self::Tcp(stream) => Pin::new(stream).poll_shutdown(cx),
Self::Shadowsocks(stream) => Pin::new(stream.as_mut()).poll_shutdown(cx),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UpstreamRouteKind {
Direct,
Socks4,
Socks5,
Shadowsocks,
}
#[derive(Debug, Clone)]
@@ -194,6 +276,7 @@ pub struct UpstreamApiSummarySnapshot {
pub direct_total: usize,
pub socks4_total: usize,
pub socks5_total: usize,
pub shadowsocks_total: usize,
}
#[derive(Debug, Clone)]
@@ -253,7 +336,8 @@ impl UpstreamManager {
connect_failfast_hard_errors: bool,
stats: Arc<Stats>,
) -> Self {
let states = configs.into_iter()
let states = configs
.into_iter()
.filter(|c| c.enabled)
.map(UpstreamState::new)
.collect();
@@ -311,20 +395,13 @@ impl UpstreamManager {
summary.unhealthy_total += 1;
}
let (route_kind, address) = match &upstream.config.upstream_type {
UpstreamType::Direct { .. } => {
summary.direct_total += 1;
(UpstreamRouteKind::Direct, "direct".to_string())
}
UpstreamType::Socks4 { address, .. } => {
summary.socks4_total += 1;
(UpstreamRouteKind::Socks4, address.clone())
}
UpstreamType::Socks5 { address, .. } => {
summary.socks5_total += 1;
(UpstreamRouteKind::Socks5, address.clone())
}
};
let (route_kind, address) = Self::describe_upstream(&upstream.config.upstream_type);
match route_kind {
UpstreamRouteKind::Direct => summary.direct_total += 1,
UpstreamRouteKind::Socks4 => summary.socks4_total += 1,
UpstreamRouteKind::Socks5 => summary.socks5_total += 1,
UpstreamRouteKind::Shadowsocks => summary.shadowsocks_total += 1,
}
let mut dc = Vec::with_capacity(NUM_DCS);
for dc_idx in 0..NUM_DCS {
@@ -352,6 +429,18 @@ impl UpstreamManager {
Some(UpstreamApiSnapshot { summary, upstreams })
}
fn describe_upstream(upstream_type: &UpstreamType) -> (UpstreamRouteKind, String) {
match upstream_type {
UpstreamType::Direct { .. } => (UpstreamRouteKind::Direct, "direct".to_string()),
UpstreamType::Socks4 { address, .. } => (UpstreamRouteKind::Socks4, address.clone()),
UpstreamType::Socks5 { address, .. } => (UpstreamRouteKind::Socks5, address.clone()),
UpstreamType::Shadowsocks { url, .. } => (
UpstreamRouteKind::Shadowsocks,
sanitize_shadowsocks_url(url).unwrap_or_else(|_| "invalid".to_string()),
),
}
}
pub fn api_policy_snapshot(&self) -> UpstreamApiPolicySnapshot {
UpstreamApiPolicySnapshot {
connect_retry_attempts: self.connect_retry_attempts,
@@ -539,44 +628,44 @@ impl UpstreamManager {
// Scope filter:
// If scope is set: only scoped and matched items
// If scope is not set: only unscoped items
let filtered_upstreams : Vec<usize> = upstreams.iter()
let filtered_upstreams: Vec<usize> = upstreams
.iter()
.enumerate()
.filter(|(_, u)| {
scope.map_or(
u.config.scopes.is_empty(),
|req_scope| {
u.config.scopes
.split(',')
.map(str::trim)
.any(|s| s == req_scope)
}
)
scope.map_or(u.config.scopes.is_empty(), |req_scope| {
u.config
.scopes
.split(',')
.map(str::trim)
.any(|s| s == req_scope)
})
})
.map(|(i, _)| i)
.collect();
// Healthy filter
let healthy: Vec<usize> = filtered_upstreams.iter()
let healthy: Vec<usize> = filtered_upstreams
.iter()
.filter(|&&i| upstreams[i].healthy)
.copied()
.collect();
if filtered_upstreams.is_empty() {
if Self::should_emit_warn(
self.no_upstreams_warn_epoch_ms.as_ref(),
5_000,
) {
warn!(scope = scope, "No upstreams available! Using first (direct?)");
if Self::should_emit_warn(self.no_upstreams_warn_epoch_ms.as_ref(), 5_000) {
warn!(
scope = scope,
"No upstreams available! Using first (direct?)"
);
}
return None;
}
if healthy.is_empty() {
if Self::should_emit_warn(
self.no_healthy_warn_epoch_ms.as_ref(),
5_000,
) {
warn!(scope = scope, "No healthy upstreams available! Using random.");
if Self::should_emit_warn(self.no_healthy_warn_epoch_ms.as_ref(), 5_000) {
warn!(
scope = scope,
"No healthy upstreams available! Using random."
);
}
return Some(filtered_upstreams[rand::rng().gen_range(0..filtered_upstreams.len())]);
}
@@ -585,14 +674,18 @@ impl UpstreamManager {
return Some(healthy[0]);
}
let weights: Vec<(usize, f64)> = healthy.iter().map(|&i| {
let base = upstreams[i].config.weight as f64;
let latency_factor = upstreams[i].effective_latency(dc_idx)
.map(|ms| if ms > 1.0 { 1000.0 / ms } else { 1000.0 })
.unwrap_or(1.0);
let weights: Vec<(usize, f64)> = healthy
.iter()
.map(|&i| {
let base = upstreams[i].config.weight as f64;
let latency_factor = upstreams[i]
.effective_latency(dc_idx)
.map(|ms| if ms > 1.0 { 1000.0 / ms } else { 1000.0 })
.unwrap_or(1.0);
(i, base * latency_factor)
}).collect();
(i, base * latency_factor)
})
.collect();
let total: f64 = weights.iter().map(|(_, w)| w).sum();
@@ -620,8 +713,34 @@ impl UpstreamManager {
}
/// Connect to target through a selected upstream.
pub async fn connect(&self, target: SocketAddr, dc_idx: Option<i16>, scope: Option<&str>) -> Result<TcpStream> {
let (stream, _) = self.connect_with_details(target, dc_idx, scope).await?;
pub async fn connect(
&self,
target: SocketAddr,
dc_idx: Option<i16>,
scope: Option<&str>,
) -> Result<UpstreamStream> {
let idx = self
.select_upstream(dc_idx, scope)
.await
.ok_or_else(|| ProxyError::Config("No upstreams available".to_string()))?;
let mut upstream = {
let guard = self.upstreams.read().await;
guard[idx].config.clone()
};
if let Some(s) = scope {
upstream.selected_scope = s.to_string();
}
let bind_rr = {
let guard = self.upstreams.read().await;
guard.get(idx).map(|u| u.bind_rr.clone())
};
let (stream, _) = self
.connect_selected_upstream(idx, upstream, target, dc_idx, bind_rr)
.await?;
Ok(stream)
}
@@ -632,7 +751,9 @@ impl UpstreamManager {
dc_idx: Option<i16>,
scope: Option<&str>,
) -> Result<(TcpStream, UpstreamEgressInfo)> {
let idx = self.select_upstream(dc_idx, scope).await
let idx = self
.select_upstream(dc_idx, scope)
.await
.ok_or_else(|| ProxyError::Config("No upstreams available".to_string()))?;
let mut upstream = {
@@ -650,6 +771,20 @@ impl UpstreamManager {
guard.get(idx).map(|u| u.bind_rr.clone())
};
let (stream, egress) = self
.connect_selected_upstream(idx, upstream, target, dc_idx, bind_rr)
.await?;
Ok((stream.into_tcp()?, egress))
}
async fn connect_selected_upstream(
&self,
idx: usize,
upstream: UpstreamConfig,
target: SocketAddr,
dc_idx: Option<i16>,
bind_rr: Option<Arc<AtomicUsize>>,
) -> Result<(UpstreamStream, UpstreamEgressInfo)> {
let connect_started_at = Instant::now();
let mut last_error: Option<ProxyError> = None;
let mut attempts_used = 0u32;
@@ -662,8 +797,8 @@ impl UpstreamManager {
break;
}
let remaining_budget = self.connect_budget.saturating_sub(elapsed);
let attempt_timeout = Duration::from_secs(DIRECT_CONNECT_TIMEOUT_SECS)
.min(remaining_budget);
let attempt_timeout =
Duration::from_secs(DIRECT_CONNECT_TIMEOUT_SECS).min(remaining_budget);
if attempt_timeout.is_zero() {
last_error = Some(ProxyError::ConnectionTimeout {
addr: target.to_string(),
@@ -786,9 +921,12 @@ impl UpstreamManager {
target: SocketAddr,
bind_rr: Option<Arc<AtomicUsize>>,
connect_timeout: Duration,
) -> Result<(TcpStream, UpstreamEgressInfo)> {
) -> Result<(UpstreamStream, UpstreamEgressInfo)> {
match &config.upstream_type {
UpstreamType::Direct { interface, bind_addresses } => {
UpstreamType::Direct {
interface,
bind_addresses,
} => {
let bind_ip = Self::resolve_bind_address(
interface,
bind_addresses,
@@ -796,9 +934,7 @@ impl UpstreamManager {
bind_rr.as_deref(),
true,
);
if bind_ip.is_none()
&& bind_addresses.as_ref().is_some_and(|v| !v.is_empty())
{
if bind_ip.is_none() && bind_addresses.as_ref().is_some_and(|v| !v.is_empty()) {
return Err(ProxyError::Config(format!(
"No valid bind_addresses for target family {target}"
)));
@@ -813,8 +949,10 @@ impl UpstreamManager {
socket.set_nonblocking(true)?;
match socket.connect(&target.into()) {
Ok(()) => {},
Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
Ok(()) => {}
Err(err)
if err.raw_os_error() == Some(libc::EINPROGRESS)
|| err.kind() == std::io::ErrorKind::WouldBlock => {}
Err(err) => return Err(ProxyError::Io(err)),
}
@@ -836,7 +974,7 @@ impl UpstreamManager {
let local_addr = stream.local_addr().ok();
Ok((
stream,
UpstreamStream::Tcp(stream),
UpstreamEgressInfo {
upstream_id,
route_kind: UpstreamRouteKind::Direct,
@@ -846,8 +984,12 @@ impl UpstreamManager {
socks_proxy_addr: None,
},
))
},
UpstreamType::Socks4 { address, interface, user_id } => {
}
UpstreamType::Socks4 {
address,
interface,
user_id,
} => {
// Try to parse as SocketAddr first (IP:port), otherwise treat as hostname:port
let mut stream = if let Ok(proxy_addr) = address.parse::<SocketAddr>() {
// IP:port format - use socket with optional interface binding
@@ -863,8 +1005,10 @@ impl UpstreamManager {
socket.set_nonblocking(true)?;
match socket.connect(&proxy_addr.into()) {
Ok(()) => {},
Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
Ok(()) => {}
Err(err)
if err.raw_os_error() == Some(libc::EINPROGRESS)
|| err.kind() == std::io::ErrorKind::WouldBlock => {}
Err(err) => return Err(ProxyError::Io(err)),
}
@@ -888,14 +1032,16 @@ impl UpstreamManager {
// Hostname:port format - use tokio DNS resolution
// Note: interface binding is not supported for hostnames
if interface.is_some() {
warn!("SOCKS4 interface binding is not supported for hostname addresses, ignoring");
warn!(
"SOCKS4 interface binding is not supported for hostname addresses, ignoring"
);
}
Self::connect_hostname_with_dns_override(address, connect_timeout).await?
};
// replace socks user_id with config.selected_scope, if set
let scope: Option<&str> = Some(config.selected_scope.as_str())
.filter(|s| !s.is_empty());
let scope: Option<&str> =
Some(config.selected_scope.as_str()).filter(|s| !s.is_empty());
let _user_id: Option<&str> = scope.or(user_id.as_deref());
let bound = match tokio::time::timeout(
@@ -915,7 +1061,7 @@ impl UpstreamManager {
let local_addr = stream.local_addr().ok();
let socks_proxy_addr = stream.peer_addr().ok();
Ok((
stream,
UpstreamStream::Tcp(stream),
UpstreamEgressInfo {
upstream_id,
route_kind: UpstreamRouteKind::Socks4,
@@ -925,8 +1071,13 @@ impl UpstreamManager {
socks_proxy_addr,
},
))
},
UpstreamType::Socks5 { address, interface, username, password } => {
}
UpstreamType::Socks5 {
address,
interface,
username,
password,
} => {
// Try to parse as SocketAddr first (IP:port), otherwise treat as hostname:port
let mut stream = if let Ok(proxy_addr) = address.parse::<SocketAddr>() {
// IP:port format - use socket with optional interface binding
@@ -942,8 +1093,10 @@ impl UpstreamManager {
socket.set_nonblocking(true)?;
match socket.connect(&proxy_addr.into()) {
Ok(()) => {},
Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) || err.kind() == std::io::ErrorKind::WouldBlock => {},
Ok(()) => {}
Err(err)
if err.raw_os_error() == Some(libc::EINPROGRESS)
|| err.kind() == std::io::ErrorKind::WouldBlock => {}
Err(err) => return Err(ProxyError::Io(err)),
}
@@ -967,15 +1120,17 @@ impl UpstreamManager {
// Hostname:port format - use tokio DNS resolution
// Note: interface binding is not supported for hostnames
if interface.is_some() {
warn!("SOCKS5 interface binding is not supported for hostname addresses, ignoring");
warn!(
"SOCKS5 interface binding is not supported for hostname addresses, ignoring"
);
}
Self::connect_hostname_with_dns_override(address, connect_timeout).await?
};
debug!(config = ?config, "Socks5 connection");
// replace socks user:pass with config.selected_scope, if set
let scope: Option<&str> = Some(config.selected_scope.as_str())
.filter(|s| !s.is_empty());
let scope: Option<&str> =
Some(config.selected_scope.as_str()).filter(|s| !s.is_empty());
let _username: Option<&str> = scope.or(username.as_deref());
let _password: Option<&str> = scope.or(password.as_deref());
@@ -996,7 +1151,7 @@ impl UpstreamManager {
let local_addr = stream.local_addr().ok();
let socks_proxy_addr = stream.peer_addr().ok();
Ok((
stream,
UpstreamStream::Tcp(stream),
UpstreamEgressInfo {
upstream_id,
route_kind: UpstreamRouteKind::Socks5,
@@ -1006,7 +1161,22 @@ impl UpstreamManager {
socks_proxy_addr,
},
))
},
}
UpstreamType::Shadowsocks { url, interface } => {
let stream = connect_shadowsocks(url, interface, target, connect_timeout).await?;
let local_addr = stream.get_ref().local_addr().ok();
Ok((
UpstreamStream::Shadowsocks(Box::new(stream)),
UpstreamEgressInfo {
upstream_id,
route_kind: UpstreamRouteKind::Shadowsocks,
local_addr,
direct_bind_ip: None,
socks_bound_addr: None,
socks_proxy_addr: None,
},
))
}
}
}
@@ -1023,7 +1193,9 @@ impl UpstreamManager {
) -> Vec<StartupPingResult> {
let upstreams: Vec<(usize, UpstreamConfig, Arc<AtomicUsize>)> = {
let guard = self.upstreams.read().await;
guard.iter().enumerate()
guard
.iter()
.enumerate()
.map(|(i, u)| (i, u.config.clone(), u.bind_rr.clone()))
.collect()
};
@@ -1051,6 +1223,11 @@ impl UpstreamManager {
}
UpstreamType::Socks4 { address, .. } => format!("socks4://{}", address),
UpstreamType::Socks5 { address, .. } => format!("socks5://{}", address),
UpstreamType::Shadowsocks { url, .. } => {
let address =
sanitize_shadowsocks_url(url).unwrap_or_else(|_| "invalid".to_string());
format!("shadowsocks://{address}")
}
};
let mut v6_results = Vec::with_capacity(NUM_DCS);
@@ -1061,8 +1238,14 @@ impl UpstreamManager {
let result = tokio::time::timeout(
Duration::from_secs(DC_PING_TIMEOUT_SECS),
self.ping_single_dc(*upstream_idx, upstream_config, Some(bind_rr.clone()), addr_v6)
).await;
self.ping_single_dc(
*upstream_idx,
upstream_config,
Some(bind_rr.clone()),
addr_v6,
),
)
.await;
let ping_result = match result {
Ok(Ok(rtt_ms)) => {
@@ -1112,8 +1295,14 @@ impl UpstreamManager {
let result = tokio::time::timeout(
Duration::from_secs(DC_PING_TIMEOUT_SECS),
self.ping_single_dc(*upstream_idx, upstream_config, Some(bind_rr.clone()), addr_v4)
).await;
self.ping_single_dc(
*upstream_idx,
upstream_config,
Some(bind_rr.clone()),
addr_v4,
),
)
.await;
let ping_result = match result {
Ok(Ok(rtt_ms)) => {
@@ -1162,7 +1351,7 @@ impl UpstreamManager {
Err(_) => {
warn!(dc = %dc_key, "Invalid dc_overrides key, skipping");
continue;
},
}
_ => continue,
};
let dc_idx = dc_num as usize;
@@ -1175,8 +1364,14 @@ impl UpstreamManager {
}
let result = tokio::time::timeout(
Duration::from_secs(DC_PING_TIMEOUT_SECS),
self.ping_single_dc(*upstream_idx, upstream_config, Some(bind_rr.clone()), addr)
).await;
self.ping_single_dc(
*upstream_idx,
upstream_config,
Some(bind_rr.clone()),
addr,
),
)
.await;
let ping_result = match result {
Ok(Ok(rtt_ms)) => DcPingResult {
@@ -1205,7 +1400,9 @@ impl UpstreamManager {
v4_results.push(ping_result);
}
}
Err(_) => warn!(dc = %dc_idx, addr = %addr_str, "Invalid dc_overrides address, skipping"),
Err(_) => {
warn!(dc = %dc_idx, addr = %addr_str, "Invalid dc_overrides address, skipping")
}
}
}
}
@@ -1381,12 +1578,8 @@ impl UpstreamManager {
ipv6_enabled: bool,
dc_overrides: HashMap<String, Vec<String>>,
) {
let groups = Self::build_health_check_groups(
prefer_ipv6,
ipv4_enabled,
ipv6_enabled,
&dc_overrides,
);
let groups =
Self::build_health_check_groups(prefer_ipv6, ipv4_enabled, ipv6_enabled, &dc_overrides);
let required_healthy_groups = Self::required_healthy_group_count(groups.len());
let mut endpoint_rotation: HashMap<(usize, i16, bool), usize> = HashMap::new();
@@ -1416,13 +1609,16 @@ impl UpstreamManager {
let mut group_ok = false;
let mut group_rtt_ms = None;
for (is_primary, endpoints) in [(true, &group.primary), (false, &group.fallback)] {
for (is_primary, endpoints) in
[(true, &group.primary), (false, &group.fallback)]
{
if endpoints.is_empty() {
continue;
}
let rotation_key = (i, group.dc_idx, is_primary);
let start_idx = *endpoint_rotation.entry(rotation_key).or_insert(0) % endpoints.len();
let start_idx =
*endpoint_rotation.entry(rotation_key).or_insert(0) % endpoints.len();
let mut next_idx = (start_idx + 1) % endpoints.len();
for step in 0..endpoints.len() {
@@ -1544,8 +1740,7 @@ impl UpstreamManager {
return None;
}
UpstreamState::dc_array_idx(dc_idx)
.map(|idx| guard[0].dc_ip_pref[idx])
UpstreamState::dc_array_idx(dc_idx).map(|idx| guard[0].dc_ip_pref[idx])
}
/// Get preferred DC address based on config preference
@@ -1566,6 +1761,12 @@ impl UpstreamManager {
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use crate::stats::Stats;
const TEST_SHADOWSOCKS_URL: &str =
"ss://2022-blake3-aes-256-gcm:MDEyMzQ1Njc4OTAxMjM0NTY3ODkwMTIzNDU2Nzg5MDE=@127.0.0.1:8388";
#[test]
fn required_healthy_group_count_applies_three_group_threshold() {
@@ -1596,15 +1797,18 @@ mod tests {
assert!(dc2.primary.iter().all(|addr| addr.is_ipv6()));
assert!(dc2.fallback.iter().all(|addr| addr.is_ipv4()));
assert!(dc2
.primary
.contains(&"[2001:db8::10]:443".parse::<SocketAddr>().unwrap()));
assert!(dc2
.fallback
.contains(&"203.0.113.10:443".parse::<SocketAddr>().unwrap()));
assert!(dc2
.fallback
.contains(&"203.0.113.11:443".parse::<SocketAddr>().unwrap()));
assert!(
dc2.primary
.contains(&"[2001:db8::10]:443".parse::<SocketAddr>().unwrap())
);
assert!(
dc2.fallback
.contains(&"203.0.113.10:443".parse::<SocketAddr>().unwrap())
);
assert!(
dc2.fallback
.contains(&"203.0.113.11:443".parse::<SocketAddr>().unwrap())
);
}
#[test]
@@ -1626,12 +1830,14 @@ mod tests {
.expect("override-only dc group must be present");
assert_eq!(dc9.primary.len(), 2);
assert!(dc9
.primary
.contains(&"198.51.100.1:443".parse::<SocketAddr>().unwrap()));
assert!(dc9
.primary
.contains(&"198.51.100.2:443".parse::<SocketAddr>().unwrap()));
assert!(
dc9.primary
.contains(&"198.51.100.1:443".parse::<SocketAddr>().unwrap())
);
assert!(
dc9.primary
.contains(&"198.51.100.2:443".parse::<SocketAddr>().unwrap())
);
assert!(dc9.fallback.is_empty());
}
@@ -1678,4 +1884,36 @@ mod tests {
assert_eq!(bind, None);
}
#[test]
fn api_snapshot_reports_shadowsocks_as_sanitized_route() {
let manager = UpstreamManager::new(
vec![UpstreamConfig {
upstream_type: UpstreamType::Shadowsocks {
url: TEST_SHADOWSOCKS_URL.to_string(),
interface: None,
},
weight: 2,
enabled: true,
scopes: String::new(),
selected_scope: String::new(),
}],
1,
100,
1000,
1,
false,
Arc::new(Stats::new()),
);
let snapshot = manager.try_api_snapshot().expect("snapshot");
assert_eq!(snapshot.summary.configured_total, 1);
assert_eq!(snapshot.summary.shadowsocks_total, 1);
assert_eq!(snapshot.upstreams.len(), 1);
assert_eq!(
snapshot.upstreams[0].route_kind,
UpstreamRouteKind::Shadowsocks
);
assert_eq!(snapshot.upstreams[0].address, "127.0.0.1:8388");
}
}