feat(compute_ctl): use TLS if configured (#10972)

Closes: https://github.com/neondatabase/cloud/issues/22998

If control-plane reports that TLS should be used, load the certificates
(and watch for updates), make sure postgres use them, and detects
updates.

Procedure:
1. Load certificates
2. Reconfigure postgres/pgbouncer
3. Loop on a timer until certificates have loaded
4. Go to 1

Notes:
1. We only run this procedure if requested on startup by control plane.
2. We needed to compile pgbouncer with openssl enabled
3. Postgres doesn't allow tls keys to be globally accessible - must be
read only to the postgres user. I couldn't convince the autoscaling team
to let me put this logic into the VM settings, so instead compute_ctl
will copy the keys to be read-only by postgres.
4. To mitigate a race condition, we also verify that the key matches the
cert.

proxy/src/binary/local_proxy.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::{Context, bail, ensure};
+use arc_swap::ArcSwapOption;
 use camino::{Utf8Path, Utf8PathBuf};
 use clap::Parser;
 use compute_api::spec::LocalProxySpec;
@@ -27,6 +28,7 @@ use crate::config::{
 };
 use crate::control_plane::locks::ApiLocks;
 use crate::control_plane::messages::{EndpointJwksResponse, JwksSettings};
+use crate::ext::TaskExt;
 use crate::http::health_server::AppMetrics;
 use crate::intern::RoleNameInt;
 use crate::metrics::{Metrics, ThreadPoolMetrics};
@@ -190,7 +192,11 @@ pub async fn run() -> anyhow::Result<()> {
     // 2. The config file is written but the signal hook is not yet received
     // 3. local_proxy completes startup but has no config loaded, despite there being a registerd config.
     refresh_config_notify.notify_one();
-    tokio::spawn(refresh_config_loop(args.config_path, refresh_config_notify));
+    tokio::spawn(refresh_config_loop(
+        config,
+        args.config_path,
+        refresh_config_notify,
+    ));
 
     maintenance_tasks.spawn(crate::http::health_server::task_main(
         metrics_listener,
@@ -269,7 +275,7 @@ fn build_config(args: &LocalProxyCliArgs) -> anyhow::Result<&'static ProxyConfig
     };
 
     Ok(Box::leak(Box::new(ProxyConfig {
-        tls_config: None,
+        tls_config: ArcSwapOption::from(None),
         metric_collection: None,
         http_config,
         authentication_config: AuthenticationConfig {
@@ -311,14 +317,16 @@ enum RefreshConfigError {
     Parse(#[from] serde_json::Error),
     #[error(transparent)]
     Validate(anyhow::Error),
+    #[error(transparent)]
+    Tls(anyhow::Error),
 }
 
-async fn refresh_config_loop(path: Utf8PathBuf, rx: Arc<Notify>) {
+async fn refresh_config_loop(config: &ProxyConfig, path: Utf8PathBuf, rx: Arc<Notify>) {
     let mut init = true;
     loop {
         rx.notified().await;
 
-        match refresh_config_inner(&path).await {
+        match refresh_config_inner(config, &path).await {
             Ok(()) => {}
             // don't log for file not found errors if this is the first time we are checking
             // for computes that don't use local_proxy, this is not an error.
@@ -327,6 +335,9 @@ async fn refresh_config_loop(path: Utf8PathBuf, rx: Arc<Notify>) {
             {
                 debug!(error=?e, ?path, "could not read config file");
             }
+            Err(RefreshConfigError::Tls(e)) => {
+                error!(error=?e, ?path, "could not read TLS certificates");
+            }
             Err(e) => {
                 error!(error=?e, ?path, "could not read config file");
             }
@@ -336,7 +347,10 @@ async fn refresh_config_loop(path: Utf8PathBuf, rx: Arc<Notify>) {
     }
 }
 
-async fn refresh_config_inner(path: &Utf8Path) -> Result<(), RefreshConfigError> {
+async fn refresh_config_inner(
+    config: &ProxyConfig,
+    path: &Utf8Path,
+) -> Result<(), RefreshConfigError> {
     let bytes = tokio::fs::read(&path).await?;
     let data: LocalProxySpec = serde_json::from_slice(&bytes)?;
 
@@ -406,5 +420,20 @@ async fn refresh_config_inner(path: &Utf8Path) -> Result<(), RefreshConfigError>
     info!("successfully loaded new config");
     JWKS_ROLE_MAP.store(Some(Arc::new(EndpointJwksResponse { jwks: jwks_set })));
 
+    if let Some(tls_config) = data.tls {
+        let tls_config = tokio::task::spawn_blocking(move || {
+            crate::tls::server_config::configure_tls(
+                &tls_config.key_path,
+                &tls_config.cert_path,
+                None,
+                false,
+            )
+        })
+        .await
+        .propagate_task_panic()
+        .map_err(RefreshConfigError::Tls)?;
+        config.tls_config.store(Some(Arc::new(tls_config)));
+    }
+
     Ok(())
 }

proxy/src/binary/proxy.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::bail;
+use arc_swap::ArcSwapOption;
 use futures::future::Either;
 use remote_storage::RemoteStorageConfig;
 use tokio::net::TcpListener;
@@ -563,6 +564,7 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
         (None, None) => None,
         _ => bail!("either both or neither tls-key and tls-cert must be specified"),
     };
+    let tls_config = ArcSwapOption::from(tls_config.map(Arc::new));
 
     let backup_metric_collection_config = config::MetricBackupCollectionConfig {
         remote_storage_config: args.metric_backup_collection_remote_storage.clone(),

proxy/src/config.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::{Context, Ok, bail, ensure};
+use arc_swap::ArcSwapOption;
 use clap::ValueEnum;
 use remote_storage::RemoteStorageConfig;
 
@@ -17,7 +18,7 @@ pub use crate::tls::server_config::{TlsConfig, configure_tls};
 use crate::types::Host;
 
 pub struct ProxyConfig {
-    pub tls_config: Option<TlsConfig>,
+    pub tls_config: ArcSwapOption<TlsConfig>,
     pub metric_collection: Option<MetricCollectionConfig>,
     pub http_config: HttpConfig,
     pub authentication_config: AuthenticationConfig,

proxy/src/console_redirect_proxy.rs

@@ -177,7 +177,8 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
     let proto = ctx.protocol();
     let request_gauge = metrics.connection_requests.guard(proto);
 
-    let tls = config.tls_config.as_ref();
+    let tls = config.tls_config.load();
+    let tls = tls.as_deref();
 
     let record_handshake_error = !ctx.has_private_peer_addr();
     let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Client);

proxy/src/proxy/handshake.rs

@@ -114,7 +114,7 @@ pub(crate) async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
 
                         let mut read_buf = read_buf.reader();
                         let mut res = Ok(());
-                        let accept = tokio_rustls::TlsAcceptor::from(tls.to_server_config())
+                        let accept = tokio_rustls::TlsAcceptor::from(tls.pg_config.clone())
                             .accept_with(raw, |session| {
                                 // push the early data to the tls session
                                 while !read_buf.get_ref().is_empty() {

proxy/src/proxy/mod.rs

@@ -278,7 +278,8 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
     let proto = ctx.protocol();
     let request_gauge = metrics.connection_requests.guard(proto);
 
-    let tls = config.tls_config.as_ref();
+    let tls = config.tls_config.load();
+    let tls = tls.as_deref();
 
     let record_handshake_error = !ctx.has_private_peer_addr();
     let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Client);

proxy/src/proxy/tests/mod.rs

@@ -96,16 +96,18 @@ fn generate_tls_config<'a>(
                 .with_safe_default_protocol_versions()
                 .context("ring should support the default protocol versions")?
                 .with_no_client_auth()
-                .with_single_cert(vec![cert.clone()], key.clone_key())?
-                .into();
+                .with_single_cert(vec![cert.clone()], key.clone_key())?;
 
         let mut cert_resolver = CertResolver::new();
         cert_resolver.add_cert(key, vec![cert], true)?;
 
         let common_names = cert_resolver.get_common_names();
 
+        let config = Arc::new(config);
+
         TlsConfig {
-            config,
+            http_config: config.clone(),
+            pg_config: config,
             common_names,
             cert_resolver: Arc::new(cert_resolver),
         }

proxy/src/serverless/mod.rs

@@ -19,6 +19,7 @@ use std::pin::{Pin, pin};
 use std::sync::Arc;
 
 use anyhow::Context;
+use arc_swap::ArcSwapOption;
 use async_trait::async_trait;
 use atomic_take::AtomicTake;
 use bytes::Bytes;
@@ -117,18 +118,7 @@ pub async fn task_main(
         auth_backend,
         endpoint_rate_limiter: Arc::clone(&endpoint_rate_limiter),
     });
-    let tls_acceptor: Arc<dyn MaybeTlsAcceptor> = match config.tls_config.as_ref() {
-        Some(config) => {
-            let mut tls_server_config = rustls::ServerConfig::clone(&config.to_server_config());
-            // prefer http2, but support http/1.1
-            tls_server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
-            Arc::new(tls_server_config)
-        }
-        None => {
-            warn!("TLS config is missing");
-            Arc::new(NoTls)
-        }
-    };
+    let tls_acceptor: Arc<dyn MaybeTlsAcceptor> = Arc::new(&config.tls_config);
 
     let connections = tokio_util::task::task_tracker::TaskTracker::new();
     connections.close(); // allows `connections.wait to complete`
@@ -216,22 +206,20 @@ pub(crate) type AsyncRW = Pin<Box<dyn AsyncReadWrite>>;
 
 #[async_trait]
 trait MaybeTlsAcceptor: Send + Sync + 'static {
-    async fn accept(self: Arc<Self>, conn: ChainRW<TcpStream>) -> std::io::Result<AsyncRW>;
+    async fn accept(&self, conn: ChainRW<TcpStream>) -> std::io::Result<AsyncRW>;
 }
 
 #[async_trait]
-impl MaybeTlsAcceptor for rustls::ServerConfig {
-    async fn accept(self: Arc<Self>, conn: ChainRW<TcpStream>) -> std::io::Result<AsyncRW> {
-        Ok(Box::pin(TlsAcceptor::from(self).accept(conn).await?))
-    }
-}
-
-struct NoTls;
-
-#[async_trait]
-impl MaybeTlsAcceptor for NoTls {
-    async fn accept(self: Arc<Self>, conn: ChainRW<TcpStream>) -> std::io::Result<AsyncRW> {
-        Ok(Box::pin(conn))
+impl MaybeTlsAcceptor for &'static ArcSwapOption<crate::config::TlsConfig> {
+    async fn accept(&self, conn: ChainRW<TcpStream>) -> std::io::Result<AsyncRW> {
+        match &*self.load() {
+            Some(config) => Ok(Box::pin(
+                TlsAcceptor::from(config.http_config.clone())
+                    .accept(conn)
+                    .await?,
+            )),
+            None => Ok(Box::pin(conn)),
+        }
     }
 }
 

proxy/src/serverless/sql_over_http.rs

@@ -614,7 +614,9 @@ async fn handle_inner(
         &config.authentication_config,
         ctx,
         request.headers(),
-        config.tls_config.as_ref(),
+        // todo: race condition?
+        // we're unlikely to change the common names.
+        config.tls_config.load().as_deref(),
     )?;
     info!(
         user = conn_info.conn_info.user_info.user.as_str(),

proxy/src/tls/server_config.rs

@@ -9,17 +9,14 @@ use rustls::pki_types::{CertificateDer, PrivateKeyDer};
 use super::{PG_ALPN_PROTOCOL, TlsServerEndPoint};
 
 pub struct TlsConfig {
-    pub config: Arc<rustls::ServerConfig>,
+    // unfortunate split since we cannot change the ALPN on demand.
+    // <https://github.com/rustls/rustls/issues/2260>
+    pub http_config: Arc<rustls::ServerConfig>,
+    pub pg_config: Arc<rustls::ServerConfig>,
     pub common_names: HashSet<String>,
     pub cert_resolver: Arc<CertResolver>,
 }
 
-impl TlsConfig {
-    pub fn to_server_config(&self) -> Arc<rustls::ServerConfig> {
-        self.config.clone()
-    }
-}
-
 /// Configure TLS for the main endpoint.
 pub fn configure_tls(
     key_path: &str,
@@ -71,8 +68,15 @@ pub fn configure_tls(
         config.key_log = Arc::new(rustls::KeyLogFile::new());
     }
 
+    let mut http_config = config.clone();
+    let mut pg_config = config;
+
+    http_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
+    pg_config.alpn_protocols = vec![b"postgresql".to_vec()];
+
     Ok(TlsConfig {
-        config: Arc::new(config),
+        http_config: Arc::new(http_config),
+        pg_config: Arc::new(pg_config),
         common_names,
         cert_resolver,
     })