subscription: implement ClientFilterType and ClientFilterFlags

cli/src/subscriptions.rs

@@ -617,7 +617,7 @@ async fn edit_filter(subscription: &mut SubscriptionData, matches: &ArgMatches)
             warn!("'{}' filter has been set without principals making this subscription apply to nothing.", op)
         }
 
-        subscription.set_client_filter(Some(ClientFilter::new(op, princs)));
+        subscription.set_client_filter(Some(ClientFilter::new_legacy(op, princs)));
         return Ok(());
     }
 

common/Cargo.toml

@@ -26,6 +26,8 @@ deadpool-sqlite = "0.5.0"
 openssl = "0.10.66"
 postgres-openssl = "0.5.0"
 strum = { version = "0.26.1", features = ["derive"] }
+bitflags = { version = "2.6.0", features = ["serde"] }
+glob = "0.3.1"
 
 [dev-dependencies]
 tempfile = "3.14.0"
@@ -68,4 +70,4 @@ assets = [
     { source = "../openwec.conf.sample.toml", dest = "/usr/share/doc/openwec/", mode = "0644", doc = true },
     { source = "../README.md", dest = "/usr/share/doc/openwec/", mode = "0644", doc = true },
     { source = "../doc/*", dest = "/usr/share/doc/openwec/doc/", mode = "0644", doc = true },
-]
+]

common/src/database/mod.rs

