From aca7e8c187bda06d4a031b22b125882d133b9e13 Mon Sep 17 00:00:00 2001 From: Raphael Date: Thu, 9 Jul 2026 16:34:55 +0200 Subject: [PATCH] fix: static mode didnt lookup token --- src/configuration/service.rs | 58 ++++++++++++++++++- .../action_transfer_service_server_impl.rs | 27 ++++++++- 2 files changed, 83 insertions(+), 2 deletions(-) diff --git a/src/configuration/service.rs b/src/configuration/service.rs index cac2f39..cf76c57 100644 --- a/src/configuration/service.rs +++ b/src/configuration/service.rs @@ -131,12 +131,13 @@ impl ServiceConfiguration { pub fn get_action_configuration( &self, + token: &String, action_identifier: &String, ) -> Vec { 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() + ); + } } diff --git a/src/server/action_transfer_service_server_impl.rs b/src/server/action_transfer_service_server_impl.rs index 9b0e74b..0f423b2 100644 --- a/src/server/action_transfer_service_server_impl.rs +++ b/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>, + 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, 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) => {