GeezerVison

The World Wide Web, restored to 800 pixels of glory.

GeezerVison is showing github.com — [dbsp] Exchange data between hosts over TLS with client certificates by blp · Pull Request #7713 · feldera/feldera

[dbsp] Exchange data between hosts over TLS with client certificates #7713

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

blp merged 5 commits into main from exchange-tls

Nov 4, 2026

Uh oh!

There was an error while loading. Please reload this page.

Merged

[dbsp] Exchange data between hosts over TLS with client certificates #7713

Changes from 1 commit

Commits

Show all changes

5 commits

Select commit Hold shift + click to select a range

4da9b2b

[feldera-mtls] Add TLS configurations for the hosts of a multihost pi…

blp Oct 29, 2026

475412a

[dbsp] Exchange data between hosts over TLS, if configured.

blp Oct 29, 2026

af7fdf4

[connectors] Use TLS for multihost exchange with client certificates.

blp Oct 29, 2026

89b92f5

[feldera-mtls] Refuse TLS 1.2 between hosts and explain setup errors.

blp Oct 30, 2026

c0fcb7e

[connectors] Say where exchange TLS files come from when they are wrong.

blp Oct 30, 2026

File filter

Filter by extension .rs (1) All 1 file type selected

Viewed files

Clear filters

Conversations

Failed to load comments.

Uh oh!

There was an error while loading. Please reload this page.

Jump to

Jump to file

Failed to load files.

Uh oh!

There was an error while loading. Please reload this page.

Diff view

Unified

Split

Hide whitespace

Diff view

Unified

Split

Hide whitespace

Next commit

[feldera-mtls] Add TLS configurations for the hosts of a multihost pi…

…peline.

The hosts of a multihost pipeline exchange data over raw TCP.  To move that
to TLS, each host needs a client and a server configuration that
authenticate the other host as a host of the same pipeline.

The identity to check is already in each pipeline certificate.  An X.509
certificate lists the names that it is valid for in its "subject
alternative name" (SAN) extension.  Most SAN entries are DNS names or IP
addresses, which name a network location, but an entry can also be a URI.
The runner gives each pipeline certificate the URI entry
`urn:feldera:pipeline:<pipeline id>`.  Because the private CA signs it, a
peer that completes a TLS handshake with that certificate proves that it
belongs to that pipeline, wherever it runs.  The control plane's
certificate has `urn:feldera:control-plane` instead.  `Caller` reads these
URIs, and the compiler, runner, pipeline and coordinator servers already
use it to admit clients.

`PipelinePeerTls::from_chain` builds both configurations from the pipeline
certificate chain and its key.  Both ends present the pipeline
certificate.  Both trust only the private CA, the first certificate after
the leaf, as the existing client certificate verifier does.  Both admit
only a peer whose certificate has the same pipeline URI, so the control
plane, external clients and other pipelines are refused.  The server
requires a client certificate.

The client needs a custom server certificate verifier, which rustls allows
only through `ClientConfig::dangerous()`.  The built-in verifier checks
that a server's DNS name or IP address entries include the name that the
client connected to.  Hosts connect by pod IP address, which no
certificate lists, and the pipeline URI is the identity that matters.  The
custom verifier does what the built-in one does without CRLs (chain to the
private CA, validity period, server auth extended key usage, handshake
signatures), except that it checks the pipeline URI instead of the name.
Two alternatives avoid `dangerous()`, but each is worse:

- Verify against a DNS name that matches the wildcard DNS entry of our own
  certificate, `*.<pipeline service>.<namespace>.svc.cluster.local`.  This
  makes the runner's DNS naming the identity, separate from the pipeline
  URI that the servers check.

- Verify against a name that every pipeline certificate has, such as
  `localhost`, and check the peer's pipeline URI after the handshake.  This
  splits verification in two, and the connection is only safe if every
  caller remembers the second check.

The tests run in-memory handshakes.  The client and the server side are
each checked on their own, against another pipeline, the control plane, a
sibling CA, the root CA, and no certificate.  Each test fails if the check
it covers is removed.

