mirror of
https://github.com/shadowsocks/shadowsocks-rust.git
synced 2026-11-06 08:14:07 +00:00
feat(local): Local server builder pattern
This commit is contained in:
zonyitoo
@@ -51,7 +51,7 @@
|
||||
use std::time::Duration;
|
||||
|
||||
#[cfg(feature = "local")]
|
||||
pub use self::local::{create as create_local, run as run_local};
|
||||
pub use self::local::run as run_local;
|
||||
|
||||
#[cfg(feature = "manager")]
|
||||
pub use self::manager::run as run_manager;
|
||||
|
||||
@@ -29,6 +29,18 @@ use self::{
|
||||
loadbalancing::{PingBalancer, PingBalancerBuilder},
|
||||
};
|
||||
|
||||
#[cfg(feature = "local-dns")]
|
||||
use self::dns::{Dns, DnsBuilder};
|
||||
#[cfg(feature = "local-http")]
|
||||
use self::http::{Http, HttpBuilder};
|
||||
#[cfg(feature = "local-redir")]
|
||||
use self::redir::{Redir, RedirBuilder};
|
||||
use self::socks::{Socks, SocksBuilder};
|
||||
#[cfg(feature = "local-tun")]
|
||||
use self::tun::{Tun, TunBuilder};
|
||||
#[cfg(feature = "local-tunnel")]
|
||||
use self::tunnel::{Tunnel, TunnelBuilder};
|
||||
|
||||
pub mod context;
|
||||
#[cfg(feature = "local-dns")]
|
||||
pub mod dns;
|
||||
@@ -73,25 +85,396 @@ impl Future for ServerHandle {
|
||||
|
||||
/// Local Server instance
|
||||
pub struct Server {
|
||||
vfut: Vec<ServerHandle>,
|
||||
balancer: PingBalancer,
|
||||
socks_servers: Vec<Socks>,
|
||||
#[cfg(feature = "local-tunnel")]
|
||||
tunnel_servers: Vec<Tunnel>,
|
||||
#[cfg(feature = "local-http")]
|
||||
http_servers: Vec<Http>,
|
||||
#[cfg(feature = "local-tun")]
|
||||
tun_servers: Vec<Tun>,
|
||||
#[cfg(feature = "local-dns")]
|
||||
dns_servers: Vec<Dns>,
|
||||
#[cfg(feature = "local-redir")]
|
||||
redir_servers: Vec<Redir>,
|
||||
local_stat_addr: Option<LocalFlowStatAddress>,
|
||||
flow_stat: Arc<FlowStat>,
|
||||
}
|
||||
|
||||
impl Server {
|
||||
/// Create a shadowsocks local server
|
||||
pub async fn create(config: Config) -> io::Result<Server> {
|
||||
create(config).await
|
||||
pub async fn new(config: Config) -> io::Result<Server> {
|
||||
assert!(config.config_type == ConfigType::Local && !config.local.is_empty());
|
||||
|
||||
trace!("{:?}", config);
|
||||
|
||||
// Warning for Stream Ciphers
|
||||
#[cfg(feature = "stream-cipher")]
|
||||
for inst in config.server.iter() {
|
||||
let server = &inst.config;
|
||||
|
||||
if server.method().is_stream() {
|
||||
log::warn!("stream cipher {} for server {} have inherent weaknesses (see discussion in https://github.com/shadowsocks/shadowsocks-org/issues/36). \
|
||||
DO NOT USE. It will be removed in the future.", server.method(), server.addr());
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "android")))]
|
||||
if let Some(nofile) = config.nofile {
|
||||
use crate::sys::set_nofile;
|
||||
if let Err(err) = set_nofile(nofile) {
|
||||
log::warn!("set_nofile {} failed, error: {}", nofile, err);
|
||||
}
|
||||
}
|
||||
|
||||
// Global ServiceContext template
|
||||
// Each Local instance will hold a copy of its fields
|
||||
let mut context = ServiceContext::new();
|
||||
|
||||
let mut connect_opts = ConnectOpts {
|
||||
#[cfg(any(target_os = "linux", target_os = "android"))]
|
||||
fwmark: config.outbound_fwmark,
|
||||
|
||||
#[cfg(target_os = "android")]
|
||||
vpn_protect_path: config.outbound_vpn_protect_path,
|
||||
|
||||
bind_interface: config.outbound_bind_interface,
|
||||
bind_local_addr: config.outbound_bind_addr,
|
||||
|
||||
..Default::default()
|
||||
};
|
||||
connect_opts.tcp.send_buffer_size = config.outbound_send_buffer_size;
|
||||
connect_opts.tcp.recv_buffer_size = config.outbound_recv_buffer_size;
|
||||
connect_opts.tcp.nodelay = config.no_delay;
|
||||
connect_opts.tcp.fastopen = config.fast_open;
|
||||
connect_opts.tcp.keepalive = config.keep_alive.or(Some(LOCAL_DEFAULT_KEEPALIVE_TIMEOUT));
|
||||
connect_opts.tcp.mptcp = config.mptcp;
|
||||
context.set_connect_opts(connect_opts);
|
||||
|
||||
let mut accept_opts = AcceptOpts {
|
||||
ipv6_only: config.ipv6_only,
|
||||
..Default::default()
|
||||
};
|
||||
accept_opts.tcp.send_buffer_size = config.inbound_send_buffer_size;
|
||||
accept_opts.tcp.recv_buffer_size = config.inbound_recv_buffer_size;
|
||||
accept_opts.tcp.nodelay = config.no_delay;
|
||||
accept_opts.tcp.fastopen = config.fast_open;
|
||||
accept_opts.tcp.keepalive = config.keep_alive.or(Some(LOCAL_DEFAULT_KEEPALIVE_TIMEOUT));
|
||||
accept_opts.tcp.mptcp = config.mptcp;
|
||||
context.set_accept_opts(accept_opts);
|
||||
|
||||
if let Some(resolver) = build_dns_resolver(config.dns, config.ipv6_first, context.connect_opts_ref()).await {
|
||||
context.set_dns_resolver(Arc::new(resolver));
|
||||
}
|
||||
|
||||
if config.ipv6_first {
|
||||
context.set_ipv6_first(config.ipv6_first);
|
||||
}
|
||||
|
||||
if let Some(acl) = config.acl {
|
||||
context.set_acl(Arc::new(acl));
|
||||
}
|
||||
|
||||
context.set_security_config(&config.security);
|
||||
|
||||
assert!(!config.local.is_empty(), "no valid local server configuration");
|
||||
|
||||
// Create a service balancer for choosing between multiple servers
|
||||
let balancer = {
|
||||
let mut mode = Mode::TcpOnly;
|
||||
|
||||
for local in &config.local {
|
||||
mode = mode.merge(local.config.mode);
|
||||
}
|
||||
|
||||
// Load balancer will hold an individual ServiceContext
|
||||
let mut balancer_builder = PingBalancerBuilder::new(Arc::new(context.clone()), mode);
|
||||
|
||||
// max_server_rtt have to be set before add_server
|
||||
if let Some(rtt) = config.balancer.max_server_rtt {
|
||||
balancer_builder.max_server_rtt(rtt);
|
||||
}
|
||||
|
||||
if let Some(intv) = config.balancer.check_interval {
|
||||
balancer_builder.check_interval(intv);
|
||||
}
|
||||
|
||||
if let Some(intv) = config.balancer.check_best_interval {
|
||||
balancer_builder.check_best_interval(intv);
|
||||
}
|
||||
|
||||
for server in config.server {
|
||||
balancer_builder.add_server(server.config);
|
||||
}
|
||||
|
||||
balancer_builder.build().await?
|
||||
};
|
||||
|
||||
let mut local_server = Server {
|
||||
balancer: balancer.clone(),
|
||||
socks_servers: Vec::new(),
|
||||
#[cfg(feature = "local-tunnel")]
|
||||
tunnel_servers: Vec::new(),
|
||||
#[cfg(feature = "local-http")]
|
||||
http_servers: Vec::new(),
|
||||
#[cfg(feature = "local-tun")]
|
||||
tun_servers: Vec::new(),
|
||||
#[cfg(feature = "local-dns")]
|
||||
dns_servers: Vec::new(),
|
||||
#[cfg(feature = "local-redir")]
|
||||
redir_servers: Vec::new(),
|
||||
local_stat_addr: config.local_stat_addr,
|
||||
flow_stat: context.flow_stat(),
|
||||
};
|
||||
|
||||
for local_instance in config.local {
|
||||
let local_config = local_instance.config;
|
||||
|
||||
// Clone from global ServiceContext instance
|
||||
// It will shares Shadowsocks' global context, and FlowStat, DNS reverse cache
|
||||
let mut context = context.clone();
|
||||
|
||||
// Private ACL
|
||||
if let Some(acl) = local_instance.acl {
|
||||
context.set_acl(Arc::new(acl))
|
||||
}
|
||||
|
||||
let context = Arc::new(context);
|
||||
let balancer = balancer.clone();
|
||||
|
||||
match local_config.protocol {
|
||||
ProtocolType::Socks => {
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "socks requires local address")),
|
||||
};
|
||||
|
||||
let mut server_builder = SocksBuilder::with_context(context.clone(), client_addr, balancer);
|
||||
server_builder.set_mode(local_config.mode);
|
||||
server_builder.set_socks5_auth(local_config.socks5_auth);
|
||||
|
||||
if let Some(c) = config.udp_max_associations {
|
||||
server_builder.set_udp_capacity(c);
|
||||
}
|
||||
if let Some(d) = config.udp_timeout {
|
||||
server_builder.set_udp_expiry_duration(d);
|
||||
}
|
||||
if let Some(b) = local_config.udp_addr {
|
||||
server_builder.set_udp_bind_addr(b.clone());
|
||||
}
|
||||
|
||||
let server = server_builder.build().await?;
|
||||
local_server.socks_servers.push(server);
|
||||
}
|
||||
#[cfg(feature = "local-tunnel")]
|
||||
ProtocolType::Tunnel => {
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "tunnel requires local address")),
|
||||
};
|
||||
|
||||
let forward_addr = local_config.forward_addr.expect("tunnel requires forward address");
|
||||
|
||||
let mut server_builder =
|
||||
TunnelBuilder::with_context(context.clone(), forward_addr.clone(), client_addr, balancer);
|
||||
|
||||
if let Some(c) = config.udp_max_associations {
|
||||
server_builder.set_udp_capacity(c);
|
||||
}
|
||||
if let Some(d) = config.udp_timeout {
|
||||
server_builder.set_udp_expiry_duration(d);
|
||||
}
|
||||
server_builder.set_mode(local_config.mode);
|
||||
if let Some(udp_addr) = local_config.udp_addr {
|
||||
server_builder.set_udp_bind_addr(udp_addr);
|
||||
}
|
||||
|
||||
let server = server_builder.build().await?;
|
||||
local_server.tunnel_servers.push(server);
|
||||
}
|
||||
#[cfg(feature = "local-http")]
|
||||
ProtocolType::Http => {
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "http requires local address")),
|
||||
};
|
||||
|
||||
let builder = HttpBuilder::with_context(context.clone(), client_addr, balancer);
|
||||
let server = builder.build().await?;
|
||||
local_server.http_servers.push(server);
|
||||
}
|
||||
#[cfg(feature = "local-redir")]
|
||||
ProtocolType::Redir => {
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "redir requires local address")),
|
||||
};
|
||||
|
||||
let mut server_builder = RedirBuilder::with_context(context.clone(), client_addr, balancer);
|
||||
if let Some(c) = config.udp_max_associations {
|
||||
server_builder.set_udp_capacity(c);
|
||||
}
|
||||
if let Some(d) = config.udp_timeout {
|
||||
server_builder.set_udp_expiry_duration(d);
|
||||
}
|
||||
server_builder.set_mode(local_config.mode);
|
||||
server_builder.set_tcp_redir(local_config.tcp_redir);
|
||||
server_builder.set_udp_redir(local_config.udp_redir);
|
||||
if let Some(udp_addr) = local_config.udp_addr {
|
||||
server_builder.set_udp_bind_addr(udp_addr);
|
||||
}
|
||||
|
||||
let server = server_builder.build().await?;
|
||||
local_server.redir_servers.push(server);
|
||||
}
|
||||
#[cfg(feature = "local-dns")]
|
||||
ProtocolType::Dns => {
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "dns requires local address")),
|
||||
};
|
||||
|
||||
let mut server_builder = {
|
||||
let local_addr = local_config.local_dns_addr.expect("missing local_dns_addr");
|
||||
let remote_addr = local_config.remote_dns_addr.expect("missing remote_dns_addr");
|
||||
|
||||
DnsBuilder::with_context(
|
||||
context.clone(),
|
||||
client_addr,
|
||||
local_addr.clone(),
|
||||
remote_addr.clone(),
|
||||
balancer,
|
||||
)
|
||||
};
|
||||
server_builder.set_mode(local_config.mode);
|
||||
|
||||
let server = server_builder.build().await?;
|
||||
local_server.dns_servers.push(server);
|
||||
}
|
||||
#[cfg(feature = "local-tun")]
|
||||
ProtocolType::Tun => {
|
||||
use log::info;
|
||||
use shadowsocks::net::UnixListener;
|
||||
|
||||
let mut builder = TunBuilder::new(context.clone(), balancer);
|
||||
if let Some(address) = local_config.tun_interface_address {
|
||||
builder.address(address);
|
||||
}
|
||||
if let Some(address) = local_config.tun_interface_destination {
|
||||
builder.destination(address);
|
||||
}
|
||||
if let Some(name) = local_config.tun_interface_name {
|
||||
builder.name(&name);
|
||||
}
|
||||
if let Some(c) = config.udp_max_associations {
|
||||
builder.udp_capacity(c);
|
||||
}
|
||||
if let Some(d) = config.udp_timeout {
|
||||
builder.udp_expiry_duration(d);
|
||||
}
|
||||
builder.mode(local_config.mode);
|
||||
#[cfg(unix)]
|
||||
if let Some(fd) = local_config.tun_device_fd {
|
||||
builder.file_descriptor(fd);
|
||||
} else if let Some(ref fd_path) = local_config.tun_device_fd_from_path {
|
||||
use std::fs;
|
||||
|
||||
let _ = fs::remove_file(fd_path);
|
||||
|
||||
let listener = match UnixListener::bind(fd_path) {
|
||||
Ok(l) => l,
|
||||
Err(err) => {
|
||||
log::error!("failed to bind uds path \"{}\", error: {}", fd_path.display(), err);
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
|
||||
info!("waiting tun's file descriptor from {}", fd_path.display());
|
||||
|
||||
loop {
|
||||
let (mut stream, peer_addr) = listener.accept().await?;
|
||||
trace!("accepted {:?} for receiving tun file descriptor", peer_addr);
|
||||
|
||||
let mut buffer = [0u8; 1024];
|
||||
let mut fd_buffer = [0];
|
||||
|
||||
match stream.recv_with_fd(&mut buffer, &mut fd_buffer).await {
|
||||
Ok((n, fd_size)) => {
|
||||
if fd_size == 0 {
|
||||
log::error!(
|
||||
"client {:?} didn't send file descriptors with buffer.size {} bytes",
|
||||
peer_addr,
|
||||
n
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
info!("got file descriptor {} for tun from {:?}", fd_buffer[0], peer_addr);
|
||||
|
||||
builder.file_descriptor(fd_buffer[0]);
|
||||
break;
|
||||
}
|
||||
Err(err) => {
|
||||
log::error!(
|
||||
"failed to receive file descriptors from {:?}, error: {}",
|
||||
peer_addr,
|
||||
err
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let server = builder.build().await?;
|
||||
local_server.tun_servers.push(server);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(local_server)
|
||||
}
|
||||
|
||||
/// Run local server
|
||||
#[deprecated]
|
||||
pub async fn run(self) -> io::Result<()> {
|
||||
self.wait_until_exit().await
|
||||
}
|
||||
let mut vfut = Vec::new();
|
||||
|
||||
/// Wait until any of the servers were exited
|
||||
pub async fn wait_until_exit(self) -> io::Result<()> {
|
||||
let (res, ..) = future::select_all(self.vfut).await;
|
||||
for svr in self.socks_servers {
|
||||
vfut.push(ServerHandle(tokio::spawn(svr.run())));
|
||||
}
|
||||
|
||||
#[cfg(feature = "local-tunnel")]
|
||||
for svr in self.tunnel_servers {
|
||||
vfut.push(ServerHandle(tokio::spawn(svr.run())));
|
||||
}
|
||||
|
||||
#[cfg(feature = "local-http")]
|
||||
for svr in self.http_servers {
|
||||
vfut.push(ServerHandle(tokio::spawn(svr.run())));
|
||||
}
|
||||
|
||||
#[cfg(feature = "local-tun")]
|
||||
for svr in self.tun_servers {
|
||||
vfut.push(ServerHandle(tokio::spawn(svr.run())));
|
||||
}
|
||||
|
||||
#[cfg(feature = "local-dns")]
|
||||
for svr in self.dns_servers {
|
||||
vfut.push(ServerHandle(tokio::spawn(svr.run())));
|
||||
}
|
||||
|
||||
#[cfg(feature = "local-redir")]
|
||||
for svr in self.redir_servers {
|
||||
vfut.push(ServerHandle(tokio::spawn(svr.run())));
|
||||
}
|
||||
|
||||
#[cfg(feature = "local-flow-stat")]
|
||||
if let Some(stat_addr) = self.local_stat_addr {
|
||||
// For Android's flow statistic
|
||||
|
||||
let report_fut = flow_report_task(stat_addr, self.flow_stat);
|
||||
vfut.push(ServerHandle(tokio::spawn(report_fut)));
|
||||
}
|
||||
|
||||
let (res, ..) = future::select_all(vfut).await;
|
||||
res
|
||||
}
|
||||
|
||||
@@ -99,342 +482,41 @@ impl Server {
|
||||
pub fn server_balancer(&self) -> &PingBalancer {
|
||||
&self.balancer
|
||||
}
|
||||
}
|
||||
|
||||
/// Starts a shadowsocks local server
|
||||
pub async fn create(config: Config) -> io::Result<Server> {
|
||||
assert!(config.config_type == ConfigType::Local && !config.local.is_empty());
|
||||
|
||||
trace!("{:?}", config);
|
||||
|
||||
// Warning for Stream Ciphers
|
||||
#[cfg(feature = "stream-cipher")]
|
||||
for inst in config.server.iter() {
|
||||
let server = &inst.config;
|
||||
|
||||
if server.method().is_stream() {
|
||||
log::warn!("stream cipher {} for server {} have inherent weaknesses (see discussion in https://github.com/shadowsocks/shadowsocks-org/issues/36). \
|
||||
DO NOT USE. It will be removed in the future.", server.method(), server.addr());
|
||||
}
|
||||
/// Get SOCKS server instances
|
||||
pub fn socks_servers(&self) -> &[Socks] {
|
||||
&self.socks_servers
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "android")))]
|
||||
if let Some(nofile) = config.nofile {
|
||||
use crate::sys::set_nofile;
|
||||
if let Err(err) = set_nofile(nofile) {
|
||||
log::warn!("set_nofile {} failed, error: {}", nofile, err);
|
||||
}
|
||||
/// Get Tunnel server instances
|
||||
#[cfg(feature = "local-tunnel")]
|
||||
pub fn tunnel_servers(&self) -> &[Tunnel] {
|
||||
&self.tunnel_servers
|
||||
}
|
||||
|
||||
// Global ServiceContext template
|
||||
// Each Local instance will hold a copy of its fields
|
||||
let mut context = ServiceContext::new();
|
||||
|
||||
let mut connect_opts = ConnectOpts {
|
||||
#[cfg(any(target_os = "linux", target_os = "android"))]
|
||||
fwmark: config.outbound_fwmark,
|
||||
|
||||
#[cfg(target_os = "android")]
|
||||
vpn_protect_path: config.outbound_vpn_protect_path,
|
||||
|
||||
bind_interface: config.outbound_bind_interface,
|
||||
bind_local_addr: config.outbound_bind_addr,
|
||||
|
||||
..Default::default()
|
||||
};
|
||||
connect_opts.tcp.send_buffer_size = config.outbound_send_buffer_size;
|
||||
connect_opts.tcp.recv_buffer_size = config.outbound_recv_buffer_size;
|
||||
connect_opts.tcp.nodelay = config.no_delay;
|
||||
connect_opts.tcp.fastopen = config.fast_open;
|
||||
connect_opts.tcp.keepalive = config.keep_alive.or(Some(LOCAL_DEFAULT_KEEPALIVE_TIMEOUT));
|
||||
connect_opts.tcp.mptcp = config.mptcp;
|
||||
context.set_connect_opts(connect_opts);
|
||||
|
||||
let mut accept_opts = AcceptOpts {
|
||||
ipv6_only: config.ipv6_only,
|
||||
..Default::default()
|
||||
};
|
||||
accept_opts.tcp.send_buffer_size = config.inbound_send_buffer_size;
|
||||
accept_opts.tcp.recv_buffer_size = config.inbound_recv_buffer_size;
|
||||
accept_opts.tcp.nodelay = config.no_delay;
|
||||
accept_opts.tcp.fastopen = config.fast_open;
|
||||
accept_opts.tcp.keepalive = config.keep_alive.or(Some(LOCAL_DEFAULT_KEEPALIVE_TIMEOUT));
|
||||
accept_opts.tcp.mptcp = config.mptcp;
|
||||
context.set_accept_opts(accept_opts);
|
||||
|
||||
if let Some(resolver) = build_dns_resolver(config.dns, config.ipv6_first, context.connect_opts_ref()).await {
|
||||
context.set_dns_resolver(Arc::new(resolver));
|
||||
/// Get HTTP server instances
|
||||
#[cfg(feature = "local-http")]
|
||||
pub fn http_servers(&self) -> &[Http] {
|
||||
&self.http_servers
|
||||
}
|
||||
|
||||
if config.ipv6_first {
|
||||
context.set_ipv6_first(config.ipv6_first);
|
||||
/// Get Tun server instances
|
||||
#[cfg(feature = "local-tun")]
|
||||
pub fn tun_servers(&self) -> &[Tun] {
|
||||
&self.tun_servers
|
||||
}
|
||||
|
||||
if let Some(acl) = config.acl {
|
||||
context.set_acl(Arc::new(acl));
|
||||
/// Get DNS server instances
|
||||
#[cfg(feature = "local-dns")]
|
||||
pub fn dns_servers(&self) -> &[Dns] {
|
||||
&self.dns_servers
|
||||
}
|
||||
|
||||
context.set_security_config(&config.security);
|
||||
|
||||
assert!(!config.local.is_empty(), "no valid local server configuration");
|
||||
|
||||
let mut vfut = Vec::new();
|
||||
|
||||
// Create a service balancer for choosing between multiple servers
|
||||
let balancer = {
|
||||
let mut mode = Mode::TcpOnly;
|
||||
|
||||
for local in &config.local {
|
||||
mode = mode.merge(local.config.mode);
|
||||
}
|
||||
|
||||
// Load balancer will hold an individual ServiceContext
|
||||
let mut balancer_builder = PingBalancerBuilder::new(Arc::new(context.clone()), mode);
|
||||
|
||||
// max_server_rtt have to be set before add_server
|
||||
if let Some(rtt) = config.balancer.max_server_rtt {
|
||||
balancer_builder.max_server_rtt(rtt);
|
||||
}
|
||||
|
||||
if let Some(intv) = config.balancer.check_interval {
|
||||
balancer_builder.check_interval(intv);
|
||||
}
|
||||
|
||||
if let Some(intv) = config.balancer.check_best_interval {
|
||||
balancer_builder.check_best_interval(intv);
|
||||
}
|
||||
|
||||
for server in config.server {
|
||||
balancer_builder.add_server(server.config);
|
||||
}
|
||||
|
||||
balancer_builder.build().await?
|
||||
};
|
||||
|
||||
#[cfg(feature = "local-flow-stat")]
|
||||
if let Some(stat_addr) = config.local_stat_addr {
|
||||
// For Android's flow statistic
|
||||
|
||||
let report_fut = flow_report_task(stat_addr, context.flow_stat());
|
||||
vfut.push(ServerHandle(tokio::spawn(report_fut)));
|
||||
/// Get Redir server instances
|
||||
#[cfg(feature = "local-redir")]
|
||||
pub fn redir_servers(&self) -> &[Redir] {
|
||||
&self.redir_servers
|
||||
}
|
||||
|
||||
for local_instance in config.local {
|
||||
let local_config = local_instance.config;
|
||||
|
||||
// Clone from global ServiceContext instance
|
||||
// It will shares Shadowsocks' global context, and FlowStat, DNS reverse cache
|
||||
let mut context = context.clone();
|
||||
|
||||
// Private ACL
|
||||
if let Some(acl) = local_instance.acl {
|
||||
context.set_acl(Arc::new(acl))
|
||||
}
|
||||
|
||||
let context = Arc::new(context);
|
||||
let balancer = balancer.clone();
|
||||
|
||||
match local_config.protocol {
|
||||
ProtocolType::Socks => {
|
||||
use self::socks::SocksBuilder;
|
||||
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "socks requires local address")),
|
||||
};
|
||||
|
||||
let mut server_builder = SocksBuilder::with_context(context.clone(), client_addr, balancer);
|
||||
server_builder.set_mode(local_config.mode);
|
||||
server_builder.set_socks5_auth(local_config.socks5_auth);
|
||||
|
||||
if let Some(c) = config.udp_max_associations {
|
||||
server_builder.set_udp_capacity(c);
|
||||
}
|
||||
if let Some(d) = config.udp_timeout {
|
||||
server_builder.set_udp_expiry_duration(d);
|
||||
}
|
||||
if let Some(b) = local_config.udp_addr {
|
||||
server_builder.set_udp_bind_addr(b.clone());
|
||||
}
|
||||
|
||||
let server = server_builder.build().await?;
|
||||
vfut.push(ServerHandle(tokio::spawn(async move { server.run().await })));
|
||||
}
|
||||
#[cfg(feature = "local-tunnel")]
|
||||
ProtocolType::Tunnel => {
|
||||
use self::tunnel::TunnelBuilder;
|
||||
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "tunnel requires local address")),
|
||||
};
|
||||
|
||||
let forward_addr = local_config.forward_addr.expect("tunnel requires forward address");
|
||||
|
||||
let mut server_builder =
|
||||
TunnelBuilder::with_context(context.clone(), forward_addr.clone(), client_addr, balancer);
|
||||
|
||||
if let Some(c) = config.udp_max_associations {
|
||||
server_builder.set_udp_capacity(c);
|
||||
}
|
||||
if let Some(d) = config.udp_timeout {
|
||||
server_builder.set_udp_expiry_duration(d);
|
||||
}
|
||||
server_builder.set_mode(local_config.mode);
|
||||
if let Some(udp_addr) = local_config.udp_addr {
|
||||
server_builder.set_udp_bind_addr(udp_addr);
|
||||
}
|
||||
|
||||
let server = server_builder.build().await?;
|
||||
vfut.push(ServerHandle(tokio::spawn(async move { server.run().await })));
|
||||
}
|
||||
#[cfg(feature = "local-http")]
|
||||
ProtocolType::Http => {
|
||||
use self::http::HttpBuilder;
|
||||
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "http requires local address")),
|
||||
};
|
||||
|
||||
let builder = HttpBuilder::with_context(context.clone(), client_addr, balancer);
|
||||
let server = builder.build().await?;
|
||||
vfut.push(ServerHandle(tokio::spawn(async move { server.run().await })));
|
||||
}
|
||||
#[cfg(feature = "local-redir")]
|
||||
ProtocolType::Redir => {
|
||||
use self::redir::RedirBuilder;
|
||||
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "redir requires local address")),
|
||||
};
|
||||
|
||||
let mut server_builder = RedirBuilder::with_context(context.clone(), client_addr, balancer);
|
||||
if let Some(c) = config.udp_max_associations {
|
||||
server_builder.set_udp_capacity(c);
|
||||
}
|
||||
if let Some(d) = config.udp_timeout {
|
||||
server_builder.set_udp_expiry_duration(d);
|
||||
}
|
||||
server_builder.set_mode(local_config.mode);
|
||||
server_builder.set_tcp_redir(local_config.tcp_redir);
|
||||
server_builder.set_udp_redir(local_config.udp_redir);
|
||||
if let Some(udp_addr) = local_config.udp_addr {
|
||||
server_builder.set_udp_bind_addr(udp_addr);
|
||||
}
|
||||
|
||||
let server = server_builder.build().await?;
|
||||
vfut.push(ServerHandle(tokio::spawn(async move { server.run().await })));
|
||||
}
|
||||
#[cfg(feature = "local-dns")]
|
||||
ProtocolType::Dns => {
|
||||
use self::dns::DnsBuilder;
|
||||
|
||||
let client_addr = match local_config.addr {
|
||||
Some(a) => a,
|
||||
None => return Err(io::Error::new(ErrorKind::Other, "dns requires local address")),
|
||||
};
|
||||
|
||||
let mut server_builder = {
|
||||
let local_addr = local_config.local_dns_addr.expect("missing local_dns_addr");
|
||||
let remote_addr = local_config.remote_dns_addr.expect("missing remote_dns_addr");
|
||||
|
||||
DnsBuilder::with_context(
|
||||
context.clone(),
|
||||
client_addr,
|
||||
local_addr.clone(),
|
||||
remote_addr.clone(),
|
||||
balancer,
|
||||
)
|
||||
};
|
||||
server_builder.set_mode(local_config.mode);
|
||||
|
||||
let server = server_builder.build().await?;
|
||||
vfut.push(ServerHandle(tokio::spawn(async move { server.run().await })));
|
||||
}
|
||||
#[cfg(feature = "local-tun")]
|
||||
ProtocolType::Tun => {
|
||||
use log::info;
|
||||
use shadowsocks::net::UnixListener;
|
||||
|
||||
use self::tun::TunBuilder;
|
||||
|
||||
let mut builder = TunBuilder::new(context.clone(), balancer);
|
||||
if let Some(address) = local_config.tun_interface_address {
|
||||
builder.address(address);
|
||||
}
|
||||
if let Some(address) = local_config.tun_interface_destination {
|
||||
builder.destination(address);
|
||||
}
|
||||
if let Some(name) = local_config.tun_interface_name {
|
||||
builder.name(&name);
|
||||
}
|
||||
if let Some(c) = config.udp_max_associations {
|
||||
builder.udp_capacity(c);
|
||||
}
|
||||
if let Some(d) = config.udp_timeout {
|
||||
builder.udp_expiry_duration(d);
|
||||
}
|
||||
builder.mode(local_config.mode);
|
||||
#[cfg(unix)]
|
||||
if let Some(fd) = local_config.tun_device_fd {
|
||||
builder.file_descriptor(fd);
|
||||
} else if let Some(ref fd_path) = local_config.tun_device_fd_from_path {
|
||||
use std::fs;
|
||||
|
||||
let _ = fs::remove_file(fd_path);
|
||||
|
||||
let listener = match UnixListener::bind(fd_path) {
|
||||
Ok(l) => l,
|
||||
Err(err) => {
|
||||
log::error!("failed to bind uds path \"{}\", error: {}", fd_path.display(), err);
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
|
||||
info!("waiting tun's file descriptor from {}", fd_path.display());
|
||||
|
||||
loop {
|
||||
let (mut stream, peer_addr) = listener.accept().await?;
|
||||
trace!("accepted {:?} for receiving tun file descriptor", peer_addr);
|
||||
|
||||
let mut buffer = [0u8; 1024];
|
||||
let mut fd_buffer = [0];
|
||||
|
||||
match stream.recv_with_fd(&mut buffer, &mut fd_buffer).await {
|
||||
Ok((n, fd_size)) => {
|
||||
if fd_size == 0 {
|
||||
log::error!(
|
||||
"client {:?} didn't send file descriptors with buffer.size {} bytes",
|
||||
peer_addr,
|
||||
n
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
info!("got file descriptor {} for tun from {:?}", fd_buffer[0], peer_addr);
|
||||
|
||||
builder.file_descriptor(fd_buffer[0]);
|
||||
break;
|
||||
}
|
||||
Err(err) => {
|
||||
log::error!(
|
||||
"failed to receive file descriptors from {:?}, error: {}",
|
||||
peer_addr,
|
||||
err
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let server = builder.build().await?;
|
||||
vfut.push(ServerHandle(tokio::spawn(async move { server.run().await })));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(Server { vfut, balancer })
|
||||
}
|
||||
|
||||
#[cfg(feature = "local-flow-stat")]
|
||||
@@ -515,5 +597,5 @@ async fn flow_report_task(stat_addr: LocalFlowStatAddress, flow_stat: Arc<FlowSt
|
||||
|
||||
/// Create then run a Local Server
|
||||
pub async fn run(config: Config) -> io::Result<()> {
|
||||
create(config).await?.wait_until_exit().await
|
||||
Server::new(config).await?.run().await
|
||||
}
|
||||
|
||||
@@ -22,8 +22,7 @@ use shadowsocks_service::{
|
||||
ProtocolType,
|
||||
ServerInstanceConfig,
|
||||
},
|
||||
create_local,
|
||||
local::loadbalancing::PingBalancer,
|
||||
local::{loadbalancing::PingBalancer, Server},
|
||||
shadowsocks::{
|
||||
config::{Mode, ServerAddr, ServerConfig},
|
||||
crypto::{available_ciphers, CipherKind},
|
||||
@@ -869,14 +868,14 @@ pub fn main(matches: &ArgMatches) -> ExitCode {
|
||||
runtime.block_on(async move {
|
||||
let config_path = config.config_path.clone();
|
||||
|
||||
let instance = create_local(config).await.expect("create local");
|
||||
let instance = Server::new(config).await.expect("create local");
|
||||
|
||||
if let Some(config_path) = config_path {
|
||||
launch_reload_server_task(config_path, instance.server_balancer().clone());
|
||||
}
|
||||
|
||||
let abort_signal = monitor::create_signal_monitor();
|
||||
let server = instance.wait_until_exit();
|
||||
let server = instance.run();
|
||||
|
||||
tokio::pin!(abort_signal);
|
||||
tokio::pin!(server);
|
||||
|
||||
Reference in New Issue
Block a user