diff --git a/changelog.d/tls_server_session_resumption.fix.md b/changelog.d/tls_server_session_resumption.fix.md new file mode 100644 index 0000000000000..ad79c90646a0e --- /dev/null +++ b/changelog.d/tls_server_session_resumption.fix.md @@ -0,0 +1,5 @@ +TLS-enabled sources with `tls.verify_certificate: true` no longer fail the handshake of clients +that resume a previous TLS session. Such clients previously had their connection rejected with a +`session id context uninitialized` error. + +authors: guilload diff --git a/lib/vector-core/src/tls/client_cert_verifier.rs b/lib/vector-core/src/tls/client_cert_verifier.rs new file mode 100644 index 0000000000000..3d159e3e88a51 --- /dev/null +++ b/lib/vector-core/src/tls/client_cert_verifier.rs @@ -0,0 +1,430 @@ +//! Application-level authorization of TLS client certificates. +//! +//! Chain validation against the configured CA only establishes that a client holds *some* +//! certificate issued by that CA. When a CA is shared by many workloads, a server may need to +//! restrict which of those certificates it accepts. A [`ClientCertVerifier`] installed with +//! [`TlsSettings::with_client_cert_verifier`](super::TlsSettings::with_client_cert_verifier) +//! makes that decision during the handshake. + +use openssl::{ + stack::StackRef, + x509::{X509, X509StoreContextRef, X509VerifyResult}, +}; + +/// Decides whether a TLS server accepts a client certificate. +/// +/// It runs during the handshake, once per connection (session resumption is disabled while a +/// verifier is installed). OpenSSL calls it after building the client's chain to the trust anchor +/// and checking its extensions, signatures, and validity periods, but before checking name +/// constraints and certificate policies: a chain the verifier accepts can still be rejected by +/// those later checks, so it must not rely on them having passed. It never relaxes validation: a +/// client whose chain fails an earlier check is rejected without calling the verifier. +/// +/// Implementations run on the connection's handshake path, so they must be fast and must not +/// block. They must not panic: the call crosses an FFI boundary, where a panic aborts the process. +pub trait ClientCertVerifier: Send + Sync { + /// Returns whether to accept the client. `chain` is the verified chain, leaf first. + fn verify(&self, chain: &StackRef) -> bool; +} + +impl ClientCertVerifier for F +where + F: Fn(&StackRef) -> bool + Send + Sync, +{ + fn verify(&self, chain: &StackRef) -> bool { + self(chain) + } +} + +/// The OpenSSL verify callback. OpenSSL calls it once per certificate in the chain, from the trust +/// anchor (highest depth) down to the leaf (depth 0); `preverified` is its own verdict for that +/// certificate. +pub(super) fn verify_client_cert( + verifier: &dyn ClientCertVerifier, + preverified: bool, + context: &mut X509StoreContextRef, +) -> bool { + if !preverified { + return false; + } + // Only the leaf identifies the client. Above it, OpenSSL's verdict stands. + if context.error_depth() != 0 { + return true; + } + if context.chain().is_some_and(|chain| verifier.verify(chain)) { + return true; + } + // Without an error code, OpenSSL reports an unspecified verification error. + context.set_error(X509VerifyResult::APPLICATION_VERIFICATION); + false +} + +#[cfg(test)] +mod tests { + use std::{ + net::SocketAddr, + pin::Pin, + sync::{Arc, Mutex, MutexGuard}, + }; + + use openssl::{ + ssl::{ + SslConnector, SslConnectorBuilder, SslFiletype, SslMethod, SslSession, + SslSessionCacheMode, + }, + stack::StackRef, + x509::{X509, X509VerifyResult}, + }; + use tokio::net::TcpStream; + + use super::ClientCertVerifier; + use crate::tls::{ + MaybeTls, MaybeTlsSettings, TEST_PEM_CA_PATH, TEST_PEM_CLIENT_CRT_PATH, + TEST_PEM_CLIENT_KEY_PATH, TEST_PEM_CRT_PATH, TEST_PEM_KEY_PATH, TlsAcceptorReloader, + TlsConfig, TlsEnableableConfig, TlsError, TlsSettings, + test_util::{common_name, self_signed}, + }; + + /// A verifier that records how often it ran and the chain it last saw. + #[derive(Default)] + struct RecordingVerifier { + accept: bool, + inner: Mutex, + } + + #[derive(Default)] + struct RecordingVerifierInner { + calls: usize, + leaf_common_name: Option, + chain_len: usize, + } + + impl RecordingVerifier { + fn new(accept: bool) -> Arc { + Arc::new(Self { + accept, + ..Self::default() + }) + } + + fn inner(&self) -> MutexGuard<'_, RecordingVerifierInner> { + self.inner.lock().unwrap() + } + } + + impl ClientCertVerifier for RecordingVerifier { + fn verify(&self, chain: &StackRef) -> bool { + let mut inner = self.inner(); + inner.calls += 1; + inner.leaf_common_name = chain.iter().next().map(common_name); + inner.chain_len = chain.len(); + self.accept + } + } + + /// Server settings trusting the test CA. `verify_certificate` is left unset (off by default + /// for servers): installing a verifier must request and require client certificates anyway. + fn server_settings() -> TlsSettings { + server_settings_from(server_config()) + } + + fn server_config() -> TlsConfig { + TlsConfig { + ca_file: Some(TEST_PEM_CA_PATH.into()), + crt_file: Some(TEST_PEM_CRT_PATH.into()), + key_file: Some(TEST_PEM_KEY_PATH.into()), + ..Default::default() + } + } + + fn server_settings_from(options: TlsConfig) -> TlsSettings { + let config = TlsEnableableConfig { + enabled: Some(true), + options, + }; + match MaybeTlsSettings::from_config(Some(&config), true).unwrap() { + MaybeTls::Tls(settings) => settings, + MaybeTls::Raw(()) => unreachable!("TLS should be enabled"), + } + } + + #[derive(Clone, Copy)] + enum Client { + /// Presents the test client certificate, issued under the test CA. + Trusted, + /// Presents a certificate the server's CA did not issue. + Untrusted, + /// Presents no certificate. + Anonymous, + } + + /// Serves one connection with `settings` and returns whether the server-side handshake + /// succeeded. The server side is the reliable signal: under TLS 1.3 the client may finish its + /// handshake before the server rejects its certificate. + async fn server_accepts(settings: &TlsSettings, client: Client) -> bool { + server_handshake(settings, client).await.is_ok() + } + + /// Serves one connection with `settings` and returns the outcome of the server-side handshake. + async fn server_handshake(settings: &TlsSettings, client: Client) -> crate::tls::Result<()> { + let addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let mut listener = MaybeTlsSettings::Tls(settings.clone()) + .bind(&addr) + .await + .unwrap(); + let local_addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let mut stream = listener.accept().await.unwrap(); + stream.handshake().await + }); + connect(local_addr, client).await; + server.await.unwrap() + } + + async fn connect(addr: SocketAddr, client: Client) { + handshake(&connector_builder(client).build(), addr, None).await; + } + + fn connector_builder(client: Client) -> SslConnectorBuilder { + let mut builder = SslConnector::builder(SslMethod::tls()).unwrap(); + builder.set_ca_file(TEST_PEM_CA_PATH).unwrap(); + match client { + Client::Trusted => { + builder + .set_certificate_chain_file(TEST_PEM_CLIENT_CRT_PATH) + .unwrap(); + builder + .set_private_key_file(TEST_PEM_CLIENT_KEY_PATH, SslFiletype::PEM) + .unwrap(); + } + Client::Untrusted => { + let (cert, key) = self_signed("localhost"); + builder.set_certificate(&cert).unwrap(); + builder.set_private_key(&key).unwrap(); + } + Client::Anonymous => {} + } + builder + } + + /// Runs one client handshake, offering `session` for resumption. Returns whether the session + /// was resumed. + async fn handshake( + connector: &SslConnector, + addr: SocketAddr, + session: Option<&SslSession>, + ) -> bool { + let mut ssl = connector + .configure() + .unwrap() + .into_ssl("localhost") + .unwrap(); + if let Some(session) = session { + // SAFETY: the session was issued to a connection of the same connector. + unsafe { ssl.set_session(session).unwrap() }; + } + let tcp = TcpStream::connect(addr).await.unwrap(); + let mut stream = tokio_openssl::SslStream::new(ssl, tcp).unwrap(); + // Rejections surface on either side depending on the TLS version; the server decides. + Pin::new(&mut stream).connect().await.ok(); + // Also lets the client receive the server's TLS 1.3 session tickets. + tokio::io::AsyncReadExt::read(&mut stream, &mut [0; 1]) + .await + .ok(); + // OpenSSL marks the session of a connection that was not shut down cleanly unresumable. + tokio::io::AsyncWriteExt::shutdown(&mut stream).await.ok(); + stream.ssl().session_reused() + } + + /// Connects as `Client::Trusted` twice, offering on the second connection any session the + /// server issued on the first. Returns whether the server accepted each handshake and whether + /// the second one resumed the session. + async fn server_accepts_reconnection(settings: &TlsSettings) -> (bool, bool, bool) { + let addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let mut listener = MaybeTlsSettings::Tls(settings.clone()) + .bind(&addr) + .await + .unwrap(); + let local_addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let mut accepted = [false; 2]; + for accepted in &mut accepted { + let mut stream = listener.accept().await.unwrap(); + *accepted = stream.handshake().await.is_ok(); + } + accepted + }); + + let session = Arc::new(Mutex::new(None)); + let mut builder = connector_builder(Client::Trusted); + builder.set_session_cache_mode(SslSessionCacheMode::CLIENT); + let issued = Arc::clone(&session); + builder.set_new_session_callback(move |_, session| { + *issued.lock().unwrap() = Some(session); + }); + let connector = builder.build(); + + handshake(&connector, local_addr, None).await; + let session = session.lock().unwrap().take(); + let resumed = handshake(&connector, local_addr, session.as_ref()).await; + let [first, second] = server.await.unwrap(); + (first, second, resumed) + } + + #[tokio::test] + async fn accepting_verifier_sees_the_verified_chain_leaf_first() { + let verifier = RecordingVerifier::new(true); + let settings = server_settings().with_client_cert_verifier(verifier.clone()); + + assert!(server_accepts(&settings, Client::Trusted).await); + let inner = verifier.inner(); + assert_eq!(inner.calls, 1, "runs once, for the leaf only"); + assert_eq!(inner.leaf_common_name.as_deref(), Some("localhost")); + // Leaf, intermediate CA, and root CA. + assert_eq!(inner.chain_len, 3); + } + + #[tokio::test] + async fn rejecting_verifier_fails_the_handshake() { + let verifier = RecordingVerifier::new(false); + let settings = server_settings().with_client_cert_verifier(verifier.clone()); + + let error = server_handshake(&settings, Client::Trusted) + .await + .expect_err("the verifier rejects the client"); + assert!( + matches!( + error, + TlsError::PeerCertificateRejected { + reason: X509VerifyResult::APPLICATION_VERIFICATION, + .. + } + ), + "unexpected error: {error}" + ); + assert_eq!(verifier.inner().calls, 1); + } + + #[tokio::test] + async fn verifier_never_runs_for_an_untrusted_chain() { + let verifier = RecordingVerifier::new(true); + let settings = server_settings().with_client_cert_verifier(verifier.clone()); + + assert!(!server_accepts(&settings, Client::Untrusted).await); + assert_eq!(verifier.inner().calls, 0); + } + + #[tokio::test] + async fn verifier_requires_a_client_certificate() { + let verifier = RecordingVerifier::new(true); + let settings = server_settings().with_client_cert_verifier(verifier.clone()); + + assert!(!server_accepts(&settings, Client::Anonymous).await); + assert_eq!(verifier.inner().calls, 0); + } + + #[tokio::test] + async fn verifier_runs_on_every_connection() { + let verifier = RecordingVerifier::new(true); + let settings = server_settings().with_client_cert_verifier(verifier.clone()); + + let (first, second, resumed) = server_accepts_reconnection(&settings).await; + assert!(first && second); + assert!(!resumed, "resuming a session would skip the verifier"); + assert_eq!(verifier.inner().calls, 2); + } + + #[tokio::test] + async fn verified_client_can_resume_its_session() { + let settings = server_settings_from(TlsConfig { + verify_certificate: Some(true), + ..server_config() + }); + + let (first, second, resumed) = server_accepts_reconnection(&settings).await; + assert!(first && second); + assert!(resumed); + } + + #[test] + fn verifier_requires_a_ca() { + let settings = server_settings_from(TlsConfig { + ca_file: None, + ..server_config() + }) + .with_client_cert_verifier(RecordingVerifier::new(true)); + + assert!(matches!( + settings.acceptor(), + Err(TlsError::ClientCertVerifierWithoutCa) + )); + } + + #[tokio::test] + async fn anonymous_client_is_accepted_without_a_verifier() { + // Control for `verifier_requires_a_client_certificate`: the same settings without a + // verifier do not request client certificates. + assert!(server_accepts(&server_settings(), Client::Anonymous).await); + } + + #[tokio::test] + async fn closures_can_be_verifiers() { + let settings = server_settings() + .with_client_cert_verifier(Arc::new(|chain: &StackRef| !chain.is_empty())); + + assert!(server_accepts(&settings, Client::Trusted).await); + } + + #[tokio::test] + async fn reload_keeps_the_verifier() { + let verifier = RecordingVerifier::new(false); + let reloader = + MaybeTlsSettings::Tls(server_settings().with_client_cert_verifier(verifier.clone())) + .reloadable_acceptor() + .unwrap() + .unwrap(); + // Rotation rebuilds settings from configuration, without the verifier. Reloading through a + // weak handle, as a background rotation task would, must keep it too. + reloader + .downgrade() + .upgrade() + .unwrap() + .reload(&server_settings()) + .unwrap(); + + assert!(!server_accepts_after_reload(reloader).await); + assert_eq!(verifier.inner().calls, 1); + } + + #[tokio::test] + async fn reload_does_not_add_a_verifier() { + let verifier = RecordingVerifier::new(false); + let reloader = MaybeTlsSettings::Tls(server_settings()) + .reloadable_acceptor() + .unwrap() + .unwrap(); + reloader + .reload(&server_settings().with_client_cert_verifier(verifier.clone())) + .unwrap(); + + assert!(server_accepts_after_reload(reloader).await); + assert_eq!(verifier.inner().calls, 0); + } + + /// Like [`server_accepts`] for `Client::Trusted`, serving the acceptor `reloader` holds. + async fn server_accepts_after_reload(reloader: TlsAcceptorReloader) -> bool { + let addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let mut listener = MaybeTlsSettings::Tls(server_settings()) + .bind_reloadable(&addr, Some(reloader)) + .await + .unwrap(); + let local_addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let mut stream = listener.accept().await.unwrap(); + stream.handshake().await.is_ok() + }); + connect(local_addr, Client::Trusted).await; + server.await.unwrap() + } +} diff --git a/lib/vector-core/src/tls/incoming.rs b/lib/vector-core/src/tls/incoming.rs index b93e268312456..bd67d40bcb41f 100644 --- a/lib/vector-core/src/tls/incoming.rs +++ b/lib/vector-core/src/tls/incoming.rs @@ -12,7 +12,7 @@ use futures::{FutureExt, Stream, future::BoxFuture, stream}; use ipnet::IpNet; use openssl::{ ssl::{Ssl, SslAcceptor, SslMethod}, - x509::X509, + x509::{X509, X509VerifyResult}, }; use snafu::ResultExt; use tokio::{ @@ -24,8 +24,8 @@ use tokio_openssl::SslStream; use tonic::transport::{Certificate, server::Connected}; use super::{ - CreateAcceptorSnafu, HandshakeSnafu, IncomingListenerSnafu, MaybeTlsSettings, MaybeTlsStream, - SslBuildSnafu, TcpBindSnafu, TlsAcceptorReloader, TlsError, TlsSettings, + CreateAcceptorSnafu, IncomingListenerSnafu, MaybeTlsSettings, MaybeTlsStream, SslBuildSnafu, + TcpBindSnafu, TlsAcceptorReloader, TlsError, TlsSettings, }; use crate::tcp::{self, TcpKeepaliveConfig}; @@ -59,21 +59,25 @@ impl MaybeTlsSettings { } /// Bind a listener, optionally sharing a [`TlsAcceptorReloader`] so its acceptor can be swapped - /// at runtime. When `reloader` is `Some` and TLS is enabled, connections accepted by the - /// returned listener use the acceptor the reloader currently holds; otherwise the acceptor is - /// fixed for the lifetime of the listener. + /// at runtime. When `reloader` is `Some`, connections accepted by the returned listener use the + /// acceptor the reloader currently holds; otherwise the acceptor is fixed for the lifetime of + /// the listener. + /// + /// Returns [`TlsError::ReloaderWithoutTls`] when `reloader` is `Some` but TLS is disabled: + /// silently serving plaintext would bypass whatever the reloader's acceptor enforces, such as + /// client certificate verification. pub async fn bind_reloadable( &self, addr: &SocketAddr, reloader: Option, ) -> crate::tls::Result { - let listener = TcpListener::bind(addr).await.context(TcpBindSnafu)?; - let acceptor = match (self, reloader) { - (Self::Raw(()), _) => None, + (Self::Raw(()), Some(_)) => return Err(TlsError::ReloaderWithoutTls), + (Self::Raw(()), None) => None, (Self::Tls(_), Some(reloader)) => Some(reloader.shared()), (Self::Tls(tls), None) => Some(Arc::new(ArcSwap::from_pointee(tls.acceptor()?))), }; + let listener = TcpListener::bind(addr).await.context(TcpBindSnafu)?; Ok(MaybeTlsListener { listener, @@ -286,10 +290,15 @@ impl MaybeTlsIncomingStream { async move { let ssl = Ssl::new(acceptor.context()).context(SslBuildSnafu)?; let mut stream = SslStream::new(ssl, stream).context(SslBuildSnafu)?; - Pin::new(&mut stream) - .accept() - .await - .context(HandshakeSnafu)?; + if let Err(source) = Pin::new(&mut stream).accept().await { + // The error itself only says verification failed; the verify result says why. + let reason = stream.ssl().verify_result(); + return Err(if reason == X509VerifyResult::OK { + TlsError::Handshake { source } + } else { + TlsError::PeerCertificateRejected { source, reason } + }); + } Ok(stream) } .boxed(), diff --git a/lib/vector-core/src/tls/mod.rs b/lib/vector-core/src/tls/mod.rs index ff84c156bdd5c..13f6187221bc2 100644 --- a/lib/vector-core/src/tls/mod.rs +++ b/lib/vector-core/src/tls/mod.rs @@ -5,6 +5,7 @@ use std::{fmt::Debug, net::SocketAddr, num::TryFromIntError, path::PathBuf, time use openssl::{ error::ErrorStack, ssl::{ConnectConfiguration, SslConnector, SslConnectorBuilder, SslMethod}, + x509::X509VerifyResult, }; use snafu::{ResultExt, Snafu}; use tokio::net::TcpStream; @@ -12,12 +13,16 @@ use tokio_openssl::SslStream; use crate::tcp::{self, TcpKeepaliveConfig}; +mod client_cert_verifier; mod incoming; mod maybe_tls; mod outgoing; mod reload; mod settings; +#[cfg(test)] +mod test_util; +pub use client_cert_verifier::ClientCertVerifier; pub use incoming::{CertificateMetadata, MaybeTlsIncomingStream, MaybeTlsListener}; pub use maybe_tls::MaybeTls; pub use reload::{TlsAcceptorReloader, WeakTlsAcceptorReloader}; @@ -81,8 +86,23 @@ pub enum TlsError { }, #[snafu(display("TLS configuration requires a certificate when enabled"))] MissingRequiredIdentity, + #[snafu(display("A reloadable TLS acceptor requires TLS to be enabled"))] + ReloaderWithoutTls, + #[snafu(display( + "A client certificate verifier requires a CA file to validate client certificates" + ))] + ClientCertVerifierWithoutCa, #[snafu(display("TLS handshake failed: {}", source))] Handshake { source: openssl::ssl::Error }, + #[snafu(display( + "TLS handshake failed: peer certificate rejected: {}: {}", + reason, + source + ))] + PeerCertificateRejected { + source: openssl::ssl::Error, + reason: X509VerifyResult, + }, #[snafu(display("Incoming listener failed: {}", source))] IncomingListener { source: tokio::io::Error }, #[snafu(display("Creating the TLS acceptor failed: {}", source))] @@ -101,6 +121,8 @@ pub enum TlsError { AddCertToStore { source: ErrorStack }, #[snafu(display("Error setting up the verification certificate: {}", source))] SetVerifyCert { source: ErrorStack }, + #[snafu(display("Error setting the TLS session id context: {}", source))] + SetSessionIdContext { source: ErrorStack }, #[snafu(display("Error setting SNI: {}", source))] SetSni { source: ErrorStack }, #[snafu(display("Error setting ALPN protocols: {}", source))] diff --git a/lib/vector-core/src/tls/reload.rs b/lib/vector-core/src/tls/reload.rs index ac781d02d9818..331197c68a20c 100644 --- a/lib/vector-core/src/tls/reload.rs +++ b/lib/vector-core/src/tls/reload.rs @@ -12,23 +12,28 @@ use std::sync::{Arc, Weak}; use arc_swap::ArcSwap; use openssl::ssl::SslAcceptor; -use super::{MaybeTlsSettings, TlsSettings}; +use super::{ClientCertVerifier, MaybeTlsSettings, TlsSettings}; /// A cloneable handle to a server TLS acceptor that can be swapped at runtime. /// /// Connections accepted after [`reload`](Self::reload) use the new acceptor; connections already /// established keep whatever they negotiated at handshake time. +/// +/// The reloader owns the [`ClientCertVerifier`] of the settings it was created from, if any, and +/// applies it to every acceptor it builds. #[derive(Clone)] pub struct TlsAcceptorReloader { acceptor: Arc>, + client_cert_verifier: Option>, } impl TlsAcceptorReloader { - /// Wrap an initial acceptor in a swappable cell. - pub(super) fn new(acceptor: SslAcceptor) -> Self { - Self { - acceptor: Arc::new(ArcSwap::from_pointee(acceptor)), - } + /// Build the initial acceptor from `settings` and wrap it in a swappable cell. + fn new(settings: &TlsSettings) -> crate::tls::Result { + Ok(Self { + acceptor: Arc::new(ArcSwap::from_pointee(settings.acceptor()?)), + client_cert_verifier: settings.client_cert_verifier.clone(), + }) } /// The shared cell the bound listener reads from on each accept. @@ -38,7 +43,14 @@ impl TlsAcceptorReloader { /// Swap in a freshly built acceptor from `settings`. New connections pick it up /// immediately; the previous acceptor is dropped once its last in-flight handshake completes. + /// + /// The acceptor applies this reloader's client certificate verifier in place of any `settings` + /// carries, so rotating certificates never changes which clients are authorized. pub fn reload(&self, settings: &TlsSettings) -> crate::tls::Result<()> { + let mut settings = settings.clone(); + settings + .client_cert_verifier + .clone_from(&self.client_cert_verifier); self.acceptor.store(Arc::new(settings.acceptor()?)); Ok(()) } @@ -47,6 +59,7 @@ impl TlsAcceptorReloader { pub fn downgrade(&self) -> WeakTlsAcceptorReloader { WeakTlsAcceptorReloader { acceptor: Arc::downgrade(&self.acceptor), + client_cert_verifier: self.client_cert_verifier.clone(), } } } @@ -55,14 +68,16 @@ impl TlsAcceptorReloader { #[derive(Clone)] pub struct WeakTlsAcceptorReloader { acceptor: Weak>, + client_cert_verifier: Option>, } impl WeakTlsAcceptorReloader { /// Return the live [`TlsAcceptorReloader`], or `None` once the bound listener has been dropped. pub fn upgrade(&self) -> Option { - self.acceptor - .upgrade() - .map(|acceptor| TlsAcceptorReloader { acceptor }) + self.acceptor.upgrade().map(|acceptor| TlsAcceptorReloader { + acceptor, + client_cert_verifier: self.client_cert_verifier.clone(), + }) } } @@ -71,10 +86,11 @@ impl MaybeTlsSettings { /// /// Pass the returned handle to [`bind_reloadable`](Self::bind_reloadable) so the bound listener /// serves it, and keep a clone to call [`reload`](TlsAcceptorReloader::reload) when the - /// certificate material rotates. + /// certificate material rotates. The handle keeps the settings' client certificate verifier + /// across reloads. pub fn reloadable_acceptor(&self) -> crate::tls::Result> { match self { - Self::Tls(tls) => Ok(Some(TlsAcceptorReloader::new(tls.acceptor()?))), + Self::Tls(tls) => Ok(Some(TlsAcceptorReloader::new(tls)?)), Self::Raw(()) => Ok(None), } } @@ -85,17 +101,15 @@ mod test { use std::{net::SocketAddr, pin::Pin}; use openssl::{ - asn1::Asn1Time, - bn::{BigNum, MsbOption}, - hash::MessageDigest, - nid::Nid, - pkey::PKey, - rsa::Rsa, + pkey::{PKeyRef, Private}, ssl::{SslConnector, SslMethod, SslVerifyMode}, - x509::{X509, X509NameBuilder}, + x509::X509Ref, }; - use crate::tls::{MaybeTls, MaybeTlsSettings, TlsConfig, TlsEnableableConfig}; + use crate::tls::{ + MaybeTls, MaybeTlsSettings, TlsConfig, TlsEnableableConfig, TlsError, + test_util::{common_name, self_signed}, + }; #[test] fn no_reloadable_acceptor_without_tls() { @@ -108,6 +122,25 @@ mod test { ); } + #[tokio::test] + async fn binding_a_reloader_without_tls_fails() { + let reloader = + MaybeTlsSettings::from_config(Some(&TlsEnableableConfig::test_config()), true) + .unwrap() + .reloadable_acceptor() + .unwrap(); + assert!(reloader.is_some()); + + let addr = "127.0.0.1:0".parse().unwrap(); + let result = MaybeTlsSettings::Raw(()) + .bind_reloadable(&addr, reloader) + .await; + assert!( + matches!(result, Err(TlsError::ReloaderWithoutTls)), + "must not silently serve plaintext" + ); + } + #[tokio::test] async fn reloadable_acceptor_swaps_and_detects_shutdown() { let settings = @@ -206,18 +239,14 @@ mod test { .ssl() .peer_certificate() .expect("server presents a certificate"); - cert.subject_name() - .entries_by_nid(Nid::COMMONNAME) - .next() - .unwrap() - .data() - .to_string() - .unwrap() + common_name(&cert) } - fn server_settings(crt_pem: &str, key_pem: &str) -> MaybeTlsSettings { + fn server_settings(crt: &X509Ref, key: &PKeyRef) -> MaybeTlsSettings { // `crt_file`/`key_file` accept inline PEM (detected by the `-----BEGIN ` marker), so no // temp files are needed. + let crt_pem = String::from_utf8(crt.to_pem().unwrap()).unwrap(); + let key_pem = String::from_utf8(key.private_key_to_pem_pkcs8().unwrap()).unwrap(); let config = TlsEnableableConfig { enabled: Some(true), options: TlsConfig { @@ -228,38 +257,4 @@ mod test { }; MaybeTlsSettings::from_config(Some(&config), true).unwrap() } - - /// Generate a self-signed certificate/key pair (PEM) with the given common name. - fn self_signed(common_name: &str) -> (String, String) { - let key = PKey::from_rsa(Rsa::generate(2048).unwrap()).unwrap(); - - let mut name = X509NameBuilder::new().unwrap(); - name.append_entry_by_text("CN", common_name).unwrap(); - let name = name.build(); - - let mut serial = BigNum::new().unwrap(); - serial.rand(128, MsbOption::MAYBE_ZERO, false).unwrap(); - - let mut builder = X509::builder().unwrap(); - builder.set_version(2).unwrap(); - builder - .set_serial_number(&serial.to_asn1_integer().unwrap()) - .unwrap(); - builder.set_subject_name(&name).unwrap(); - builder.set_issuer_name(&name).unwrap(); - builder.set_pubkey(&key).unwrap(); - builder - .set_not_before(&Asn1Time::days_from_now(0).unwrap()) - .unwrap(); - builder - .set_not_after(&Asn1Time::days_from_now(1).unwrap()) - .unwrap(); - builder.sign(&key, MessageDigest::sha256()).unwrap(); - let cert = builder.build(); - - ( - String::from_utf8(cert.to_pem().unwrap()).unwrap(), - String::from_utf8(key.private_key_to_pem_pkcs8().unwrap()).unwrap(), - ) - } } diff --git a/lib/vector-core/src/tls/settings.rs b/lib/vector-core/src/tls/settings.rs index d946f35771685..9917c9a0817f9 100644 --- a/lib/vector-core/src/tls/settings.rs +++ b/lib/vector-core/src/tls/settings.rs @@ -4,21 +4,25 @@ use std::{ fs::File, io::Read, path::{Path, PathBuf}, - sync::{LazyLock, Mutex}, + sync::{Arc, LazyLock, Mutex}, }; use super::{ - AddCertToStoreSnafu, AddExtraChainCertSnafu, CaStackPushSnafu, EncodeAlpnProtocolsSnafu, - FileOpenFailedSnafu, FileReadFailedSnafu, MaybeTls, NewCaStackSnafu, NewStoreBuilderSnafu, - ParsePkcs12Snafu, PrivateKeyParseSnafu, Result, SetAlpnProtocolsSnafu, SetCertificateSnafu, - SetPrivateKeySnafu, SetVerifyCertSnafu, TlsError, X509ParseSnafu, + AddCertToStoreSnafu, AddExtraChainCertSnafu, CaStackPushSnafu, ClientCertVerifier, + EncodeAlpnProtocolsSnafu, FileOpenFailedSnafu, FileReadFailedSnafu, MaybeTls, NewCaStackSnafu, + NewStoreBuilderSnafu, ParsePkcs12Snafu, PrivateKeyParseSnafu, Result, SetAlpnProtocolsSnafu, + SetCertificateSnafu, SetPrivateKeySnafu, SetSessionIdContextSnafu, SetVerifyCertSnafu, + TlsError, X509ParseSnafu, client_cert_verifier::verify_client_cert, }; use cfg_if::cfg_if; use lookup::lookup_v2::OptionalValuePath; use openssl::{ pkcs12::Pkcs12, pkey::{PKey, Private}, - ssl::{AlpnError, ConnectConfiguration, SslContextBuilder, SslVerifyMode, select_next_proto}, + ssl::{ + AlpnError, ConnectConfiguration, SslContextBuilder, SslOptions, SslSessionCacheMode, + SslVerifyMode, select_next_proto, + }, stack::Stack, x509::{X509, store::X509StoreBuilder, verify::X509CheckFlags}, }; @@ -178,6 +182,7 @@ pub struct TlsSettings { pub(super) identity: Option, alpn_protocols: Option>, server_name: Option, + pub(super) client_cert_verifier: Option>, } /// Identity store in PEM format @@ -220,9 +225,27 @@ impl TlsSettings { identity: options.load_identity()?, alpn_protocols: options.parse_alpn_protocols()?, server_name: options.server_name.clone(), + client_cert_verifier: None, }) } + /// Authorize TLS clients with `verifier`, on top of chain validation. + /// + /// Only applies when these settings build a server acceptor. Installing a verifier makes the + /// server request and require a client certificate, whatever `verify_certificate` says, so a + /// verifier can never be bypassed by connecting without one. It also disables session + /// resumption, which would skip the verifier. Building the acceptor fails unless the settings + /// have a CA to validate client chains against. + /// + /// A [`TlsAcceptorReloader`](super::TlsAcceptorReloader) created from these settings keeps the + /// verifier across reloads, so settings rebuilt from configuration on certificate rotation need + /// not carry it. + #[must_use] + pub fn with_client_cert_verifier(mut self, verifier: Arc) -> Self { + self.client_cert_verifier = Some(verifier); + self + } + /// The configured SNI server name override, if any. pub fn server_name(&self) -> Option<&str> { self.server_name.as_deref() @@ -281,11 +304,39 @@ impl TlsSettings { context: &mut SslContextBuilder, for_server: bool, ) -> Result<()> { - context.set_verify(if self.verify_certificate { - SslVerifyMode::PEER | SslVerifyMode::FAIL_IF_NO_PEER_CERT - } else { - SslVerifyMode::NONE - }); + match &self.client_cert_verifier { + Some(verifier) if for_server => { + // Without a CA, client chains would be checked against the system roots, which + // accept any publicly issued certificate, or against an empty store, which rejects + // every client. + if self.authorities.is_empty() { + return Err(TlsError::ClientCertVerifierWithoutCa); + } + let verifier = Arc::clone(verifier); + context.set_verify_callback( + SslVerifyMode::PEER | SslVerifyMode::FAIL_IF_NO_PEER_CERT, + move |preverified, store_context| { + verify_client_cert(verifier.as_ref(), preverified, store_context) + }, + ); + // A resumed session skips certificate verification, so the verifier would not run. + context.set_session_cache_mode(SslSessionCacheMode::OFF); + context.set_options(SslOptions::NO_TICKET); + } + _ => context.set_verify(if self.verify_certificate { + SslVerifyMode::PEER | SslVerifyMode::FAIL_IF_NO_PEER_CERT + } else { + SslVerifyMode::NONE + }), + } + if for_server { + // When it requests client certificates, OpenSSL fails the handshake of any client + // resuming a session unless a session id context is set. Each context has its own + // session cache and ticket keys, so a constant is enough. + context + .set_session_id_context(b"vector") + .context(SetSessionIdContextSnafu)?; + } if let Some(identity) = &self.identity { context .set_certificate(&identity.cert) @@ -609,6 +660,10 @@ impl fmt::Debug for TlsSettings { f.debug_struct("TlsSettings") .field("verify_certificate", &self.verify_certificate) .field("verify_hostname", &self.verify_hostname) + .field( + "has_client_cert_verifier", + &self.client_cert_verifier.is_some(), + ) .finish_non_exhaustive() } } diff --git a/lib/vector-core/src/tls/test_util.rs b/lib/vector-core/src/tls/test_util.rs new file mode 100644 index 0000000000000..f063b4e25e519 --- /dev/null +++ b/lib/vector-core/src/tls/test_util.rs @@ -0,0 +1,52 @@ +//! Certificate helpers shared by the TLS tests. + +use openssl::{ + asn1::Asn1Time, + bn::{BigNum, MsbOption}, + hash::MessageDigest, + nid::Nid, + pkey::{PKey, Private}, + rsa::Rsa, + x509::{X509, X509NameBuilder, X509Ref}, +}; + +/// Generates a self-signed certificate, valid for a day, with the given common name, and its key. +pub(super) fn self_signed(common_name: &str) -> (X509, PKey) { + let key = PKey::from_rsa(Rsa::generate(2048).unwrap()).unwrap(); + + let mut name = X509NameBuilder::new().unwrap(); + name.append_entry_by_text("CN", common_name).unwrap(); + let name = name.build(); + + let mut serial = BigNum::new().unwrap(); + serial.rand(128, MsbOption::MAYBE_ZERO, false).unwrap(); + + let mut builder = X509::builder().unwrap(); + builder.set_version(2).unwrap(); + builder + .set_serial_number(&serial.to_asn1_integer().unwrap()) + .unwrap(); + builder.set_subject_name(&name).unwrap(); + builder.set_issuer_name(&name).unwrap(); + builder.set_pubkey(&key).unwrap(); + builder + .set_not_before(&Asn1Time::days_from_now(0).unwrap()) + .unwrap(); + builder + .set_not_after(&Asn1Time::days_from_now(1).unwrap()) + .unwrap(); + builder.sign(&key, MessageDigest::sha256()).unwrap(); + + (builder.build(), key) +} + +/// Returns the common name of `cert`'s subject. +pub(super) fn common_name(cert: &X509Ref) -> String { + cert.subject_name() + .entries_by_nid(Nid::COMMONNAME) + .next() + .unwrap() + .data() + .to_string() + .unwrap() +} diff --git a/src/sources/heroku_logs.rs b/src/sources/heroku_logs.rs index 5c85b98b9bc80..c994e094cf7df 100644 --- a/src/sources/heroku_logs.rs +++ b/src/sources/heroku_logs.rs @@ -202,6 +202,7 @@ impl SourceConfig for LogplexConfig { StatusCode::OK, true, self.tls.as_ref(), + None, self.auth.as_ref(), cx, self.acknowledgements, diff --git a/src/sources/http_server.rs b/src/sources/http_server.rs index 0e0f34b696e20..eef3f1ea04cd3 100644 --- a/src/sources/http_server.rs +++ b/src/sources/http_server.rs @@ -26,7 +26,7 @@ use crate::{ HttpSource, http::{HttpMethod, add_headers, add_query_parameters}, }, - tls::TlsEnableableConfig, + tls::{TlsAcceptorReloader, TlsEnableableConfig}, }; /// Configuration for the `http_server` source. @@ -295,10 +295,25 @@ pub fn build_param_matcher(list: &[String]) -> crate::Result>>() } -#[async_trait::async_trait] -#[typetag::serde(name = "http_server")] -impl SourceConfig for SimpleHttpConfig { - async fn build(&self, cx: SourceContext) -> crate::Result { +impl SimpleHttpConfig { + /// The source's TLS configuration, if any. Exposed so a wrapping source can build a + /// [`TlsAcceptorReloader`] and watch the certificate files for rotation. + #[must_use] + pub const fn tls_config(&self) -> Option<&TlsEnableableConfig> { + self.tls.as_ref() + } + + /// Build the source serving a runtime-swappable TLS acceptor when `tls_reloader` is set. + /// + /// # Errors + /// + /// Returns an error if the decoder, TLS, or authentication cannot be built, or if `tls_reloader` + /// is set but TLS is disabled. + pub fn build_with_tls_reloader( + &self, + cx: SourceContext, + tls_reloader: Option, + ) -> crate::Result { let log_namespace = cx.log_namespace(self.log_namespace); let decoder = self .get_decoding_config() @@ -323,12 +338,21 @@ impl SourceConfig for SimpleHttpConfig { self.response_code, self.strict_path, self.tls.as_ref(), + tls_reloader, self.auth.as_ref(), cx, self.acknowledgements, self.keepalive.clone(), ) } +} + +#[async_trait::async_trait] +#[typetag::serde(name = "http_server")] +impl SourceConfig for SimpleHttpConfig { + async fn build(&self, cx: SourceContext) -> crate::Result { + self.build_with_tls_reloader(cx, None) + } fn outputs(&self, global_log_namespace: LogNamespace) -> Vec { // There is a global and per-source `log_namespace` config. @@ -492,6 +516,7 @@ mod tests { owned_value_path, path, }, schema::Definition, + tls::{MaybeTlsSettings, TlsEnableableConfig, TlsError}, }; use vrl::{ path::ValuePath as _, @@ -1679,6 +1704,25 @@ mod tests { assert_eq!(200, send_with_headers(addr, "", headers).await); } + #[tokio::test] + async fn tls_reloader_requires_tls() { + let reloader = + MaybeTlsSettings::from_config(Some(&TlsEnableableConfig::test_config()), true) + .unwrap() + .reloadable_acceptor() + .unwrap(); + assert!(reloader.is_some()); + + let (sender, _recv) = SourceSender::new_test_finalize(EventStatus::Delivered); + let result = SimpleHttpConfig::default() + .build_with_tls_reloader(SourceContext::new_test(sender, None), reloader); + let error = result.err().expect("must not silently serve plaintext"); + assert!( + matches!(error.downcast_ref(), Some(TlsError::ReloaderWithoutTls)), + "unexpected error: {error}" + ); + } + #[test] fn output_schema_definition_vector_namespace() { let config = SimpleHttpConfig { diff --git a/src/sources/prometheus/pushgateway.rs b/src/sources/prometheus/pushgateway.rs index 1134d43edcdad..aa654f741049b 100644 --- a/src/sources/prometheus/pushgateway.rs +++ b/src/sources/prometheus/pushgateway.rs @@ -94,6 +94,7 @@ impl SourceConfig for PrometheusPushgatewayConfig { http::StatusCode::OK, false, self.tls.as_ref(), + None, self.auth.as_ref(), cx, self.acknowledgements, diff --git a/src/sources/prometheus/remote_write.rs b/src/sources/prometheus/remote_write.rs index c57570c8fb930..dc747ee5f48cf 100644 --- a/src/sources/prometheus/remote_write.rs +++ b/src/sources/prometheus/remote_write.rs @@ -130,6 +130,7 @@ impl SourceConfig for PrometheusRemoteWriteConfig { StatusCode::OK, true, self.tls.as_ref(), + None, self.auth.as_ref(), cx, self.acknowledgements, diff --git a/src/sources/util/http/prelude.rs b/src/sources/util/http/prelude.rs index 024e851891a8e..2ad644c1dd512 100644 --- a/src/sources/util/http/prelude.rs +++ b/src/sources/util/http/prelude.rs @@ -32,7 +32,10 @@ use crate::{ HttpBadRequest, HttpBytesReceived, HttpEventsReceived, HttpInternalError, StreamClosedError, }, sources::util::http::HttpMethod, - tls::{MaybeTlsIncomingStream, MaybeTlsSettings, TlsEnableableConfig}, + tls::{ + MaybeTlsIncomingStream, MaybeTlsSettings, TlsAcceptorReloader, TlsEnableableConfig, + TlsError, + }, }; pub trait HttpSource: Clone + Send + Sync + 'static { @@ -119,10 +122,13 @@ pub trait HttpSource: Clone + Send + Sync + 'static { decompress_body(encoding_header, body) } + /// Serves the acceptor held by `tls_reloader` when set, so the caller can swap in new TLS + /// material at runtime. /// /// # Errors /// - /// Returns an error if TLS or authentication cannot be configured. + /// Returns an error if TLS or authentication cannot be configured, or if `tls_reloader` is set + /// but TLS is disabled. #[allow( clippy::too_many_lines, reason = "Splitting this source routine is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659" @@ -136,12 +142,17 @@ pub trait HttpSource: Clone + Send + Sync + 'static { response_code: StatusCode, strict_path: bool, tls: Option<&TlsEnableableConfig>, + tls_reloader: Option, auth: Option<&HttpServerAuthConfig>, cx: SourceContext, acknowledgements: SourceAcknowledgementsConfig, keepalive_settings: KeepaliveConfig, ) -> crate::Result { let tls = MaybeTlsSettings::from_config(tls, true)?; + // Binding would catch this too, but only once the source runs. + if tls_reloader.is_some() && tls.is_raw() { + return Err(TlsError::ReloaderWithoutTls.into()); + } let protocol = tls.http_protocol_name(); let auth_matcher = auth .map(|a| a.build(&cx.enrichment_tables, &cx.metrics_storage)) @@ -286,7 +297,7 @@ pub trait HttpSource: Clone + Send + Sync + 'static { info!(message = "Building HTTP server.", address = %address); let listener = tls - .bind(&address) + .bind_reloadable(&address, tls_reloader) .await .inspect_err(|err| error!("An error occurred: {err:?}."))? .with_keepalive(keepalive_settings.tcp_keepalive);