Signed-off-by: Ben Pfaff <[email protected]>

Loading branch information

Uh oh!

There was an error while loading. Please reload this page.

blp committed Nov 3, 2026

commit 4da9b2bd2ffae5f418d504bd3fa31fbd2a1cce82

There are no files selected for viewing

373 changes: 365 additions & 8 deletions 373 crates/feldera-mtls/src/lib.rs

Show comments

View file

Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -14,11 +14,16 @@ use actix_web::middleware::Next;
		use actix_web::rt::net::TcpStream;
		use actix_web::{HttpResponse, ResponseError};
		use feldera_types::error::{DetailedError, ErrorResponse};
		use rustls::RootCertStore;
		use rustls::crypto::CryptoProvider;
		use rustls::pki_types::CertificateDer;
		use rustls::server::WebPkiClientVerifier;
		use rustls::server::danger::ClientCertVerifier;
		use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
		use rustls::client::verify_server_cert_signed_by_trust_anchor;
		use rustls::crypto::{CryptoProvider, verify_tls12_signature, verify_tls13_signature};
		use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime};
		use rustls::server::danger::{ClientCertVerified, ClientCertVerifier};
		use rustls::server::{ParsedCertificate, WebPkiClientVerifier};
		use rustls::{
		ClientConfig, DigitallySignedStruct, DistinguishedName, RootCertStore, ServerConfig,
		SignatureScheme,
		};
		use serde::Serialize;
		use std::any::Any;
		use std::borrow::Cow;
