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 (and
98            // serves the port's passthrough listeners too, by SNI).
99            entry.0 = entry.0.min(kind);
100            entry.1.push(listener.clone());
101        }
102        *self.wanted.lock().unwrap_or_else(|e| e.into_inner()) = wanted.clone();
103        let mut ports = self.ports.lock().unwrap_or_else(|e| e.into_inner());
104        ports.retain(|key, _| wanted.contains_key(key));
105        for (key, (kind, listeners)) in wanted {
106            if let Some(running) = ports.get(&key) {
107                update(running, kind, listeners);
108                continue;
109            }
110            match serve(&self.bind_ip, key.0, kind, listeners) {
111                Ok(running) => {
112                    ports.insert(key, running);
113                }
114                Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => self.retry(key),
115                Err(e) => eprintln!("[gateway-api] can't serve port {}: {e}", key.0),
116            }
117        }
118    }
119
120    /// Bind `key` again shortly, while it's still wanted and not served.
121    fn retry(&self, key: (u16, bool)) {
122        let (ports, wanted, bind_ip) = (Arc::clone(&self.ports), Arc::clone(&self.wanted), self.bind_ip.clone());
123        tokio::spawn(async move {
124            for _ in 0..BIND_RETRIES {
125                tokio::time::sleep(BIND_RETRY_EVERY).await;
126                let Some((kind, listeners)) = wanted.lock().unwrap_or_else(|e| e.into_inner()).get(&key).cloned() else { return };
127                let mut ports = ports.lock().unwrap_or_else(|e| e.into_inner());
128                if ports.contains_key(&key) {
129                    return;
130                }
131                match serve(&bind_ip, key.0, kind, listeners) {
132                    Ok(running) => {
133                        ports.insert(key, running);
134                        return;
135                    }
136                    Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => continue,
137                    Err(e) => {
138                        eprintln!("[gateway-api] can't serve port {}: {e}", key.0);
139                        return;
140                    }
141                }
142            }
143            eprintln!("[gateway-api] can't serve port {}: still in use", key.0);
144        });
145    }
146}
147
148/// The TLS acceptor a kind of port needs.
149fn acceptor_for(kind: Kind, listeners: &[ListenerConfig]) -> Option<tokio_rustls::TlsAcceptor> {
150    match kind {
151        Kind::Http => tls_acceptor(listeners, true),
152        Kind::TlsTerminate => tls_acceptor(listeners, false),
153        _ => None,
154    }
155}
156
157/// A running port takes its new listeners, certificates and kind.
158fn update(running: &Port, kind: Kind, listeners: Vec<ListenerConfig>) {
159    let acceptor = acceptor_for(kind, &listeners);
160    *running.listeners.write().unwrap_or_else(|e| e.into_inner()) = Arc::new(listeners);
161    *running.tls.write().unwrap_or_else(|e| e.into_inner()) = acceptor;
162    *running.kind.write().unwrap_or_else(|e| e.into_inner()) = kind;
163}
164
165fn serve(bind_ip: &str, port: u16, kind: Kind, listeners: Vec<ListenerConfig>) -> std::io::Result<Port> {
166    let tls = Arc::new(RwLock::new(acceptor_for(kind, &listeners)));
167    let listeners: PortListeners = Arc::new(RwLock::new(Arc::new(listeners)));
168    let kind_now = Arc::new(RwLock::new(kind));
169    if kind == Kind::Udp {
170        let socket = std::net::UdpSocket::bind((bind_ip, port))?;
171        socket.set_nonblocking(true)?;
172        let socket = Arc::new(tokio::net::UdpSocket::from_std(socket)?);
173        let task = tokio::spawn(super::l4::serve_udp(socket, Arc::clone(&listeners)));
174        return Ok(Port { kind: kind_now, listeners, tls, task });
175    }
176    let socket = std::net::TcpListener::bind((bind_ip, port))?;
177    socket.set_nonblocking(true)?;
178    let socket = tokio::net::TcpListener::from_std(socket)?;
179    let router = PortRouter::new(Arc::clone(&listeners));
180    let (listeners_for_loop, tls_for_loop, kind_for_loop) = (Arc::clone(&listeners), Arc::clone(&tls), Arc::clone(&kind_now));
181    let counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));
182    let task = tokio::spawn(async move {
183        loop {
184            let Ok((stream, peer)) = socket.accept().await else {
185                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
186                continue;
187            };
188            let acceptor = tls_for_loop.read().unwrap_or_else(|e| e.into_inner()).clone();
189            let kind = *kind_for_loop.read().unwrap_or_else(|e| e.into_inner());
190            let (router, listeners, counter) = (router.clone(), Arc::clone(&listeners_for_loop), Arc::clone(&counter));
191            tokio::spawn(async move {
192                match (kind, acceptor) {
193                    (Kind::Http, Some(acceptor)) => crate::server::serve_tls_connection(stream, peer, acceptor, router).await,
194                    (Kind::Http, None) => crate::server::serve_plain_connection(stream, peer, router).await,
195                    (Kind::TlsTerminate, acceptor) => super::l4::serve_tls_mixed(stream, acceptor, listeners, counter).await,
196                    (Kind::TlsPassthrough, _) => super::l4::serve_tls_passthrough(stream, listeners, counter).await,
197                    (Kind::Tcp, _) => super::l4::serve_tcp(stream, listeners, counter).await,
198                    _ => {}
199                }
200            });
201        }
202    });
203    Ok(Port { kind: kind_now, listeners, tls, task })
204}
205
206/// Certificates by listener hostname, for SNI.
207#[derive(Debug)]
208struct ListenerCerts {
209    /// (hostname or empty for any, certificate).
210    certs: Vec<(String, Arc<CertifiedKey>)>,
211}
212
213impl ResolvesServerCert for ListenerCerts {
214    fn resolve(&self, hello: ClientHello) -> Option<Arc<CertifiedKey>> {
215        let name = hello.server_name().map(str::to_ascii_lowercase).unwrap_or_default();
216        self.certs
217            .iter()
218            .filter_map(|(pattern, cert)| if pattern.is_empty() { Some(((0, 0), cert)) } else { host_score(pattern, &name).map(|s| (s, cert)) })
219            .max_by_key(|(score, _)| *score)
220            .map(|(_, cert)| Arc::clone(cert))
221    }
222}
223
224/// A TLS acceptor for the port's listeners that terminate TLS (HTTPS, or
225/// TLS in `Terminate` mode); `None` when none do. `http` offers h2 and
226/// http/1.1 by ALPN. With a client CA (per port, so the same for all of
227/// them) client certificates are asked for: required and verified, or under
228/// `AllowInsecureFallback` asked for and let through either way.
229fn tls_acceptor(listeners: &[ListenerConfig], http: bool) -> Option<tokio_rustls::TlsAcceptor> {
230    let certs: Vec<(String, Arc<CertifiedKey>)> = listeners
231        .iter()
232        .filter(|l| matches!(l.protocol.as_str(), "HTTPS" | "TLS"))
233        .filter_map(|l| {
234            let (cert, key) = l.tls.as_ref()?;
235            match crate::tls::certified_key_from_pem(cert, key) {
236                Ok(certified) => Some((l.hostname.clone().unwrap_or_default(), Arc::new(certified))),
237                Err(e) => {
238                    eprintln!("[gateway-api] listener {}: {e}", l.name);
239                    None
240                }
241            }
242        })
243        .collect();
244    if certs.is_empty() {
245        return None;
246    }
247    let _ = crate::crypto_provider().install_default();
248    let builder = rustls::ServerConfig::builder();
249    let builder = match listeners.iter().find_map(|l| l.client_ca.as_ref()).map(client_verifier) {
250        Some(Ok(verifier)) => builder.with_client_cert_verifier(verifier),
251        Some(Err(e)) => {
252            eprintln!("[gateway-api] client CA: {e}");
253            return None; // fail closed: no connection without the asked-for validation
254        }
255        None => builder.with_no_client_auth(),
256    };
257    let mut config = builder.with_cert_resolver(Arc::new(ListenerCerts { certs }));
258    if http {
259        config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
260    }
261    Some(tokio_rustls::TlsAcceptor::from(Arc::new(config)))
262}
263
264fn client_verifier(ca: &ClientCa) -> Result<Arc<dyn rustls::server::danger::ClientCertVerifier>, String> {
265    use rustls::pki_types::pem::PemObject;
266    let mut roots = rustls::RootCertStore::empty();
267    for cert in rustls::pki_types::CertificateDer::pem_slice_iter(&ca.pem) {
268        roots.add(cert.map_err(|e| e.to_string())?).map_err(|e| e.to_string())?;
269    }
270    let verifier = rustls::server::WebPkiClientVerifier::builder(Arc::new(roots)).build().map_err(|e| e.to_string())?;
271    Ok(if ca.insecure_fallback { Arc::new(InsecureFallback(verifier)) } else { verifier })
272}
273
274/// `AllowInsecureFallback`: asks for a client certificate (naming the CAs)
275/// but lets the connection through without one or with one that doesn't
276/// verify; handshake signatures are still checked.
277#[derive(Debug)]
278struct InsecureFallback(Arc<dyn rustls::server::danger::ClientCertVerifier>);
279
280impl rustls::server::danger::ClientCertVerifier for InsecureFallback {
281    fn root_hint_subjects(&self) -> &[rustls::DistinguishedName] {
282        self.0.root_hint_subjects()
283    }
284
285    fn client_auth_mandatory(&self) -> bool {
286        false
287    }
288
289    fn verify_client_cert(
290        &self,
291        _end_entity: &rustls::pki_types::CertificateDer<'_>,
292        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
293        _now: rustls::pki_types::UnixTime,
294    ) -> Result<rustls::server::danger::ClientCertVerified, rustls::Error> {
295        Ok(rustls::server::danger::ClientCertVerified::assertion())
296    }
297
298    fn verify_tls12_signature(
299        &self,
300        message: &[u8],
301        cert: &rustls::pki_types::CertificateDer<'_>,
302        dss: &rustls::DigitallySignedStruct,
303    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
304        self.0.verify_tls12_signature(message, cert, dss)
305    }
306
307    fn verify_tls13_signature(
308        &self,
309        message: &[u8],
310        cert: &rustls::pki_types::CertificateDer<'_>,
311        dss: &rustls::DigitallySignedStruct,
312    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
313        self.0.verify_tls13_signature(message, cert, dss)
314    }
315
316    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
317        self.0.supported_verify_schemes()
318    }
319}
320
321#[cfg(test)]
322mod tests;
rws_gateway_api/gateway_api/dataplane.rs

rws_gateway_api/gateway_api/
dataplane.rs
