used shared helper for setting up Edge connections

Author: wojcik91

Cargo.lock

@@ -16,6 +16,7 @@ defguard_web_ui = { workspace = true }
 defguard_version = { workspace = true }
 model_derive = { workspace = true }
 defguard_certs = { workspace = true }
+defguard_grpc_tls = { workspace = true }
 defguard_static_ip = { workspace = true }
 
 # external dependencies
@@ -30,6 +31,7 @@ bytes = { workspace = true }
 chrono = { workspace = true }
 futures = { workspace = true }
 humantime = { workspace = true }
+hyper-rustls = { workspace = true }
 # match version used by sqlx
 ipnetwork = { workspace = true }
 jsonwebkey = { workspace = true }

crates/defguard_core/Cargo.toml

@@ -1,5 +1,5 @@
 use std::{
-    collections::VecDeque,
+    collections::{HashMap, VecDeque},
     convert::Infallible,
     sync::{Arc, Mutex, PoisonError},
     time::Duration,
@@ -29,6 +29,7 @@ use defguard_common::{
     types::proxy::ProxyControlMessage,
     utils::strip_scheme,
 };
+use defguard_grpc_tls::certs::proxy_mtls_channel;
 use defguard_proto::{
     common::{CertBundle, CertificateInfo},
     gateway::gateway_setup_client::GatewaySetupClient,
@@ -43,7 +44,10 @@ use reqwest::Url;
 use serde::{Deserialize, Serialize};
 use sqlx::PgPool;
 use tokio::{
-    sync::mpsc::{Sender, UnboundedReceiver, UnboundedSender, unbounded_channel},
+    sync::{
+        mpsc::{Sender, UnboundedReceiver, UnboundedSender, unbounded_channel},
+        oneshot, watch,
+    },
     time::{Instant, sleep_until, timeout},
 };
 use tokio_stream::StreamExt;
@@ -1175,8 +1179,7 @@ fn public_proxy_hostname() -> Result<String, String> {
 /// collected during the ACME run (sent by the proxy via an [`AcmeLogs`] event).
 async fn call_proxy_trigger_acme(
     pool: &PgPool,
-    proxy_host: &str,
-    proxy_port: u16,
+    proxy: &Proxy<Id>,
     domain: String,
     account_credentials_json: String,
     progress_tx: UnboundedSender<AcmeStep>,
@@ -1191,32 +1194,29 @@ async fn call_proxy_trigger_acme(
         )
     })?;
 
-    let cert_pem = der_to_pem(&ca_cert_der, defguard_certs::PemLabel::Certificate)
-        .map_err(|e| (format!("Failed to convert CA cert to PEM: {e}"), Vec::new()))?;
-
-    let endpoint_str = format!("https://{proxy_host}:{proxy_port}");
-    let endpoint = Endpoint::from_shared(endpoint_str)
-        .map_err(|e| (format!("Failed to build Edge endpoint: {e}"), Vec::new()))?
-        .http2_keep_alive_interval(Duration::from_secs(5))
-        .tcp_keepalive(Some(Duration::from_secs(5)))
-        .keep_alive_while_idle(true);
-
-    let tls = ClientTlsConfig::new().ca_certificate(Certificate::from_pem(cert_pem));
-    let endpoint = endpoint.tls_config(tls).map_err(|e| {
+    let cert_serial = proxy.certificate_serial.as_deref().ok_or_else(|| {
         (
-            format!("Failed to configure TLS for Edge endpoint: {e}"),
+            "Edge certificate serial not provisioned".to_string(),
             Vec::new(),
         )
     })?;
 
+    // Seed a one-shot serial map so the rustls verifier validates the server cert serial.
+    let (_, certs_rx) = watch::channel(Arc::new(HashMap::from([(
+        proxy.id,
+        cert_serial.to_string(),
+    )])));
+
+    let channel = proxy_mtls_channel(proxy, &ca_cert_der, certs_rx)
+        .map_err(|e| (format!("Failed to build mTLS channel: {e}"), Vec::new()))?;
+
     let version = Version::parse(VERSION)
         .map_err(|e| (format!("Failed to parse core version: {e}"), Vec::new()))?;
     let version_interceptor = ClientVersionInterceptor::new(version);
 
-    let mut client =
-        ProxyClient::with_interceptor(endpoint.connect_lazy(), move |req: Request<()>| {
-            version_interceptor.clone().call(req)
-        });
+    let mut client = ProxyClient::with_interceptor(channel, move |req: Request<()>| {
+        version_interceptor.clone().call(req)
+    });
 
     let mut stream = client
         .trigger_acme(AcmeChallenge {
@@ -1300,7 +1300,7 @@ pub async fn stream_proxy_acme(
 
         let account_credentials_json = certs.acme_account_credentials.clone().unwrap_or_default();
 
-        let proxies = match Proxy::list(&pool).await {
+        let proxies = match Proxy::all_enabled(&pool).await {
             Ok(list) => list,
             Err(e) => {
                 yield Ok(acme_error_event(
@@ -1323,8 +1323,8 @@ pub async fn stream_proxy_acme(
             return;
         };
 
-        let proxy_host = proxy.address.clone();
-        let proxy_port = proxy.port as u16;
+        let proxy_host = &proxy.address;
+        let proxy_port = proxy.port;
         info!(
             "Triggering ACME HTTP-01 via Edge gRPC TriggerAcme for domain: {domain} \
              Edge={proxy_host}:{proxy_port}"
@@ -1333,16 +1333,15 @@ pub async fn stream_proxy_acme(
         let (progress_tx, mut progress_rx) =
             unbounded_channel::<AcmeStep>();
         let (result_tx, result_rx) =
-            tokio::sync::oneshot::channel::<Result<(String, String, String), (String, Vec<String>)>>();
+            oneshot::channel::<Result<(String, String, String), (String, Vec<String>)>>();
 
         let pool_clone = pool.clone();
         let domain_clone = domain.clone();
         let acct_creds_clone = account_credentials_json.clone();
         tokio::spawn(async move {
             let result = call_proxy_trigger_acme(
                 &pool_clone,
-                &proxy_host,
-                proxy_port,
+                &proxy,
                 domain_clone,
                 acct_creds_clone,
                 progress_tx,

crates/defguard_core/src/handlers/component_setup.rs

@@ -10,6 +10,7 @@ rust-version.workspace = true
 [dependencies]
 defguard_common.workspace = true
 http = "1.1"
+hyper-rustls.workspace = true
 rustls = { version = "0.23", features = ["ring"] }
 thiserror.workspace = true
 tokio.workspace = true

crates/defguard_grpc_tls/Cargo.toml

@@ -8,9 +8,10 @@
 //! - A lightweight in-memory cache (refreshed periodically) avoids database access
 //!   during the handshake and keeps verification synchronous.
 
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 
-use defguard_common::db::Id;
+use defguard_common::db::{Id, models::proxy::Proxy};
+use hyper_rustls::HttpsConnectorBuilder;
 use rustls::{
     CertificateError, DistinguishedName, Error as RustlsError, RootCertStore, SignatureScheme,
     client::{
@@ -22,10 +23,14 @@ use rustls::{
 };
 use thiserror::Error;
 use tokio::sync::watch;
-use tonic::transport::{Certificate, Identity, ServerTlsConfig};
+use tonic::transport::{Certificate, Channel, Endpoint, Identity, ServerTlsConfig};
 use tracing::error;
 use x509_parser::parse_x509_certificate;
 
+use crate::connector::HttpsSchemeConnector;
+
+const TEN_SECS: Duration = Duration::from_secs(10);
+
 /// Errors that can occur while building a TLS config with a pinned verifier.
 #[derive(Debug, Error)]
 pub enum CertConfigError {
@@ -208,3 +213,53 @@ pub fn client_config(
         )));
     Ok(config)
 }
+
+/// Build an mTLS [`Channel`] to a proxy using its stored per-component client certificate.
+///
+/// * `proxy` — the full `Proxy<Id>` row from the database; `core_client_cert_der`,
+///   `core_client_cert_key_der`, and `certificate_serial` must all be `Some`.
+/// * `ca_cert_der` — the core CA certificate in DER form, used as the only trusted root.
+/// * `certs_rx` — watch channel carrying the current `{ proxy_id → cert_serial }` map.
+///   Pass a long-lived receiver for persistent connections (serial revocation is picked up
+///   dynamically) or a one-shot channel seeded with the proxy's current serial for
+///   short-lived calls.
+///
+/// The returned channel uses an `http://` endpoint scheme; TLS is applied by the
+/// internal [`HttpsSchemeConnector`](crate::connector::HttpsSchemeConnector).
+pub fn proxy_mtls_channel(
+    proxy: &Proxy<Id>,
+    ca_cert_der: &[u8],
+    certs_rx: watch::Receiver<Arc<HashMap<Id, String>>>,
+) -> Result<Channel, CertConfigError> {
+    let cert_der = proxy.core_client_cert_der.as_deref().ok_or_else(|| {
+        CertConfigError::TlsConfig(format!(
+            "core client certificate not provisioned for proxy id={}",
+            proxy.id
+        ))
+    })?;
+    let key_der = proxy.core_client_cert_key_der.as_deref().ok_or_else(|| {
+        CertConfigError::TlsConfig(format!(
+            "core client certificate key not provisioned for proxy id={}",
+            proxy.id
+        ))
+    })?;
+
+    let tls_config = client_config(ca_cert_der, certs_rx, proxy.id, cert_der, key_der)?;
+
+    let connector = HttpsConnectorBuilder::new()
+        .with_tls_config(tls_config)
+        .https_only()
+        .enable_http2()
+        .build();
+    let connector = HttpsSchemeConnector::new(connector);
+
+    // Use http:// scheme — the HttpsSchemeConnector rewrites it to https:// internally.
+    let endpoint_str = format!("http://{}:{}", proxy.address, proxy.port);
+    let endpoint = Endpoint::from_shared(endpoint_str)
+        .map_err(|e| CertConfigError::TlsConfig(format!("invalid proxy endpoint URL: {e}")))?
+        .http2_keep_alive_interval(TEN_SECS)
+        .tcp_keepalive(Some(TEN_SECS))
+        .keep_alive_while_idle(true);
+
+    Ok(endpoint.connect_with_connector_lazy(connector))
+}