socks_hub_core/
lib.rs

1cfg_if::cfg_if! {
2    if #[cfg(feature = "acl")] {
3        mod acl;
4        pub use acl::AccessControl;
5    }
6}
7
8//
9// can't use cfg_if here, because it will cause cbindgen to can't generate ffi.h correctly.
10// see this issue: https://github.com/mozilla/cbindgen/issues/935
11//
12// cfg_if::cfg_if! {
13//     if #[cfg(feature = "sockshub")] {
14
15#[cfg(feature = "sockshub")]
16mod config;
17#[cfg(feature = "sockshub")]
18pub use config::{ArgVerbosity, Config};
19
20#[cfg(feature = "httpproxy")]
21mod tokiort;
22#[cfg(feature = "httpproxy")]
23pub use socks5_impl::protocol::{Address, Command, ProxyParameters, ProxyType, UserKey};
24#[cfg(feature = "sockshub")]
25use socks5_impl::protocol::{AsyncStreamOperation, Reply, Response as SocksResponse, UdpHeader};
26#[cfg(feature = "sockshub")]
27use tokio::{io::AsyncWriteExt, net::UdpSocket};
28#[cfg(feature = "httpproxy")]
29use tokiort::TokioIo;
30
31#[cfg(feature = "sockshub")]
32mod http2socks;
33#[cfg(feature = "sockshub")]
34mod socks2socks;
35
36#[cfg(feature = "httpproxy")]
37mod httpproxy;
38#[cfg(feature = "httpproxy")]
39pub use httpproxy::run_http_service;
40
41#[cfg(feature = "sockshub")]
42mod api;
43#[cfg(feature = "sockshub")]
44mod dump_logger;
45#[cfg(feature = "sockshub")]
46mod ffi;
47
48#[cfg(feature = "httpproxy")]
49pub type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
50#[cfg(feature = "sockshub")]
51pub type Result<T, E = BoxError> = std::result::Result<T, E>;
52
53#[cfg(feature = "sockshub")]
54pub async fn main_entry<F>(config: &Config, cancel_token: tokio_util::sync::CancellationToken, callback: Option<F>) -> Result<(), BoxError>
55where
56    F: FnOnce(std::net::SocketAddr) + Send + Sync + 'static,
57{
58    if config.remote_server.proxy_type != ProxyType::Socks5 {
59        return Err("remote server must be socks5".into());
60    }
61    if let Some(middle_server) = &config.middle_server {
62        if middle_server.proxy_type != ProxyType::Socks5 {
63            return Err("middle server must be socks5".into());
64        }
65    }
66    match config.listen_proxy_role.proxy_type {
67        ProxyType::Http => http2socks::main_entry(config, cancel_token, callback).await,
68        ProxyType::Socks5 => socks2socks::main_entry(config, cancel_token, callback).await,
69        ProxyType::None => mixed_main_entry(config, cancel_token, callback).await,
70        _ => Err("listen proxy must be http, socks5, or none (mixed)".into()),
71    }
72}
73
74#[cfg(feature = "sockshub")]
75pub async fn mixed_main_entry<F>(
76    config: &Config,
77    cancel_token: tokio_util::sync::CancellationToken,
78    callback: Option<F>,
79) -> Result<(), BoxError>
80where
81    F: FnOnce(std::net::SocketAddr) + Send + Sync + 'static,
82{
83    let listen_addr: std::net::SocketAddr = config.listen_proxy_role.addr.clone().try_into()?;
84
85    let listener = tokio::net::TcpListener::bind(listen_addr).await?;
86
87    if let Some(callback) = callback {
88        callback(listener.local_addr()?);
89    } else {
90        log::info!("Listening on {}", config.listen_proxy_role);
91    }
92
93    let config = std::sync::Arc::new(config.clone());
94
95    loop {
96        let config = config.clone();
97        tokio::select! {
98            _ = cancel_token.cancelled() => {
99                log::info!("quit signal received");
100                break;
101            }
102            result = listener.accept() => {
103                let (stream, incoming) = result?;
104                tokio::task::spawn(async move {
105                    if let Err(err) = service_connection(stream, config).await {
106                        log::error!("service on incoming {incoming} error: {err}");
107                    }
108                });
109            }
110        }
111    }
112    Ok(())
113}
114
115#[cfg(feature = "sockshub")]
116async fn service_connection(mut stream: tokio::net::TcpStream, config: std::sync::Arc<Config>) -> Result<(), BoxError> {
117    let mut peek_buf = [0u8; 10];
118    let n = stream.peek(&mut peek_buf).await?;
119    if n == 0 {
120        return Ok(());
121    }
122
123    match peek_buf[0] {
124        0x05 => {
125            log::trace!("socks5 client detected");
126            let credentials = config.get_listen_credentials();
127            let auth: socks5_impl::server::auth::AuthAdaptor = if credentials.to_string().is_empty() {
128                std::sync::Arc::new(socks5_impl::server::auth::NoAuth)
129            } else {
130                std::sync::Arc::new(socks5_impl::server::auth::UserKeyAuth::from(credentials))
131            };
132            let req = socks5_impl::server::socks5_service_handshake(&mut stream, auth).await?;
133            match req.command {
134                Command::Connect => crate::socks2socks::handle_socks5_connect(stream, req.address, config.remote_server.clone()).await,
135                Command::Bind => {
136                    let resp = SocksResponse::new(Reply::CommandNotSupported, Address::unspecified());
137                    resp.write_to_async_stream(&mut stream).await.map_err(std_io_error_other)?;
138                    stream.shutdown().await.map_err(std_io_error_other)?;
139                    Ok(())
140                }
141                Command::UdpAssociate => {
142                    let client_ip = stream.local_addr()?.ip();
143                    let udp_listener = UdpSocket::bind(std::net::SocketAddr::from((client_ip, 0))).await?;
144                    let listen_addr = udp_listener.local_addr()?;
145                    let resp = SocksResponse::new(Reply::Succeeded, Address::from(listen_addr));
146                    resp.write_to_async_stream(&mut stream).await.map_err(std_io_error_other)?;
147
148                    let buf_size = 1500 - UdpHeader::max_serialized_len();
149                    let listen_udp = std::sync::Arc::new(socks5_impl::server::AssociatedUdpSocket::from((udp_listener, buf_size)));
150                    let s5_udp_client =
151                        create_s5_udp_client(config.remote_server.clone(), CONNECT_TIMEOUT, config.middle_server.clone()).await?;
152
153                    crate::socks2socks::run_udp_associate_relay(listen_udp, s5_udp_client, &mut stream, listen_addr).await
154                }
155            }
156        }
157        0x04 => {
158            log::warn!("socks4 client detected, but only SOCKS5/HTTP mixed mode is supported");
159            let _ = stream.shutdown().await;
160            Ok(())
161        }
162        _ => {
163            let first_bytes = &peek_buf[..n];
164            let is_http = if let Ok(text) = std::str::from_utf8(first_bytes) {
165                let methods = [
166                    "CONNECT", "GET", "POST", "HEAD", "PUT", "OPTIONS", "DELETE", "TRACE", "PATCH", "LOCK", "UNLOCK", "PROPFIND", "MKCOL",
167                    "COPY", "MOVE",
168                ];
169                methods
170                    .iter()
171                    .any(|method| text[..method.len()].eq_ignore_ascii_case(method) && text.as_bytes().get(method.len()) == Some(&b' '))
172            } else {
173                false
174            };
175
176            if !is_http {
177                log::warn!("unknown client type detected, first byte: 0x{:02x}", first_bytes[0]);
178                let _ = stream.shutdown().await;
179                return Ok(());
180            }
181            log::trace!("http client detected by method peek");
182
183            let server = config.remote_server.clone();
184            let credentials = config.listen_proxy_role.credentials.clone().unwrap_or_default();
185            let middle_server = config.middle_server.clone();
186            let middle_server_for_conn = middle_server.clone();
187            let connector: crate::httpproxy::HttpConnector = std::sync::Arc::new(move |dst| {
188                let server = server.clone();
189                let middle_server = middle_server_for_conn.clone();
190                Box::pin(async move {
191                    let stream = crate::create_s5_connect(server, CONNECT_TIMEOUT, &dst, middle_server).await?;
192                    Ok(Box::new(stream) as crate::httpproxy::BoxedStream)
193                })
194            });
195            httpproxy::run_http_service(stream, connector, credentials).await?;
196            Ok(())
197        }
198    }
199}
200
201#[cfg(feature = "sockshub")]
202pub(crate) const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
203
204#[cfg(feature = "sockshub")]
205pub(crate) async fn create_s5_connect(
206    server: ProxyParameters,
207    dur: std::time::Duration,
208    dst: &socks5_impl::protocol::Address,
209    middle_server: Option<ProxyParameters>,
210) -> std::io::Result<tokio::io::BufStream<tokio::net::TcpStream>> {
211    let auth = server.credentials.clone();
212    let mut stream = connect_proxy_stream(server, dur, middle_server).await?;
213    socks5_impl::client::connect(&mut stream, dst, auth)
214        .await
215        .map_err(std_io_error_other)?;
216    Ok(stream)
217}
218
219#[cfg(feature = "sockshub")]
220async fn connect_proxy_stream(
221    server: ProxyParameters,
222    dur: std::time::Duration,
223    middle_server: Option<ProxyParameters>,
224) -> std::io::Result<tokio::io::BufStream<tokio::net::TcpStream>> {
225    let stream = if let Some(middle_server) = middle_server {
226        let middle_addr: std::net::SocketAddr = middle_server.addr.try_into()?;
227        let stream = tokio::time::timeout(dur, tokio::net::TcpStream::connect(middle_addr)).await??;
228        let mut stream = tokio::io::BufStream::new(stream);
229        let middle_auth = middle_server.credentials.clone();
230        socks5_impl::client::connect(&mut stream, server.addr, middle_auth)
231            .await
232            .map_err(std_io_error_other)?;
233        stream
234    } else {
235        let server_addr: std::net::SocketAddr = server.addr.try_into()?;
236        let stream = tokio::time::timeout(dur, tokio::net::TcpStream::connect(server_addr)).await??;
237        tokio::io::BufStream::new(stream)
238    };
239    Ok(stream)
240}
241
242#[cfg(feature = "sockshub")]
243pub(crate) async fn create_s5_udp_client(
244    server: ProxyParameters,
245    dur: std::time::Duration,
246    middle_server: Option<ProxyParameters>,
247) -> std::io::Result<socks5_impl::client::SocksUdpClient> {
248    let stream = connect_proxy_stream(server.clone(), dur, middle_server).await?;
249    let client_addr = if server.addr.is_ipv4() { "0.0.0.0:0" } else { "[::]:0" };
250    let client = tokio::net::UdpSocket::bind(client_addr).await?;
251    let auth = server.credentials.clone();
252    socks5_impl::client::SocksDatagram::udp_associate(stream, client, auth)
253        .await
254        .map_err(std_io_error_other)
255}
256
257#[cfg(feature = "httpproxy")]
258pub(crate) fn std_io_error_other<E: Into<BoxError>>(err: E) -> std::io::Error {
259    std::io::Error::other(err)
260}
261
262//     }
263// }
socks_hub_core/lib.rs

socks_hub_core/
lib.rs
