body validation

Cargo.toml

@@ -14,6 +14,7 @@ tonic = "0.14.1"
 tucana = { version = "0.0.77", features = ["aquila", "sagittarius_gateway"] }
 code0-flow = { version = "0.0.42", features = ["flow_config", "flow_health", "flow_telemetry"] }
 serde_json = "1.0.140"
+lupus = "0.0.2"
 async-nats = "0.50.0"
 tonic-health = "0.14.1"
 tokio-stream = "0.1.17"

src/main.rs

@@ -19,6 +19,7 @@ pub mod sagittarius;
 pub mod server;
 pub mod startup;
 pub mod telemetry;
+pub mod validation;
 pub mod version;
 
 const CONFIG_PATH_ENV: &str = "AQUILA_CONFIG_PATH";

src/sagittarius/flow_service_client_impl/mod.rs

@@ -128,7 +128,8 @@ impl SagittariusFlowClient {
                     dev_export::overwrite(&self.flow_export_path, flows.clone()).await;
                 }
 
-                let (purged_count, stored_count) = flow_store::replace_all(&self.store, flows).await;
+                let (purged_count, stored_count) =
+                    flow_store::replace_all(&self.store, flows).await;
                 log::info!(
                     "Finished replacing stored flows received_count={} purged_count={} stored_count={}",
                     received_count,

src/sagittarius/module_configuration_client_impl.rs

@@ -109,6 +109,8 @@ impl SagittariusModuleConfigurationClient {
         }
 
         log::warn!("Sagittarius closed the module configuration stream; reconnecting");
-        Err(tonic::Status::unavailable("module configuration stream ended"))
+        Err(tonic::Status::unavailable(
+            "module configuration stream ended",
+        ))
     }
 }

src/sagittarius/test_execution_client_impl/flow_id_cache.rs

