socks_hub_core/
httpproxy.rs

1use crate::{TokioIo, std_io_error_other};
2use bytes::Bytes;
3use http_body_util::{BodyExt, combinators::BoxBody};
4use hyper::{
5    Method, Request, Response,
6    header::{CONNECTION, HeaderMap, HeaderValue, PROXY_AUTHENTICATE, PROXY_AUTHORIZATION},
7    service::service_fn,
8    upgrade::Upgraded,
9};
10use socks5_impl::protocol::{Address, UserKey};
11use std::{future::Future, pin::Pin, sync::Arc};
12
13const HTTP_DEFAULT_PORT: u16 = 80;
14
15pub trait TokioStream: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Sync + Unpin {}
16impl<T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Sync + Unpin> TokioStream for T {}
17
18pub type BoxedStream = Box<dyn TokioStream>;
19pub type BoxedConnectFuture = Pin<Box<dyn Future<Output = std::io::Result<BoxedStream>> + Send + 'static>>;
20pub type HttpConnector = Arc<dyn Fn(Address) -> BoxedConnectFuture + Send + Sync>;
21
22#[cfg(feature = "acl")]
23use crate::acl::TargetDecision;
24
25#[cfg(feature = "acl")]
26static ACL_CENTER: std::sync::OnceLock<Option<crate::acl::AccessControl>> = std::sync::OnceLock::new();
27
28pub async fn run_http_service(stream: tokio::net::TcpStream, connector: HttpConnector, credentials: UserKey) -> std::io::Result<()> {
29    let io = TokioIo::new(stream);
30    hyper::server::conn::http1::Builder::new()
31        .preserve_header_case(true)
32        .title_case_headers(true)
33        .serve_connection(
34            io,
35            service_fn(|req: Request<hyper::body::Incoming>| {
36                let connector = connector.clone();
37                let credentials = credentials.clone();
38                async move { proxy(req, connector, credentials).await }
39            }),
40        )
41        .with_upgrades()
42        .await
43        .map_err(std_io_error_other)?;
44    Ok(())
45}
46
47async fn proxy(
48    mut req: Request<hyper::body::Incoming>,
49    connector: HttpConnector,
50    credentials: UserKey,
51) -> std::io::Result<Response<BoxBody<Bytes, hyper::Error>>> {
52    //
53    // https://github.com/hyperium/hyper/blob/90eb95f62a32981cb662b0f750027231d8a2586b/examples/http_proxy.rs#L51
54    //
55    log::trace!("req: {req:?}");
56
57    let auth_value = req.headers().get(PROXY_AUTHORIZATION);
58    // Some clients may omit proxy auth on the first CONNECT request and retry after a 407 challenge.
59    if !is_proxy_authorized(&credentials, auth_value) {
60        log::warn!("authorization fail");
61        let mut resp = Response::new(empty());
62        *resp.status_mut() = hyper::StatusCode::PROXY_AUTHENTICATION_REQUIRED;
63        resp.headers_mut()
64            .insert(PROXY_AUTHENTICATE, HeaderValue::from_static("Basic realm=\"socks-hub\""));
65        return Ok(resp);
66    }
67    if auth_value.is_some() {
68        let _ = req.headers_mut().remove(PROXY_AUTHORIZATION);
69    }
70
71    if Method::CONNECT == req.method() {
72        if let Some(host) = req.uri().host() {
73            let port = req.uri().port_u16().unwrap_or(HTTP_DEFAULT_PORT);
74            let up_addr = Address::from((host, port));
75
76            #[cfg(feature = "acl")]
77            {
78                if let Some(Some(acl)) = ACL_CENTER.get() {
79                    match acl.decide_target(&up_addr).await {
80                        TargetDecision::Proxy | TargetDecision::Bypass => {}
81                        TargetDecision::Block => {
82                            let mut resp = Response::new(full("blocked by ACL"));
83                            *resp.status_mut() = hyper::http::StatusCode::FORBIDDEN;
84                            return Ok(resp);
85                        }
86                    }
87                }
88            }
89
90            let connector = connector.clone();
91            tokio::task::spawn(async move {
92                match hyper::upgrade::on(req).await {
93                    Ok(upgraded) => {
94                        if let Err(e) = tunnel(upgraded, up_addr, connector).await {
95                            log::error!("server io error: {e}");
96                        };
97                    }
98                    Err(e) => log::error!("upgrade error: {e}"),
99                }
100            });
101            Ok(Response::new(empty()))
102        } else {
103            log::error!("CONNECT host is not socket addr: {:?}", req.uri());
104            let mut resp = Response::new(full("CONNECT must be to a socket address"));
105            *resp.status_mut() = hyper::http::StatusCode::BAD_REQUEST;
106            Ok(resp)
107        }
108    } else {
109        remove_hop_by_hop_headers(req.headers_mut());
110        let host = req.uri().host().unwrap_or_default();
111        let port = req.uri().port_u16().unwrap_or(HTTP_DEFAULT_PORT);
112        let up_addr = Address::from((host, port));
113
114        log::debug!("destination address {up_addr}");
115
116        #[cfg(feature = "acl")]
117        {
118            let mut must_proxied = true;
119            if let Some(Some(acl)) = ACL_CENTER.get() {
120                match acl.decide_target(&up_addr).await {
121                    TargetDecision::Proxy => must_proxied = true,
122                    TargetDecision::Bypass => must_proxied = false,
123                    TargetDecision::Block => {
124                        let mut resp = Response::new(full("blocked by ACL"));
125                        *resp.status_mut() = hyper::http::StatusCode::FORBIDDEN;
126                        return Ok(resp);
127                    }
128                }
129            }
130            if !must_proxied {
131                log::debug!("connect to destination address {up_addr:?} without proxy");
132                let stream = tokio::net::TcpStream::connect((host, port)).await?;
133                return proxy_internal(stream, req).await;
134            }
135        }
136
137        log::debug!("connect to upstream proxy for {up_addr:?}");
138        let stream = connector(up_addr).await?;
139        proxy_internal(stream, req).await
140    }
141}
142
143async fn proxy_internal<S>(stream: S, req: Request<hyper::body::Incoming>) -> Result<Response<BoxBody<Bytes, hyper::Error>>, std::io::Error>
144where
145    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Sync + 'static + Unpin,
146{
147    let io = TokioIo::new(stream);
148    let (mut sender, conn) = hyper::client::conn::http1::Builder::new()
149        .preserve_header_case(true)
150        .title_case_headers(true)
151        .handshake(io)
152        .await
153        .map_err(std_io_error_other)?;
154    tokio::task::spawn(async move {
155        if let Err(err) = conn.await {
156            log::error!("Connection failed: {err:?}");
157        }
158    });
159    let resp = sender.send_request(req).await.map_err(std_io_error_other)?;
160    Ok(resp.map(|b| b.boxed()))
161}
162
163fn empty() -> BoxBody<Bytes, hyper::Error> {
164    http_body_util::Empty::<Bytes>::new().map_err(|never| match never {}).boxed()
165}
166
167fn full<T: Into<Bytes>>(chunk: T) -> BoxBody<Bytes, hyper::Error> {
168    http_body_util::Full::new(chunk.into()).map_err(|never| match never {}).boxed()
169}
170
171fn remove_hop_by_hop_headers(headers: &mut HeaderMap<HeaderValue>) {
172    let connection_values = headers
173        .get(CONNECTION)
174        .and_then(|connection| connection.to_str().ok())
175        .map(|connection_value| connection_value.to_owned());
176
177    if let Some(connection_value) = connection_values {
178        for name in connection_value.split(',') {
179            let name = name.trim();
180            if !name.is_empty() {
181                headers.remove(name);
182            }
183        }
184    }
185
186    for header_name in &[
187        "connection",
188        "proxy-authorization",
189        "proxy-authenticate",
190        "proxy-connection",
191        "keep-alive",
192        "transfer-encoding",
193        "te",
194        "trailer",
195        "upgrade",
196    ] {
197        headers.remove(*header_name);
198    }
199}
200
201// Create a TCP connection to host:port, build a tunnel between the connection and
202// the upgraded connection
203async fn tunnel(upgraded: Upgraded, dst: Address, connector: HttpConnector) -> std::io::Result<()> {
204    #[cfg(feature = "acl")]
205    {
206        let mut must_proxied = true;
207        if let Some(Some(acl)) = ACL_CENTER.get() {
208            match acl.decide_target(&dst).await {
209                TargetDecision::Proxy => must_proxied = true,
210                TargetDecision::Bypass => must_proxied = false,
211                TargetDecision::Block => {
212                    return Err(std_io_error_other("blocked by ACL"));
213                }
214            }
215        }
216        if !must_proxied {
217            log::debug!("connect to destination address {dst:?} without proxy");
218            let mut upgraded = TokioIo::new(upgraded);
219            use std::net::ToSocketAddrs;
220            let addr = dst.to_socket_addrs()?.next().ok_or(std_io_error_other("no address found"))?;
221            let mut server = tokio::net::TcpStream::connect(addr).await?;
222            let (from_client, from_server) = tokio::io::copy_bidirectional(&mut upgraded, &mut server).await?;
223            log::debug!("client wrote {from_client} bytes and received {from_server} bytes");
224            return Ok(());
225        }
226    }
227
228    let mut upgraded = TokioIo::new(upgraded);
229    let mut server = connector(dst).await?;
230    let (from_client, from_server) = tokio::io::copy_bidirectional(&mut upgraded, &mut server).await?;
231    log::debug!("client wrote {from_client} bytes and received {from_server} bytes");
232    Ok(())
233}
234
235fn verify_basic_authorization(credentials: &UserKey, header_value: Option<&HeaderValue>) -> bool {
236    if header_value.is_none() && credentials.to_string().is_empty() {
237        return true;
238    }
239    header_value
240        .and_then(|v| v.to_str().ok())
241        .and_then(|s| {
242            let s = s.trim();
243            let mut parts = s.splitn(2, ' ');
244            let scheme = parts.next()?;
245            let encoded = parts.next()?.trim();
246            if !scheme.eq_ignore_ascii_case("Basic") {
247                return None;
248            }
249            base64easy::decode(encoded, base64easy::EngineKind::Standard).ok()
250        })
251        .is_some_and(|v| v == credentials.to_string().as_bytes().to_vec())
252}
253
254fn is_proxy_authorized(credentials: &UserKey, header_value: Option<&HeaderValue>) -> bool {
255    credentials.to_string().is_empty() || verify_basic_authorization(credentials, header_value)
256}
257
258#[cfg(test)]
259mod tests {
260    use super::*;
261    use hyper::header::{AUTHORIZATION, PROXY_AUTHORIZATION, UPGRADE};
262
263    #[test]
264    fn connect_requires_proxy_auth_when_credentials_are_configured() {
265        let credentials = UserKey::new("alice", "secret");
266        assert!(!is_proxy_authorized(&credentials, None));
267    }
268
269    #[test]
270    fn connect_allows_missing_proxy_auth_when_no_credentials_are_configured() {
271        let credentials = UserKey::default();
272        assert!(is_proxy_authorized(&credentials, None));
273    }
274
275    #[test]
276    fn proxy_authorization_header_is_used_for_proxy_auth() {
277        let credentials = UserKey::new("alice", "secret");
278        let value = HeaderValue::from_str("Basic YWxpY2U6c2VjcmV0").unwrap();
279        assert!(is_proxy_authorized(&credentials, Some(&value)));
280    }
281
282    #[test]
283    fn authorization_header_is_not_used_for_proxy_auth() {
284        let mut headers = HeaderMap::new();
285        headers.insert(AUTHORIZATION, HeaderValue::from_static("Basic YWxpY2U6c2VjcmV0"));
286        assert!(headers.get(PROXY_AUTHORIZATION).is_none());
287    }
288
289    #[test]
290    fn remove_hop_by_hop_headers_strips_connection_values() {
291        let mut headers = HeaderMap::new();
292        headers.insert(CONNECTION, HeaderValue::from_static("keep-alive, Upgrade"));
293        headers.insert("keep-alive", HeaderValue::from_static("timeout=5, max=1000"));
294        headers.insert(UPGRADE, HeaderValue::from_static("websocket"));
295        headers.insert(PROXY_AUTHORIZATION, HeaderValue::from_static("Basic test"));
296
297        remove_hop_by_hop_headers(&mut headers);
298
299        assert!(headers.get(CONNECTION).is_none());
300        assert!(headers.get("keep-alive").is_none());
301        assert!(headers.get(UPGRADE).is_none());
302        assert!(headers.get(PROXY_AUTHORIZATION).is_none());
303    }
304}
socks_hub_core/httpproxy.rs

socks_hub_core/
httpproxy.rs
