inn_network/
proxy.rs

1//-------------------------------------------------------------------
2// MIT License
3// Copyright (c) 2022 black-mongo
4// @author CameronYang
5// @doc
6//
7// @end
8// Created : 2022-05-25T00:49:22+08:00
9//-------------------------------------------------------------------
10
11use std::collections::HashMap;
12use std::convert::Infallible;
13use std::sync::Arc;
14
15use actix::{Addr, Recipient};
16use http::uri::PathAndQuery;
17use hyper::client::HttpConnector;
18use hyper::server::conn::{AddrStream, Http};
19use hyper::service::{make_service_fn, service_fn};
20use hyper::upgrade::Upgraded;
21use hyper::{Body, Client, Method, Request, Response, Server};
22use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
23use inn_common::genca::CertAuthority;
24use log::{debug, error, info};
25use tokio::net::TcpStream;
26use tokio_rustls::TlsAcceptor;
27
28use crate::server::ProxyServer;
29use crate::{ToProxyServer, WsHttpReq};
30#[derive(Clone)]
31pub struct Proxy {
32    ca: Arc<CertAuthority>,
33    client: Client<HttpsConnector<HttpConnector>>,
34    server: Recipient<ToProxyServer>,
35}
36// To try this example:
37// 1. cargo run --example http_proxy
38// 2. config http_proxy in command line
39//    $ export http_proxy=http://127.0.0.1:8100
40//    $ export https_proxy=http://127.0.0.1:8100
41// 3. send requests
42//    $ curl -i https://www.some_domain.com/
43impl Proxy {
44    pub async fn start_proxy(ip: &str, cacert: &str, cakey: &str, server: Addr<ProxyServer>) {
45        let addr = ip.parse().expect("invalid ip");
46        let https = HttpsConnectorBuilder::new()
47            .with_webpki_roots()
48            .https_or_http()
49            .enable_http1()
50            .enable_http2()
51            .build();
52        let client = Client::builder()
53            .http1_title_case_headers(true)
54            .http1_preserve_header_case(true)
55            .build(https);
56        let ca = Arc::new(CertAuthority::new(cacert.to_string(), cakey.to_string()));
57        let make_service = make_service_fn(move |_conn: &AddrStream| {
58            let client = client.clone();
59            let server = server.clone();
60            let ca = Arc::clone(&ca);
61            async move {
62                Ok::<_, Infallible>(service_fn(move |req| {
63                    Proxy {
64                        ca: Arc::clone(&ca),
65                        client: client.clone(),
66                        server: server.clone().recipient(),
67                    }
68                    .proxy(req)
69                }))
70            }
71        });
72
73        let server = Server::bind(&addr)
74            .http1_preserve_header_case(true)
75            .http1_title_case_headers(true)
76            .serve(make_service);
77
78        info!("Https proxy server, Listening on http://{}", addr);
79
80        if let Err(e) = server.await {
81            eprintln!("server error: {}", e);
82        }
83    }
84
85    async fn proxy(self, req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
86        if Method::CONNECT == req.method() {
87            // Received an HTTP request like:
88            // ```
89            // CONNECT www.domain.com:443 HTTP/1.1
90            // Host: www.domain.com:443
91            // Proxy-Connection: Keep-Alive
92            // ```
93            //
94            // When HTTP method is CONNECT we should return an empty body
95            // then we can eventually upgrade the connection and talk a new protocol.
96            //
97            // Note: only after client received an empty body with STATUS_OK can the
98            // connection be upgraded, so we can't return a response inside
99            // `on_upgrade` future.
100            if let Some(addr) = Proxy::host_addr(req.uri()) {
101                tokio::task::spawn(async move {
102                    match hyper::upgrade::on(req).await {
103                        Ok(upgraded) => {
104                            let host: Vec<&str> = addr.split(':').collect();
105                            if Proxy::mitm_match(host[0], host[1]) {
106                                // Man in the middle
107                                let server_config = self.ca.dynamic_gen_cert(host[0]).await;
108                                match TlsAcceptor::from(server_config).accept(upgraded).await {
109                                    Ok(stream) => {
110                                        if let Err(e) = self.serve_https(stream).await {
111                                            error!("addr = {} serve_https error = {}", addr, e);
112                                        }
113                                    }
114                                    Err(e) => {
115                                        error!("addr = {} TlsAcceptor error = {}", addr, e);
116                                    }
117                                }
118                            } else {
119                                debug!("addr = {}, tunnel", addr);
120                                let _ = Proxy::tunnel(upgraded, &addr).await;
121                            }
122                        }
123                        Err(e) => error!("addr = {}, upgrade error: {}", addr, e),
124                    }
125                });
126                Ok(Response::new(Body::empty()))
127            } else {
128                error!("CONNECT host is not socket addr: {:?}", req.uri());
129                let mut resp = Response::new(Body::from("CONNECT must be to a socket address"));
130                *resp.status_mut() = http::StatusCode::BAD_REQUEST;
131                Ok(resp)
132            }
133        } else {
134            self.request(req).await
135        }
136    }
137
138    async fn serve_https(
139        self,
140        stream: tokio_rustls::server::TlsStream<Upgraded>,
141    ) -> Result<(), hyper::Error> {
142        let service = service_fn(|mut req| {
143            if req.version() == http::Version::HTTP_10 || req.version() == http::Version::HTTP_11 {
144                let authority = req
145                    .headers()
146                    .get(http::header::HOST)
147                    .expect("Host is a required header")
148                    .to_str()
149                    .expect("Failed to convert host to str");
150
151                let uri = http::uri::Builder::new()
152                    .scheme(http::uri::Scheme::HTTPS)
153                    .authority(authority)
154                    .path_and_query(
155                        req.uri()
156                            .path_and_query()
157                            .unwrap_or(&PathAndQuery::from_static("/"))
158                            .to_owned(),
159                    )
160                    .build()
161                    .expect("Failed to build URI");
162
163                let (mut parts, body) = req.into_parts();
164                parts.uri = uri;
165                req = Request::from_parts(parts, body)
166            };
167            // self.client.request(req)
168            self.clone().request(req)
169        });
170
171        Http::new()
172            .serve_connection(stream, service)
173            .with_upgrades()
174            .await
175    }
176    fn host_addr(uri: &http::Uri) -> Option<String> {
177        uri.authority().map(|auth| auth.to_string())
178    }
179    ///
180    ///
181    async fn request(self, req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
182        let uri = req.uri().clone();
183        let headers = req.headers().clone();
184        let body = format!("{:?}", req.body());
185        let method = req.method().as_str().to_string();
186        let ver = format!("{:?}", req.version());
187        let host = Proxy::host_addr(req.uri()).unwrap();
188        let rs = self.client.request(req).await;
189        match &rs {
190            Ok(resp) => {
191                let mut h = HashMap::new();
192                for (k, v) in &headers {
193                    h.insert(k.to_string(), v.to_str().unwrap().to_string());
194                }
195                let mut resp_h = HashMap::new();
196                for (k, v) in resp.headers() {
197                    resp_h.insert(k.to_string(), v.to_str().unwrap().to_string());
198                }
199
200                self.server
201                    .do_send(ToProxyServer::HttpReq(Box::new(WsHttpReq {
202                        id: "0".to_string(),
203                        uri: uri.to_string(),
204                        headers: h,
205                        status: resp.status().as_u16(),
206                        error: "".to_owned(),
207                        method,
208                        req_body: body,
209                        server_ip: "".to_string(),
210                        protocol: ver,
211                        host,
212                        resp_headers: resp_h,
213                        resp_body: format!("{:?}", resp.body()),
214                        time: "".to_string(),
215                    })));
216            }
217            Err(_e) => {
218                // self.server.do_send(ToProxyServer::HttpReq {
219                //     uri,
220                //     headers,
221                //     status: StatusCode::NO_CONTENT,
222                //     error: format!("{}", e),
223                // });
224            }
225        }
226        rs
227    }
228    // Create a TCP connection to host:port, build a tunnel between the connection and
229    // the upgraded connection
230    async fn tunnel(mut upgraded: Upgraded, addr: &str) -> std::io::Result<()> {
231        // Connect to remote server
232        let mut server = TcpStream::connect(addr).await?;
233        let _ = tokio::io::copy_bidirectional(&mut upgraded, &mut server).await?;
234        Ok(())
235    }
236    fn mitm_match(host: &str, port: &str) -> bool {
237        matches!(
238            (host, port),
239            ("github.com", _) | ("www.github.com", _) | ("baidu.com", _) | ("www.baidu.com", _)
240        )
241    }
242}
inn_network/proxy.rs

inn_network/
proxy.rs
