Merge pull request #378 from code0-tech/#354-reliable-action-config

static mode didnt lookup token

Author: raphael-goetz

src/configuration/service.rs

@@ -131,12 +131,13 @@ impl ServiceConfiguration {
 
     pub fn get_action_configuration(
         &self,
+        token: &String,
         action_identifier: &String,
     ) -> Vec<ModuleConfigurations> {
         match self
             .actions
             .iter()
-            .find(|x| &x.service_name == action_identifier)
+            .find(|x| &x.token == token && &x.service_name == action_identifier)
         {
             Some(a) => a.config.clone(),
             None => vec![],
@@ -187,6 +188,7 @@ impl ServiceConfiguration {
 mod tests {
     use super::{
         RuntimeServiceConfiguration, SerializableActionServiceConfiguration,
+        SerializableModuleConfiguration, SerializableModuleProjectConfiguration,
         SerializableServiceConfiguration, ServiceConfiguration,
     };
 
@@ -253,6 +255,7 @@ mod tests {
             &String::from("action-identifier")
         ));
         assert!(!config.has_action(&String::from("action-token"), &String::from("action-other")));
+        assert!(!config.has_action(&String::from("example"), &String::from("example")));
     }
 
     #[test]
@@ -274,4 +277,57 @@ mod tests {
         ));
         assert!(!config.has_service(&String::from("action-token"), &String::from("taurus-x")));
     }
+
+    #[test]
+    fn get_action_configuration_requires_matching_token_and_identifier() {
+        let config: ServiceConfiguration = SerializableServiceConfiguration {
+            actions: vec![
+                SerializableActionServiceConfiguration {
+                    token: String::from("old-token"),
+                    identifier: String::from("shared-action"),
+                    configs: vec![SerializableModuleProjectConfiguration {
+                        project_id: 1,
+                        configs: vec![SerializableModuleConfiguration {
+                            identifier: String::from("endpoint"),
+                            value: serde_json::json!("old.example"),
+                        }],
+                    }],
+                },
+                SerializableActionServiceConfiguration {
+                    token: String::from("new-token"),
+                    identifier: String::from("shared-action"),
+                    configs: vec![SerializableModuleProjectConfiguration {
+                        project_id: 2,
+                        configs: vec![SerializableModuleConfiguration {
+                            identifier: String::from("endpoint"),
+                            value: serde_json::json!("new.example"),
+                        }],
+                    }],
+                },
+            ],
+            runtimes: vec![],
+        }
+        .into();
+
+        let configs = config
+            .get_action_configuration(&String::from("new-token"), &String::from("shared-action"));
+
+        assert_eq!(configs.len(), 1);
+        assert_eq!(configs[0].module_identifier, "shared-action");
+        assert_eq!(configs[0].module_configurations[0].project_id, 2);
+    }
+
+    #[test]
+    fn get_action_configuration_returns_empty_for_identifier_with_wrong_token() {
+        let config = fixture();
+
+        assert!(
+            config
+                .get_action_configuration(
+                    &String::from("wrong-token"),
+                    &String::from("action-identifier")
+                )
+                .is_empty()
+        );
+    }
 }

src/server/action_transfer_service_server_impl.rs

@@ -229,6 +229,15 @@ fn pending_reply_keys(
     keys
 }
 
+async fn send_stream_error(
+    tx: &tokio::sync::mpsc::Sender<Result<ActionTransferResponse, tonic::Status>>,
+    status: tonic::Status,
+) {
+    if tx.send(Err(status)).await.is_err() {
+        log::debug!("Action transfer response stream closed before error could be sent");
+    }
+}
+
 fn insert_pending_reply(
     pending: &mut HashMap<String, PendingReply>,
     reply_subject: Subject,
@@ -636,6 +645,11 @@ impl ActionTransferService for AquilaActionTransferServiceServer {
                                 Some(ref m) => m.identifier.clone(),
                                 None => {
                                     log::warn!("Rejected action logon reason=missing_module");
+                                    send_stream_error(
+                                        &tx,
+                                        Status::aborted("Please provide a module configuration."),
+                                    )
+                                    .await;
                                     break;
                                 }
                             };
@@ -663,6 +677,7 @@ impl ActionTransferService for AquilaActionTransferServiceServer {
                                         status.code(),
                                         status.message()
                                     );
+                                    send_stream_error(&tx, status).await;
                                     break;
                                 }
                             };
@@ -677,6 +692,11 @@ impl ActionTransferService for AquilaActionTransferServiceServer {
                         }
                         _ => {
                             log::error!("Action stream protocol violation expected=logon");
+                            send_stream_error(
+                                &tx,
+                                Status::failed_precondition("first action stream message must be logon"),
+                            )
+                            .await;
                             break;
                         }
                     }
@@ -702,7 +722,7 @@ impl ActionTransferService for AquilaActionTransferServiceServer {
 
                 if is_static {
                     let lock = actions.lock().await;
-                    let configs = lock.get_action_configuration(&identifier);
+                    let configs = lock.get_action_configuration(&token, &identifier);
                     for conf in configs {
                         if let Err(err) = cfg_tx.send(conf) {
                             log::warn!("No action configuration receivers available: {:?}", err);
@@ -716,6 +736,11 @@ impl ActionTransferService for AquilaActionTransferServiceServer {
                             "Action stream protocol violation identifier={} reason=duplicate_logon",
                             identifier
                         );
+                        send_stream_error(
+                            &tx,
+                            Status::failed_precondition("action stream logon was already accepted"),
+                        )
+                        .await;
                         break;
                     }
                     tucana::aquila::action_transfer_request::Data::Event(event) => {