gway 0.1.5

proxy
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

use std::sync::Arc;

use hyper::{Request, body::Incoming, service::service_fn};
use hyper_util::rt::{TokioExecutor, TokioIo};
use parking_lot::RwLock;
use tokio::net::TcpListener;
use tokio_rustls::rustls::{self, ServerConfig};

use crate::{
  Result, Route,
  cert_loader::{CertLoad, CertLoader},
  proxy,
};

pub async fn srv<D: CertLoad>(
  shutdown_lock: Arc<tokio::sync::RwLock<()>>,
  conn_lock: Arc<RwLock<()>>,
  listener: TcpListener,
  route: Arc<Route>,
  cert_loader: Arc<CertLoader<D>>,
) -> Result<()> {
  // let listener = TcpListener::bind(addr).await?;

  loop {
    tokio::select! {
        res = listener.accept() => {
            let (stream, _remote_addr) = match res {
                Ok(val) => val,
                Err(e) => {
                    log::warn!("h2 accept error: {e}");
                    continue;
                }
            };
            let route = route.clone();
            let cert_loader = cert_loader.clone();
            let conn_lock = conn_lock.clone();

            tokio::spawn(
            #[allow(clippy::await_holding_lock)]
            async move {
                let _guard = conn_lock.read();
                let acceptor =
                    tokio_rustls::LazyConfigAcceptor::new(rustls::server::Acceptor::default(), stream).await;

                let stream = match acceptor {
                    Ok(start_handshake) => {
                        let client_hello = start_handshake.client_hello();
                        if let Some(server_name) = client_hello.server_name()
                            && !server_name.is_empty()
                        {
                            let cert = match cert_loader.get(server_name).await {
                                Ok(cert) => cert,
                                Err(err) => {
                                    log::warn!("h2 tls: cert get error: {err}");
                                    return;
                                }
                            };

                            let mut tls_config = match ServerConfig::builder()
                                .with_no_client_auth()
                                .with_single_cert(cert.rustls.fullchain.clone(), cert.rustls.key.clone_key())
                            {
                                Ok(config) => config,
                                Err(err) => {
                                    log::warn!("h2 tls: server config error: {err}");
                                    return;
                                }
                            };

                            tls_config.alpn_protocols = vec![b"h2".to_vec()];
                            match start_handshake.into_stream(Arc::new(tls_config)).await {
                                Ok(stream) => stream,
                                Err(err) => {
                                    log::warn!("h2 tls: handshake error: {err}");
                                    return;
                                }
                            }
                        } else {
                            log::warn!("h2 tls: server name is empty");
                            return;
                        }
                    }
                    Err(err) => {
                        log::warn!("h2 tls: lazy acceptor error: {err}");
                        return;
                    }
                };

                let io = TokioIo::new(stream);
                let service = service_fn(move |req: Request<Incoming>| {
                    let route = route.clone();
                    async move { Ok::<_, hyper::Error>(proxy(req, route).await) }
                });

                let conn_builder = hyper::server::conn::http2::Builder::new(TokioExecutor::new());
                if let Err(err) = conn_builder.serve_connection(io, service).await {
                    log::warn!("h2 error: {err}");
                }
            });
        },
        _ = shutdown_lock.read() => {
            break;
        }
    }
  }

  log::info!("h2 server shut down");
  Ok(())
}