shadowsocks/shadowsocks-rust
1
0
Fork 0
mirror of https://github.com/shadowsocks/shadowsocks-rust.git synced 2026-10-12 02:22:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

[#149] Setting timeout, udp_timeout, udp_max_associations from opt

Browse Source 
Add ManagerConfig for encapsulating add manager's options
Removed Config.timeout, servers should always use ServerConfig.timeout
This commit is contained in:
zonyitoo
2020-06-01 21:55:00 +08:00
parent 4efb5daa6d
commit 62832b9ad9
12 changed files with 142 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bin/local.rs
+22 -3
View File
@@ -4,6 +4,8 @@
//! or you could specify a configuration file. The format of configuration file is defined
//! in mod `config`.
use std::time::Duration;
use clap::{clap_app, Arg};
use futures::future::{self, Either};
use log::info;
@@ -60,8 +62,9 @@ fn main() {
(@arg LOCAL_ADDR: -b --("local-addr") +takes_value {validator::validate_server_addr} "Local address, listen only to this address if specified")
(@arg SERVER_ADDR: -s --("server-addr") +takes_value {validator::validate_server_addr} requires[PASSWORD ENCRYPT_METHOD] "Server address")
(@arg PASSWORD: -k --password +takes_value requires[SERVER_ADDR ENCRYPT_METHOD] "Password")
(@arg ENCRYPT_METHOD: -m --("encrypt-method") +takes_value possible_values(&available_ciphers) requires[SERVER_ADDR PASSWORD] +next_line_help "Encryption method")
(@arg PASSWORD: -k --password +takes_value requires[SERVER_ADDR] "Server's password")
(@arg ENCRYPT_METHOD: -m --("encrypt-method") +takes_value possible_values(&available_ciphers) requires[SERVER_ADDR] +next_line_help "Server's encryption method")
(@arg TIMEOUT: --timeout +takes_value {validator::validate_u64} requires[SERVER_ADDR] "Server's timeout seconds for TCP relay")
(@arg PLUGIN: --plugin +takes_value requires[SERVER_ADDR] "SIP003 (https://shadowsocks.org/en/spec/Plugin.html) plugin")
(@arg PLUGIN_OPT: --("plugin-opts") +takes_value requires[PLUGIN] "Set SIP003 plugin options")
@@ -79,6 +82,9 @@ fn main() {
(@arg NOFILE: -n --nofile +takes_value "Set RLIMIT_NOFILE with both soft and hard limit (only for *nix systems)")
(@arg ACL: --acl +takes_value "Path to ACL (Access Control List)")
(@arg LOG_WITHOUT_TIME: --("log-without-time") "Log without datetime prefix")
(@arg UDP_TIMEOUT: --("udp-timeout") +takes_value {validator::validate_u64} "Timeout seconds for UDP relay")
(@arg UDP_MAX_ASSOCIATIONS: --("udp-max-associations") +takes_value {validator::validate_u64} "Maximum associations to be kept simultaneously for UDP relay")
);
// FIXME: -6 is not a identifier, so we cannot build it with clap_app!
@@ -187,7 +193,12 @@ fn main() {
.expect("encryption method");
let svr_addr = svr_addr.parse::<ServerAddr>().expect("server-addr");
let mut sc = ServerConfig::new(svr_addr, password.to_owned(), method, None, None);
let timeout = matches
.value_of("TIMEOUT")
.map(|t| t.parse::<u64>().expect("timeout"))
.map(Duration::from_secs);
let mut sc = ServerConfig::new(svr_addr, password.to_owned(), method, timeout, None);
if let Some(p) = matches.value_of("PLUGIN") {
let plugin = PluginConfig {
@@ -317,6 +328,14 @@ fn main() {
}
}
if let Some(udp_timeout) = matches.value_of("UDP_TIMEOUT") {
config.udp_timeout = Some(Duration::from_secs(udp_timeout.parse::<u64>().expect("udp-timeout")));
}
if let Some(udp_max_assoc) = matches.value_of("UDP_MAX_ASSOCIATIONS") {
config.udp_max_associations = Some(udp_max_assoc.parse::<usize>().expect("udp-max-associations"));
}
// DONE READING options
if config.local_addr.is_none() {
src/bin/manager.rs
+22 -7
View File
@@ -7,7 +7,10 @@
//! *It should be notice that the extented configuration file is not suitable for the server
//! side.*
use std::net::{IpAddr, SocketAddr};
use std::{
net::{IpAddr, SocketAddr},
time::Duration,
};
use clap::{clap_app, Arg};
use futures::future::{self, Either};
@@ -21,6 +24,7 @@ use shadowsocks::{
Config,
ConfigType,
ManagerAddr,
ManagerConfig,
Mode,
ServerAddr,
};
@@ -50,6 +54,7 @@ fn main() {
(@arg MANAGER_ADDRESS: --("manager-address") +takes_value {validator::validate_manager_addr} "ShadowSocks Manager (ssmgr) address, could be ip:port, domain:port or /path/to/unix.sock")
(@arg ENCRYPT_METHOD: -m --("encrypt-method") +takes_value possible_values(&available_ciphers) +next_line_help "Default encryption method")
(@arg TIMEOUT: --timeout +takes_value {validator::validate_u64} "Default timeout seconds for TCP relay")
(@arg NOFILE: -n --nofile +takes_value "Set RLIMIT_NOFILE with both soft and hard limit (only for *nix systems)")
(@arg ACL: --acl +takes_value "Path to ACL (Access Control List)")
@@ -79,10 +84,6 @@ fn main() {
None => Config::new(ConfigType::Manager),
};
if let Some(method) = matches.value_of("ENCRYPT_METHOD") {
config.manager_method = Some(method.parse().expect("encryption method"));
}
if let Some(bind_addr) = matches.value_of("BIND_ADDR") {
let bind_addr = match bind_addr.parse::<IpAddr>() {
Ok(ip) => ServerAddr::from(SocketAddr::new(ip, 0)),
@@ -109,7 +110,21 @@ fn main() {
}
if let Some(m) = matches.value_of("MANAGER_ADDRESS") {
config.manager_addr = Some(m.parse::<ManagerAddr>().expect("manager bind address"));
if let Some(ref mut manager_config) = config.manager {
manager_config.addr = m.parse::<ManagerAddr>().expect("manager-address");
} else {
config.manager = Some(ManagerConfig::new(m.parse::<ManagerAddr>().expect("manager-address")));
}
}
if let Some(ref mut manager_config) = config.manager {
if let Some(m) = matches.value_of("ENCRYPT_METHOD") {
manager_config.method = Some(m.parse::<CipherType>().expect("encrypt-method"));
}
if let Some(t) = matches.value_of("TIMEOUT") {
manager_config.timeout = Some(Duration::from_secs(t.parse::<u64>().expect("timeout")));
}
}
if let Some(nofile) = matches.value_of("NOFILE") {
@@ -132,7 +147,7 @@ fn main() {
// DONE reading options
if config.manager_addr.is_none() {
if config.manager.is_none() {
eprintln!(
"missing `manager_address`, consider specifying it by --manager-address command line option, \
or \"manager_address\" and \"manager_port\" keys in configuration file"
src/bin/server.rs
+25 -5
View File
@@ -7,7 +7,10 @@
//! *It should be notice that the extented configuration file is not suitable for the server
//! side.*
use std::net::{IpAddr, SocketAddr};
use std::{
net::{IpAddr, SocketAddr},
time::Duration,
};
use clap::{clap_app, Arg};
use futures::future::{self, Either};
@@ -22,6 +25,7 @@ use shadowsocks::{
Config,
ConfigType,
ManagerAddr,
ManagerConfig,
Mode,
ServerAddr,
ServerConfig,
@@ -46,8 +50,9 @@ fn main() {
(@arg BIND_ADDR: -b --("bind-addr") +takes_value "Bind address, outbound socket will bind this address")
(@arg SERVER_ADDR: -s --("server-addr") +takes_value {validator::validate_server_addr} requires[PASSWORD ENCRYPT_METHOD] "Server address")
(@arg PASSWORD: -k --password +takes_value requires[SERVER_ADDR ENCRYPT_METHOD] "Password")
(@arg ENCRYPT_METHOD: -m --("encrypt-method") +takes_value possible_values(&available_ciphers) requires[SERVER_ADDR PASSWORD] "Encryption method")
(@arg PASSWORD: -k --password +takes_value requires[SERVER_ADDR] "Server's password")
(@arg ENCRYPT_METHOD: -m --("encrypt-method") +takes_value possible_values(&available_ciphers) requires[SERVER_ADDR] "Server's encryption method")
(@arg TIMEOUT: --timeout +takes_value {validator::validate_u64} requires[SERVER_ADDR] "Server's timeout seconds for TCP relay")
(@arg PLUGIN: --plugin +takes_value requires[SERVER_ADDR] "SIP003 (https://shadowsocks.org/en/spec/Plugin.html) plugin")
(@arg PLUGIN_OPT: --("plugin-opts") +takes_value requires[PLUGIN] "Set SIP003 plugin options")
@@ -58,6 +63,9 @@ fn main() {
(@arg NOFILE: -n --nofile +takes_value "Set RLIMIT_NOFILE with both soft and hard limit (only for *nix systems)")
(@arg ACL: --acl +takes_value "Path to ACL (Access Control List)")
(@arg LOG_WITHOUT_TIME: --("log-without-time") "Log without datetime prefix")
(@arg UDP_TIMEOUT: --("udp-timeout") +takes_value {validator::validate_u64} "Timeout seconds for UDP relay")
(@arg UDP_MAX_ASSOCIATIONS: --("udp-max-associations") +takes_value {validator::validate_u64} "Maximum associations to be kept simultaneously for UDP relay")
);
let matches = app
@@ -91,8 +99,12 @@ fn main() {
.parse::<CipherType>()
.expect("encryption method");
let svr_addr = svr_addr.parse::<ServerAddr>().expect("server-addr");
let timeout = matches
.value_of("TIMEOUT")
.map(|t| t.parse::<u64>().expect("timeout"))
.map(Duration::from_secs);
let mut sc = ServerConfig::new(svr_addr, password.to_owned(), method, None, None);
let mut sc = ServerConfig::new(svr_addr, password.to_owned(), method, timeout, None);
if let Some(p) = matches.value_of("PLUGIN") {
let plugin = PluginConfig {
@@ -132,7 +144,7 @@ fn main() {
}
if let Some(m) = matches.value_of("MANAGER_ADDRESS") {
config.manager_addr = Some(m.parse::<ManagerAddr>().expect("manager address"));
config.manager = Some(ManagerConfig::new(m.parse::<ManagerAddr>().expect("manager address")));
}
if let Some(nofile) = matches.value_of("NOFILE") {
@@ -153,6 +165,14 @@ fn main() {
config.ipv6_first = true;
}
if let Some(udp_timeout) = matches.value_of("UDP_TIMEOUT") {
config.udp_timeout = Some(Duration::from_secs(udp_timeout.parse::<u64>().expect("udp-timeout")));
}
if let Some(udp_max_assoc) = matches.value_of("UDP_MAX_ASSOCIATIONS") {
config.udp_max_associations = Some(udp_max_assoc.parse::<usize>().expect("udp-max-associations"));
}
// DONE READING options
if config.server.is_empty() {
src/bin/validator/mod.rs
+1
View File
@@ -29,6 +29,7 @@ validate_type!(
ManagerAddr,
"should be either ip:port, domain:port or /path/to/unix.sock"
);
validate_type!(validate_u64, u64, "should be unsigned integer");
pub fn validate_server_url(v: String) -> Result<(), String> {
match ServerConfig::from_url(&v) {
src/config.rs
+49 -20
View File
@@ -911,6 +911,28 @@ impl FromStr for RedirType {
}
}
/// Configuration for Manager
#[derive(Clone, Debug)]
pub struct ManagerConfig {
/// Address of `ss-manager`. Send servers' statistic data to the manager server
pub addr: ManagerAddr,
/// Manager's default method
pub method: Option<CipherType>,
/// Timeout for TCP connections, setting to manager's created servers
pub timeout: Option<Duration>,
}
impl ManagerConfig {
/// Create a ManagerConfig with default options
pub fn new(addr: ManagerAddr) -> ManagerConfig {
ManagerConfig {
addr,
method: None,
timeout: None,
}
}
}
/// Configuration
#[derive(Clone, Debug)]
pub struct Config {
@@ -934,10 +956,8 @@ pub struct Config {
pub mode: Mode,
/// Set `TCP_NODELAY` socket option
pub no_delay: bool,
/// Address of `ss-manager`. Send servers' statistic data to the manager server
pub manager_addr: Option<ManagerAddr>,
/// Manager's default method
pub manager_method: Option<CipherType>,
/// Manager's configuration
pub manager: Option<ManagerConfig>,
/// Config is for Client or Server
pub config_type: ConfigType,
/// Timeout for UDP Associations, default is 5 minutes
@@ -946,8 +966,6 @@ pub struct Config {
pub udp_max_associations: Option<usize>,
/// `RLIMIT_NOFILE` option for *nix systems
pub nofile: Option<u64>,
/// Timeout for TCP connections, could be replaced by server*.timeout
pub timeout: Option<Duration>,
/// ACL configuration
pub acl: Option<AccessControl>,
/// Path to stat callback unix address, only for Android
@@ -1060,13 +1078,11 @@ impl Config {
dns: None,
mode: Mode::TcpOnly,
no_delay: false,
manager_addr: None,
manager_method: None,
manager: None,
config_type,
udp_timeout: None,
udp_max_associations: None,
nofile: None,
timeout: None,
acl: None,
tcp_redir: RedirType::tcp_default(),
udp_redir: RedirType::udp_default(),
@@ -1209,6 +1225,17 @@ impl Config {
}
}
// Set timeout globally
if let Some(timeout) = config.timeout {
let timeout = Duration::from_secs(timeout);
// Set as a default timeout
for svr in &mut nconfig.server {
if svr.timeout.is_none() {
svr.timeout = Some(timeout);
}
}
}
// Manager Address
if let Some(ma) = config.manager_address {
let manager = match config.manager_port {
@@ -1230,7 +1257,13 @@ impl Config {
}
};
nconfig.manager_addr = Some(manager);
let manager_config = ManagerConfig {
addr: manager,
method: None,
timeout: None,
};
nconfig.manager = Some(manager_config);
}
// DNS
@@ -1259,15 +1292,12 @@ impl Config {
// UDP
nconfig.udp_timeout = config.udp_timeout.map(Duration::from_secs);
// Maximum associations to be kept simultaneously
nconfig.udp_max_associations = config.udp_max_associations;
// RLIMIT_NOFILE
nconfig.nofile = config.nofile;
// TCP timeout
// This is mostly used for manager for creating new servers
nconfig.timeout = config.timeout.map(Duration::from_secs);
// Uses IPv6 first
if let Some(f) = config.ipv6_first {
nconfig.ipv6_first = f;
@@ -1370,7 +1400,7 @@ impl Config {
}
if self.config_type.is_manager() {
if self.manager_addr.is_some() {
if self.manager.is_some() {
return Ok(());
}
@@ -1424,7 +1454,7 @@ impl fmt::Display for Config {
jconf.password = Some(svr.password().to_string());
jconf.plugin = svr.plugin().map(|p| p.plugin.to_string());
jconf.plugin_opts = svr.plugin().and_then(|p| p.plugin_opt.clone());
jconf.timeout = svr.timeout().or(self.timeout).map(|t| t.as_secs());
jconf.timeout = svr.timeout().map(|t| t.as_secs());
}
_ => {
let mut vsvr = Vec::new();
@@ -1448,19 +1478,18 @@ impl fmt::Display for Config {
}
jconf.servers = Some(vsvr);
jconf.timeout = self.timeout.map(|t| t.as_secs());
}
}
if let Some(ref ma) = self.manager_addr {
jconf.manager_address = Some(match *ma {
if let Some(ref m) = self.manager {
jconf.manager_address = Some(match m.addr {
ManagerAddr::SocketAddr(ref saddr) => saddr.ip().to_string(),
ManagerAddr::DomainName(ref dname, ..) => dname.clone(),
#[cfg(unix)]
ManagerAddr::UnixSocketAddr(ref path) => path.display().to_string(),
});
jconf.manager_port = match *ma {
jconf.manager_port = match m.addr {
ManagerAddr::SocketAddr(ref saddr) => Some(saddr.port()),
ManagerAddr::DomainName(.., port) => Some(port),
#[cfg(unix)]
src/context.rs
+1 -1
View File
@@ -132,7 +132,7 @@ impl ServerState {
/// Create a global shared server state
pub async fn new_shared(config: &Config) -> SharedServerState {
let state = ServerState {
dns_resolver: match create_resolver(config.get_dns_config(), config.timeout, config.ipv6_first).await {
dns_resolver: match create_resolver(config.get_dns_config(), config.ipv6_first).await {
Ok(resolver) => Some(resolver),
Err(..) => None,
},
src/lib.rs
+1 -1
View File
@@ -82,7 +82,7 @@ use std::io;
pub const VERSION: &str = env!("CARGO_PKG_VERSION");
pub use self::{
config::{ClientConfig, Config, ConfigType, ManagerAddr, Mode, ServerAddr, ServerConfig},
config::{ClientConfig, Config, ConfigType, ManagerAddr, ManagerConfig, Mode, ServerAddr, ServerConfig},
relay::{
local::run as run_local,
manager::run as run_manager,
src/relay/dns_resolver/trust_dns_resolver.rs
+1 -9
View File
@@ -3,7 +3,6 @@
use std::{
io::{self, Error, ErrorKind},
net::SocketAddr,
time::Duration,
};
use log::{error, trace};
@@ -16,15 +15,8 @@ use super::tokio_dns_resolver::resolve as tokio_resolve;
use crate::context::Context;
/// Create a `trust-dns` asynchronous DNS resolver
pub async fn create_resolver(
dns: Option<ResolverConfig>,
timeout: Option<Duration>,
ipv6_first: bool,
) -> io::Result<TokioAsyncResolver> {
pub async fn create_resolver(dns: Option<ResolverConfig>, ipv6_first: bool) -> io::Result<TokioAsyncResolver> {
let mut resolver_opts = ResolverOpts::default();
if let Some(d) = timeout {
resolver_opts.timeout = d;
}
if ipv6_first {
resolver_opts.ip_strategy = LookupIpStrategy::Ipv6thenIpv4;
src/relay/manager.rs
+11 -10
View File
@@ -382,12 +382,12 @@ impl ManagerService {
async fn handle_add(&mut self, p: protocol::ServerConfig) -> io::Result<Option<Vec<u8>>> {
trace!("ACTION \"add\" {:?}", p);
let manager_config = self.context.config().manager.as_ref().expect("manager config");
let server_port = p.server_port;
let method = match p.method {
None => self.context
.config()
.manager_method
None => manager_config.method
// Default method as shadowsocks-libev's ss-server
// Just for compatiblity, some shadowsocks manager relies on this default method
.unwrap_or(CipherType::ChaCha20IetfPoly1305),
@@ -405,7 +405,7 @@ impl ManagerService {
ServerAddr::from(bind_addr),
p.password,
method,
None,
manager_config.timeout,
match p.plugin {
Some(pp) => Some(PluginConfig {
plugin: pp,
@@ -437,9 +437,9 @@ impl ManagerService {
config.no_delay = self.context.config().no_delay;
}
// timeouts
// UDP configurations
config.udp_timeout = self.context.config().udp_timeout;
config.timeout = self.context.config().timeout;
config.udp_max_associations = self.context.config().udp_max_associations;
// Mode
config.mode = self.context.config().mode;
@@ -557,15 +557,16 @@ pub async fn run(config: Config) -> io::Result<()> {
// Create a context containing a DNS resolver and server running state flag.
let state = ServerState::new_shared(&config).await;
let context = Context::new_with_state_shared(config, state.clone());
let bind_addr = match context.config().manager_addr {
Some(ref a) => a,
let manager_config = match context.config().manager {
Some(ref mc) => mc,
None => {
let err = Error::new(ErrorKind::Other, "missing `manager_addr` in configuration");
let err = Error::new(ErrorKind::Other, "missing `manager` in configuration");
return Err(err);
}
};
let bind_addr = &manager_config.addr;
let mut service = ManagerService::bind(bind_addr, context.clone()).await?;
// Creates known servers in configuration
src/relay/server.rs
+3 -2
View File
@@ -93,7 +93,7 @@ pub(crate) async fn run_with(
// If specified manager-address, reports transmission statistic to it
//
// Dont do that if server is created by manager
if context.config().manager_addr.is_some() {
if context.config().manager.is_some() {
let report_fut = manager_report_task(context.clone(), flow_stat);
vf.push(report_fut.boxed());
}
@@ -108,7 +108,8 @@ pub(crate) async fn run_with(
}
async fn manager_report_task(context: SharedContext, flow_stat: SharedMultiServerFlowStatistic) -> io::Result<()> {
let manager_addr = context.config().manager_addr.as_ref().unwrap();
let manager_config = context.config().manager.as_ref().unwrap();
let manager_addr = &manager_config.addr;
let mut socket = ManagerDatagram::bind_for(manager_addr).await?;
while context.server_running() {
src/relay/tcprelay/proxy_stream.rs
+5 -6
View File
@@ -285,14 +285,13 @@ impl ProxyStream {
pub async fn connect_direct(context: SharedContext, addr: &Address) -> io::Result<ProxyStream> {
debug!("connect to {} directly (bypassed)", addr);
// NOTE: Direct connection's timeout is controlled by the global key
let timeout = context.config().timeout;
// FIXME: No timeout for direct connections
let stream = match *addr {
Address::SocketAddress(ref saddr) => try_timeout(tcp_stream_connect(&saddr, &context), timeout).await?,
Address::SocketAddress(ref saddr) => tcp_stream_connect(&saddr, &context).await?,
Address::DomainNameAddress(ref domain, port) => {
lookup_then!(context, domain, port, |saddr| {
try_timeout(tcp_stream_connect(&saddr, &context), timeout).await
tcp_stream_connect(&saddr, &context).await
})?
.1
}
@@ -300,7 +299,7 @@ impl ProxyStream {
Ok(ProxyStream {
context,
connection: ProxyConnection::Direct(Connection::new(stream, timeout)),
connection: ProxyConnection::Direct(Connection::new(stream, None)),
})
}
@@ -457,7 +456,7 @@ async fn connect_proxy_server_internal(
/// Connect to proxy server with `ServerConfig`
async fn connect_proxy_server(context: &Context, svr_cfg: &ServerConfig) -> io::Result<STcpStream> {
let timeout = svr_cfg.timeout().or(context.config().timeout);
let timeout = svr_cfg.timeout();
let svr_addr = match context.config().config_type {
ConfigType::Server => svr_cfg.addr(),
src/relay/tcprelay/server.rs
+1 -1
View File
@@ -32,7 +32,7 @@ async fn handle_client(
socket: TcpStream,
peer_addr: SocketAddr,
) -> io::Result<()> {
let timeout = svr_cfg.timeout().or(context.config().timeout);
let timeout = svr_cfg.timeout();
if let Err(err) = socket.set_keepalive(timeout) {
error!("failed to set keep alive: {:?}", err);