feat(ACL): Add liveliness token filters

DEFAULT_CONFIG.json5

@@ -225,6 +225,7 @@
   //       "messages": [
   //         "put", "delete", "declare_subscriber",
   //         "query", "reply", "declare_queryable",
+  //         "liveliness_token", "liveliness_query", "declare_liveliness_subscriber",
   //       ],
   //       "flows":["egress","ingress"],
   //       "permission": "allow",

commons/zenoh-config/src/lib.rs

@@ -170,6 +170,9 @@ pub enum AclMessage {
     Query,
     DeclareQueryable,
     Reply,
+    LivelinessToken,
+    DeclareLivelinessSubscriber,
+    LivelinessQuery,
 }
 
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, Hash, PartialEq)]

zenoh/src/api/session.rs

@@ -1252,7 +1252,9 @@ impl SessionInner {
                     primitives.send_interest(Interest {
                         id: pub_state.remote_id,
                         mode: InterestMode::Final,
-                        options: InterestOptions::empty(),
+                        // Note: InterestMode::Final options are undefined in the current protocol specification,
+                        //       they are initialized here for internal use by local egress interceptors.
+                        options: InterestOptions::SUBSCRIBERS,
                         wire_expr: None,
                         ext_qos: declare::ext::QoSType::DEFAULT,
                         ext_tstamp: None,
@@ -1458,7 +1460,9 @@ impl SessionInner {
                     primitives.send_interest(Interest {
                         id: sub_state.id,
                         mode: InterestMode::Final,
-                        options: InterestOptions::empty(),
+                        // Note: InterestMode::Final options are undefined in the current protocol specification,
+                        //       they are initialized here for internal use by local egress interceptors.
+                        options: InterestOptions::TOKENS,
                         wire_expr: None,
                         ext_qos: declare::ext::QoSType::DEFAULT,
                         ext_tstamp: None,

zenoh/src/net/routing/dispatcher/token.rs

@@ -156,6 +156,7 @@ pub(crate) fn undeclare_token(
     {
         tracing::debug!("{} Undeclare token {} ({})", face, id, res.expr());
     } else {
-        tracing::error!("{} Undeclare unknown token {}", face, id);
+        // NOTE: This is expected behavior if liveliness tokens are denied with ingress ACL interceptor.
+        tracing::debug!("{} Undeclare unknown token {}", face, id);
     }
 }

zenoh/src/net/routing/hat/client/interests.rs

@@ -212,7 +212,9 @@ impl HatInterestTrait for HatCode {
                                 Interest {
                                     id,
                                     mode: InterestMode::Final,
-                                    options: InterestOptions::empty(),
+                                    // Note: InterestMode::Final options are undefined in the current protocol specification,
+                                    //       they are initialized here for internal use by local egress interceptors.
+                                    options: interest.options,
                                     wire_expr: None,
                                     ext_qos: ext::QoSType::DECLARE,
                                     ext_tstamp: None,

zenoh/src/net/routing/hat/p2p_peer/interests.rs

@@ -238,7 +238,9 @@ impl HatInterestTrait for HatCode {
                                 Interest {
                                     id,
                                     mode: InterestMode::Final,
-                                    options: InterestOptions::empty(),
+                                    // Note: InterestMode::Final options are undefined in the current protocol specification,
+                                    //       they are initialized here for internal use by local egress interceptors.
+                                    options: interest.options,
                                     wire_expr: None,
                                     ext_qos: ext::QoSType::DECLARE,
                                     ext_tstamp: None,

zenoh/src/net/routing/interceptor/access_control.rs

@@ -26,7 +26,10 @@ use zenoh_config::{
 };
 use zenoh_protocol::{
     core::ZenohIdProto,
-    network::{Declare, DeclareBody, NetworkBody, NetworkMessage, Push, Request, Response},
+    network::{
+        interest::InterestMode, Declare, DeclareBody, Interest, NetworkBody, NetworkMessage, Push,
+        Request, Response,
+    },
     zenoh::{PushBody, RequestBody},
 };
 use zenoh_result::ZResult;
@@ -334,6 +337,75 @@ impl InterceptorTrait for IngressAclEnforcer {
                     }
                 }
             }
+            NetworkBody::Declare(Declare {
+                body: DeclareBody::DeclareToken(_),
+                ..
+            }) => {
+                if self.action(
+                    AclMessage::LivelinessToken,
+                    "Liveliness Token (ingress)",
+                    key_expr?,
+                ) == Permission::Deny
+                {
+                    return None;
+                }
+            }
+
+            NetworkBody::Declare(Declare {
+                body: DeclareBody::UndeclareToken(_),
+                ..
+            }) => {
+                // Undeclaration filtering diverges between ingress and egress:
+                // Undeclarations in ingress are only filtered if the ext_wire_expr is set.
+                // If it's not set, we let the undeclaration pass, it will be rejected by the routing logic
+                // if its associated declaration was denied.
+                if let Some(key_expr) = key_expr {
+                    if !key_expr.is_empty()
+                        && self.action(
+                            AclMessage::LivelinessToken,
+                            "Undeclare Liveliness Token (ingress)",
+                            key_expr,
+                        ) == Permission::Deny
+                    {
+                        return None;
+                    }
+                }
+            }
+            NetworkBody::Interest(Interest {
+                mode: InterestMode::Current,
+                options,
+                ..
+            }) if options.tokens() => {
+                if self.action(
+                    AclMessage::LivelinessQuery,
+                    "Liveliness Query (ingress)",
+                    key_expr?,
+                ) == Permission::Deny
+                {
+                    return None;
+                }
+            }
+            NetworkBody::Interest(Interest {
+                mode: InterestMode::Future | InterestMode::CurrentFuture,
+                options,
+                ..
+            }) if options.tokens() => {
+                if self.action(
+                    AclMessage::DeclareLivelinessSubscriber,
+                    "Declare Liveliness Subscriber (ingress)",
+                    key_expr?,
+                ) == Permission::Deny
+                {
+                    return None;
+                }
+            }
+            NetworkBody::Interest(Interest {
+                mode: InterestMode::Final,
+                ..
+            }) => {
+                // InterestMode::Final filtering diverges between ingress and egress:
+                // InterestMode::Final ingress is always allowed, it will be rejected by routing logic if its associated Interest was denied
+            }
             // Unfiltered Declare messages
             NetworkBody::Declare(Declare {
                 body: DeclareBody::DeclareKeyExpr(_),
@@ -342,19 +414,11 @@ impl InterceptorTrait for IngressAclEnforcer {
             | NetworkBody::Declare(Declare {
                 body: DeclareBody::DeclareFinal(_),
                 ..
-            })
-            | NetworkBody::Declare(Declare {
-                body: DeclareBody::DeclareToken(_),
-                ..
             }) => {}
             // Unfiltered Undeclare messages
             NetworkBody::Declare(Declare {
                 body: DeclareBody::UndeclareKeyExpr(_),
                 ..
-            })
-            | NetworkBody::Declare(Declare {
-                body: DeclareBody::UndeclareToken(_),
-                ..
             }) => {}
             // Unfiltered remaining message types
             NetworkBody::Interest(_) | NetworkBody::OAM(_) | NetworkBody::ResponseFinal(_) => {}
@@ -470,6 +534,80 @@ impl InterceptorTrait for EgressAclEnforcer {
                     return None;
                 }
             }
+            NetworkBody::Declare(Declare {
+                body: DeclareBody::DeclareToken(_),
+                ..
+            }) => {
+                if self.action(
+                    AclMessage::LivelinessToken,
+                    "Liveliness Token (egress)",
+                    key_expr?,
+                ) == Permission::Deny
+                {
+                    return None;
+                }
+            }
+            NetworkBody::Declare(Declare {
+                body: DeclareBody::UndeclareToken(_),
+                ..
+            }) => {
+                // Undeclaration filtering diverges between ingress and egress:
+                // in egress the keyexpr has to be provided in the RoutingContext
+                if self.action(
+                    AclMessage::LivelinessToken,
+                    "Undeclare Liveliness Token (egress)",
+                    key_expr?,
+                ) == Permission::Deny
+                {
+                    return None;
+                }
+            }
+            NetworkBody::Interest(Interest {
+                mode: InterestMode::Current,
+                options,
+                ..
+            }) if options.tokens() => {
+                if self.action(
+                    AclMessage::LivelinessQuery,
+                    "Liveliness Query (egress)",
+                    key_expr?,
+                ) == Permission::Deny
+                {
+                    return None;
+                }
+            }
+            NetworkBody::Interest(Interest {
+                mode: InterestMode::Future | InterestMode::CurrentFuture,
+                options,
+                ..
+            }) if options.tokens() => {
+                if self.action(
+                    AclMessage::DeclareLivelinessSubscriber,
+                    "Declare Liveliness Subscriber (egress)",
+                    key_expr?,
+                ) == Permission::Deny
+                {
+                    return None;
+                }
+            }
+            NetworkBody::Interest(Interest {
+                mode: InterestMode::Final,
+                options,
+                ..
+            }) if options.tokens() => {
+                // Note: options are set for InterestMode::Final for internal use only by egress interceptors.
+
+                // InterestMode::Final filtering diverges between ingress and egress:
+                // in egress the keyexpr has to be provided in the RoutingContext
+                if self.action(
+                    AclMessage::DeclareLivelinessSubscriber,
+                    "Undeclare Liveliness Subscriber (egress)",
+                    key_expr?,
+                ) == Permission::Deny
+                {
+                    return None;
+                }
+            }
             // Unfiltered Declare messages
             NetworkBody::Declare(Declare {
                 body: DeclareBody::DeclareKeyExpr(_),
@@ -478,19 +616,11 @@ impl InterceptorTrait for EgressAclEnforcer {
             | NetworkBody::Declare(Declare {
                 body: DeclareBody::DeclareFinal(_),
                 ..
-            })
-            | NetworkBody::Declare(Declare {
-                body: DeclareBody::DeclareToken(_),
-                ..
             }) => {}
             // Unfiltered Undeclare messages
             NetworkBody::Declare(Declare {
                 body: DeclareBody::UndeclareKeyExpr(_),
                 ..
-            })
-            | NetworkBody::Declare(Declare {
-                body: DeclareBody::UndeclareToken(_),
-                ..
             }) => {}
             // Unfiltered remaining message types
             NetworkBody::Interest(_) | NetworkBody::OAM(_) | NetworkBody::ResponseFinal(_) => {}

zenoh/src/net/routing/interceptor/authorization.rs

@@ -188,6 +188,9 @@ struct ActionPolicy {
     declare_subscriber: PermissionPolicy,
     declare_queryable: PermissionPolicy,
     reply: PermissionPolicy,
+    liveliness_token: PermissionPolicy,
+    declare_liveliness_sub: PermissionPolicy,
+    liveliness_query: PermissionPolicy,
 }
 
 impl ActionPolicy {
@@ -199,6 +202,9 @@ impl ActionPolicy {
             AclMessage::Delete => &self.delete,
             AclMessage::DeclareSubscriber => &self.declare_subscriber,
             AclMessage::DeclareQueryable => &self.declare_queryable,
+            AclMessage::LivelinessToken => &self.liveliness_token,
+            AclMessage::DeclareLivelinessSubscriber => &self.declare_liveliness_sub,
+            AclMessage::LivelinessQuery => &self.liveliness_query,
         }
     }
     fn action_mut(&mut self, action: AclMessage) -> &mut PermissionPolicy {
@@ -209,6 +215,9 @@ impl ActionPolicy {
             AclMessage::Delete => &mut self.delete,
             AclMessage::DeclareSubscriber => &mut self.declare_subscriber,
             AclMessage::DeclareQueryable => &mut self.declare_queryable,
+            AclMessage::LivelinessToken => &mut self.liveliness_token,
+            AclMessage::DeclareLivelinessSubscriber => &mut self.declare_liveliness_sub,
+            AclMessage::LivelinessQuery => &mut self.liveliness_query,
         }
     }
 }