Merge pull request #406 from code0-tech/#261-body-validation

body validation

Author: raphael-goetz

Cargo.lock

@@ -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"

Cargo.toml

@@ -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/main.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/flow_service_client_impl/mod.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/module_configuration_client_impl.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/flow_id_cache.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/mod.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/sagittarius/test_execution_client_impl/response_sender.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/logon.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};