Expand Down Expand Up		@@ -292,17 +297,226 @@ impl PipelineClientAuth {
		}
		}

		/// TLS configurations for the connections between the hosts of a multihost pipeline.
		#[derive(Clone, Debug)]
		pub struct PipelinePeerTls {
		pub client: Arc<ClientConfig>,
		pub server: Arc<ServerConfig>,
		}

		impl PipelinePeerTls {
		/// Builds the configurations from the pipeline certificate chain and its key. Both ends
		/// present the pipeline certificate and admit only a peer whose certificate the private
		/// CA signed for the same pipeline. Peers connect by IP address, so the server name is
		/// not checked: the pipeline URI SAN is the identity.
		pub fn from_chain(
		chain: Vec<CertificateDer<'static>>,
		key: PrivateKeyDer<'static>,
		) -> Result<Self, String> {
		let verifier = Arc::new(PipelinePeerVerifier::from_chain(&chain)?);
		let provider = verifier.provider.clone();
		let server = ServerConfig::builder_with_provider(provider.clone())
		.with_protocol_versions(&[&rustls::version::TLS13])
		.map_err(\|e\| format!("peer TLS server: {e}"))?
Comment thread Outdated gz Oct 30, 2026 Copy link Copy Markdown Contributor There was a problem hiding this comment. Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. maybe we could make these more actionable, I wonder if they can fail for some setup problem (helm) etc. we can tell people where to look Sorry, something went wrong. Uh oh! There was an error while loading. Please reload this page. All reactions blp Oct 30, 2026 Copy link Copy Markdown Member Author There was a problem hiding this comment. Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. Done in `2b9fff0` and `832af79`. The errors that a setup problem can cause now say what is wrong: a private key that does not match the pipeline certificate, or a certificate file without the CA chain, which the runner appends from the private CA in the Helm value `caSecretRef`. The panic in `server.rs` now names both file paths and says that the runner writes them from the pipeline certificate, which it signs with that CA. The TLS 1.3 error can only come from a code bug, so it just says that the crypto provider does not support TLS 1.3. Sorry, something went wrong. Uh oh! There was an error while loading. Please reload this page. All reactions
		.with_client_cert_verifier(verifier.clone())
		.with_single_cert(chain.clone(), key.clone_key())
		.map_err(\|e\| format!("peer TLS server: {e}"))?;
		let client = ClientConfig::builder_with_provider(provider)
		.with_protocol_versions(&[&rustls::version::TLS13])
		.map_err(\|e\| format!("peer TLS client: {e}"))?
		// rustls puts every custom server certificate verifier behind `dangerous()`,
		// because a custom verifier can skip checks. This one is needed because the
		// built-in verifier identifies a server only by its DNS name or IP address, and
		// hosts connect by an IP address that no certificate names. It skips nothing
		// but that name check, and checks the pipeline URI SAN instead (see
		// `verify_server_cert` below).
		.dangerous()
		.with_custom_certificate_verifier(verifier)
		.with_client_auth_cert(chain, key)
		.map_err(\|e\| format!("peer TLS client: {e}"))?;
		Ok(Self {
		client: Arc::new(client),
		server: Arc::new(server),
		})
		}
		}

		/// Verifies the certificate of another host of the same pipeline, on both ends of a
		/// connection.
		#[derive(Debug)]
		struct PipelinePeerVerifier {
		own: Caller,
		roots: Arc<RootCertStore>,
		provider: Arc<CryptoProvider>,
		client_verifier: Arc<dyn ClientCertVerifier>,
		}

		impl PipelinePeerVerifier {
		/// Admits peers with the pipeline identity of the leaf of `chain`, signed by the private
		/// CA, the first certificate after the leaf.
		fn from_chain(chain: &[CertificateDer<'static>]) -> Result<Self, String> {
		let (leaf, ca_chain) = chain
		.split_first()
		.ok_or("the certificate file holds no certificate")?;
		let own @ Caller::Pipeline(_) = Caller::from_certificate(leaf) else {
		return Err(format!(
		"the certificate names no pipeline: peer certificates need the URI SAN {PIPELINE_CERTIFICATE_URI_PREFIX}<id>, which the runner adds"
		));
		};
		let issuing_ca = ca_chain
		.first()
		.ok_or("the certificate file holds no CA chain after the leaf to verify peers with")?;
		let mut roots = RootCertStore::empty();
		roots
		.add(issuing_ca.clone())
		.map_err(\|e\| format!("invalid CA certificate: {e}"))?;
		let roots = Arc::new(roots);
		let provider = crypto_provider();
		Ok(Self {
		own,
		client_verifier: WebPkiClientVerifier::builder_with_provider(
		roots.clone(),
		provider.clone(),
		)
		.build()
		.map_err(\|e\| format!("peer certificate verifier: {e}"))?,
		roots,
		provider,
		})
		}

		fn check_identity(&self, end_entity: &CertificateDer<'_>) -> Result<(), rustls::Error> {
		let caller = Caller::from_certificate(end_entity);
		if caller == self.own {
		Ok(())
		} else {
		Err(rustls::Error::General(format!(
		"peer certificate identifies {caller:?}, not this pipeline ({:?})",
		self.own
		)))
		}
		}
		}

		impl ServerCertVerifier for PipelinePeerVerifier {
		/// Does what rustls's `WebPkiServerVerifier` does, which checks the chain to the
		/// private CA, the validity period and the server auth extended key usage, except
		/// that it checks the pipeline identity instead of `server_name`. Like
		/// `WebPkiServerVerifier` without CRLs, which Feldera does not use, it does not
		/// check revocation, and it ignores OCSP responses.
		fn verify_server_cert(
		&self,
		end_entity: &CertificateDer<'_>,
		intermediates: &[CertificateDer<'_>],
		_server_name: &ServerName<'_>,
		_ocsp_response: &[u8],
		now: UnixTime,
		) -> Result<ServerCertVerified, rustls::Error> {
		verify_server_cert_signed_by_trust_anchor(
		&ParsedCertificate::try_from(end_entity)?,
		&self.roots,
		intermediates,
		now,
		self.provider.signature_verification_algorithms.all,
		)?;
		self.check_identity(end_entity)?;
		Ok(ServerCertVerified::assertion())
		}

		fn verify_tls12_signature(
		&self,
		message: &[u8],
		cert: &CertificateDer<'_>,
		dss: &DigitallySignedStruct,
		) -> Result<HandshakeSignatureValid, rustls::Error> {
		verify_tls12_signature(
Comment thread Outdated gz Oct 30, 2026 Copy link Copy Markdown Contributor There was a problem hiding this comment. Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. maybe we can just block 1.2? Sorry, something went wrong. Uh oh! There was an error while loading. Please reload this page. All reactions blp Oct 30, 2026 Copy link Copy Markdown Member Author There was a problem hiding this comment. Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. Done in `2b9fff0`. The configurations already negotiate only TLS 1.3, but the rustls traits require `verify_tls12_signature`. Both verifiers now refuse in it instead of checking the signature, so TLS 1.2 is blocked twice. `peer_tls_refuses_tls12` checks that a host refuses TLS 1.2 as client and as server. When I let the configurations allow TLS 1.2, the test failed, and the handshake stopped at the new refusal in the verifier. Sorry, something went wrong. Uh oh! There was an error while loading. Please reload this page. All reactions
		message,
		cert,
		dss,
		&self.provider.signature_verification_algorithms,
		)
		}

		fn verify_tls13_signature(
		&self,
		message: &[u8],
		cert: &CertificateDer<'_>,
		dss: &DigitallySignedStruct,
		) -> Result<HandshakeSignatureValid, rustls::Error> {
		verify_tls13_signature(
		message,
		cert,
		dss,
		&self.provider.signature_verification_algorithms,
		)
		}

		fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
		self.provider
		.signature_verification_algorithms
		.supported_schemes()
		}
		}

		impl ClientCertVerifier for PipelinePeerVerifier {
		fn root_hint_subjects(&self) -> &[DistinguishedName] {
		self.client_verifier.root_hint_subjects()
		}

		fn verify_client_cert(
		&self,
		end_entity: &CertificateDer<'_>,
		intermediates: &[CertificateDer<'_>],
		now: UnixTime,
		) -> Result<ClientCertVerified, rustls::Error> {
		let verified = self
		.client_verifier
		.verify_client_cert(end_entity, intermediates, now)?;
		self.check_identity(end_entity)?;
		Ok(verified)
		}

		fn verify_tls12_signature(
		&self,
		message: &[u8],
		cert: &CertificateDer<'_>,
		dss: &DigitallySignedStruct,
		) -> Result<HandshakeSignatureValid, rustls::Error> {
		self.client_verifier
Comment thread Outdated gz Oct 30, 2026 Copy link Copy Markdown Contributor There was a problem hiding this comment. Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. same q Sorry, something went wrong. Uh oh! There was an error while loading. Please reload this page. All reactions blp Oct 30, 2026 Copy link Copy Markdown Member Author There was a problem hiding this comment. Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. Same change, in `2b9fff0`: the client certificate verifier refuses TLS 1.2 signatures too. Sorry, something went wrong. Uh oh! There was an error while loading. Please reload this page. All reactions
		.verify_tls12_signature(message, cert, dss)
		}

		fn verify_tls13_signature(
		&self,
		message: &[u8],
		cert: &CertificateDer<'_>,
		dss: &DigitallySignedStruct,
		) -> Result<HandshakeSignatureValid, rustls::Error> {
		self.client_verifier
		.verify_tls13_signature(message, cert, dss)
		}

		fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
		self.client_verifier.supported_verify_schemes()
		}
		}

		#[cfg(test)]
		mod tests {
		use super::crypto_provider;
		use super::{
		CONTROL_PLANE_CERTIFICATE_URI, Caller, PipelineClientAuth, client_certificate_verifier,
		pipeline_certificate_uri,
		CONTROL_PLANE_CERTIFICATE_URI, Caller, PipelineClientAuth, PipelinePeerTls,
		PipelinePeerVerifier, client_certificate_verifier, pipeline_certificate_uri,
		};
		use rcgen::{
		BasicConstraints, CertificateParams, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer,
		KeyPair, SanType,
		};
		use rustls::pki_types::{CertificateDer, UnixTime};
		use rustls::pki_types::{
		CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName, UnixTime,
		};
		use rustls::{ClientConfig, ClientConnection, ServerConfig, ServerConnection};
		use std::sync::Arc;
		use uuid::Uuid;

		const PIPELINE_A: Uuid = Uuid::from_u128(0xa);
Expand Down Expand Up		@@ -456,4 +670,147 @@ mod tests {
		);
		}
		}

		/// A pipeline certificate chain with server and client auth, as the runner issues it: the
		/// leaf, then `ca_chain`.
		fn peer_chain(
		issuer: &Issuer<'static, KeyPair>,
		ca_chain: &[CertificateDer<'static>],
		uri: &str,
		) -> (Vec<CertificateDer<'static>>, PrivateKeyDer<'static>) {
		let mut params = CertificateParams::default();
		params.subject_alt_names = vec![SanType::URI(uri.try_into().unwrap())];
		params.extended_key_usages = vec![
		ExtendedKeyUsagePurpose::ServerAuth,
		ExtendedKeyUsagePurpose::ClientAuth,
		];
		let key = KeyPair::generate().unwrap();
		let leaf = params.signed_by(&key, issuer).unwrap().der().clone();
		let chain = std::iter::once(leaf)
		.chain(ca_chain.iter().cloned())
		.collect();
		(chain, PrivatePkcs8KeyDer::from(key.serialize_der()).into())
		}

		fn peer_tls(
		issuer: &Issuer<'static, KeyPair>,
		ca_chain: &[CertificateDer<'static>],
		uri: &str,
		) -> PipelinePeerTls {
		let (chain, key) = peer_chain(issuer, ca_chain, uri);
		PipelinePeerTls::from_chain(chain, key).unwrap()
		}

		/// Runs a TLS handshake between `client` and `server` in memory.
		fn handshake(
		client: Arc<ClientConfig>,
		server: Arc<ServerConfig>,
		) -> Result<(), rustls::Error> {
		let mut client = ClientConnection::new(client, ServerName::try_from("127.0.0.1").unwrap())?;
		let mut server = ServerConnection::new(server)?;
		for _ in 0..10 {
		let mut buf = Vec::new();
		while client.wants_write() {
		client.write_tls(&mut buf).unwrap();
		}
		server.read_tls(&mut buf.as_slice()).unwrap();
		server.process_new_packets()?;

		let mut buf = Vec::new();
		while server.wants_write() {
		server.write_tls(&mut buf).unwrap();
		}
		client.read_tls(&mut buf.as_slice()).unwrap();
		client.process_new_packets()?;

		if !client.is_handshaking() && !server.is_handshaking() {
		return Ok(());
		}
		}
		panic!("handshake did not finish");
		}

		/// Hosts of one pipeline connect to each other. A host refuses a peer that is another
		/// pipeline, the control plane, has no certificate, or has a certificate from a CA other
		/// than the private CA, as client and as server.
		#[test]
		fn peer_tls_admits_only_own_pipeline() {
		let (root, root_cert) = named_ca("root", None);
		let (private_ca, private_ca_cert) = named_ca("private", Some(&root));
		let (other_ca, other_ca_cert) = named_ca("other", Some(&root));
		let ca_chain = [private_ca_cert, root_cert.clone()];
		let a = pipeline_certificate_uri(PIPELINE_A);
		let b = pipeline_certificate_uri(PIPELINE_B);

		let host0 = peer_tls(&private_ca, &ca_chain, &a);
		let host1 = peer_tls(&private_ca, &ca_chain, &a);
		handshake(host0.client.clone(), host1.server.clone()).unwrap();
		handshake(host1.client.clone(), host0.server.clone()).unwrap();

		// A client that admits host 0 as its server, so that only host 0 can refuse. It
		// needs `dangerous()` for the same reason as `PipelinePeerTls::from_chain`.
		let verifier = Arc::new(
		PipelinePeerVerifier::from_chain(&peer_chain(&private_ca, &ca_chain, &a).0).unwrap(),
		);
		let client_builder = \|\| {
		ClientConfig::builder_with_provider(crypto_provider())
		.with_protocol_versions(&[&rustls::version::TLS13])
		.unwrap()
		.dangerous()
		.with_custom_certificate_verifier(verifier.clone())
		};
		let anonymous = client_builder().with_no_client_auth();
		assert!(
		handshake(Arc::new(anonymous), host0.server.clone()).is_err(),
		"server admitted a client without a certificate"
		);

		let refused = [
		("another pipeline", peer_chain(&private_ca, &ca_chain, &b)),
		(
		"the control plane",
		peer_chain(&private_ca, &ca_chain, CONTROL_PLANE_CERTIFICATE_URI),
		),
		(
		"the same pipeline from a sibling CA",
		peer_chain(&other_ca, &[other_ca_cert, root_cert.clone()], &a),
		),
		(
		"the same pipeline from the root CA",
		peer_chain(&root, std::slice::from_ref(&root_cert), &a),
		),
		];
		for (name, (chain, key)) in refused {
		let client = client_builder()
		.with_client_auth_cert(chain.clone(), key.clone_key())
		.unwrap();
		assert!(
		handshake(Arc::new(client), host0.server.clone()).is_err(),
		"server admitted {name}"
		);

		// A server that requires no client certificate, so that only host 0 can refuse.
		let server = ServerConfig::builder_with_provider(crypto_provider())
		.with_protocol_versions(&[&rustls::version::TLS13])
		.unwrap()
		.with_no_client_auth()
		.with_single_cert(chain, key)
		.unwrap();
		assert!(
		handshake(host0.client.clone(), Arc::new(server)).is_err(),
		"client admitted {name}"
		);
		}
		}

		/// Peer configurations need a pipeline certificate followed by the CA chain.
		#[test]
		fn peer_tls_needs_pipeline_chain() {
		let (issuer, ca_cert) = ca();
		let a = pipeline_certificate_uri(PIPELINE_A);
		let (chain, key) = peer_chain(&issuer, &[], &a);
		assert!(PipelinePeerTls::from_chain(chain, key).is_err());
		let (chain, key) = peer_chain(&issuer, &[ca_cert], CONTROL_PLANE_CERTIFICATE_URI);
		assert!(PipelinePeerTls::from_chain(chain, key).is_err());
		}
		}

Add this suggestion to a batch that can be applied as a single commit. This suggestion is invalid because no changes were made to the code. Suggestions cannot be applied while the pull request is closed. Suggestions cannot be applied while viewing a subset of changes. Only one suggestion per line can be applied in a batch. Add this suggestion to a batch that can be applied as a single commit. Applying suggestions on deleted lines is not supported. You must change the existing code in this line in order to create a valid suggestion. Outdated suggestions cannot be applied. This suggestion has been applied or marked resolved. Suggestions cannot be applied from pending reviews. Suggestions cannot be applied on multi-line comments. Suggestions cannot be applied while the pull request is queued to merge. Suggestion cannot be applied right now. Please check back later.

You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.

GeezerVison

[dbsp] Exchange data between hosts over TLS with client certificates #7713

Uh oh!

[dbsp] Exchange data between hosts over TLS with client certificates #7713

Uh oh!

Uh oh!

Diff view

Diff view

Uh oh!

There are no files selected for viewing

gz Oct 30, 2026

Choose a reason for hiding this comment

Uh oh!

blp Oct 30, 2026

Choose a reason for hiding this comment

Uh oh!

gz Oct 30, 2026

Choose a reason for hiding this comment

Uh oh!

blp Oct 30, 2026

Choose a reason for hiding this comment

Uh oh!

gz Oct 30, 2026

Choose a reason for hiding this comment

Uh oh!

blp Oct 30, 2026

Choose a reason for hiding this comment

Uh oh!