rws_gateway_api/gateway_api/
dataplane.rs

1//! The data plane: one accept loop (or UDP socket) per container port of a
2//! served listener, started, updated and stopped as the controller's
3//! configuration changes. HTTP and HTTPS connections go through rws's own
4//! `serve_plain_connection` / `serve_tls_connection` (HTTP/1.1, h2c, TLS with
5//! HTTP/2, PROXY protocol, timeouts, metrics) with a [`PortRouter`] as the
6//! application; TLS, TCP and UDP listeners through [`super::l4`].
7
8use std::collections::BTreeMap;
9use std::sync::{Arc, Mutex, RwLock};
10
11use rustls::server::{ClientHello, ResolvesServerCert};
12use rustls::sign::CertifiedKey;
13
14use super::router::{host_score, PortListeners, PortRouter};
15use super::translate::{ClientCa, GatewayConfig, ListenerConfig};
16
17/// How a port's connections are served.
18#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
19enum Kind {
20    /// HTTP or HTTPS: rws's HTTP server with a [`PortRouter`].
21    Http,
22    /// TLS, terminated with the listeners' certificates (TLSRoute).
23    TlsTerminate,
24    /// TLS passed through by SNI (TLSRoute).
25    TlsPassthrough,
26    /// TCPRoute.
27    Tcp,
28    /// UDPRoute.
29    Udp,
30}
31
32impl Kind {
33    fn of(listener: &ListenerConfig) -> Option<Kind> {
34        Some(match listener.protocol.as_str() {
35            "HTTP" | "HTTPS" => Kind::Http,
36            "TLS" if listener.tls.is_some() => Kind::TlsTerminate,
37            "TLS" => Kind::TlsPassthrough,
38            "TCP" => Kind::Tcp,
39            "UDP" => Kind::Udp,
40            _ => return None,
41        })
42    }
43}
44
45/// A port being served.
46struct Port {
47    /// How connections are served; a change between TCP kinds (say HTTP to
48    /// TLS passthrough when the port goes to another Gateway) keeps the socket.
49    kind: Arc<RwLock<Kind>>,
50    listeners: PortListeners,
51    tls: Arc<RwLock<Option<tokio_rustls::TlsAcceptor>>>,
52    task: tokio::task::JoinHandle<()>,
53}
54
55impl Drop for Port {
56    fn drop(&mut self) {
57        self.task.abort(); // closes the listening socket (once the task is dropped)
58    }
59}
60
61/// What each port should serve, by (container port, UDP?).
62type Wanted = BTreeMap<(u16, bool), (Kind, Vec<ListenerConfig>)>;
63
64/// The ports being served, by container port and transport (a TCP and a
65/// UDP listener can share a number).
66#[derive(Default)]
67pub struct DataPlane {
68    ports: Arc<Mutex<BTreeMap<(u16, bool), Port>>>,
69    /// The last configuration applied, for the bind retries.
70    wanted: Arc<Mutex<Wanted>>,
71    /// The address to bind (`0.0.0.0` in a pod).
72    bind_ip: String,
73}
74
75/// How long a port whose bind failed (the socket of a port just given up,
76/// not closed yet) is retried, and how often.
77const BIND_RETRIES: usize = 100;
78const BIND_RETRY_EVERY: std::time::Duration = std::time::Duration::from_millis(100);
79
80impl DataPlane {
81    pub fn new(bind_ip: &str) -> DataPlane {
82        DataPlane { bind_ip: bind_ip.to_string(), ..Default::default() }
83    }
84
85    /// The container ports currently served.
86    pub fn ports(&self) -> Vec<u16> {
87        self.ports.lock().unwrap_or_else(|e| e.into_inner()).keys().map(|(port, _)| *port).collect()
88    }
89
90    /// Serve `config`: bind new ports, update the routes, certificates and
91    /// kind of running ones, stop the rest. Must run inside a tokio runtime.
92    pub fn apply(&self, config: &GatewayConfig) {
93        let mut wanted: Wanted = BTreeMap::new();
94        for listener in &config.listeners {
95            let Some(kind) = Kind::of(listener) else { continue };
96            let entry = wanted.entry((listener.container_port, kind == Kind::Udp)).or_insert((kind, Vec::new()));
97            // One port, one way of serving: a terminating TLS listener wins.
98            entry.0 = entry.0.min(kind);
99            entry.1.push(listener.clone());
100        }
101        *self.wanted.lock().unwrap_or_else(|e| e.into_inner()) = wanted.clone();
102        let mut ports = self.ports.lock().unwrap_or_else(|e| e.into_inner());
103        ports.retain(|key, _| wanted.contains_key(key));
104        for (key, (kind, listeners)) in wanted {
105            if let Some(running) = ports.get(&key) {
106                update(running, kind, listeners);
107                continue;
108            }
109            match serve(&self.bind_ip, key.0, kind, listeners) {
110                Ok(running) => {
111                    ports.insert(key, running);
112                }
113                Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => self.retry(key),
114                Err(e) => eprintln!("[gateway-api] can't serve port {}: {e}", key.0),
115            }
116        }
117    }
118
119    /// Bind `key` again shortly, while it's still wanted and not served.
120    fn retry(&self, key: (u16, bool)) {
121        let (ports, wanted, bind_ip) = (Arc::clone(&self.ports), Arc::clone(&self.wanted), self.bind_ip.clone());
122        tokio::spawn(async move {
123            for _ in 0..BIND_RETRIES {
124                tokio::time::sleep(BIND_RETRY_EVERY).await;
125                let Some((kind, listeners)) = wanted.lock().unwrap_or_else(|e| e.into_inner()).get(&key).cloned() else { return };
126                let mut ports = ports.lock().unwrap_or_else(|e| e.into_inner());
127                if ports.contains_key(&key) {
128                    return;
129                }
130                match serve(&bind_ip, key.0, kind, listeners) {
131                    Ok(running) => {
132                        ports.insert(key, running);
133                        return;
134                    }
135                    Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => continue,
136                    Err(e) => {
137                        eprintln!("[gateway-api] can't serve port {}: {e}", key.0);
138                        return;
139                    }
140                }
141            }
142            eprintln!("[gateway-api] can't serve port {}: still in use", key.0);
143        });
144    }
145}
146
147/// The TLS acceptor a kind of port needs.
148fn acceptor_for(kind: Kind, listeners: &[ListenerConfig]) -> Option<tokio_rustls::TlsAcceptor> {
149    match kind {
150        Kind::Http => tls_acceptor(listeners, true),
151        Kind::TlsTerminate => tls_acceptor(listeners, false),
152        _ => None,
153    }
154}
155
156/// A running port takes its new listeners, certificates and kind.
157fn update(running: &Port, kind: Kind, listeners: Vec<ListenerConfig>) {
158    let acceptor = acceptor_for(kind, &listeners);
159    *running.listeners.write().unwrap_or_else(|e| e.into_inner()) = Arc::new(listeners);
160    *running.tls.write().unwrap_or_else(|e| e.into_inner()) = acceptor;
161    *running.kind.write().unwrap_or_else(|e| e.into_inner()) = kind;
162}
163
164fn serve(bind_ip: &str, port: u16, kind: Kind, listeners: Vec<ListenerConfig>) -> std::io::Result<Port> {
165    let tls = Arc::new(RwLock::new(acceptor_for(kind, &listeners)));
166    let listeners: PortListeners = Arc::new(RwLock::new(Arc::new(listeners)));
167    let kind_now = Arc::new(RwLock::new(kind));
168    if kind == Kind::Udp {
169        let socket = std::net::UdpSocket::bind((bind_ip, port))?;
170        socket.set_nonblocking(true)?;
171        let socket = Arc::new(tokio::net::UdpSocket::from_std(socket)?);
172        let task = tokio::spawn(super::l4::serve_udp(socket, Arc::clone(&listeners)));
173        return Ok(Port { kind: kind_now, listeners, tls, task });
174    }
175    let socket = std::net::TcpListener::bind((bind_ip, port))?;
176    socket.set_nonblocking(true)?;
177    let socket = tokio::net::TcpListener::from_std(socket)?;
178    let router = PortRouter::new(Arc::clone(&listeners));
179    let (listeners_for_loop, tls_for_loop, kind_for_loop) = (Arc::clone(&listeners), Arc::clone(&tls), Arc::clone(&kind_now));
180    let counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));
181    let task = tokio::spawn(async move {
182        loop {
183            let Ok((stream, peer)) = socket.accept().await else {
184                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
185                continue;
186            };
187            let acceptor = tls_for_loop.read().unwrap_or_else(|e| e.into_inner()).clone();
188            let kind = *kind_for_loop.read().unwrap_or_else(|e| e.into_inner());
189            let (router, listeners, counter) = (router.clone(), Arc::clone(&listeners_for_loop), Arc::clone(&counter));
190            tokio::spawn(async move {
191                match (kind, acceptor) {
192                    (Kind::Http, Some(acceptor)) => crate::server::serve_tls_connection(stream, peer, acceptor, router).await,
193                    (Kind::Http, None) => crate::server::serve_plain_connection(stream, peer, router).await,
194                    (Kind::TlsTerminate, Some(acceptor)) => super::l4::serve_tls_terminate(stream, acceptor, listeners, counter).await,
195                    (Kind::TlsPassthrough, _) => super::l4::serve_tls_passthrough(stream, listeners, counter).await,
196                    (Kind::Tcp, _) => super::l4::serve_tcp(stream, listeners, counter).await,
197                    _ => {}
198                }
199            });
200        }
201    });
202    Ok(Port { kind: kind_now, listeners, tls, task })
203}
204
205/// Certificates by listener hostname, for SNI.
206#[derive(Debug)]
207struct ListenerCerts {
208    /// (hostname or empty for any, certificate).
209    certs: Vec<(String, Arc<CertifiedKey>)>,
210}
211
212impl ResolvesServerCert for ListenerCerts {
213    fn resolve(&self, hello: ClientHello) -> Option<Arc<CertifiedKey>> {
214        let name = hello.server_name().map(str::to_ascii_lowercase).unwrap_or_default();
215        self.certs
216            .iter()
217            .filter_map(|(pattern, cert)| if pattern.is_empty() { Some(((0, 0), cert)) } else { host_score(pattern, &name).map(|s| (s, cert)) })
218            .max_by_key(|(score, _)| *score)
219            .map(|(_, cert)| Arc::clone(cert))
220    }
221}
222
223/// A TLS acceptor for the port's listeners that terminate TLS (HTTPS, or
224/// TLS in `Terminate` mode); `None` when none do. `http` offers h2 and
225/// http/1.1 by ALPN. With a client CA (per port, so the same for all of
226/// them) client certificates are asked for: required and verified, or under
227/// `AllowInsecureFallback` asked for and let through either way.
228fn tls_acceptor(listeners: &[ListenerConfig], http: bool) -> Option<tokio_rustls::TlsAcceptor> {
229    let certs: Vec<(String, Arc<CertifiedKey>)> = listeners
230        .iter()
231        .filter(|l| matches!(l.protocol.as_str(), "HTTPS" | "TLS"))
232        .filter_map(|l| {
233            let (cert, key) = l.tls.as_ref()?;
234            match crate::tls::certified_key_from_pem(cert, key) {
235                Ok(certified) => Some((l.hostname.clone().unwrap_or_default(), Arc::new(certified))),
236                Err(e) => {
237                    eprintln!("[gateway-api] listener {}: {e}", l.name);
238                    None
239                }
240            }
241        })
242        .collect();
243    if certs.is_empty() {
244        return None;
245    }
246    let _ = crate::crypto_provider().install_default();
247    let builder = rustls::ServerConfig::builder();
248    let builder = match listeners.iter().find_map(|l| l.client_ca.as_ref()).map(client_verifier) {
249        Some(Ok(verifier)) => builder.with_client_cert_verifier(verifier),
250        Some(Err(e)) => {
251            eprintln!("[gateway-api] client CA: {e}");
252            return None; // fail closed: no connection without the asked-for validation
253        }
254        None => builder.with_no_client_auth(),
255    };
256    let mut config = builder.with_cert_resolver(Arc::new(ListenerCerts { certs }));
257    if http {
258        config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
259    }
260    Some(tokio_rustls::TlsAcceptor::from(Arc::new(config)))
261}
262
263fn client_verifier(ca: &ClientCa) -> Result<Arc<dyn rustls::server::danger::ClientCertVerifier>, String> {
264    use rustls::pki_types::pem::PemObject;
265    let mut roots = rustls::RootCertStore::empty();
266    for cert in rustls::pki_types::CertificateDer::pem_slice_iter(&ca.pem) {
267        roots.add(cert.map_err(|e| e.to_string())?).map_err(|e| e.to_string())?;
268    }
269    let verifier = rustls::server::WebPkiClientVerifier::builder(Arc::new(roots)).build().map_err(|e| e.to_string())?;
270    Ok(if ca.insecure_fallback { Arc::new(InsecureFallback(verifier)) } else { verifier })
271}
272
273/// `AllowInsecureFallback`: asks for a client certificate (naming the CAs)
274/// but lets the connection through without one or with one that doesn't
275/// verify; handshake signatures are still checked.
276#[derive(Debug)]
277struct InsecureFallback(Arc<dyn rustls::server::danger::ClientCertVerifier>);
278
279impl rustls::server::danger::ClientCertVerifier for InsecureFallback {
280    fn root_hint_subjects(&self) -> &[rustls::DistinguishedName] {
281        self.0.root_hint_subjects()
282    }
283
284    fn client_auth_mandatory(&self) -> bool {
285        false
286    }
287
288    fn verify_client_cert(
289        &self,
290        _end_entity: &rustls::pki_types::CertificateDer<'_>,
291        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
292        _now: rustls::pki_types::UnixTime,
293    ) -> Result<rustls::server::danger::ClientCertVerified, rustls::Error> {
294        Ok(rustls::server::danger::ClientCertVerified::assertion())
295    }
296
297    fn verify_tls12_signature(
298        &self,
299        message: &[u8],
300        cert: &rustls::pki_types::CertificateDer<'_>,
301        dss: &rustls::DigitallySignedStruct,
302    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
303        self.0.verify_tls12_signature(message, cert, dss)
304    }
305
306    fn verify_tls13_signature(
307        &self,
308        message: &[u8],
309        cert: &rustls::pki_types::CertificateDer<'_>,
310        dss: &rustls::DigitallySignedStruct,
311    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
312        self.0.verify_tls13_signature(message, cert, dss)
313    }
314
315    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
316        self.0.supported_verify_schemes()
317    }
318}
319
320#[cfg(test)]
321mod tests;
rws_gateway_api/gateway_api/dataplane.rs

rws_gateway_api/gateway_api/
dataplane.rs