@@ -123,7 +123,9 @@ fn prune_expired(entries: &mut HashMap<String, ExecutionFlowIdMapping>, now: Ins
 
 /// Evicts whichever entry is closest to expiring, since that's the best
 /// approximation of "oldest" without tracking insertion order separately.
-fn remove_soonest_to_expire(entries: &mut HashMap<String, ExecutionFlowIdMapping>) -> Option<String> {
+fn remove_soonest_to_expire(
+    entries: &mut HashMap<String, ExecutionFlowIdMapping>,
+) -> Option<String> {
     let soonest_execution_id = entries
         .iter()
         .min_by_key(|(_, mapping)| mapping.expires_at)

src/sagittarius/test_execution_client_impl/mod.rs

@@ -28,7 +28,9 @@ use tucana::sagittarius_gateway::execution_service_client::ExecutionServiceClien
 use tucana::sagittarius_gateway::{ExecutionLogonRequest, Logon};
 use tucana::shared::{ExecutionFlow, ValidationFlow};
 
-use crate::{authorization::authorization::get_authentication_metadata, flow::key_has_flow_id};
+use crate::{
+    authorization::authorization::get_authentication_metadata, flow::key_has_flow_id, validation,
+};
 
 pub struct SagittariusTestExecutionServiceClient {
     nats_client: async_nats::Client,
@@ -205,8 +207,44 @@ impl SagittariusTestExecutionServiceClient {
                             }
                         };
 
-                        // TODO: When the new validator is ready, the body needs to be validated at this
-                        // point.
+                        if validation::is_rest_flow(&validation_flow) {
+                            let input_schema = validation::extract_input_schema(&validation_flow);
+                            if let Err(err) = validation::validate_body_against_schema(
+                                input_schema,
+                                request.body.as_ref(),
+                            ) {
+                                log::warn!(
+                                    "Rejecting Sagittarius execution request due to input schema validation failure requested_execution_id={} flow_id={} error={}",
+                                    request.execution_identifier,
+                                    request.flow_id,
+                                    err
+                                );
+
+                                let execution_id = if request.execution_identifier.is_empty() {
+                                    uuid::Uuid::new_v4().to_string()
+                                } else {
+                                    request.execution_identifier.clone()
+                                };
+
+                                let rejection = validation::rejection_result(
+                                    execution_id,
+                                    request.flow_id,
+                                    &err,
+                                );
+
+                                if let Err(status) =
+                                    self.response_sender.send_execution_result(rejection).await
+                                {
+                                    log::error!(
+                                        "Failed to send input schema validation rejection result flow_id={} error={:?}",
+                                        request.flow_id,
+                                        status
+                                    );
+                                }
+
+                                continue;
+                            }
+                        }
 
                         let execution_id = if request.execution_identifier.is_empty() {
                             uuid::Uuid::new_v4().to_string()

src/sagittarius/test_execution_client_impl/response_sender.rs

@@ -6,8 +6,8 @@ use std::sync::Arc;
 
 use tokio::sync::Mutex;
 use tonic::Status;
-use tucana::sagittarius_gateway::execution_logon_request::Data;
 use tucana::sagittarius_gateway::ExecutionLogonRequest;
+use tucana::sagittarius_gateway::execution_logon_request::Data;
 use tucana::shared::ExecutionResult;
 
 use super::flow_id_cache::ExecutionFlowIdCache;
@@ -50,7 +50,9 @@ impl SagittariusExecutionResponseSender {
     }
 
     pub(super) async fn remember_execution_flow(&self, execution_id: &str, flow_id: i64) {
-        self.execution_flow_ids.remember(execution_id, flow_id).await;
+        self.execution_flow_ids
+            .remember(execution_id, flow_id)
+            .await;
     }
 
     pub(super) async fn forget_execution_flow(&self, execution_id: &str) {

src/server/action_transfer/logon.rs

@@ -46,15 +46,21 @@ pub(super) fn extract_token(
 /// Whether a broadcasted config update is meant for `action_identifier`, since
 /// [`spawn_cfg_forwarder`] subscribes to a single broadcast channel shared by
 /// every connected action.
-fn applies_to_action(configs: &tucana::shared::ModuleConfigurations, action_identifier: &str) -> bool {
+fn applies_to_action(
+    configs: &tucana::shared::ModuleConfigurations,
+    action_identifier: &str,
+) -> bool {
     configs.module_identifier == action_identifier
 }
 
 /// Rewrites every definition's `definition_source` on the module an action
 /// logs on with, so downstream consumers can tell it came from this action
 /// rather than from whatever source the action's module definition was
 /// authored against.
-fn overwrite_module_definition_sources(module: &mut tucana::shared::Module, action_identifier: &str) {
+fn overwrite_module_definition_sources(
+    module: &mut tucana::shared::Module,
+    action_identifier: &str,
+) {
     let source = format!("action.{}", action_identifier);
 
     for flow_type in &mut module.flow_types {

src/server/action_transfer/mod.rs

@@ -21,13 +21,13 @@ use tokio_stream::wrappers::ReceiverStream;
 use tonic::Status;
 use tracing::Instrument;
 use tucana::aquila::{
-    ActionTransferRequest, ActionTransferResponse, action_transfer_service_server::ActionTransferService,
+    ActionTransferRequest, ActionTransferResponse,
+    action_transfer_service_server::ActionTransferService,
 };
 
 use crate::{
     configuration::service::ServiceConfiguration,
-    sagittarius::module_service_client_impl::SagittariusModuleServiceClient,
-    telemetry::metrics,
+    sagittarius::module_service_client_impl::SagittariusModuleServiceClient, telemetry::metrics,
 };
 
 use logon::{extract_token, handle_logon};

src/server/action_transfer/nats_bridge.rs

@@ -6,7 +6,9 @@ use async_nats::{Subject, Subscriber};
 use futures::StreamExt;
 use prost::Message;
 use tucana::{
-    aquila::{ActionEvent, ActionExecutionRequest, ActionExecutionResponse, ActionTransferResponse},
+    aquila::{
+        ActionEvent, ActionExecutionRequest, ActionExecutionResponse, ActionTransferResponse,
+    },
     shared::{ExecutionFlow, Flows, ValidationFlow, Value},
 };
 

src/server/dynamic_server.rs

@@ -10,8 +10,8 @@ use crate::{
         test_execution_client_impl::SagittariusExecutionResponseSender,
     },
     server::{
-        action_transfer::AquilaActionTransferServiceServer,
-        create_readiness_interceptor, module_service_server_impl::AquilaModuleServiceServer,
+        action_transfer::AquilaActionTransferServiceServer, create_readiness_interceptor,
+        module_service_server_impl::AquilaModuleServiceServer,
         runtime_execution_service_server_impl::AquilaExecutionServiceServer,
         runtime_status_service_server_impl::AquilaRuntimeStatusServiceServer,
     },

src/server/runtime_status_service_server_impl/monitor.rs

@@ -33,7 +33,11 @@ pub(super) fn spawn(
             interval.tick().await;
 
             let timeout_updates = registry
-                .collect_timeout_updates(Instant::now(), not_responding_after, stopped_after_not_responding)
+                .collect_timeout_updates(
+                    Instant::now(),
+                    not_responding_after,
+                    stopped_after_not_responding,
+                )
                 .await;
 
             if timeout_updates.is_empty() {

src/server/runtime_status_service_server_impl/registry.rs

@@ -133,7 +133,12 @@ impl TrackedRuntimeRegistry {
         stopped_after_not_responding: Duration,
     ) -> Vec<RuntimeStatusUpdateRequest> {
         let mut tracked = self.tracked.lock().await;
-        collect_timeout_updates(&mut tracked, now, not_responding_after, stopped_after_not_responding)
+        collect_timeout_updates(
+            &mut tracked,
+            now,
+            not_responding_after,
+            stopped_after_not_responding,
+        )
     }
 }
 

src/server/static_server.rs

@@ -4,10 +4,7 @@
 
 use crate::{
     configuration::{config::Config, service::ServiceConfiguration, state::AppReadiness},
-    server::{
-        action_transfer::AquilaActionTransferServiceServer,
-        create_readiness_interceptor,
-    },
+    server::{action_transfer::AquilaActionTransferServiceServer, create_readiness_interceptor},
 };
 use async_nats::jetstream::kv::Store;
 use log::info;

src/startup/dynamic_mode.rs

@@ -203,7 +203,10 @@ pub async fn run(
                 action_config_tx_for_module_configuration.clone(),
             );
 
-            match module_configuration_client.init_configuration_stream().await {
+            match module_configuration_client
+                .init_configuration_stream()
+                .await
+            {
                 Ok(_) => {
                     log::warn!(
                         "Sagittarius module configuration stream ended normally; reconnecting"