@@ -196,6 +196,8 @@ pub mod tests {
             .set_ignore_channel_error(false)
             .set_client_filter(Some(ClientFilter::from(
                 "Only".to_string(),
+                "KerberosPrinc".to_string(),
+                None,
                 Some("couscous,boulette".to_string()),
             )?))
             .set_outputs(vec![
@@ -232,7 +234,7 @@ pub mod tests {
         );
         assert_eq!(
             tata.client_filter().unwrap().targets(),
-            &HashSet::from(["couscous".to_string(), "boulette".to_string()])
+            HashSet::from(["couscous", "boulette"])
         );
 
         assert_eq!(
@@ -298,10 +300,10 @@ pub mod tests {
         );
         assert_eq!(
             tata2.client_filter().unwrap().targets(),
-            &HashSet::from([
-                "couscous".to_string(),
-                "boulette".to_string(),
-                "semoule".to_string()
+            HashSet::from([
+                "couscous",
+                "boulette",
+                "semoule"
             ])
         );
         assert_eq!(tata2.is_active_for("couscous"), true);
@@ -329,7 +331,7 @@ pub mod tests {
         );
         assert_eq!(
             tata2_clone.client_filter().unwrap().targets(),
-            &HashSet::from(["boulette".to_string(), "semoule".to_string()])
+            HashSet::from(["boulette", "semoule"])
         );
 
         assert_eq!(tata2_clone.is_active_for("couscous"), true);

common/src/database/postgres.rs

@@ -216,7 +216,11 @@ fn row_to_subscription(row: &Row) -> Result<SubscriptionData> {
     let client_filter_op: Option<String> = row.try_get("client_filter_op")?;
 
     let client_filter = match client_filter_op {
-        Some(op) => Some(ClientFilter::from(op, row.try_get("client_filter_value")?)?),
+        Some(op) =>  {
+            let client_filter_type: Option<_> = row.try_get("client_filter_type").unwrap();
+            let client_filter_type = client_filter_type.unwrap_or("KerberosPrinc".to_owned());
+            Some(ClientFilter::from(op, client_filter_type, row.try_get("client_filter_flags")?, row.try_get("client_filter_value")?)?)
+        },
         None => None
     };
 
@@ -628,6 +632,8 @@ impl Database for PostgresDatabase {
 
         let max_envelope_size: i32 = subscription.max_envelope_size().try_into()?;
         let client_filter_op: Option<String> = subscription.client_filter().map(|f| f.operation().to_string());
+        let client_filter_type = subscription.client_filter().map(|f| f.kind().to_string());
+        let client_filter_flags = subscription.client_filter().map(|f| f.flags().to_string());
         let client_filter_value = subscription.client_filter().and_then(|f| f.targets_to_opt_string());
 
         let count = self
@@ -638,9 +644,9 @@ impl Database for PostgresDatabase {
                 r#"INSERT INTO subscriptions (uuid, version, revision, name, uri, query,
                     heartbeat_interval, connection_retry_count, connection_retry_interval,
                     max_time, max_elements, max_envelope_size, enabled, read_existing_events, content_format,
-                    ignore_channel_error, client_filter_op, client_filter_value, outputs, locale,
+                    ignore_channel_error, client_filter_op, client_filter_type, client_filter_flags, client_filter_value, outputs, locale,
                     data_locale)
-                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21)
+                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23)
                     ON CONFLICT (uuid) DO UPDATE SET
                         version = excluded.version,
                         revision = excluded.revision,
@@ -658,6 +664,8 @@ impl Database for PostgresDatabase {
                         content_format = excluded.content_format,
                         ignore_channel_error = excluded.ignore_channel_error,
                         client_filter_op = excluded.client_filter_op,
+                        client_filter_type = excluded.client_filter_type,
+                        client_filter_flags = excluded.client_filter_flags,
                         client_filter_value = excluded.client_filter_value,
                         outputs = excluded.outputs,
                         locale = excluded.locale,
@@ -680,6 +688,8 @@ impl Database for PostgresDatabase {
                     &subscription.content_format().to_string(),
                     &subscription.ignore_channel_error(),
                     &client_filter_op,
+                    &client_filter_type,
+                    &client_filter_flags,
                     &client_filter_value,
                     &serde_json::to_string(subscription.outputs())?.as_str(),
                     &subscription.locale(),

common/src/database/schema/postgres/_014_alter_client_filter_in_subscriptions.rs

@@ -16,12 +16,16 @@ impl PostgresMigration for AlterClientFilterInSubscriptionsTable {
     async fn up(&self, tx: &mut Transaction) -> Result<()> {
         tx.execute("ALTER TABLE subscriptions RENAME COLUMN princs_filter_op TO client_filter_op", &[]).await?;
         tx.execute("ALTER TABLE subscriptions RENAME COLUMN princs_filter_value TO client_filter_value", &[]).await?;
+        tx.execute("ALTER TABLE subscriptions ADD COLUMN client_filter_type TEXT", &[]).await?;
+        tx.execute("ALTER TABLE subscriptions ADD COLUMN client_filter_flags TEXT", &[]).await?;
         Ok(())
     }
 
     async fn down(&self, tx: &mut Transaction) -> Result<()> {
         tx.execute("ALTER TABLE subscriptions RENAME COLUMN client_filter_op TO princs_filter_op", &[]).await?;
         tx.execute("ALTER TABLE subscriptions RENAME COLUMN client_filter_value TO princs_filter_value", &[]).await?;
+        tx.execute("ALTER TABLE subscriptions DROP COLUMN client_filter_type", &[]).await?;
+        tx.execute("ALTER TABLE subscriptions DROP COLUMN client_filter_flags", &[]).await?;
         Ok(())
     }
 }

common/src/database/schema/sqlite/_014_alter_client_filter_in_subscriptions.rs

@@ -17,6 +17,10 @@ impl SQLiteMigration for AlterClientFilterInSubscriptionsTable {
             .map_err(|err| anyhow!("SQLiteError: {}", err))?;
         conn.execute("ALTER TABLE subscriptions RENAME COLUMN princs_filter_value TO client_filter_value", [])
             .map_err(|err| anyhow!("SQLiteError: {}", err))?;
+        conn.execute("ALTER TABLE subscriptions ADD COLUMN client_filter_type TEXT", [])
+            .map_err(|err| anyhow!("SQLiteError: {}", err))?;
+        conn.execute("ALTER TABLE subscriptions ADD COLUMN client_filter_flags TEXT", [])
+            .map_err(|err| anyhow!("SQLiteError: {}", err))?;
         Ok(())
     }
 
@@ -25,6 +29,10 @@ impl SQLiteMigration for AlterClientFilterInSubscriptionsTable {
             .map_err(|err| anyhow!("SQLiteError: {}", err))?;
         conn.execute("ALTER TABLE subscriptions RENAME COLUMN client_filter_value TO princs_filter_value", [])
             .map_err(|err| anyhow!("SQLiteError: {}", err))?;
+        conn.execute("ALTER TABLE subscriptions DROP COLUMN client_filter_type", [])
+            .map_err(|err| anyhow!("SQLiteError: {}", err))?;
+        conn.execute("ALTER TABLE subscriptions DROP COLUMN client_filter_flags", [])
+            .map_err(|err| anyhow!("SQLiteError: {}", err))?;
         Ok(())
     }
 }

common/src/database/sqlite.rs

@@ -169,7 +169,11 @@ fn row_to_subscription(row: &Row) -> Result<SubscriptionData> {
     let client_filter_op: Option<String> = row.get("client_filter_op")?;
 
     let client_filter = match client_filter_op {
-        Some(op) => Some(ClientFilter::from(op, row.get("client_filter_value")?)?),
+        Some(op) =>  {
+            let client_filter_type: Option<_> = row.get("client_filter_type")?;
+            let client_filter_type = client_filter_type.unwrap_or("KerberosPrinc".to_owned());
+            Some(ClientFilter::from(op, client_filter_type, row.get("client_filter_flags")?, row.get("client_filter_value")?)?)
+        },
         None => None
     };
 
@@ -553,6 +557,8 @@ impl Database for SQLiteDatabase {
     async fn store_subscription(&self, subscription: &SubscriptionData) -> Result<()> {
         let subscription = subscription.clone();
         let client_filter_op: Option<String> = subscription.client_filter().map(|f| f.operation().to_string());
+        let client_filter_type = subscription.client_filter().map(|f| f.kind().to_string());
+        let client_filter_flags = subscription.client_filter().map(|f| f.flags().to_string());
         let client_filter_value = subscription.client_filter().and_then(|f| f.targets_to_opt_string());
 
         let count = self
@@ -564,12 +570,12 @@ impl Database for SQLiteDatabase {
                     r#"INSERT INTO subscriptions (uuid, version, revision, name, uri, query,
                     heartbeat_interval, connection_retry_count, connection_retry_interval,
                     max_time, max_elements, max_envelope_size, enabled, read_existing_events, content_format,
-                    ignore_channel_error, client_filter_op, client_filter_value, outputs, locale,
+                    ignore_channel_error, client_filter_op, client_filter_type, client_filter_flags, client_filter_value, outputs, locale,
                     data_locale)
                     VALUES (:uuid, :version, :revision, :name, :uri, :query,
                         :heartbeat_interval, :connection_retry_count, :connection_retry_interval,
                         :max_time, :max_elements, :max_envelope_size, :enabled, :read_existing_events, :content_format,
-                        :ignore_channel_error, :client_filter_op, :client_filter_value, :outputs,
+                        :ignore_channel_error, :client_filter_op, :client_filter_type, :client_filter_flags, :client_filter_value, :outputs,
                         :locale, :data_locale)
                     ON CONFLICT (uuid) DO UPDATE SET
                         version = excluded.version,
@@ -588,6 +594,8 @@ impl Database for SQLiteDatabase {
                         content_format = excluded.content_format,
                         ignore_channel_error = excluded.ignore_channel_error,
                         client_filter_op = excluded.client_filter_op,
+                        client_filter_type = excluded.client_filter_type,
+                        client_filter_flags = excluded.client_filter_flags,
                         client_filter_value = excluded.client_filter_value,
                         outputs = excluded.outputs,
                         locale = excluded.locale,
@@ -610,6 +618,8 @@ impl Database for SQLiteDatabase {
                         ":content_format": subscription.content_format().to_string(),
                         ":ignore_channel_error": subscription.ignore_channel_error(),
                         ":client_filter_op": client_filter_op,
+                        ":client_filter_type": client_filter_type,
+                        ":client_filter_flags": client_filter_flags,
                         ":client_filter_value": client_filter_value,
                         ":outputs": serde_json::to_string(subscription.outputs())?,
                         ":locale": subscription.locale(),

common/src/models/config.rs

@@ -178,6 +178,10 @@ impl From<ClientFilterOperation> for crate::subscription::ClientFilterOperation
 #[serde(deny_unknown_fields)]
 struct ClientFilter {
     pub operation: ClientFilterOperation,
+    #[serde(rename = "type", default)]
+    pub kind: crate::subscription::ClientFilterType,
+    #[serde(default)]
+    pub flags: crate::subscription::ClientFilterFlags,
     #[serde(alias = "cert_subjects", alias = "princs")]
     pub targets: HashSet<String>,
 }
@@ -186,7 +190,7 @@ impl TryFrom<ClientFilter> for crate::subscription::ClientFilter {
     type Error = anyhow::Error;
 
     fn try_from(value: ClientFilter) -> std::prelude::v1::Result<Self, Self::Error> {
-        Ok(crate::subscription::ClientFilter::new(value.operation.into(), value.targets))
+        crate::subscription::ClientFilter::try_new(value.operation.into(), value.kind, value.flags, value.targets)
     }
 }
 
@@ -506,7 +510,9 @@ path = "/whatever/you/{ip}/want/{principal}/{ip:2}/{node}/end"
         let mut targets = HashSet::new();
         targets.insert("toto@windomain.local".to_string());
         targets.insert("tutu@windomain.local".to_string());
-        let filter = crate::subscription::ClientFilter::new(crate::subscription::ClientFilterOperation::Only, targets);
+        let kind = crate::subscription::ClientFilterType::KerberosPrinc;
+        let flags = crate::subscription::ClientFilterFlags::empty();
+        let filter = crate::subscription::ClientFilter::try_new(crate::subscription::ClientFilterOperation::Only, kind, flags, targets)?;
 
         expected.set_client_filter(Some(filter));
 

common/src/models/export.rs

@@ -202,7 +202,7 @@ mod v1 {
                 return None;
             };
 
-            Some(crate::subscription::ClientFilter::new(op.into(), self.princs))
+            Some(crate::subscription::ClientFilter::new_legacy(op.into(), self.princs))
         }
     }
 
@@ -549,15 +549,15 @@ pub mod v2 {
                 return None;
             };
 
-            Some(crate::subscription::ClientFilter::new(op.into(), self.princs))
+            Some(crate::subscription::ClientFilter::new_legacy(op.into(), self.princs))
         }
     }
 
     impl From<Option<crate::subscription::ClientFilter>> for PrincsFilter {
         fn from(value: Option<crate::subscription::ClientFilter>) -> Self {
             Self {
                 operation: value.as_ref().and_then(|f| Some(f.operation().clone().into())),
-                princs: value.map_or(HashSet::new(), |f| f.targets().clone()),
+                princs: value.map_or(HashSet::new(), |f| f.targets().iter().cloned().map(String::from).collect()),
             }
         }
     }
@@ -712,10 +712,12 @@ mod tests {
             .set_max_elements(Some(100))
             .set_read_existing_events(false)
             .set_uri(Some("toto".to_string()))
-            .set_client_filter(Some(crate::subscription::ClientFilter::new(
+            .set_client_filter(Some(crate::subscription::ClientFilter::try_new(
                 crate::subscription::ClientFilterOperation::Except,
+                crate::subscription::ClientFilterType::KerberosPrinc,
+                crate::subscription::ClientFilterFlags::CaseSensitive,
                 targets,
-            )))
+            )?))
             .set_outputs(vec![crate::subscription::SubscriptionOutput::new(
                 crate::subscription::SubscriptionOutputFormat::Json,
                 crate::subscription::SubscriptionOutputDriver::Tcp(