Close TLS and QUIC links on certificate expiration

Cargo.toml

@@ -167,6 +167,7 @@ socket2 = { version = "0.5.7", features = ["all"] }
 stop-token = "0.7.0"
 syn = "2.0"
 tide = "0.16.0"
+time = "0.3.36"
 token-cell = { version = "1.5.0", default-features = false }
 tokio = { version = "1.40.0", default-features = false } # Default features are disabled due to some crates' requirements
 tokio-util = "0.7.12"

DEFAULT_CONFIG.json5

@@ -466,6 +466,10 @@
         // This could be dangerous because your CA can have signed a server cert for foo.com, that's later being used to host a server at baz.com. If you wan't your
         // ca to verify that the server at baz.com is actually baz.com, let this be true (default).
         verify_name_on_connect: true,
+        // Whether or not to close links when remote certificates expires.
+        // If set to true, links that require certificates (tls/quic) will automatically disconnect when the time of expiration of the remote certificate chain is reached
+        // note that mTLS (client authentication) is required for a listener to disconnect a client on expiration
+        close_link_on_expiration: false,
       },
     },
     /// Shared memory configuration.

commons/zenoh-config/src/lib.rs

@@ -483,6 +483,7 @@ validated_struct::validator! {
                     connect_private_key: Option<String>,
                     connect_certificate: Option<String>,
                     verify_name_on_connect: Option<bool>,
+                    close_link_on_expiration: Option<bool>,
                     // Skip serializing field because they contain secrets
                     #[serde(skip_serializing)]
                     root_ca_certificate_base64: Option<SecretValue>,

io/zenoh-link-commons/Cargo.toml

@@ -35,6 +35,7 @@ futures = { workspace = true }
 rustls = { workspace = true, optional = true }
 rustls-webpki = { workspace = true, optional = true }
 serde = { workspace = true, features = ["default"] }
+time = { workspace = true }
 tokio = { workspace = true, features = [
   "fs",
   "io-util",

io/zenoh-link-commons/src/tls.rs

@@ -86,3 +86,124 @@ impl WebPkiVerifierAnyServerName {
         Self { roots }
     }
 }
+
+pub mod expiration {
+    use std::{
+        net::SocketAddr,
+        sync::{atomic::AtomicBool, Weak},
+    };
+
+    use async_trait::async_trait;
+    use time::OffsetDateTime;
+    use tokio::{sync::Mutex as AsyncMutex, task::JoinHandle};
+    use tokio_util::sync::CancellationToken;
+    use zenoh_result::ZResult;
+
+    #[async_trait]
+    pub trait LinkWithCertExpiration: Send + Sync {
+        async fn expire(&self) -> ZResult<()>;
+    }
+
+    #[derive(Debug)]
+    pub struct LinkCertExpirationManager {
+        token: CancellationToken,
+        handle: AsyncMutex<Option<JoinHandle<ZResult<()>>>>,
+        /// Closing the link is a critical section that requires exclusive access to expiration_task
+        /// or the transport. `link_closing` is used to synchronize the access to this operation.
+        link_closing: AtomicBool,
+    }
+
+    impl LinkCertExpirationManager {
+        pub fn new(
+            link: Weak<dyn LinkWithCertExpiration>,
+            src_addr: SocketAddr,
+            dst_addr: SocketAddr,
+            link_type: &'static str,
+            expiration_time: OffsetDateTime,
+        ) -> Self {
+            let token = CancellationToken::new();
+            let handle = zenoh_runtime::ZRuntime::Acceptor.spawn(expiration_task(
+                link,
+                src_addr,
+                dst_addr,
+                link_type,
+                expiration_time,
+                token.clone(),
+            ));
+            Self {
+                token,
+                handle: AsyncMutex::new(Some(handle)),
+                link_closing: AtomicBool::new(false),
+            }
+        }
+
+        /// Takes exclusive access to closing the link.
+        ///
+        /// Returns `true` if successful, `false` if another task is (or finished) closing the link.
+        pub fn set_closing(&self) -> bool {
+            !self
+                .link_closing
+                .swap(true, std::sync::atomic::Ordering::Relaxed)
+        }
+
+        /// Sends cancelation signal to expiration_task
+        pub fn cancel_expiration_task(&self) {
+            self.token.cancel()
+        }
+
+        /// Waits for expiration task to complete, returning its return value.
+        pub async fn wait_for_expiration_task(&self) -> ZResult<()> {
+            let mut lock = self.handle.lock().await;
+            let handle = lock.take().expect("handle should be set");
+            handle.await?
+        }
+    }
+
+    async fn expiration_task(
+        link: Weak<dyn LinkWithCertExpiration>,
+        src_addr: SocketAddr,
+        dst_addr: SocketAddr,
+        link_type: &'static str,
+        expiration_time: OffsetDateTime,
+        token: CancellationToken,
+    ) -> ZResult<()> {
+        tracing::trace!(
+            "Expiration task started for {} link {:?} => {:?}",
+            link_type.to_uppercase(),
+            src_addr,
+            dst_addr,
+        );
+        tokio::select! {
+            _ = token.cancelled() => {},
+            _ = sleep_until_date(expiration_time) => {
+                // expire the link
+                if let Some(link) = link.upgrade() {
+                    tracing::warn!(
+                        "Closing {} link {:?} => {:?} : remote certificate chain expired",
+                        link_type.to_uppercase(),
+                        src_addr,
+                        dst_addr,
+                    );
+                    return link.expire().await;
+                }
+            },
+        }
+        Ok(())
+    }
+
+    async fn sleep_until_date(wakeup_time: OffsetDateTime) {
+        const MAX_SLEEP_DURATION: tokio::time::Duration = tokio::time::Duration::from_secs(600);
+        loop {
+            let now = OffsetDateTime::now_utc();
+            if wakeup_time <= now {
+                break;
+            }
+            // next sleep duration is the minimum between MAX_SLEEP_DURATION and the duration till wakeup
+            // this mitigates the unsoundness of using `tokio::time::sleep` with long durations
+            let wakeup_duration = std::time::Duration::try_from(wakeup_time - now)
+                .expect("wakeup_time should be greater than now");
+            let sleep_duration = tokio::time::Duration::min(MAX_SLEEP_DURATION, wakeup_duration);
+            tokio::time::sleep(sleep_duration).await;
+        }
+    }
+}

io/zenoh-links/zenoh-link-quic/Cargo.toml

@@ -33,6 +33,7 @@ rustls-pemfile = { workspace = true }
 rustls-pki-types = { workspace = true }
 rustls-webpki = { workspace = true }
 secrecy = { workspace = true }
+time = { workspace = true }
 tokio = { workspace = true, features = [
   "fs",
   "io-util",

io/zenoh-links/zenoh-link-quic/src/lib.rs

@@ -112,4 +112,7 @@ pub mod config {
 
     pub const TLS_VERIFY_NAME_ON_CONNECT: &str = "verify_name_on_connect";
     pub const TLS_VERIFY_NAME_ON_CONNECT_DEFAULT: bool = true;
+
+    pub const TLS_CLOSE_LINK_ON_EXPIRATION: &str = "close_link_on_expiration";
+    pub const TLS_CLOSE_LINK_ON_EXPIRATION_DEFAULT: bool = false;
 }