[ZEN-530] Add link protocols subject to interceptors (#1850)

* Rework AuthId types

* Add InterceptorLink config enum

* Add link protocol subject to ACL

* Add ACL link_protocols tests

* Remove LinkAuthId::None variant

* Add link_protocols subject to Downsampling

* Update DEFAULT_CONFIG

* Address clippy warning

* Fix typo in enum variant

* Address clippy warning

* Remove outdated doc

* Use NEVec to avoid boilerplate config-checking code

* Support link-protocols in qos-overwrite

* clippy fix

* update DEFAULT_CONFIG.json

* Homogenize DEFAULT_CONFIG.json5 doc comments

* Update tracing

* Log line number when error parsing json config

---------

Co-authored-by: Denis Biryukov <[email protected]>
Co-authored-by: OlivierHecart <[email protected]>

Author: 3 people

Cargo.lock

@@ -121,6 +121,7 @@ libloading = "0.8"
 tracing = "0.1"
 lz4_flex = "0.11"
 nix = { version = "0.29.0", features = ["fs"] }
+nonempty-collections = {version = "0.3.0", features = ["serde"]}
 num_cpus = "1.16.0"
 num-traits = { version = "0.2.19", default-features = false }
 once_cell = "1.19.0"

Cargo.toml

@@ -261,6 +261,9 @@
   //          "lo0",
   //          "en0",
   //        ],
+  //        /// Optional list of link protocols. Transports with at least one of these links will have their qos overwritten.
+  //        /// If absent, the overwrite will be applied to all transports. An empty list is invalid.
+  //        link_protocols: [ "tcp", "udp", "tls", "quic", "ws", "serial", "unixsock-stream", "unixpipe", "vsock"],
   //        /// List of message types to apply to.
   //        messages: [
   //          "put", // put publications
@@ -312,8 +315,11 @@
   //      /// Optional Id, has to be unique
   //      "id": "wlan0egress",
   //      /// Optional list of network interfaces messages will be processed on, the rest will be passed as is.
-  //      /// If absent, the rules will be applied to all interfaces, in case of an empty list it means that they will not be applied to any.
+  //      /// If absent, the rules will be applied to all interfaces. An empty list is invalid.
   //      interfaces: [ "wlan0" ],
+  //      /// Optional list of link protocols. Transports with at least one of these links will have their messages filtered.
+  //      /// If absent, the rules will be applied to all transports. An empty list is invalid.
+  //      link_protocols: [ "tcp", "udp", "tls", "quic", "ws", "serial", "unixsock-stream", "unixpipe", "vsock"],
   //      /// Optional list of data flows messages will be processed on ("egress" and/or "ingress").
   //      /// If absent, the rules will be applied to both flows.
   //      flow: ["ingress", "egress"],
@@ -412,6 +418,12 @@
   //       "id": "subject3",
   //       /// An empty subject combination is a wildcard
   //     },
+  //     {
+  //       "id": "subject4",
+  //      /// link protocols can also be used to identify transports to filter messages on.
+  //      /// If absent, the rules will be applied to all transports. An empty list is invalid.
+  //      link_protocols: [ "tcp", "udp", "tls", "quic", "ws", "serial", "unixsock-stream", "unixpipe", "vsock"],
+  //     },
   //   ],
   //   /// The policies list associates rules to subjects
   //   "policies":
@@ -426,7 +438,7 @@
   //      },
   //      {
   //         "rules": ["rule2"],
-  //         "subjects": ["subject3"],
+  //         "subjects": ["subject3", "subject4"],
   //      },
   //   ]
   //},

DEFAULT_CONFIG.json5

@@ -37,6 +37,7 @@ serde_json = { workspace = true }
 serde_with = { workspace = true }
 serde_yaml = { workspace = true }
 validated_struct = { workspace = true, features = ["json5", "json_get"] }
+nonempty-collections = {workspace = true }
 zenoh-core = { workspace = true }
 zenoh-keyexpr = { workspace = true }
 zenoh-protocol = { workspace = true }

commons/zenoh-config/Cargo.toml

@@ -32,7 +32,8 @@ use std::{
 };
 
 use include::recursive_include;
-use qos::{PublisherQoSConfList, QosOverwriteItemConf};
+use nonempty_collections::NEVec;
+use qos::{PublisherQoSConfList, QosOverwriteMessage, QosOverwrites};
 use secrecy::{CloneableSecret, DebugSecret, Secret, SerializableSecret, Zeroize};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
@@ -111,30 +112,54 @@ pub struct DownsamplingItemConf {
     pub id: Option<String>,
     /// A list of interfaces to which the downsampling will be applied
     /// Downsampling will be applied for all interfaces if the parameter is None
-    pub interfaces: Option<Vec<String>>,
+    pub interfaces: Option<NEVec<String>>,
+    /// A list of link types, transports having one of those link types will have the downsampling applied
+    /// Downsampling will be applied for all link types if the parameter is None
+    pub link_protocols: Option<NEVec<InterceptorLink>>,
     // list of message types on which the downsampling will be applied
-    pub messages: Vec<DownsamplingMessage>,
-    /// A list of interfaces to which the downsampling will be applied.
-    pub rules: Vec<DownsamplingRuleConf>,
+    pub messages: NEVec<DownsamplingMessage>,
+    /// A list of downsampling rules: key_expression and the maximum frequency in Hertz
+    pub rules: NEVec<DownsamplingRuleConf>,
     /// Downsampling flow directions: egress and/or ingress
-    pub flows: Option<Vec<InterceptorFlow>>,
+    pub flows: Option<NEVec<InterceptorFlow>>,
 }
 
 #[derive(Serialize, Debug, Deserialize, Clone)]
 pub struct AclConfigRule {
     pub id: String,
-    pub key_exprs: Vec<String>,
-    pub messages: Vec<AclMessage>,
-    pub flows: Option<Vec<InterceptorFlow>>,
+    pub key_exprs: NEVec<OwnedKeyExpr>,
+    pub messages: NEVec<AclMessage>,
+    pub flows: Option<NEVec<InterceptorFlow>>,
     pub permission: Permission,
 }
 
 #[derive(Serialize, Debug, Deserialize, Clone)]
 pub struct AclConfigSubjects {
     pub id: String,
-    pub interfaces: Option<Vec<Interface>>,
-    pub cert_common_names: Option<Vec<CertCommonName>>,
-    pub usernames: Option<Vec<Username>>,
+    pub interfaces: Option<NEVec<Interface>>,
+    pub cert_common_names: Option<NEVec<CertCommonName>>,
+    pub usernames: Option<NEVec<Username>>,
+    pub link_protocols: Option<NEVec<InterceptorLink>>,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct QosOverwriteItemConf {
+    /// Optional identifier for the qos modification configuration item.
+    pub id: Option<String>,
+    /// A list of interfaces to which the qos will be applied.
+    /// QosOverwrite will be applied for all interfaces if the parameter is None.
+    pub interfaces: Option<NEVec<String>>,
+    /// A list of link types, transports having one of those link types will have the qos overwrite applied
+    /// Qos overwrite will be applied for all link types if the parameter is None.
+    pub link_protocols: Option<NEVec<InterceptorLink>>,
+    /// List of message types on which the qos overwrite will be applied.
+    pub messages: NEVec<QosOverwriteMessage>,
+    /// List of key expressions to apply qos overwrite.
+    pub key_exprs: Vec<OwnedKeyExpr>,
+    // The qos value to overwrite with.
+    pub overwrite: QosOverwrites,
+    /// QosOverwrite flow directions: egress and/or ingress.
+    pub flows: Option<NEVec<InterceptorFlow>>,
 }
 
 #[derive(Serialize, Debug, Deserialize, Clone, PartialEq, Eq, Hash)]
@@ -164,6 +189,26 @@ impl std::fmt::Display for Username {
     }
 }
 
+#[derive(Serialize, Debug, Deserialize, Clone, PartialEq, Eq, Hash)]
+#[serde(rename_all = "kebab-case")]
+pub enum InterceptorLink {
+    Tcp,
+    Udp,
+    Tls,
+    Quic,
+    Serial,
+    Unixpipe,
+    UnixsockStream,
+    Vsock,
+    Ws,
+}
+
+impl std::fmt::Display for InterceptorLink {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Transport({:?})", self)
+    }
+}
+
 #[derive(Serialize, Debug, Deserialize, Clone, PartialEq, Eq, Hash)]
 pub struct AclConfigPolicyEntry {
     pub id: Option<String>,
@@ -174,7 +219,7 @@ pub struct AclConfigPolicyEntry {
 #[derive(Clone, Serialize, Debug, Deserialize)]
 pub struct PolicyRule {
     pub subject_id: usize,
-    pub key_expr: String,
+    pub key_expr: OwnedKeyExpr,
     pub message: AclMessage,
     pub permission: Permission,
     pub flow: InterceptorFlow,
@@ -908,13 +953,13 @@ impl Config {
                     Some("json") | Some("json5") => match json5::Deserializer::from_str(&content) {
                         Ok(mut d) => Config::from_deserializer(&mut d).map_err(|e| match e {
                             Ok(c) => zerror!("Invalid configuration: {}", c).into(),
-                            Err(e) => zerror!("JSON error: {}", e).into(),
+                            Err(e) => zerror!("JSON error: {:?}", e).into(),
                         }),
                         Err(e) => bail!(e),
                     },
                     Some("yaml") | Some("yml") => Config::from_deserializer(serde_yaml::Deserializer::from_str(&content)).map_err(|e| match e {
                         Ok(c) => zerror!("Invalid configuration: {}", c).into(),
-                        Err(e) => zerror!("YAML error: {}", e).into(),
+                        Err(e) => zerror!("YAML error: {:?}", e).into(),
                     }),
                     Some(other) => bail!("Unsupported file type '.{}' (.json, .json5 and .yaml are supported)", other),
                     None => bail!("Unsupported file type. Configuration files must have an extension (.json, .json5 and .yaml supported)")

commons/zenoh-config/src/lib.rs

@@ -15,8 +15,6 @@ use serde::{Deserialize, Serialize};
 use zenoh_keyexpr::keyexpr_tree::{IKeyExprTreeMut, KeBoxTree};
 use zenoh_protocol::core::{key_expr::OwnedKeyExpr, CongestionControl, Priority, Reliability};
 
-use crate::InterceptorFlow;
-
 #[derive(Debug, Deserialize, Default, Serialize, Clone)]
 pub struct PublisherQoSConfList(pub(crate) Vec<PublisherQoSConf>);
 
@@ -134,23 +132,6 @@ pub enum PublisherLocalityConf {
     Any,
 }
 
-#[derive(Default, Debug, Deserialize, Serialize, Clone)]
-pub struct QosOverwriteItemConf {
-    /// Optional identifier for the qos modification configuration item.
-    pub id: Option<String>,
-    /// A list of interfaces to which the qos will be applied.
-    /// QosOverwrite will be applied for all interfaces if the parameter is None.
-    pub interfaces: Option<Vec<String>>,
-    /// List of message types on which the qos overwrite will be applied.
-    pub messages: Vec<QosOverwriteMessage>,
-    /// List of key expressions to apply qos overwrite.
-    pub key_exprs: Vec<OwnedKeyExpr>,
-    // The qos value to overwrite with.
-    pub overwrite: QosOverwrites,
-    /// QosOverwrite flow directions: egress and/or ingress.
-    pub flows: Option<Vec<InterceptorFlow>>,
-}
-
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, Hash, PartialEq)]
 #[serde(rename_all = "snake_case")]
 pub enum QosOverwriteMessage {

commons/zenoh-config/src/qos.rs

@@ -105,7 +105,7 @@ impl Link {
             mtu: link.get_mtu(),
             is_streamed: false,
             interfaces: vec![],
-            auth_identifier: LinkAuthId::default(),
+            auth_identifier: link.get_auth_id().clone(),
             priorities: None,
             reliability: None,
         }

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

@@ -27,6 +27,8 @@ use zenoh_protocol::{
 };
 use zenoh_result::{zerror, ZResult};
 
+use crate::LinkAuthId;
+
 /*************************************/
 /*             MANAGER               */
 /*************************************/
@@ -48,6 +50,7 @@ pub trait LinkMulticastTrait: Send + Sync {
     fn get_mtu(&self) -> BatchSize;
     fn get_src(&self) -> &Locator;
     fn get_dst(&self) -> &Locator;
+    fn get_auth_id(&self) -> &LinkAuthId;
     fn is_reliable(&self) -> bool;
     async fn write(&self, buffer: &[u8]) -> ZResult<usize>;
     async fn write_all(&self, buffer: &[u8]) -> ZResult<()>;

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

@@ -123,74 +123,14 @@ pub fn get_ip_interface_names(addr: &SocketAddr) -> Vec<String> {
 }
 
 #[derive(Clone, Debug, Serialize, Hash, PartialEq, Eq)]
-pub enum LinkAuthType {
-    Tls,
-    Quic,
-    None,
-}
-
-#[derive(Clone, Debug, Serialize, Hash, PartialEq, Eq)]
-pub struct LinkAuthId {
-    auth_type: LinkAuthType,
-    auth_value: Option<String>,
-}
-
-impl LinkAuthId {
-    pub const NONE: Self = Self {
-        auth_type: LinkAuthType::None,
-        auth_value: None,
-    };
-    pub fn get_type(&self) -> &LinkAuthType {
-        &self.auth_type
-    }
-    pub fn get_value(&self) -> &Option<String> {
-        &self.auth_value
-    }
-    pub fn builder() -> LinkAuthIdBuilder {
-        LinkAuthIdBuilder::new()
-    }
-}
-
-impl Default for LinkAuthId {
-    fn default() -> Self {
-        LinkAuthId::NONE.clone()
-    }
-}
-
-#[derive(Debug)]
-pub struct LinkAuthIdBuilder {
-    pub auth_type: LinkAuthType,    // HAS to be provided when building
-    pub auth_value: Option<String>, // actual value added to the above type; is None for None type
-}
-
-impl Default for LinkAuthIdBuilder {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl LinkAuthIdBuilder {
-    pub fn new() -> LinkAuthIdBuilder {
-        LinkAuthIdBuilder {
-            auth_type: LinkAuthType::None,
-            auth_value: None,
-        }
-    }
-
-    pub fn auth_type(mut self, auth_type: LinkAuthType) -> Self {
-        self.auth_type = auth_type;
-        self
-    }
-
-    pub fn auth_value(mut self, auth_value: Option<String>) -> Self {
-        self.auth_value = auth_value;
-        self
-    }
-
-    pub fn build(self) -> LinkAuthId {
-        LinkAuthId {
-            auth_type: self.auth_type.clone(),
-            auth_value: self.auth_value.clone(),
-        }
-    }
+pub enum LinkAuthId {
+    Tls(Option<String>),
+    Quic(Option<String>),
+    Tcp,
+    Udp,
+    Serial,
+    Unixpipe,
+    UnixsockStream,
+    Vsock,
+    Ws,
 }